Matilda
Contact
Client SDK · Chat

Streaming.

Full event streaming from the chat API.

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