> ## Documentation Index
> Fetch the complete documentation index at: https://thethirdpenco-feat-tool-lifecycle-events.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.

# 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

```python theme={null}
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.

```python theme={null}
@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

<ParamField path="idle_timeout" type="float | None" default="300.0">
  Seconds before evicting idle agents (no connections). Set to `None` to disable auto-eviction.
</ParamField>

<ParamField path="max_agents" type="int | None" default="None">
  Maximum concurrent agents. Returns 429 Too Many Requests if limit reached. `None` = unlimited.
</ParamField>

<ParamField path="auth" type="Callable[[Request], Awaitable[bool]] | None" default="None">
  Async function to validate requests. Return `True` to allow, `False` to reject (401).
</ParamField>

<ParamField path="allowed_origins" type="list[str] | None" default="None">
  CORS allowed origins. `None` = allow all (`*`).
</ParamField>

<ParamField path="base_path" type="str" default="/agent">
  Base path for agent routes.
</ParamField>

<ParamField path="enable_rest_api" type="bool" default="True">
  Enable REST endpoints (GET/PUT state, DELETE agent).
</ParamField>

<ParamField path="enable_list_agents" type="bool" default="False">
  Enable `GET /agents` endpoint. Off by default for security.
</ParamField>

***

## AgentServer

```python theme={null}
class AgentServer:
    def __init__(
        self,
        target: Agent | AgentRegistry,
        config: AgentServerConfig | None = None,
    )
```

### Parameters

<ParamField path="target" type="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.
</ParamField>

<ParamField path="config" type="AgentServerConfig | None">
  Optional configuration. Uses defaults if not provided.
</ParamField>

***

## Methods

### get\_or\_create

```python theme={null}
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

```python theme={null}
async def evict(self, agent_id: str) -> None
```

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

### list\_agents

```python theme={null}
def list_agents(self) -> list[str]
```

Return list of active agent IDs.

### serve

```python theme={null}
def serve(
    self,
    host: str = "localhost",
    port: int = 8080,
) -> None
```

Start the multi-agent server (blocking).

### serve\_async

```python theme={null}
async def serve_async(
    self,
    host: str = "localhost",
    port: int = 8080,
) -> None
```

Start the multi-agent server (async).

### create\_app

```python theme={null}
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:**

```python theme={null}
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

```python theme={null}
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:**

```python theme={null}
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

```python theme={null}
async def on_agent_create(self, agent: Agent) -> None
```

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

### on\_agent\_evict

```python theme={null}
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:

```json theme={null}
// Chat
{"action": "chat", "message": "Hello!"}

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

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

**POST `{base_path}/{id}/chat`**

Standard chat request:

```json theme={null}
{"message": "Hello, how can you help me?"}
```

Response:

```json theme={null}
{"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:

```bash theme={null}
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`**

```json theme={null}
{"task": "summarize", "text": "Long document..."}
// or
{"payload": {"task": "summarize", "text": "..."}}
```

Response:

```json theme={null}
{"agent_id": "agent-123", "result": {"summary": "..."}}
```

***

## Agent.serve\_many

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

```python theme={null}
def serve_many(
    self,
    host: str = "localhost",
    port: int = 8080,
    config: AgentServerConfig | None = None,
) -> None
```

Equivalent to:

```python theme={null}
AgentServer(agent, config=config).serve(host=host, port=port)
```

***

## Example

### Basic Usage

```python theme={null}
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

```python theme={null}
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

```python theme={null}
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)
```