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

from ai_query.agents import Agent

Constructor

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

id · str · required

Stable agent identifier. Used for storage keys, routing, and RPC.

model · LanguageModel | None

Default model for chat(), stream(), run(), and turn().

system · str

Default system prompt.

tools · ToolSet | None

Default tools available to model calls.

storage · Storage | None

Storage backend. Defaults to MemoryStorage().

initial_state · StateT | None

State used when storage has no saved state for this agent.

stop_when · StopCondition | list[StopCondition] | None

Default stop condition for tool loops.

provider_options · ProviderOptions | None

Default provider-specific options.

reasoning · ReasoningConfig | None

Default reasoning controls.

retry · RetryPolicy | None

Default provider-call retry policy for chat(), stream(), run(), and turn().

on_reasoning_event · OnReasoningEvent | None

Callback for provider reasoning events. Agent also emits these as reasoning events.

hooks · AgentHooks | None

Default runtime hooks for step control, tool policy, tool result overrides, and reasoning observation.

transport · AgentTransport | None

Transport used for remote agent calls.

event_bus · EventBus | None

Optional event bus used by server/transport layers.

Class Attributes

enable_event_log · bool

When True, emitted events are persisted for replay.

state · dict | BaseModel

Optional subclass blueprint for initial or typed state.

Properties

id · str

Agent ID.

storage · Storage

Storage backend.

state · StateT

Current loaded state. Raises RuntimeError before start().

messages · list[Message]

Current message history.

start

async def start(self) -> None

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

Idempotent and concurrency-safe.

stop

async def stop(self) -> None

Stops the mailbox processor and calls on_stop().

Context Manager

async with agent:
    ...

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

chat

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:

ValueError · Exception

Raised if self.model is not configured.

AbortError · Exception

Raised if signal aborts.

stream

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

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

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:

ValueError · Exception

Raised if self.model is not configured.

set_state

async def set_state(self, state: StateT) -> None

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

update_state

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

async def clear(self) -> None

Clears message history and persists an empty message list.

emit

async def emit(
    self,
    event: str,
    data: dict[str, Any],
    *,
    replay: bool = True,
) -> int

Emits an agent event.

Parameters:

event · str · required

Event type to deliver to connected clients.

data · dict[str, Any] · required

Event payload.

replay · 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.

Returns:

event_id · int

Monotonic event ID for this agent instance.

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

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

async def clear_event_log(self) -> None

Clears event log and resets event counter.

handle_request

async def handle_request(self, request: dict[str, Any]) -> dict[str, Any]

Serverless request dispatcher.

Supported actions:

• chat
• action
• invoke
• state

handle_request_stream

async def handle_request_stream(
    self,
    request: dict[str, Any],
) -> AsyncIterator[str]

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

call

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

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

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

Starts an AgentServer for this agent class.

Lifecycle Hooks

Override in subclasses:

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

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

agent = Agent("assistant", model=openai("gpt-4o"))

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

Persistent agent

agent = Agent(
    "assistant",
    model=openai("gpt-4o"),
    storage=SQLiteStorage("agents.sqlite3"),
)

Live turn

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
• AgentTurn
• Agent Server
• Stateful Agents
• Live Turns