> ## 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.

# 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

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

<CardGroup cols={2}>
  <Card title="Per-User Assistants" icon="user">
    Each user gets their own AI with isolated state and history
  </Card>

  <Card title="Task Workers" icon="robot">
    Spawn independent agents for parallel task execution
  </Card>

  <Card title="Multi-Tenant APIs" icon="building">
    Isolate AI state per customer or organization
  </Card>

  <Card title="Session Agents" icon="clock">
    Create agents for specific workflows or sessions
  </Card>
</CardGroup>

## 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](/reference/agent-server) and [AgentRegistry API Reference](/reference/agent-registry).

## 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.

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

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

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

<Tabs>
  <Tab title="Python (Client SDK)">
    The recommended way to interact with the server.

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

  <Tab title="REST (HTTP)">
    ```bash theme={null}
    # 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..."}'
    ```
  </Tab>

  <Tab title="SSE (Server-Sent Events)">
    ```javascript theme={null}
    // 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!" })
    });
    ```
  </Tab>
</Tabs>

## Lifecycle Management

### Idle Timeout

Agents with no connections are automatically evicted after `idle_timeout` seconds:

```python theme={null}
config = AgentServerConfig(idle_timeout=600)  # 10 minutes
```

### Max Agents Limit

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

```python theme={null}
config = AgentServerConfig(max_agents=100)
```

### Lifecycle Hooks

Override hooks on `AgentServer` for custom logic:

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

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