# Stream

> Iterate the full agent event stream as events arrive.

Source: https://maincode.com/docs/python-agent-sdk-stream
Section: Agent SDK · Matilda documentation

---

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