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

> API reference for the Agent class

`Agent` is the stateful orchestration class in `ai-query`.

Use it when you need model configuration, tools, persistent state, message history, events, or transport/RPC behavior attached to a stable agent ID.

## Import

```python theme={null}
from ai_query.agents import Agent
```

## Constructor

```python theme={null}
class Agent(Generic[StateT]):
    def __init__(
        self,
        id: str,
        *,
        model: LanguageModel | None = None,
        system: str = "You are a helpful assistant.",
        tools: ToolSet | None = None,
        storage: Storage | None = None,
        initial_state: StateT | None = None,
        stop_when: StopCondition | list[StopCondition] | None = None,
        provider_options: ProviderOptions | None = None,
        reasoning: ReasoningConfig | None = None,
        retry: RetryPolicy | None = None,
        on_reasoning_event: OnReasoningEvent | None = None,
        hooks: AgentHooks | None = None,
        transport: AgentTransport | None = None,
        event_bus: EventBus | None = None,
    ) -> None
```

### Parameters

<ParamField path="id" type="str" required>
  Stable agent identifier. Used for storage keys, routing, and RPC.
</ParamField>

<ParamField path="model" type="LanguageModel | None">
  Default model for `chat()`, `stream()`, `run()`, and `turn()`.
</ParamField>

<ParamField path="system" type="str">
  Default system prompt.
</ParamField>

<ParamField path="tools" type="ToolSet | None">
  Default tools available to model calls.
</ParamField>

<ParamField path="storage" type="Storage | None">
  Storage backend. Defaults to `MemoryStorage()`.
</ParamField>

<ParamField path="initial_state" type="StateT | None">
  State used when storage has no saved state for this agent.
</ParamField>

<ParamField path="stop_when" type="StopCondition | list[StopCondition] | None">
  Default stop condition for tool loops.
</ParamField>

<ParamField path="provider_options" type="ProviderOptions | None">
  Default provider-specific options.
</ParamField>

<ParamField path="reasoning" type="ReasoningConfig | None">
  Default reasoning controls.
</ParamField>

<ParamField path="retry" type="RetryPolicy | None">
  Default provider-call retry policy for `chat()`, `stream()`, `run()`, and `turn()`.
</ParamField>

<ParamField path="on_reasoning_event" type="OnReasoningEvent | None">
  Callback for provider reasoning events. Agent also emits these as `reasoning` events.
</ParamField>

<ParamField path="hooks" type="AgentHooks | None">
  Default runtime hooks for step control, tool policy, tool result overrides, and reasoning observation.
</ParamField>

<ParamField path="transport" type="AgentTransport | None">
  Transport used for remote agent calls.
</ParamField>

<ParamField path="event_bus" type="EventBus | None">
  Optional event bus used by server/transport layers.
</ParamField>

## Class Attributes

<ParamField path="enable_event_log" type="bool">
  When `True`, emitted events are persisted for replay.
</ParamField>

<ParamField path="state" type="dict | BaseModel">
  Optional subclass blueprint for initial or typed state.
</ParamField>

## Properties

<ResponseField name="id" type="str">
  Agent ID.
</ResponseField>

<ResponseField name="storage" type="Storage">
  Storage backend.
</ResponseField>

<ResponseField name="state" type="StateT">
  Current loaded state. Raises `RuntimeError` before `start()`.
</ResponseField>

<ResponseField name="messages" type="list[Message]">
  Current message history.
</ResponseField>

## start

```python theme={null}
async def start(self) -> None
```

Loads state, messages, event log, action registry, and starts the mailbox processor.

Idempotent and concurrency-safe.

## stop

```python theme={null}
async def stop(self) -> None
```

Stops the mailbox processor and calls `on_stop()`.

## Context Manager

```python theme={null}
async with agent:
    ...
```

Equivalent to `await agent.start()` followed by `await agent.stop()`.

## chat

```python theme={null}
async def chat(
    self,
    message: Content,
    *,
    signal: AbortSignal | None = None,
) -> str
```

Appends a user message, runs the model/tool loop, persists messages, and returns final text.

Raises:

<ResponseField name="ValueError" type="Exception">
  Raised if `self.model` is not configured.
</ResponseField>

<ResponseField name="AbortError" type="Exception">
  Raised if `signal` aborts.
</ResponseField>

## stream

```python theme={null}
async def stream(
    self,
    message: Content,
    *,
    signal: AbortSignal | None = None,
) -> AsyncIterator[str]
```

Streams text chunks while persisting final message history.

Raises the same errors as `chat()`.

## run

```python theme={null}
async def run(
    self,
    message: Content,
    *,
    options: TurnOptions | None = None,
) -> TurnResult
```

Runs an `AgentTurn` and returns its structured result.

Use this when you need `steps`, `usage`, `finish_reason`, or `output_message`.

## turn

```python theme={null}
def turn(
    self,
    message: Content,
    *,
    options: TurnOptions | None = None,
) -> AgentTurn
```

Creates a live turn. The turn starts lazily when `events()`, `text_stream()`, or `result()` is consumed.

Raises:

<ResponseField name="ValueError" type="Exception">
  Raised if `self.model` is not configured.
</ResponseField>

## set\_state

```python theme={null}
async def set_state(self, state: StateT) -> None
```

Replaces current state, persists it, and emits a `state` event.

## update\_state

```python theme={null}
async def update_state(self, **kwargs: Any) -> None
```

Merges keyword updates into state, persists the state, and emits a `state` event.

For Pydantic state, uses `model_copy(update=kwargs)` when available.

## clear

```python theme={null}
async def clear(self) -> None
```

Clears message history and persists an empty message list.

## emit

```python theme={null}
async def emit(
    self,
    event: str,
    data: dict[str, Any],
    *,
    replay: bool = True,
) -> int
```

Emits an agent event.

Parameters:

<ParamField path="event" type="str" required>
  Event type to deliver to connected clients.
</ParamField>

<ParamField path="data" type="dict[str, Any]" required>
  Event payload.
</ParamField>

<ParamField path="replay" type="bool">
  When `True`, the event is stored in the replay log and may be returned by `replay_events()`. When `False`, the event is delivered live to connected clients but omitted from replay persistence.
</ParamField>

Returns:

<ResponseField name="event_id" type="int">
  Monotonic event ID for this agent instance.
</ResponseField>

If `enable_event_log = True`, events are persisted to storage.

Notes:

* Event IDs remain monotonic even for `replay=False` events.
* Replay streams may therefore contain gaps in event IDs.
* Use `replay=False` for ephemeral UI commands such as `open_document` that should not re-run on reconnect.

## replay\_events

```python theme={null}
async def replay_events(self, after_id: int = 0) -> AsyncIterator[Event]
```

Yields events with `id > after_id` from the in-memory/persisted event log.

## clear\_event\_log

```python theme={null}
async def clear_event_log(self) -> None
```

Clears event log and resets event counter.

## handle\_request

```python theme={null}
async def handle_request(self, request: dict[str, Any]) -> dict[str, Any]
```

Serverless request dispatcher.

Supported actions:

* `chat`
* `action`
* `invoke`
* `state`

## handle\_request\_stream

```python theme={null}
async def handle_request_stream(
    self,
    request: dict[str, Any],
) -> AsyncIterator[str]
```

Serverless streaming request handler. Yields SSE-formatted strings for chat requests.

## call

```python theme={null}
def call(self, agent_id: str, *, agent_cls: type[T]) -> AgentCallProxy[T]
```

Returns a proxy for calling another agent through the configured transport.

## handle\_invoke

```python theme={null}
async def handle_invoke(self, payload: dict[str, Any]) -> dict[str, Any]
```

Handles inbound calls from other agents or transports.

Resolution order:

1. registered `@action` method
2. legacy `on_call_{method}` handler
3. error response

## serve

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

Starts an `AgentServer` for this agent class.

## Lifecycle Hooks

Override in subclasses:

```python theme={null}
async def on_start(self) -> None: ...
async def on_stop(self) -> None: ...
async def on_connect(self, connection, ctx) -> None: ...
async def on_message(self, connection, message) -> None: ...
async def on_close(self, connection, code: int, reason: str) -> None: ...
```

## @action

```python theme={null}
from ai_query.agents import action


@action
async def method_name(self, **kwargs): ...
```

Marks an agent method as externally callable via action/RPC routes.

## Examples

### Basic agent

```python theme={null}
agent = Agent("assistant", model=openai("gpt-4o"))

async with agent:
    text = await agent.chat("Hello")
```

### Persistent agent

```python theme={null}
agent = Agent(
    "assistant",
    model=openai("gpt-4o"),
    storage=SQLiteStorage("agents.sqlite3"),
)
```

### Live turn

```python theme={null}
turn = agent.turn("Fix the failing test")

async for event in turn.events():
    if event.type == "text.delta":
        print(event.text, end="")
```

## See Also

* [Hooks](/core/hooks)
* [AgentTurn](/reference/agent-turn)
* [Agent Server](/reference/agent-server)
* [Stateful Agents](/core/stateful-agents)
* [Live Turns](/core/live-turns)