# Streaming

> Full event streaming from the chat API.

Source: https://maincode.com/docs/python-client-sdk-streaming
Section: Client SDK · Matilda documentation

---

## `chat.stream(...)`

An async generator that yields `ChatEvent` objects as they arrive over SSE. This is the full event stream — tool calls, usage, status changes, safety replacements, and more.

```python title="stream.py"
async def main():
    async with MatildaClient(token="...") as client:
        async for event in client.chat.stream(input="Explain quantum computing."):
            if event.type == "response.created":
                print(f"Stream started: {event.stream_id}")
            elif event.type == "response.output_text.delta":
                print(event.delta, end="", flush=True)
            elif event.type == "response.tool_call.started":
                print(f"\nTool: {event.tool}")
            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}")
```

Each event is a frozen dataclass with a `type: ClassVar[str]` discriminator, so you can dispatch on `event.type` (as above) or with `isinstance(event, OutputTextDelta)`.

## `ChatEvent`

A union of 14 event dataclasses:

### `ResponseCreated` — `type="response.created"`

Emitted once at stream start with the durable stream ID.

```python
@dataclass(frozen=True)
class ResponseCreated:
    stream_id: str
    # type = "response.created"
```

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

### `OutputTextReplace` — `type="response.output_text.replace"`

The server replaced the output (e.g. safety filter). The `content` field holds the replacement text; `categories` lists the safety categories that triggered the replacement.

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

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

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

### `ToolCallCompleted` — `type="response.tool_call.completed"`

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

### `GenerationStatus` — `type="response.generation_status"`

Generation phase update.

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

### `UsageEvent` — `type="response.usage"`

Token usage data for the turn. The `usage` field is a `Usage` dataclass:

```python
@dataclass(frozen=True)
class Usage:
    output_tokens: int
    context_pct: float | None = None              # context window usage (0-100)
    context_messages_trimmed: int | None = None   # messages trimmed to fit context budget
    context_budget_tokens: int | None = None      # total context budget in tokens
```

### `CursorEvent` — `type="response.cursor"`

Durable stream cursor (Redis stream entry 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"
```