AgentServer

API reference for multi-agent server with unified transport layer routing

The AgentServer class manages multiple agent instances on a single server, routing clients to the correct agent based on URL path using the unified transport layer. It supports both local agent classes and remote agent transports through the AgentRegistry.

Import

from ai_query.agents import AgentServer
from ai_query.agents import AgentServerConfig
from ai_query.agents import AgentRegistry
from ai_query.agents.transport import HTTPTransport

AgentServerConfig

Configuration dataclass for AgentServer lifecycle and security.

@dataclass
class AgentServerConfig:
    # Lifecycle
    idle_timeout: float | None = 300.0
    max_agents: int | None = None

    # Security
    auth: Callable[[Request], Awaitable[bool]] | None = None
    allowed_origins: list[str] | None = None

    # Routes
    base_path: str = "/agent"
    enable_rest_api: bool = True
    enable_list_agents: bool = False

Parameters

idle_timeout · float | None · default: "300.0"

Seconds before evicting idle agents (no connections). Set to None to disable auto-eviction.

max_agents · int | None · default: "None"

Maximum concurrent agents. Returns 429 Too Many Requests if limit reached. None = unlimited.

auth · Callable[[Request], Awaitable[bool]] | None · default: "None"

Async function to validate requests. Return True to allow, False to reject (401).

allowed_origins · list[str] | None · default: "None"

CORS allowed origins. None = allow all (*).

base_path · str · default: "/agent"

Base path for agent routes.

enable_rest_api · bool · default: "True"

Enable REST endpoints (GET/PUT state, DELETE agent).

enable_list_agents · bool · default: "False"

Enable GET /agents endpoint. Off by default for security.

---

AgentServer

class AgentServer:
    def __init__(
        self,
        target: Agent | AgentRegistry,
        config: AgentServerConfig | None = None,
    )

Parameters

target · Agent | AgentRegistry · required

Either a single Agent instance to use as a template for each client, or an AgentRegistry for multi-agent routing with different agent types and transports.

config · AgentServerConfig | None

Optional configuration. Uses defaults if not provided.

---

Methods

get_or_create

def get_or_create(self, agent_id: str) -> Agent

Get or lazily create an agent by ID.

Raises: HTTPTooManyRequests if max_agents limit is reached.

evict

async def evict(self, agent_id: str) -> None

Evict an agent, closing all connections and removing it from the registry.

list_agents

def list_agents(self) -> list[str]

Return list of active agent IDs.

serve

def serve(
    self,
    host: str = "localhost",
    port: int = 8080,
) -> None

Start the multi-agent server (blocking).

serve_async

async def serve_async(
    self,
    host: str = "localhost",
    port: int = 8080,
) -> None

Start the multi-agent server (async).

create_app

def create_app(self) -> web.Application

Create and return the configured aiohttp Application with all agent routes registered.

Use this method when you need full control over the app, such as:

• Adding custom routes or middleware
• Integrating with existing aiohttp applications
• Running with custom server configurations

Returns: Configured aiohttp.web.Application instance.

Example:

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

agent = Agent("my-bot", model=openai("gpt-4o"), storage=MemoryStorage())
server = AgentServer(agent)
app = server.create_app()

# Add custom routes
app.router.add_get("/health", health_handler)

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

---

Lifecycle Hooks

Override these methods in a subclass for custom behavior:

on_app_setup

def on_app_setup(self, app: web.Application) -> None

Called after the aiohttp app is created but before serving. Use this to add custom routes, middleware, or configuration.

Example:

class MyServer(AgentServer):
    def on_app_setup(self, app):
        app.router.add_get("/health", self.health_check)
        app.router.add_post("/webhook", self.handle_webhook)

    async def health_check(self, request):
        return web.json_response({"status": "ok"})

on_agent_create

async def on_agent_create(self, agent: Agent) -> None

Called when a new agent is created (on first connection).

on_agent_evict

async def on_agent_evict(self, agent: Agent) -> None

Called when an agent is about to be evicted.

---

Endpoints

When the server is running, these endpoints are available:

Real-time Endpoints

Method	Path	Description
GET	{base_path}/{id}/ws	WebSocket connection
GET	{base_path}/{id}/events	SSE stream for AI streaming

REST API Endpoints

These are enabled when enable_rest_api=True (the default):

Method	Path	Description
POST	{base_path}/{id}	Generic request handler
POST	{base_path}/{id}/chat	Chat with the agent (supports ?stream=true)
POST	{base_path}/{id}/invoke	Invoke agent task
GET	{base_path}/{id}/state	Get agent state
PUT	{base_path}/{id}/state	Update agent state
DELETE	{base_path}/{id}	Evict agent
GET	/agents	List agents (if enable_list_agents=True)

Request Formats

POST {base_path}/{id} (Generic)

Accepts a standard action format:

// Chat
{"action": "chat", "message": "Hello!"}

// Invoke
{"action": "invoke", "payload": {"task": "summarize", "text": "..."}}

// Get state
{"action": "state"}

POST {base_path}/{id}/chat

Standard chat request:

{"message": "Hello, how can you help me?"}

Response:

{"agent_id": "agent-123", "response": "I can help with..."}

Streaming Chat (?stream=true)

Send a request to {base_path}/{id}/chat?stream=true to get a Server-Sent Events (SSE) stream.

Request:

curl -X POST "http://localhost:8080/agent/user-1/chat?stream=true" \
  -H "Content-Type: application/json" \
  -d '{"message": "Tell me a story"}'

Response (Content-Type: text/event-stream):

event: start
data:

event: chunk
data: Once upon a time

event: chunk
data: , in a land far away...

event: end
data: "Once upon a time, in a land far away..."

POST {base_path}/{id}/invoke

{"task": "summarize", "text": "Long document..."}
// or
{"payload": {"task": "summarize", "text": "..."}}

Response:

{"agent_id": "agent-123", "result": {"summary": "..."}}

---

Agent.serve_many

Convenience method on Agent to start a multi-agent server:

def serve_many(
    self,
    host: str = "localhost",
    port: int = 8080,
    config: AgentServerConfig | None = None,
) -> None

Equivalent to:

AgentServer(agent, config=config).serve(host=host, port=port)

---

Example

Basic Usage

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

agent = Agent(
    "my-agent",
    model=openai("gpt-4o"),
    system="You are helpful.",
    storage=MemoryStorage()
)

# These are equivalent:
agent.serve_many(port=8080)
# or
AgentServer(agent).serve(port=8080)

With Configuration

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

async def auth(request):
    return request.headers.get("X-API-Key") == "secret"

config = AgentServerConfig(
    idle_timeout=600,
    max_agents=50,
    auth=auth,
    allowed_origins=["https://myapp.com"],
)

agent = Agent("my-agent", model=openai("gpt-4o"), storage=MemoryStorage())
AgentServer(agent, config=config).serve(port=8080)

Custom Server with Hooks

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

    async def on_agent_evict(self, agent):
        print(f"Agent {agent.id} evicted")

agent = Agent("my-agent", model=openai("gpt-4o"), storage=MemoryStorage())
MyServer(agent, config=config).serve(port=8080)