Agent Server

Run multiple independent agent instances on a single server with routing

Run multiple independent agent instances on a single server. Each client connects to their own isolated agent via URL path—perfect for per-user assistants, task workers, or multi-tenant AI services.

Quick Start

from ai_query.agents import Agent
from ai_query.agents import AgentServer
from ai_query.providers import openai, MemoryStorage

agent = Agent(
    "task-agent",
    model=openai("gpt-4o"),
    system="You are a task execution assistant.",
    storage=MemoryStorage(),
    initial_state={"tasks_completed": 0}
)

# Start multi-agent server
AgentServer(agent).serve(port=8080)

# Clients connect via multiple protocols:
# WebSocket: ws://localhost:8080/agent/user-alice/ws
# REST:      POST http://localhost:8080/agent/user-alice/chat
# SSE:       GET http://localhost:8080/agent/user-alice/events
# Each agent ID gets its own isolated instance

Use Cases

Per-User Assistants

Each user gets their own AI with isolated state and history

Task Workers

Spawn independent agents for parallel task execution

Multi-Tenant APIs

Isolate AI state per customer or organization

Session Agents

Create agents for specific workflows or sessions

Architecture

The Agent Server uses an Agent Registry to route requests between local and remote agents.

1. Request: POST /agent/user-123/chat
2. Registry: Matches user-123 to registered target (Agent class or Transport).
3. Execution:
   • Local: Instantiates agent and executes locally
   • Remote: Delegates to transport (HTTP, WebSocket, etc.)

For detailed API reference, see AgentServer API Reference and AgentRegistry API Reference.

Configuration

Basic (Single Agent Type)

The simplest way is to pass a single Agent class (or instance) to the server. This maps all IDs to that agent type.

# Maps all IDs (.*) to MyAgent
AgentServer(MyAgent).serve(port=8080)

Advanced (Multi-Agent Registry)

Use AgentRegistry to serve different types of agents from the same server.

from ai_query import AgentServer, AgentRegistry
from agents import ChatBot, TaskWorker, AdminAgent

registry = AgentRegistry()

# 1. Exact Match
registry.register("admin", AdminAgent)

# 2. Pattern Match (Regex)
registry.register("chat-.*", ChatBot)
registry.register("worker-.*", TaskWorker)

# Start server with registry
AgentServer(registry).serve(port=8080)

Now requests are routed dynamically:

• /agent/admin -> AdminAgent("admin")
• /agent/chat-123 -> ChatBot("chat-123")
• /agent/worker-abc -> TaskWorker("worker-abc")

Server Configuration

Control lifecycle and security with AgentServerConfig.

from ai_query.agents import AgentServerConfig

config = AgentServerConfig(
    idle_timeout=300,     # Evict after 5 min idle
    max_agents=100,       # Max 100 concurrent agents
    enable_rest_api=True, # Enable state REST endpoints
    base_path="/api/v1",  # Custom base path
)

AgentServer(registry, config=config).serve(port=8080)

Endpoints

Endpoint	Type	Description
/agent/{id}/ws	WebSocket	Bidirectional communication
/agent/{id}/events	SSE	AI streaming (server-sent events)
/agent/{id}/chat	REST	Chat with the agent (supports ?stream=true)
/agent/{id}/invoke	REST	Invoke agent tasks
/agent/{id}/state	REST	Get/PUT agent state
/agent/{id}	REST	DELETE to evict agent
/agents	REST	List active agents (if enabled)

Client Examples

Python (Client SDK)

The recommended way to interact with the server.

from ai_query import connect

# Connect to a specific agent instance
agent = connect("http://localhost:8080/agent/user-123")

# Chat
response = await agent.chat("Hello!")

# Stream
async for chunk in agent.stream("Write a poem"):
    print(chunk, end="")

REST (HTTP)

# Chat with an agent
curl -X POST http://localhost:8080/agent/user-123/chat \
  -H "Content-Type: application/json" \
  -d '{"message": "Hello, how can you help me?"}'

# Stream response
curl -X POST "http://localhost:8080/agent/user-123/chat?stream=true" \
 -H "Content-Type: application/json" \
 -d '{"message": "Write a short poem"}'

# Invoke a task
curl -X POST http://localhost:8080/agent/user-123/invoke \
 -H "Content-Type: application/json" \
 -d '{"task": "summarize", "text": "Long document here..."}'

SSE (Server-Sent Events)

// The recommended pattern: SSE for events, REST for messages

// 1. Connect to SSE first
const evtSource = new EventSource(
  "http://localhost:8080/agent/user-123/events"
);

// 2. Listen for events
evtSource.addEventListener("chunk", (e) => {
  const data = JSON.parse(e.data);
  appendToOutput(data.content);
});

// 3. Send messages via REST
fetch("http://localhost:8080/agent/user-123/chat", {
  method: "POST",
  headers: { "Content-Type": "application/json" },
  body: JSON.stringify({ message: "Hello!" })
});

Lifecycle Management

Idle Timeout

Agents with no connections are automatically evicted after idle_timeout seconds:

config = AgentServerConfig(idle_timeout=600)  # 10 minutes

Max Agents Limit

Reject new agents when limit is reached (returns 429 Too Many Requests):

config = AgentServerConfig(max_agents=100)

Lifecycle Hooks

Override hooks on AgentServer for custom logic:

class MyServer(AgentServer):
    async def on_agent_create(self, agent):
        print(f"Created agent: {agent.id}")

    async def on_agent_evict(self, agent):
        print(f"Evicting agent: {agent.id}")
        # Save important data before eviction

MyServer(agent, config=config).serve(port=8080)

Custom Routes

Add your own endpoints alongside the agent routes using create_app().

from aiohttp import web
from ai_query.agents import AgentServer

server = AgentServer(agent)
app = server.create_app()

# Add custom routes
async def health_handler(request):
    return web.json_response({"status": "ok"})

app.router.add_get("/health", health_handler)

# Run with aiohttp directly
web.run_app(app, port=8080)