Matilda
Contact
Agent SDK · Running agents

Stream.

Iterate the full agent event stream as events arrive.

runner.stream(agent, prompt, **options)

An async generator that yields AgentEvent objects as they arrive. This is the full event stream — tool calls, usage, status changes, safety replacements, and more. Accepts the same options as run() except callbacks / throw_on_stream_error.

Python
async for event in runner.stream(
    {"name": "explainer", "instructions": "Explain quantum computing."},
    "What is quantum entanglement?",
):
    if event.type == "run.started":
        print(f'Agent "{event.agent_name}" started.')
    elif event.type == "stream.started":
        print(f"Stream {event.stream_id} connected.")
    elif event.type == "response.output_text.delta":
        print(event.delta, end="", flush=True)
    elif event.type == "client.tool.requested":
        print(f"\nTool requested: {event.name}")
    elif event.type == "client.tool.result":
        print(f"Tool result: {event.result}")
    elif event.type == "response.usage":
        print(f"\nTokens: {event.usage.output_tokens}")
    elif event.type == "response.completed":
        print("\n[done]")
    elif event.type == "response.error":
        print(f"Error: {event.code} — {event.message}")

AgentEvent

A union of 20 event dataclasses — the 13 chat events shared with the client SDK, plus 7 agent-level events. Each carries a type: ClassVar[str] discriminator; dispatch on event.type or isinstance().

RunStarted — type="run.started"

Emitted once at the start of a run with the agent's name.

Python
@dataclass(frozen=True)
class RunStarted:
    agent_name: str
    # type = "run.started"

StreamStarted — type="stream.started"

Emitted once when the SSE stream connects, with the durable stream ID.

Python
@dataclass(frozen=True)
class StreamStarted:
    stream_id: str
    # type = "stream.started"

OutputTextDelta — type="response.output_text.delta"

A text chunk from the assistant.

Python
@dataclass(frozen=True)
class OutputTextDelta:
    delta: str
    # type = "response.output_text.delta"

StatusEvent — type="response.status"

Stream lifecycle status change.

Python
@dataclass(frozen=True)
class StatusEvent:
    status: str  # 'thinking' | 'streaming' | 'queued' | 'idle' | 'done' | 'error' | ...
    # type = "response.status"

QueuedEvent — type="response.queued"

Queue position update while waiting for a free slot.

Python
@dataclass(frozen=True)
class QueuedEvent:
    state: str
    position: int
    estimated_wait_seconds: float
    # type = "response.queued"

ToolCallStarted — type="response.tool_call.started"

A server-side tool invocation began.

Python
@dataclass(frozen=True)
class ToolCallStarted:
    tool: str
    input_or_args: object = None
    output: str | None = None
    # type = "response.tool_call.started"

ToolCallProgress — type="response.tool_call.progress"

Progress update from a running server-side tool.

Python
@dataclass(frozen=True)
class ToolCallProgress:
    tool: str
    message: str
    # type = "response.tool_call.progress"

ToolCallCompleted — type="response.tool_call.completed"

A server-side tool invocation finished.

Python
@dataclass(frozen=True)
class ToolCallCompleted:
    tool: str
    status: str  # 'success' | 'error'
    input: str | None = None
    output: str | None = None
    # type = "response.tool_call.completed"

ClientToolRequested — type="client.tool.requested"

The agent called a client tool. The SDK will execute the matching handler from tool_handlers.

Python
@dataclass(frozen=True)
class ClientToolRequested:
    name: str
    args: dict
    id: str | None = None
    # type = "client.tool.requested"

ClientToolExecuting — type="client.tool.executing"

The SDK is about to execute the handler for a requested client tool.

Python
@dataclass(frozen=True)
class ClientToolExecuting:
    name: str
    args: dict
    id: str | None = None
    # type = "client.tool.executing"

ClientToolResult — type="client.tool.result"

A client tool handler returned a result.

Python
@dataclass(frozen=True)
class ClientToolResult:
    name: str
    result: str
    is_error: bool
    id: str | None = None
    # type = "client.tool.result"

ClientToolRoundtrip — type="client.tool.roundtrip"

Emitted after each tool roundtrip cycle, showing progress against the maximum.

Python
@dataclass(frozen=True)
class ClientToolRoundtrip:
    turn: int
    max_turns: int
    # type = "client.tool.roundtrip"

TurnRetrying — type="turn.retrying"

A retryable error occurred and the turn is being retried.

Python
@dataclass(frozen=True)
class TurnRetrying:
    attempt: int
    max_retries: int
    error: dict
    delay_ms: float
    # type = "turn.retrying"

GenerationStatus — type="response.generation_status"

Generation phase update.

Python
@dataclass(frozen=True)
class GenerationStatus:
    phase: str
    # type = "response.generation_status"

OutputTextReplace — type="response.output_text.replace"

The server replaced the output via a safety filter. content holds the replacement text; categories lists the safety categories.

Python
@dataclass(frozen=True)
class OutputTextReplace:
    content: str | None = None
    categories: list[str] | None = None
    # type = "response.output_text.replace"

UsageEvent — type="response.usage"

Token usage data for the turn, in the usage field:

Python
@dataclass(frozen=True)
class Usage:
    output_tokens: int
    context_pct: float | None = None
    context_messages_trimmed: int | None = None
    context_budget_tokens: int | None = None

CursorEvent — type="response.cursor"

Durable stream cursor (event ID). Persist this to resume from this point.

Python
@dataclass(frozen=True)
class CursorEvent:
    last_event_id: str
    # type = "response.cursor"

Truncated — type="response.truncated"

The response was cut short.

Python
@dataclass(frozen=True)
class Truncated:
    reason: str
    # type = "response.truncated"

Completed — type="response.completed"

The stream finished successfully.

Python
Completed()

ResponseError — type="response.error"

An error occurred during the stream.

Python
@dataclass(frozen=True)
class ResponseError:
    code: str  # a ChatErrorCode
    message: str
    # type = "response.error"