---
title: "Matilda Python agent SDK"
summary: "Full API reference for the matilda-agent-sdk Python package covering agents, runner, client tools, sessions, structured output and stream resume."
url: https://help.maincode.com/products/matilda/v1/en/sdks/python-agent
product: "Matilda"
kind: reference
version: v1
discovery: {"area":"build","components":["python-agent"],"capabilities":["authentication","streaming","tools"],"tasks":["Build an agent with tools"],"platforms":[],"releases":[{"component":"python-agent","version":"0.1.0","channel":"stable"}]}
source: MaincodeHQ/matilda-core@85e7865b594539e3add99a033eedd80790a5a556
---

# Matilda Python agent SDK

**Version 0.1.0**

A Python SDK for building agentic applications on Matilda. Provides agent abstractions, client-side tool execution, multi-turn sessions, automatic retry with exponential backoff, DSML tool-call interception, and durable stream resume — all on top of the Matilda-native chat contract. Async-first (`asyncio` + `httpx`). Requires Python ≥ 3.12.

The agent SDK wraps the [client SDK](https://help.maincode.com/products/matilda/v1/en/sdks/python-client) (`matilda-client`) and adds:

- **Agent** — a named, configurable persona with dynamic instructions and purpose-based routing
- **Runner** — runs agent turns with streaming, retry, and a client-side tool execution loop
- **Session** — multi-turn conversations with auto-managed `conversation_id` and turn accumulation
- **Client tools** — register handlers the agent can invoke mid-turn; the SDK handles the roundtrip loop
- **DSML interception** — tool calls emitted as text tokens (`<｜DSML｜tool_call>`) are automatically captured and surfaced as native tool events
- **Durable stream resume** — reconnect to a detached stream from the last cursor

Agent runs send `persist: false` by default — conversations do not appear in the Matilda web app's chat history.

## What's included {#whats-included}

- **Agent** — named persona with static or dynamic instructions, purpose-based routing
- **Runner** — `run()`, `stream()`, `stream_text()`, `run_text()`, `run_object()`, `stream_object()` with retry and tool execution
- **Session** — multi-turn conversations with automatic `conversation_id` reuse and client-side transcript replay
- **Client tools** — `ToolHandlers` with automatic roundtrip loop and advertised-tool guard
- **Files** — upload (single and parallel), retrieve metadata
- **Conversations** — list, retrieve, rename, and set message feedback
- **Auth** — managed PKCE browser login, RFC 8628 device flow, token restore, persistent token storage

## What's NOT included {#whats-not-included}

- **Server-side tool execution** — server-side tools (web search, code execution, etc.) are handled by Matilda core. The agent SDK's tool loop is for client-side tools only.
- **Model/provider selection** — Matilda core owns routing, safety, and policy.

---

## Table of Contents {#table-of-contents}

1. [Installation](#s-installation)
2. [Quick Start](#s-quick-start)
3. [Configuration](#s-configuration)
4. [Authentication](#s-authentication)
5. [Agent](#s-agent)
6. [Runner — Run (Non-Streaming)](#s-runner-run-non-streaming)
7. [Runner — Stream (Full Events)](#s-runner-stream-full-events)
8. [Runner — Text Helpers](#s-runner-text-helpers)
9. [Runner — Structured Output](#s-runner-structured-output)
10. [Client Tools](#s-client-tools)
11. [Session (Multi-Turn)](#s-session-multi-turn)
12. [Stream Resume](#s-stream-resume)
13. [Files](#s-files)
14. [Conversations](#s-conversations)
15. [Feedback](#s-feedback)
17. [Error Handling](#s-error-handling)
18. [Multi-Agent Patterns](#s-multi-agent-patterns)
19. [Suggested Tasks / Recipes](#s-suggested-tasks-recipes)
20. [Exports Reference](#s-exports-reference)

---

## 1. Installation {#installation}

```sh
pip install matilda-agent-sdk==0.1.0
# or: uv add matilda-agent-sdk
```

Requires Python ≥ 3.12. The client SDK (`matilda-client`) is installed automatically as a dependency.

`pydantic` is an optional dependency — install it alongside the SDK if you want schema-validated structured output ([§9](#s-runner-structured-output)):

```sh
pip install pydantic
```

### Import {#import}

```python
from matilda_agent_sdk import Agent, Runner, run, stream
```

---

## 2. Quick Start {#quick-start}

### Minimal: run a single agent turn {#minimal-run-a-single-agent-turn}

```python
import asyncio

from matilda_agent_sdk import run


async def main():
    result = await run(
        {"name": "greeter", "instructions": "Be friendly and concise."},
        "Say hello in three languages.",
    )

    print(result.final_output)
    print(result.usage)


asyncio.run(main())
```

### Minimal: streaming {#minimal-streaming}

```python
from matilda_agent_sdk import stream


async def main():
    async for event in stream(
        {"name": "storyteller", "instructions": "Write a short sci-fi haiku."},
        "Write about a Dyson sphere.",
    ):
        if event.type == "response.output_text.delta":
            print(event.delta, end="", flush=True)
        if event.type == "response.completed":
            print("\n[done]")
```

### Authenticated: device flow + run {#authenticated-device-flow-run}

```python
import asyncio

from matilda_agent_sdk import MatildaClient, Runner


async def main():
    runner = Runner(MatildaClient(base_url="https://matilda.com/api"))

    # Authenticate via RFC 8628 device flow — prints a code to stderr
    if not await runner.auth.get_tokens():
        await runner.auth.login_with_device_flow(client_id="matilda-code")

    # Token is now managed automatically — refresh on 401 comes for free
    result = await runner.run(
        {"name": "helper", "instructions": "Be concise."},
        "What is the capital of Australia?",
    )
    print(result.final_output)


asyncio.run(main())
```

---

## 3. Configuration {#configuration}

### `configure_default_client(client)` {#configure-default-client}

Installs the shared default `MatildaClient` used by the module-level convenience functions (`run`, `stream`, `stream_text`, `run_text`, `run_object`, `stream_object`) and their shared default `Runner`. Returns nothing.

```python
import os

from matilda_agent_sdk import MatildaClient, configure_default_client

configure_default_client(
    MatildaClient(
        base_url="https://matilda.com/api",
        token=os.environ["MATILDA_ACCESS_TOKEN"],
    )
)
```

The client derives its base URL, static token, and async `get_token` provider from `MatildaConfig` — see the [client SDK docs](https://help.maincode.com/products/matilda/v1/en/sdks/python-client) for the full field list.

### `MatildaClient` {#matildaclient}

The underlying API client, re-exported from `matilda-client`. Each `Runner` can hold its own `MatildaClient` instance for isolation, or share the module-level default.

```python
import os

from matilda_agent_sdk import MatildaClient, Runner

client = MatildaClient(
    base_url="https://matilda.com/api",
    token=os.environ["MATILDA_ACCESS_TOKEN"],
)

runner = Runner(client)
```

### `Runner` {#runner}

The main agent execution class. Optionally accepts an explicit `MatildaClient` for isolation.

```python
from matilda_agent_sdk import MatildaClient, Runner

# Uses the shared default client (materialised lazily, or installed
# via configure_default_client)
default_runner = Runner()

# Uses an isolated client — independent config, auth, and token lifecycle
isolated_runner = Runner(MatildaClient(base_url="https://matilda.com/api"))
```

| Constructor parameter | Type | Description |
|---|---|---|
| `client` | `MatildaClient \| None` | Optional explicit client. If omitted, the shared default is used (created on first use if none was configured). |

### Environment URLs {#environment-urls}

| Environment | Base URL |
|---|---|
| Production | `https://matilda.com/api` |

### Instance isolation {#instance-isolation}

Each `Runner` holds its own `MatildaClient` (either explicit or the shared default). The `auth`, `files`, `conversations`, and `feedback` resources resolve their client lazily, so a client installed via `configure_default_client()` is picked up correctly.

```python
from matilda_agent_sdk import MatildaClient, Runner

runner_a = Runner(MatildaClient(base_url="https://matilda.com/api"))
runner_b = Runner(MatildaClient(base_url="https://matilda.com/api"))

# Each runner is fully isolated — independent auth, config, and token lifecycle
await runner_a.auth.login_with_device_flow(client_id="matilda-code")
await runner_b.auth.login_with_browser(client_id="matilda-code")
```

### Other configuration utilities {#other-configuration-utilities}

| Export | Description |
|---|---|
| `MatildaConfig` | Client configuration dataclass (re-exported from `matilda-client`). |
| `MATILDA_API_VERSION_HEADER` / `MATILDA_CURRENT_API_VERSION` | API version constants, importable from `matilda_client`. |

---

## 4. Authentication {#authentication}

The agent SDK provides `AgentAuth` — a managed auth tier layered on the client SDK's `TokenManager`. On successful login, a `TokenManager` is auto-configured as the runner client's token provider. Every subsequent request automatically carries a managed access token with single-flight, skew-aware auto-refresh.

### OAuth client ID {#oauth-client-id}

The agent SDK uses the `matilda-code` OAuth client ID, which supports both PKCE browser login and RFC 8628 device flow. All examples in this documentation use `matilda-code`.

### API keys {#api-keys}

Agents can also authenticate with an `mc_live_` API key instead of OAuth — useful for CI and server-side deployments where there's no browser. Keys are minted with the client SDK's `matilda-key` CLI (social-login friendly — no code required):

```sh
matilda-key create-api-key --name "my-agent"
# or without installing: uvx --from matilda-client matilda-key create-api-key --name "my-agent"
# mints an mc_live_… key after device-flow sign-in; see the client SDK docs for flags
```

Then supply it as a static access token:

```python
import os

from matilda_agent_sdk import MatildaClient, configure_default_client

configure_default_client(
    MatildaClient(
        base_url="https://matilda.com/api",
        token=os.environ["MATILDA_API_KEY"],
    )
)
```

### `runner.auth.login_with_browser(...)` {#runner-auth-login-with-browser}

Managed PKCE browser login (RFC 8252 loopback). Starts a temporary local server, opens the browser, receives the auth code, exchanges it for tokens, and auto-wires a `TokenManager` on the runner's client.

```python
tokens = await runner.auth.login_with_browser(
    client_id="matilda-code",
    open_browser=lambda url: print(f"Open: {url}"),
)
print(tokens.access_token)
```

#### Parameters {#parameters}

| Parameter | Type | Default | Description |
|---|---|---|---|
| `client_id` | `str` | **(required)** | OAuth client alias (e.g. `'matilda-code'`). |
| `scope` | `str` | `'openid offline_access'` | Space-separated OAuth scopes. |
| `identity_provider_id` | `str \| None` | `None` | Route straight to a federated IdP (e.g. Google SSO). |
| `callback_port` | `int \| None` | random ephemeral | Fixed loopback port. |
| `timeout_ms` | `int` | `300_000` (5 min) | How long to wait for the browser callback. |
| `open_browser` | `Callable[[str], None \| Awaitable[None]] \| None` | `None` | Called with the authorize URL. |
| `success_redirect` | `str` | `DEFAULT_SUCCESS_REDIRECT` (`'https://matilda.com/cli/signed-in'`) | URL the browser is 302-redirected to on success. |
| `error_redirect` | `str \| None` | `None` | URL for the error case. |
| `token_store` | `StorageAdapter \| None` | `memory_storage()` | Custom token persistence. |
| `token_lock` | `TokenLock \| None` | `None` | Cross-process critical-section lock for token refresh. |
| `on_event` | `Callable[[LoginFlowEvent], None] \| None` | `None` | Subscribe to login flow state events. |
| `http` | `httpx.AsyncClient \| None` | `None` | Override HTTP transport. |

Returns `TokenSet`.

### `runner.auth.login_with_device_flow(...)` {#runner-auth-login-with-device-flow}

Managed RFC 8628 device flow. Requests a device code, prints the user code and verification URL to stderr (by default), and polls until the user authorises.

```python
def show_login(event):
    if event.type == "user_code":
        print(f"Visit {event.verification_uri} and enter code: {event.user_code}")


tokens = await runner.auth.login_with_device_flow(
    client_id="matilda-code",
    on_event=show_login,
)
```

If no `on_event` handler is provided, the SDK prints the user code and verification URL to stderr automatically (`default_device_on_event`).

#### Parameters {#parameters-2}

| Parameter | Type | Default | Description |
|---|---|---|---|
| `client_id` | `str` | **(required)** | OAuth client alias. |
| `scope` | `str` | `'openid offline_access'` | Space-separated OAuth scopes. |
| `token_store` | `StorageAdapter \| None` | `memory_storage()` | Custom token persistence. |
| `token_lock` | `TokenLock \| None` | `None` | Cross-process lock for token refresh. |
| `on_event` | `Callable[[LoginFlowEvent], None] \| None` | `default_device_on_event` | Subscribe to login flow events. If omitted, prints user code to stderr. |
| `http` | `httpx.AsyncClient \| None` | `None` | Override HTTP transport. |

Returns `TokenSet`.

### `runner.auth.restore(...)` {#runnerauthrestore}

Adopts tokens that were already persisted (e.g. by `create_file_token_store`) without repeating the interactive login. Returns `None` when the store holds nothing usable, so a caller can fall back to `login_with_*`.

```python
from pathlib import Path

from matilda_client import create_file_token_store

token_store = create_file_token_store(Path.home() / ".matilda" / "tokens.json")

tokens = await runner.auth.restore(
    client_id="matilda-code",
    token_store=token_store.store,
    token_lock=token_store.lock,
)

if not tokens:
    # No persisted tokens — fall back to interactive login
    await runner.auth.login_with_device_flow(
        client_id="matilda-code",
        token_store=token_store.store,
        token_lock=token_store.lock,
    )
```

#### Parameters {#parameters-3}

| Parameter | Type | Description |
|---|---|---|
| `client_id` | `str` | OAuth client alias. |
| `token_store` | `StorageAdapter` | Token persistence adapter. |
| `token_lock` | `TokenLock \| None` | Cross-process lock. |
| `http` | `httpx.AsyncClient \| None` | Override HTTP transport. |
| `metadata` | `AuthServerMetadata \| None` | Pre-fetched server metadata (skips discovery). |

Returns `TokenSet | None`.

### `runner.auth.get_tokens()` {#runner-auth-get-tokens}

Returns the current token set from the managed `TokenManager`, or `None` if not authenticated.

```python
tokens = await runner.auth.get_tokens()
if tokens:
    print(f"Token expires at epoch ms: {tokens.expires_at}")
```

Returns `TokenSet | None`.

### `runner.auth.logout()` {#runnerauthlogout}

Clears the token store, destroys the `TokenManager`, and restores the client's previous token provider (important when the client is the shared default).

```python
await runner.auth.logout()
```

Returns `None`.

### Token persistence {#token-persistence}

By default, tokens are stored in memory (`memory_storage()`). For cross-process persistence (e.g. CLI sessions), use `create_file_token_store` — imported from `matilda_client`:

```python
from pathlib import Path

from matilda_agent_sdk import MatildaClient, Runner
from matilda_client import create_file_token_store


async def main():
    token_path = Path.home() / ".matilda" / "tokens.json"
    token_store = create_file_token_store(token_path)

    runner = Runner(MatildaClient(base_url="https://matilda.com/api"))

    await runner.auth.login_with_browser(
        client_id="matilda-code",
        token_store=token_store.store,
        token_lock=token_store.lock,
    )
```

The file store uses a `0600` JSON file with a lockfile-based single-writer lock to prevent cross-process refresh races.

### `StorageAdapter` protocol {#storageadapter-protocol}

Imported from `matilda_client`:

```python
class StorageAdapter(Protocol):
    def get(self, key: str) -> str | None | Awaitable[str | None]: ...
    def set(self, key: str, value: str) -> None | Awaitable[None]: ...
    def remove(self, key: str) -> None | Awaitable[None]: ...
```

### `TokenManager` {#tokenmanager}

```python
await manager.get_access_token(force_refresh=False)  # -> str
await manager.get_tokens()                           # -> TokenSet | None
await manager.set_tokens(tokens)                     # -> None
await manager.clear()                                # -> None
```

### `TokenSet` {#tokenset}

```python
@dataclass(frozen=True)
class TokenSet:
    access_token: str
    expires_at: int              # epoch milliseconds
    refresh_token: str | None
    id_token: str | None
```

### `AuthError` class {#autherror-class}

Imported from `matilda_client`:

```python
class AuthError(MatildaError):
    code: str         # OAuth error code (e.g. 'invalid_grant', 'authorization_pending')
    retryable: bool   # True for transient 5xx/network; False for revoked tokens
```

### `LoginFlowEvent` {#loginflowevent}

Imported from `matilda_client` — three frozen dataclasses with a `type` discriminator:

```python
# StateEvent        (type="state")         — status: 'idle' | 'awaiting_user' | 'exchanging' | 'authenticated' | 'error'
# AuthorizeUrlEvent (type="authorize_url") — url: str
# UserCodeEvent     (type="user_code")     — user_code, verification_uri, verification_uri_complete
```

---

## 5. Agent {#agent}

An `Agent` is a named, reusable persona with optional instructions, context, purpose, and metadata. You can pass a plain `dict` with the same keys anywhere an `Agent` is accepted — the SDK normalises it (a camelCase `responseMode` key is also accepted as an alias of `response_mode`).

### `Agent` class {#agent-class}

```python
from matilda_agent_sdk import Agent

reviewer = Agent(
    name="code-reviewer",
    purpose="code",
    instructions="Review code for correctness, security, and readability.",
    context="Project: matilda-core\nLanguage: Python",
    metadata={"team": "platform"},
)
```

The `Agent` is reusable — construct once, run many times.

#### Constructor parameters {#constructor-parameters}

| Parameter | Type | Default | Description |
|---|---|---|---|
| `name` | `str` | **(required)** | Agent name. Must be non-empty (raises `ValueError` otherwise). |
| `instructions` | `AgentInstructions` | `None` | Static string or dynamic callable (see below). |
| `context` | `str \| None` | `None` | Additional context appended to instructions under a `Context:` header. |
| `purpose` | `AgentPurpose` | `'code'` | Controls the default `response_mode` and how the server routes the request. |
| `response_mode` | `str \| None` | `None` | Override the response mode. If omitted, derived from `purpose`. |
| `metadata` | `dict` | `{}` | Custom data available to dynamic instructions. |

### `AgentPurpose` {#agentpurpose}

```python
AgentPurpose = Literal["code", "analysis", "general"]
```

| Purpose | `response_mode` default |
|---|---|
| `'code'` | `'auto'` |
| `'analysis'` | `'deep'` |
| `'general'` | `'auto'` |

### Dynamic instructions {#dynamic-instructions}

Instructions can be a callable that receives a runtime context dict, letting you customise behaviour per-call (async callables are awaited):

```python
from matilda_agent_sdk import Agent, run


def reviewer_instructions(ctx):
    lang = ctx["metadata"].get("language", "auto-detect")
    strictness = ctx["metadata"].get("strictness", "normal")
    return "\n".join(
        [
            "Review the following code.",
            f"Language: {lang}",
            f"Strictness: {strictness}",
            "Focus on: correctness, security, and readability.",
        ]
    )


agent = Agent(name="code-reviewer", purpose="code", instructions=reviewer_instructions)


async def main():
    result = await run(
        agent,
        "def add(a, b): return a + b",
        metadata={"language": "Python", "strictness": "strict"},
    )
    print(result.final_output)
```

#### `AgentInstructions` {#agentinstructions}

```python
AgentInstructions = str | Callable[[dict], str | Awaitable[str]]
```

The context dict passed to the callable:

```python
{
    "agent_name": agent.name,
    "input": trimmed_prompt,
    "purpose": purpose,
    "metadata": merged_metadata,
}
```

### How agent messages are constructed {#how-agent-messages-are-constructed}

The SDK packs the agent's identity, instructions, context, and the user's prompt into a single user message with labelled sections:

```text
Agent: code-reviewer

Instructions:
Review the following code.
Language: Python
...

Context:
Project: matilda-core

Code task:
def add(a, b): return a + b
```

### `build_agent_chat_request(...)` {#build-agent-chat-request}

Low-level async builder that constructs the chat request dict without executing it. Useful for testing, logging, or custom execution paths.

```python
import asyncio

from matilda_agent_sdk import build_agent_chat_request


async def main():
    request = await build_agent_chat_request(
        prompt="Fix the failing test",
        agent={"name": "helper", "purpose": "code"},
        instructions="Be specific.",
        context="Project root: /repo",
        conversation_id="conv-1",
        file_ids=["file-1"],
        client_tools=[{"name": "read_file", "description": "Read a file", "parameters": {"type": "object"}}],
    )

    # request["messages"], request["responseMode"], request["conversation_id"], ...
```

#### Parameters {#parameters-4}

| Parameter | Type | Description |
|---|---|---|
| `prompt` | `str` | The user's message. Required. |
| `agent` | `Agent \| dict \| None` | Optional agent to use. Defaults to a generic agent. |
| `purpose` | `AgentPurpose \| None` | Fallback purpose if agent doesn't specify one. |
| `instructions` | `AgentInstructions \| None` | Override the agent's instructions. |
| `context` | `str \| None` | Override the agent's context. |
| `metadata` | `dict \| None` | Merge with agent's metadata. |
| `conversation_id` | `str \| None` | Associate with a conversation thread. |
| `file_ids` | `list[str] \| None` | File IDs to attach. |
| `client_tools` | `list[dict] \| None` | Client tools to advertise. |
| `response_mode` | `str \| None` | Override response mode. |
| `response_schema` | `str \| None` | Raw JSON Schema string to constrain output. |
| `history` | `list[dict] \| None` | Prior-transcript messages replayed ahead of this turn. |

Returns a `dict` shaped like a chat request (`messages`, `responseMode`, optional `conversation_id` / `fileIds` / `clientTools` / `responseSchema`).

---

## 6. Runner — Run (Non-Streaming) {#runner-run-non-streaming}

### `runner.run(agent, prompt, **options)` {#runnerrunagent-prompt-options}

Runs an agent turn and returns the complete result. Internally streams and collects all events.

```python
from matilda_agent_sdk import Runner

runner = Runner()

result = await runner.run(
    {"name": "helper", "instructions": "Be concise."},
    "What is the capital of Australia?",
    conversation_id="conv-123",
    response_mode="instant",
    callbacks={
        "on_token": lambda delta: print(delta, end="", flush=True),
        "on_done": lambda: print("\n[done]"),
    },
)

print(result.final_output)
print(result.usage)
```

### Run options {#run-options}

All options are keyword-only.

| Option | Type | Default | Description |
|---|---|---|---|
| `conversation_id` | `str \| None` | auto-generated | Associates this turn with a conversation thread. |
| `file_ids` | `list[str] \| None` | — | File IDs to attach (from `files.upload()`). |
| `client_tools` | `list[dict] \| None` | — | Client tools to advertise for this turn. |
| `purpose` | `AgentPurpose \| None` | — | Fallback purpose (used when `agent` is a dict without one). |
| `response_mode` | `str \| None` | — | Override the response mode. |
| `response_schema` | `str \| None` | — | Raw JSON Schema (as a string) to grammar-constrain the response to. Prefer `runner.run_object` / `runner.stream_object`, which convert a pydantic model for you (see [§9](#s-runner-structured-output)). |
| `stall_timeout_ms` | `int \| None` | `45_000` | SSE stall watchdog timeout in ms. Pass `0` to disable. |
| `metadata` | `dict \| None` | — | Custom data available to dynamic instructions. |
| `tool_handlers` | `ToolHandlers \| None` | — | Handlers for client tools (see [§10](#s-client-tools)). |
| `max_tool_roundtrips` | `int` | `25` | Maximum tool roundtrip cycles before stopping. |
| `max_retries` | `int` | `0` | Maximum retries on retryable errors (5xx, 429, 408, network). |
| `throw_on_stream_error` | `bool` | `True` | Raise `MatildaAgentStreamError` if the stream emits an error event. |
| `callbacks` | `dict \| None` | — | Callback hooks for events (see below). |
| `history` | `list[dict] \| None` | — | Prior-transcript messages replayed ahead of this turn. `Session` threads these automatically; only set this on a standalone `Runner` when you want to inject external context. Caller-supplied entries sent on a `Session` turn appear *before* the session's own transcript. |
| `fingerprint` | `str \| None` | — | Device fingerprint for rate limiting. |

### `callbacks` {#callbacks}

Simple callback hooks, passed as a `dict`, that fire as events arrive. An alternative to manually iterating `stream()`.

```python
result = await runner.run(
    agent,
    prompt,
    callbacks={
        "on_token": lambda delta: print(delta, end="", flush=True),
        "on_tool_call": lambda name, args: print(f"Tool: {name}"),
        "on_tool_result": lambda name, result, is_error: print(f"Result: {result}"),
        "on_usage": lambda usage: print(f"Tokens: {usage.output_tokens}"),
        "on_error": lambda code, message: print(f"Error: {code}"),
        "on_retry": lambda attempt, error, delay_ms: print(f"Retry {attempt} in {delay_ms}ms"),
        "on_done": lambda: print("Done"),
    },
)
```

| Callback | Signature | Fires when |
|---|---|---|
| `on_event` | `(event: AgentEvent) -> None` | Every event — catch-all, fires before the typed hooks below. Useful for telemetry, UI plumbing, or event logging. |
| `on_token` | `(delta: str) -> None` | A text chunk arrives. |
| `on_tool_call` | `(name: str, args: dict) -> None` | The agent calls a client tool. |
| `on_tool_result` | `(name: str, result: str, is_error: bool) -> None` | A client tool handler returns. |
| `on_usage` | `(usage: Usage) -> None` | Token usage data arrives. |
| `on_error` | `(code: str, message: str) -> None` | A stream error occurs. |
| `on_retry` | `(attempt: int, error: dict, delay_ms: float) -> None` | A retryable error triggers a retry. |
| `on_done` | `() -> None` | The stream finishes. Fires per-turn in multi-turn tool loops. |

### `AgentRunResult` {#agentrunresult}

| Field | Type | Description |
|---|---|---|
| `agent_name` | `str` | The agent's name. |
| `final_output` | `str` | The full assistant response text. Accumulated from `response.output_text.delta` events. |
| `events` | `list[AgentEvent]` | Every event emitted during the run. |
| `stream_id` | `str \| None` | Durable stream ID (from the `stream.started` event). |
| `last_event_id` | `str \| None` | Last stream event ID (for resume). |
| `usage` | `UsageSummary \| None` | Token usage. Accumulated across multi-roundtrip runs. |
| `errors` | `list[StreamErrorDetail]` | Any errors emitted during the run (`code` + `message`). |
| `truncated_reason` | `str \| None` | Why the response was truncated (e.g. `'max_tokens'`, `'max_tool_roundtrips'`). |
| `safety_replace` | `dict \| None` | Set when the backend replaced the answer for safety (`{"message": ..., "categories": [...]}`). `final_output` holds the replacement text. |

### `throw_on_stream_error=False` {#throw-on-stream-error}

By default, `run()` raises `MatildaAgentStreamError` if the stream emits an error event. Pass `throw_on_stream_error=False` to suppress the raise and inspect errors on the returned result instead:

```python
result = await runner.run(agent, prompt, throw_on_stream_error=False)

if result.errors:
    for err in result.errors:
        print(f"{err.code}: {err.message}")
print("Partial output:", result.final_output or "(none)")
```

---

## 7. Runner — Stream (Full Events) {#runner-stream-full-events}

### `runner.stream(agent, prompt, **options)` {#runnerstreamagent-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` {#agentevent}

A union of 21 event dataclasses — the 14 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"` {#runstarted-typerunstarted}

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"` {#streamstarted-typestreamstarted}

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"` {#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"` {#statusevent-typeresponsestatus}

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"` {#queuedevent-typeresponsequeued}

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"` {#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"` {#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"` {#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"` {#clienttoolrequested-typeclienttoolrequested}

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"` {#clienttoolexecuting-typeclienttoolexecuting}

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"` {#clienttoolresult-typeclienttoolresult}

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"` {#clienttoolroundtrip-typeclienttoolroundtrip}

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"` {#turnretrying-typeturnretrying}

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-status}

Generation phase update.

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

#### `OutputTextReplace` — `type="response.output_text.replace"` {#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"` {#usageevent-typeresponseusage}

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 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"` {#cursorevent-typeresponsecursor}

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"` {#truncated-typeresponsetruncated}

The response was cut short.

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

#### `Completed` — `type="response.completed"` {#completed-typeresponsecompleted}

The stream finished successfully.

```python
Completed()
```

#### `ResponseError` — `type="response.error"` {#responseerror-typeresponseerror}

An error occurred during the stream.

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

---

## 8. Runner — Text Helpers {#runner-text-helpers}

These helpers filter the event stream to just text — useful when you only need the response text and don't care about tool calls, usage, or status events.

### `runner.stream_text(agent, prompt, **options)` {#runner-stream-text}

An async generator that yields raw string deltas. Raises `SafetyReplaceError` when the server replaces the output (safety filter). Raises `MatildaStreamError` on stream errors.

```python
from matilda_agent_sdk import Runner, SafetyReplaceError

runner = Runner()


async def main():
    try:
        async for chunk in runner.stream_text(
            {"name": "poet", "instructions": "Write a haiku."},
            "Write about the ocean.",
        ):
            print(chunk, end="", flush=True)
    except SafetyReplaceError as err:
        print(f"\nSafety replace: {', '.join(err.categories)}")
```

> **Why raise on safety replace?** The original text has already been yielded to the consumer by the time the replace event arrives. Raising forces the consumer to handle the replacement explicitly — silently dropping it would lose the replacement message.

### `runner.run_text(agent, prompt, **options)` {#runner-run-text}

Non-streaming convenience that returns just the final output text. Raises `SafetyReplaceError` on safety replacement, and propagates `MatildaAgentStreamError` if the stream produced error events.

```python
text = await runner.run_text(
    {"name": "helper"},
    "What is 2 + 2?",
)
print(text)  # "4"
```

### `SafetyReplaceError` {#safetyreplaceerror}

```python
class SafetyReplaceError(MatildaError):
    categories: list[str]
    # str(err) = replacement content (or empty string)
```

### Convenience functions {#convenience-functions}

The SDK exports shared-default-runner-backed convenience functions so you don't need to instantiate a `Runner` for simple use cases:

```python
from matilda_agent_sdk import run, run_text, stream, stream_text

# These are equivalent to default_runner.run(), default_runner.stream(), etc.
result = await run(agent, prompt, ...)        # AgentRunResult
text = await run_text(agent, prompt, ...)     # str

async for event in stream(agent, prompt, ...):
    ...                                       # AgentEvent

async for chunk in stream_text(agent, prompt, ...):
    ...                                       # str
```

These use the shared default client (installed via `configure_default_client()`, or created with defaults on first use). For isolated config or auth, instantiate your own `Runner`.

---

## 9. Runner — Structured Output {#runner-structured-output}

Structured output constrains the agent's response to a JSON Schema, server-side (grammar-constrained decoding), and validates it client-side against your pydantic model. Pass a pydantic model class (or JSON-schema `dict`), receive a validated object — no prompt engineering, no brittle JSON extraction.

### `runner.stream_object(agent, prompt, schema, **options)` {#runner-stream-object}

Streams exactly like `runner.stream()` — you receive every `AgentEvent` (including tool-loop and usage events) — plus one final event with the parsed, schema-validated object. Options are the run options.

```python
from pydantic import BaseModel


class Issue(BaseModel):
    severity: str  # 'low' | 'medium' | 'high'
    description: str


class Review(BaseModel):
    summary: str
    issues: list[Issue]


async def main():
    reviewer = Agent(name="reviewer", instructions="Review the code the user provides.")

    async for event in runner.stream_object(reviewer, "Review this function: ...", Review):
        if event.type == "response.output_text.delta":
            print(event.delta, end="", flush=True)
        if event.type == "object":
            review = event.object  # a validated Review instance
            print(f"\nValidated: {review.summary}")
```

The final event:

```python
@dataclass(frozen=True)
class ObjectEvent[T]:
    object: T  # your pydantic model
    # type = "object"
```

### `runner.run_object(agent, prompt, schema, **options)` {#runner-run-object}

Non-streaming convenience. Like `runner.run()`, it honours `callbacks`, `throw_on_stream_error`, retries, and the tool loop — and returns an `AgentObjectResult`: the full `AgentRunResult` plus the validated `object`.

```python
from pydantic import BaseModel


class Invoice(BaseModel):
    total: float
    currency: str


async def main():
    result = await runner.run_object(
        extractor,
        "Invoice total $1,250.00 AUD due 30 Sep.",
        Invoice,
    )

    print(result.object.total)     # 1250.0 (float)
    print(result.object.currency)  # "AUD" (str)
    print(result.final_output)     # raw JSON text as returned
    print(result.usage)            # token usage, as usual
```

#### `AgentObjectResult` {#agentobjectresult}

Extends `AgentRunResult` with one additional field:

| Field | Type | Description |
|---|---|---|
| `object` | `T \| None` | The response text parsed as JSON and validated against your schema. |

### Convenience functions {#convenience-functions-2}

Shared-default-runner-backed, like the other top-level helpers:

```python
from matilda_agent_sdk import run_object, stream_object

result = await run_object(agent, prompt, schema, ...)

async for event in stream_object(agent, prompt, schema, ...):
    ...
```

### Raw JSON Schema via `response_schema` {#raw-json-schema-via-response-schema}

The run options (and therefore `Session` turns) accept a stringified JSON Schema directly on any run or stream:

```python
import json

result = await runner.run(
    agent,
    "List three Australian birds.",
    response_schema=json.dumps({
        "type": "object",
        "properties": {"birds": {"type": "array", "items": {"type": "string"}}},
        "required": ["birds"],
        "additionalProperties": False,
    }),
)
data = json.loads(result.final_output)  # guaranteed valid, schema-conforming JSON
```

With `response_schema` set, `final_output` is guaranteed to be valid JSON conforming to the schema — but parsing and validation are up to you.

> **Safety replace and structured output.** Like `run_text()` / `stream_text()`, the object helpers raise `SafetyReplaceError` when the server replaces the output mid-stream — `str(err)` holds the replacement text and `.categories` the triggering categories. Token deltas already yielded to your consumer are not rolled back; `run_object()` is unaffected at the value level, since it raises before returning a result.

> **Truncation throws.** If the stream is truncated before the JSON completes, both helpers raise `MatildaObjectParseError` with the partial text in `.raw`. See [§17. Error Handling](#s-error-handling).

> **Stream errors throw.** If the server emits an `error` event mid-stream, `stream_object()` raises `MatildaStreamError` with the server's error code and message, and `run_object()` raises `MatildaAgentStreamError` with the partial `result` attached — matching the behaviour of the text helpers.

---

## 10. Client Tools {#client-tools}

Client tools are handlers you register that the agent can invoke mid-turn. The SDK handles the entire roundtrip loop: detecting tool calls, executing your handler, feeding the result back to the agent, and repeating until the agent stops calling tools or the roundtrip limit is reached.

### Tool execution loop {#tool-execution-loop}

```text
┌─────────────────────────────────────────────────────────┐
│  Turn 0                                                 │
│  1. Send messages + client_tools to server              │
│  2. Stream events — agent responds, may call tools      │
│  3. If tool calls detected:                             │
│     a. client.tool.requested → client.tool.executing    │
│     b. Execute handler from tool_handlers               │
│     c. client.tool.result                               │
│     d. Append result to messages as a user message      │
│     e. client.tool.roundtrip (turn + 1 / max_turns)     │
│     f. Go to Turn 1                                     │
│  4. If no tool calls: run is done                       │
│  5. If turn >= max_tool_roundtrips: truncated           │
└─────────────────────────────────────────────────────────┘
```

### Declaring and handling tools {#declaring-and-handling-tools}

```python
import asyncio
import json

from matilda_agent_sdk import ToolResult, stream

# 1. Declare the tools so the server knows they exist
client_tools = [
    {"name": "get_weather", "description": "Get current weather for a city", "parameters": {"type": "object"}},
    {"name": "calculate", "description": "Evaluate a math expression", "parameters": {"type": "object"}},
]


# 2. Register handlers — the SDK calls these when the agent invokes a tool
async def get_weather(args, ctx):
    city = args.get("city", "unknown")
    return ToolResult(content=json.dumps({"city": city, "temp": 22, "condition": "sunny"}))


async def calculate(args, ctx):
    try:
        # NOTE: eval is a stand-in for a real expression parser — never eval
        # untrusted input in production.
        result = eval(str(args["expression"]), {"__builtins__": {}}, {})
        return ToolResult(content=str(result))
    except Exception:
        return ToolResult(content="Invalid expression", is_error=True)


tool_handlers = {"get_weather": get_weather, "calculate": calculate}


# 3. Pass both to stream() or run()
async def main():
    async for event in stream(
        {"name": "assistant", "instructions": "Use the available tools to answer questions."},
        "What is the weather in Sydney, and what is 15 * 23?",
        tool_handlers=tool_handlers,
        client_tools=client_tools,
    ):
        if event.type == "client.tool.requested":
            print(f"→ Agent requested: {event.name}({json.dumps(event.args)})")
        if event.type == "client.tool.executing":
            print(f"⚙ Executing: {event.name}")
        if event.type == "client.tool.result":
            suffix = " (error)" if event.is_error else ""
            print(f"← Result: {event.result}{suffix}")
        if event.type == "client.tool.roundtrip":
            print(f"  Roundtrip {event.turn}/{event.max_turns}")
        if event.type == "response.output_text.delta":
            print(event.delta, end="", flush=True)


asyncio.run(main())
```

### `ToolHandler` {#toolhandler}

```python
ToolHandler = Callable[[dict, ToolExecutionContext], Awaitable[ToolResult]]
```

### `ToolExecutionContext` {#toolexecutioncontext}

```python
@dataclass(frozen=True)
class ToolExecutionContext:
    tool_call_id: str | None = None
```

### `ToolResult` {#toolresult}

```python
@dataclass(frozen=True)
class ToolResult:
    content: str
    is_error: bool = False
```

### `ToolHandlers` {#toolhandlers}

```python
ToolHandlers = dict[str, ToolHandler]
```

### `max_tool_roundtrips` {#max-tool-roundtrips}

Controls how many back-and-forth tool cycles the SDK allows before stopping. Default is `25` (`DEFAULT_MAX_TOOL_ROUNDTRIPS`). Lower it to prevent infinite loops or control cost.

```python
result = await runner.run(
    {"name": "tool-heavy-agent", "instructions": "Use tools to gather information."},
    "Do a task that needs tools",
    max_tool_roundtrips=5,
    tool_handlers={"search": lambda args, ctx: search_handler(args)},
    client_tools=[{"name": "search", "description": "Search", "parameters": {"type": "object"}}],
)

roundtrips = [e for e in result.events if e.type == "client.tool.roundtrip"]
print(f"Roundtrips used: {len(roundtrips)} (max was 5)")
```

### Advertised-tool guard {#advertised-tool-guard}

The SDK enforces that the agent can only call tools that were advertised for the current turn. If the model calls a tool that wasn't in `client_tools`, the SDK returns an error result instead of executing a handler:

```text
Tool not offered this turn: <name>
```

This prevents a model from talking the runner into invoking a handler that was never offered — tool output is untrusted input.

### DSML tool-call interception {#dsml-tool-call-interception}

Some models emit tool calls as text tokens wrapped in DSML markup (`<｜DSML｜tool_call>{...}</｜DSML｜tool_call>`) instead of using native function calling. The SDK automatically intercepts these text tokens, parses the JSON payload, and surfaces them as native `ClientToolRequested` events — the consumer never sees the raw markup.

This interception is fully automatic and applies to all streaming paths (the tag constants are exported as `DSML_TOOL_CALL_OPEN` / `DSML_TOOL_CALL_CLOSE`).

### Human-in-the-loop {#human-in-the-loop}

The tool execution loop makes human-in-the-loop trivial: a tool handler is just an async callable, so it can block on stdin, a UI prompt, or any other input source.

```python
import asyncio

from matilda_agent_sdk import ToolResult


async def ask_user(args, ctx):
    answer = await asyncio.to_thread(input, f"\n  Agent asks: {args['question']}\n  > ")
    return ToolResult(content=answer.strip())


tool_handlers = {"ask_user": ask_user}
client_tools = [
    {"name": "ask_user", "description": "Ask the user a question", "parameters": {"type": "object"}},
]

result = await runner.run(
    {"name": "clarifier", "instructions": "Ask the user for clarification when needed."},
    "Help me plan a trip",
    tool_handlers=tool_handlers,
    client_tools=client_tools,
)
```

---

## 11. Session (Multi-Turn) {#session-multi-turn}

A `Session` wraps an `Agent` with auto-managed `conversation_id` and accumulates turn results. Agent runs send `persist: false`, so the server does not retain turn content — instead, the session accumulates each turn's exchange **client-side** and replays the transcript with every subsequent turn, giving the model full conversation context.

### `create_session(agent, **options)` {#create-session}

```python
import asyncio

from matilda_agent_sdk import create_session


async def main():
    session = create_session(
        {"name": "tutor", "instructions": "You are a patient programming tutor. Explain concepts simply."}
    )

    print("Conversation ID:", session.conversation_id)

    # Turn 1
    r1 = await session.run("What is a closure in Python?")
    print("Turn 1:", r1.final_output[:100], "...")

    # Turn 2 — the session replays the accumulated transcript, so the model
    # sees the previous exchange
    r2 = await session.run("Can you show me a simple example?")
    print("Turn 2:", r2.final_output[:100], "...")

    # The session accumulates all turn results
    print("Total turns:", len(session.turns))
    print("Last turn stream ID:", session.last_turn.stream_id if session.last_turn else None)


asyncio.run(main())
```

### `Session` class {#session-class}

#### `session.run(prompt, **options)` {#sessionrunprompt-options}

Runs a turn and accumulates the result in `session.turns`. Options are the `run()` options (minus `conversation_id`); per-turn options merge with session defaults. Returns `AgentRunResult`.

#### `session.stream(prompt, **options)` {#sessionstreamprompt-options}

Streams a turn, yielding `AgentEvent` as they arrive. The result is accumulated in `session.turns` when the stream completes.

```python
async for event in session.stream("Explain decorators in one paragraph."):
    if event.type == "response.output_text.delta":
        print(event.delta, end="", flush=True)

print("\n[Turns accumulated]:", len(session.turns))
print("[Final output cached]:", (session.last_turn.final_output[:60] + "...") if session.last_turn else "")
```

#### `session.conversation_id` {#session-conversation-id}

The auto-generated (or provided) conversation ID. Reused across all turns.

#### `session.turns` {#sessionturns}

A `list[AgentRunResult]` — one per completed turn.

#### `session.last_turn` {#session-last-turn}

The most recent `AgentRunResult`, or `None` if no turns have run.

### Session options {#session-options}

| Option | Type | Default | Description |
|---|---|---|---|
| `runner` | `Runner \| None` | shared default | Custom runner instance. |
| `client` | `MatildaClient \| None` | — | Explicit client (wraps it in a fresh `Runner`). Ignored when `runner` is set. |
| `conversation_id` | `str \| None` | auto-generated | Explicit conversation ID. |
| `purpose` | `AgentPurpose \| None` | — | Fallback purpose when `agent` is a dict. |
| `response_mode` | `str \| None` | — | Session-level response mode override. |
| `metadata` | `dict \| None` | `{}` | Session-level metadata merged into every turn. |
| `tool_handlers` | `ToolHandlers \| None` | — | Session-level tool handlers. |
| `max_tool_roundtrips` | `int` | `25` | Session-level roundtrip cap. |

### Metadata passthrough {#metadata-passthrough}

Metadata can be set at multiple levels: `Agent` construction, `Session` construction, or per-call. Per-call metadata merges with (and overrides) session defaults.

```python
from matilda_agent_sdk import create_session


def tutor_instructions(ctx):
    env = ctx["metadata"].get("env", "unknown")
    user = ctx["metadata"].get("user", "anonymous")
    return f"Environment: {env}. User: {user}."


async def main():
    session = create_session(
        {"name": "helper", "instructions": tutor_instructions},
        metadata={"env": "staging", "user": "demo-user"},
    )

    # Session-level metadata is used by default
    await session.run("Who am I?")

    # Per-call metadata overrides session defaults
    await session.run("Who am I now?", metadata={"user": "admin"})
    # → env=staging (from session), user=admin (overridden per-call)
```

### Custom runner {#custom-runner}

A `Session` can use a custom `Runner` for dependency injection in tests or isolated configuration:

```python
import asyncio

from matilda_agent_sdk import Runner, Session


async def main():
    my_runner = Runner()
    session = Session(
        {"name": "custom-runner-agent", "instructions": "Be brief."},
        runner=my_runner,
    )

    result = await session.run("What is 2 + 2?")


asyncio.run(main())
```

---

## 12. Stream Resume {#stream-resume}

Durable streaming lets a client disconnect mid-stream and resume from where it left off. The server buffers events, keyed by a `stream_id` advertised at stream start.

### `resume_agent_stream(stream_id, last_event_id=None, *, on_event=None, ...)` {#resume-agent-stream}

Resumes a previously detached stream by replaying buffered events from `last_event_id`. Returns the accumulated `AgentRunResult`.

```python
import asyncio

from matilda_agent_sdk import resume_agent_stream


async def main():
    result = await resume_agent_stream(
        saved_stream_id,
        saved_last_event_id,
        on_event=lambda event: (
            print(event.delta, end="", flush=True)
            if event.type == "response.output_text.delta"
            else None
        ),
    )
    print("Resumed output:", result.final_output)
```

### Parameters {#parameters-5}

| Parameter | Type | Description |
|---|---|---|
| `stream_id` | `str` | The stream ID from the `stream.started` event (or `result.stream_id`). |
| `last_event_id` | `str \| None` | The last cursor received (from the `response.cursor` event or `result.last_event_id`). Omit to replay from the start. |
| `on_event` | `Callable[[AgentEvent], None] \| None` | Event handler callback. |
| `client` | `MatildaClient \| None` | Explicit client. Defaults to the shared default. |
| `access_token` | `str \| None` | Override access token. |
| `fingerprint` | `str \| None` | Device fingerprint. |
| `stall_timeout_ms` | `int \| None` | SSE stall watchdog timeout (default `45_000`). |

Returns `AgentRunResult`.

### 401 auto-refresh {#auto-refresh-on-401}

If the resume request returns 401 and the client has a `get_token` provider, the SDK automatically refreshes the token and retries once.

### Full resume example {#full-resume-example}

```python
import asyncio

from matilda_agent_sdk import resume_agent_stream, stream


async def main():
    stream_id = None
    last_event_id = None
    received_text = ""

    # Start streaming — capture IDs for potential resume
    async for event in stream({"name": "resumable-agent"}, "Tell me a fact."):
        if event.type == "stream.started":
            stream_id = event.stream_id
        if event.type == "response.cursor":
            last_event_id = event.last_event_id
        if event.type == "response.output_text.delta":
            received_text += event.delta
            print(event.delta, end="", flush=True)

    print(f"\n[Stream completed — stream_id: {stream_id}, cursor: {last_event_id}]")

    # Later — resume from the last cursor if the stream was interrupted
    if stream_id:
        result = await resume_agent_stream(
            stream_id,
            last_event_id,
            on_event=lambda event: (
                print(event.delta, end="", flush=True)
                if event.type == "response.output_text.delta"
                else None
            ),
        )
        print(f"\n[Resumed — output: {result.final_output[:60]}...]")


asyncio.run(main())
```

---

## 13. Files {#files}

The `Runner` exposes a `files` resource (the client SDK's `FilesResource`) for uploading and retrieving files. Uploaded files can be attached to agent runs via `file_ids`.

### `runner.files.upload(file, ...)` {#runnerfilesuploadfile}

Uploads a single file by path. Files at or above the chunked threshold use the parallel multipart protocol; smaller files use single-shot upload.

```python
async def main():
    result = await runner.files.upload(
        "hello.txt",
        content_type="text/plain",
        on_progress=lambda pct: print(f"Upload: {pct}%"),
    )
    print(f"File ID: {result.file_id}, Status: {result.status}")
```

#### Parameters {#parameters-6}

| Parameter | Type | Default | Description |
|---|---|---|---|
| `file` | `str \| Path` | **(required)** | Path to the file to upload. |
| `content_type` | `str` | `'application/octet-stream'` | MIME type. |
| `filename` | `str \| None` | path basename | Override the reported filename. |
| `on_progress` | `Callable[[int], None] \| None` | — | Progress callback (0–100). |
| `ephemeral` | `bool` | `False` | Upload for an incognito chat: the server deletes the file 24 hours after upload, and only incognito turns may attach it. |

Returns `FileComplete`:

```python
@dataclass
class FileComplete:
    file_id: str
    status: str  # 'pending' | 'scanning' | 'processing' | 'ready' | 'failed' | 'rejected'
```

### `runner.files.upload_many(files, *, file_concurrency=3, ...)` {#runner-files-upload-many}

Uploads multiple files in parallel. One file's failure does not abort the others — per-file outcomes arrive in the result list as `FileComplete` or `UploadError`.

```python
from matilda_client import UploadError


async def main():
    results = await runner.files.upload_many(["doc1.txt", "doc2.txt"])
    for i, result in enumerate(results):
        if isinstance(result, UploadError):
            print(f"File {i} failed: {result}")
        else:
            print(f"File {i}: {result.file_id} ({result.status})")
```

Returns `list[FileComplete | UploadError]`.

### `runner.files.retrieve(file_id)` {#runner-files-retrieve}

Retrieves metadata for a previously uploaded file.

```python
file = await runner.files.retrieve("file-abc123")
print(f"{file['filename']} — {file['status']} ({file['sizeBytes']} bytes)")
```

Returns a `dict` (a `FileAttachment`: `id`, `filename`, `contentType`, `sizeBytes`, `status`, optional `extractedText`, `thumbnailUrl`, `localUri`, `failureReason`, `createdAt`).

### Using files in agent runs {#using-files-in-agent-runs}

Upload a file, then reference its `file_id` in an agent run:

```python
async def main():
    upload = await runner.files.upload("report.txt", content_type="text/plain")

    result = await runner.run(
        {"name": "analyst", "purpose": "analysis", "instructions": "Summarise the report."},
        "What are the key findings?",
        file_ids=[upload.file_id],
    )
    print(result.final_output)
```

---

## 14. Conversations {#conversations}

The `Runner` exposes a `conversations` resource for listing, retrieving, renaming, and providing feedback on conversations. Resources return plain `dict` objects using the server's camelCase wire keys.

> **Note:** Agent runs send `persist: false` by default, so they do not appear in the Matilda web app's chat history. The conversations resource accesses conversations created by other clients (e.g. the web app). If you need agent runs to appear in chat history, you would need to override the `persist` flag — but this is not exposed as a public option in the agent SDK.

### `runner.conversations.list(...)` {#runnerconversationslist}

Lists conversations with pagination.

```python
async def main():
    result = await runner.conversations.list(limit=20, offset=0)
    for conv in result["conversations"]:
        print(f"{conv['id']}: {conv['title']} (updated {conv['updatedAt']})")
```

| Parameter | Type | Description |
|---|---|---|
| `limit` | `int \| None` | Maximum number of conversations to return. |
| `offset` | `int \| None` | Pagination offset. |

Returns a `dict` (`ConversationListResponse`: `conversations`, `total`, `limit`, `offset`).

### `runner.conversations.retrieve(conversation_id)` {#runner-conversations-retrieve}

Retrieves a full conversation thread with all messages.

```python
conv = await runner.conversations.retrieve("conv-123")
for msg in conv["messages"]:
    print(f"[{msg['role']}] {msg['content']}")
```

Returns a `dict` — the summary shape plus a `messages` list.

### `runner.conversations.update(conversation_id, *, title=...)` {#runner-conversations-update}

Updates a conversation's metadata (currently only title).

```python
await runner.conversations.update("conv-123", title="My Chat About AI")
```

Returns `None`.

### `runner.conversations.set_message_feedback(conversation_id, message_id, feedback)` {#runner-conversations-set-message-feedback}

Sets thumbs-up or thumbs-down feedback on a specific message.

```python
await runner.conversations.set_message_feedback("conv-123", "msg-456", "positive")
```

| Parameter | Type | Description |
|---|---|---|
| `conversation_id` | `str` | The conversation containing the message. |
| `message_id` | `str` | The message to rate. |
| `feedback` | `'positive' \| 'negative'` | The feedback value. |

Returns a `dict` (e.g. `{"ok": True}`).

---

## 15. Feedback {#feedback}

### `runner.feedback.report_bug(...)` {#runner-feedback-report-bug}

Files a bug report from your application. Reports land as GitHub issues in a private feedback repository — titled `[SDK] <title>` and labelled `sdk-feedback` — so they route directly to the team maintaining the SDK.

The SDK automatically stamps the report with its package name (`matilda-agent-sdk`), package version, and the Python runtime version, so you usually only need `title` and `description`. Pass the optional fields to override or supply anything else.

```python
await runner.feedback.report_bug(
    title="runner.stream() hangs after a client tool handler raises",
    description="The run coroutine never finishes when a tool handler raises asynchronously.",
    reproduction="Register a tool whose handler raises after an await, then call runner.stream()",
)
```

#### Parameters {#parameters-7}

| Parameter | Type | Description |
|---|---|---|
| `title` | `str` | One-line summary of the bug (required, max 200 chars). |
| `description` | `str` | What went wrong (required, max 4000 chars). |
| `reproduction` | `str \| None` | Optional steps to reproduce (max 4000 chars). |
| `package` | `str \| None` | Reporting package name — defaults to this SDK's package name. |
| `package_version` | `str \| None` | Reporting package version — defaults to this SDK's version. |
| `platform` | `'web' \| 'ios' \| 'android' \| 'unknown' \| None` | Client platform. |
| `app_version` | `str \| None` | Your application's version string. |
| `idempotency_key` | `str \| None` | Optional key (max 128 chars) so retries don't create duplicate issues. |

Returns a `dict` (e.g. `{"feedbackId": ..., "acknowledgedAt": ...}`).

---

## 17. Error Handling {#error-handling}

### `MatildaAgentRunError` {#matildaagentrunerror}

Raised on HTTP-level failures (non-2xx response from the server).

```python
class MatildaAgentRunError(MatildaError):
    status: int         # HTTP status code
    response_text: str  # Raw response body
```

### `MatildaAgentStreamError` {#matildaagentstreamerror}

Raised by `run()` when the stream emits an error event (unless `throw_on_stream_error=False`). Carries the full `AgentRunResult` with whatever was collected before the error.

```python
class MatildaAgentStreamError(MatildaError):
    code: str                    # ChatErrorCode of the first error ('unknown' when empty)
    errors: list[StreamErrorDetail]
    result: AgentRunResult
```

```python
from matilda_agent_sdk import MatildaAgentRunError, MatildaAgentStreamError

try:
    await runner.run(agent, prompt)
except MatildaAgentStreamError as err:
    print("Stream error code:", err.code)
    print("Partial output before error:", err.result.final_output)
    print("All errors:", err.errors)
except MatildaAgentRunError as err:
    print("HTTP error status:", err.status)
    print("Response body:", err.response_text)
```

### `SafetyReplaceError` {#safetyreplaceerror-2}

Raised by `stream_text()`, `run_text()`, `stream_object()`, and `run_object()` when the server replaces the output via a safety filter. `str(err)` contains the replacement text (or empty string), and `categories` lists the safety categories.

```python
class SafetyReplaceError(MatildaError):
    categories: list[str]
```

### `MatildaObjectParseError` {#matildaobjectparseerror}

Raised by `stream_object()` / `run_object()` when the response cannot be parsed as JSON or fails pydantic validation — e.g. a truncated stream (see [§9](#s-runner-structured-output)). `raw` holds the full response text; `cause` is the underlying `json.loads` or pydantic error. Safety replacement does not surface here — it raises `SafetyReplaceError` first.

```python
class MatildaObjectParseError(MatildaError):
    raw: str
    cause: BaseException | str
```

### Client SDK errors (re-exported) {#client-sdk-errors-re-exported}

The client SDK's error types are re-exported unchanged: `MatildaError` (shared base), `MatildaAPIError` (`status`, `response_text`), `QuotaExceededError` (structured 429 quota attributes), `MatildaStreamError` (`code`), `MatildaContentTypeError` (non-SSE 2xx from a streaming endpoint), and `UploadError`. `AuthError` (`code`, `retryable`) is importable from `matilda_client`.

### Chat error codes (`ChatErrorCode`) {#chat-error-codes-chaterrorcode}

These codes are emitted via the `ResponseError` stream event and appear in `AgentRunResult.errors` (mirrored on `StreamErrorDetail.code`):

| Code | Description |
|---|---|
| `internal_error` | Server-side failure. |
| `upstream_unavailable` | The AI model is not responding. |
| `rate_limited` | Too many requests. |
| `content_blocked` | Safety filter blocked the content. |
| `stream_aborted` | The stream was interrupted before completion. |
| `deadline_exceeded` | The response did not finish before the deadline. |
| `context_too_large` | The conversation is too long for the model. |
| `invalid_request_error` | The request was not accepted. Sending it again unchanged fails the same way. |
| `url_read_incomplete` | A URL the server tried to read did not finish loading. |
| `request_budget_exceeded` | The request exceeded its server-side budget. |
| `stalled` | No SSE events for the configured stall window. |
| `unknown` | Unclassified error (old server without typed codes). |

### Stall watchdog {#stall-watchdog}

The streaming parser arms an idle-event watchdog. If no SSE event arrives for `stall_timeout_ms` milliseconds, the stream is considered dead and aborted with a `'stalled'` error.

- **Default:** `45_000` ms (`DEFAULT_AGENT_STALL_TIMEOUT_MS`)
- **Disable:** Pass `stall_timeout_ms=0` (not recommended)

### Retry behaviour {#retry-behaviour}

The SDK retries retryable errors within a single turn. Retries are controlled by `max_retries` (default: `0` — no retries; `DEFAULT_MAX_RETRIES`).

**Retryable errors:**
- HTTP 429 (rate limited), 408 (request timeout), 5xx (server errors)
- Network errors: `httpx.ConnectError`, `httpx.ConnectTimeout`, `httpx.ReadTimeout`, `httpx.WriteTimeout`, `httpx.PoolTimeout`, `httpx.RemoteProtocolError`
- SSE stall watchdog (`stalled` error code)
- `TimeoutError`

**Retry delay:** Exponential backoff with jitter (1 s base), capped at 30 seconds.

**Important:** Retries only happen when no text has been yielded to the consumer yet. Retrying past that point would replay text the caller already has.

When a retry occurs, a `TurnRetrying` event is emitted:

```python
async for event in runner.stream(agent, prompt, max_retries=3):
    if event.type == "turn.retrying":
        print(f"Retry {event.attempt}/{event.max_retries} in {event.delay_ms}ms: {event.error['message']}")
```

### Error handling example {#error-handling-example}

```python
import asyncio

from matilda_agent_sdk import (
    MatildaAgentRunError,
    MatildaAgentStreamError,
    Runner,
    SafetyReplaceError,
)


async def main():
    runner = Runner()

    try:
        text = await runner.run_text(agent, "Hello!")
        print(text)
    except MatildaAgentStreamError as err:
        print(f"Stream error: {err.code} — {err}")
        print("Partial output:", err.result.final_output)
    except MatildaAgentRunError as err:
        if err.status == 401:
            print("Session expired — re-authenticate.")
        elif err.status == 429:
            print("Rate limited — slow down.")
        else:
            print(f"API error {err.status}: {err.response_text}")
    except SafetyReplaceError as err:
        print(f"Safety filter: {', '.join(err.categories)}")


asyncio.run(main())
```

---

## 18. Multi-Agent Patterns {#multi-agent-patterns}

The SDK has no built-in orchestrator — multi-agent emerges from composition. The `Runner` is your execution primitive, and standard Python patterns (chaining, `asyncio.gather`, tool-based delegation) build the architecture.

### Pattern 1: Sequential pipeline {#pattern-1-sequential-pipeline}

Chain `run()` calls, feeding each agent's output to the next. Each stage has a single responsibility.

```python
import asyncio

from matilda_agent_sdk import Agent, Runner


async def main():
    runner = Runner()

    researcher = Agent(
        name="researcher",
        purpose="analysis",
        instructions="Produce a structured list of key facts for a blog post. Bullet points only.",
    )

    writer = Agent(
        name="writer",
        purpose="general",
        instructions="Given research notes, write an engaging blog post draft under 400 words.",
    )

    editor = Agent(
        name="editor",
        purpose="general",
        instructions="Polish the draft for clarity, grammar, and flow. Return the full revised post.",
    )

    topic = "Why developers are adopting AI coding assistants"

    # Stage 1 → 2 → 3
    research = await runner.run(researcher, f"Research this topic: {topic}")
    draft = await runner.run(writer, research.final_output)
    edited = await runner.run(editor, draft.final_output)

    print(edited.final_output)

    # Total token usage across the pipeline
    total_tokens = sum(
        r.usage.output_tokens for r in (research, draft, edited) if r.usage
    )
    print(f"Total output tokens: {total_tokens}")


asyncio.run(main())
```

### Pattern 2: Parallel fan-out / fan-in {#pattern-2-parallel-fan-out-fan-in}

Run multiple specialist agents concurrently with `asyncio.gather`, then feed their outputs to a synthesiser.

````python
import asyncio

from matilda_agent_sdk import Agent, Runner


async def main():
    runner = Runner()

    security_reviewer = Agent(
        name="security-reviewer",
        purpose="analysis",
        instructions="Review code for vulnerabilities. Report only security issues.",
    )

    performance_reviewer = Agent(
        name="performance-reviewer",
        purpose="analysis",
        instructions="Review code for efficiency. Report only performance issues.",
    )

    synthesiser = Agent(
        name="synthesiser",
        purpose="analysis",
        instructions="Given reviews from multiple reviewers, produce a prioritised action list.",
    )

    code = "def get_user_data(user_id, db): ..."

    # Fan-out: two reviewers analyse concurrently
    security, performance = await asyncio.gather(
        runner.run(security_reviewer, f"Review this code:\n```python\n{code}\n```"),
        runner.run(performance_reviewer, f"Review this code:\n```python\n{code}\n```"),
    )

    # Fan-in: synthesiser merges the reviews
    combined_input = "\n".join(
        [
            "## Security Review", security.final_output,
            "## Performance Review", performance.final_output,
        ]
    )

    synthesis = await runner.run(synthesiser, combined_input)
    print(synthesis.final_output)


asyncio.run(main())
````

### Pattern 3: Router / delegator {#pattern-3-router-delegator}

A triage agent receives queries and decides which specialist to invoke. Each specialist is exposed as a client tool — when the agent calls a tool, the SDK handler runs the specialist agent via `run()` and returns its output.

```python
import asyncio

from matilda_agent_sdk import Agent, Runner, ToolResult

runner = Runner()

billing_specialist = Agent(
    name="billing-specialist",
    instructions="You are a billing support specialist.",
)

technical_specialist = Agent(
    name="technical-specialist",
    instructions="You are a technical support specialist. Include code examples when relevant.",
)

triage_agent = Agent(
    name="triage",
    purpose="general",
    instructions="You are a customer support triage specialist.",
)

triage_tools = [
    {
        "name": "ask_billing_specialist",
        "description": "Route billing questions to the billing specialist.",
        "parameters": {"type": "object", "properties": {"question": {"type": "string"}}, "required": ["question"]},
    },
    {
        "name": "ask_technical_specialist",
        "description": "Route technical questions to the technical specialist.",
        "parameters": {"type": "object", "properties": {"question": {"type": "string"}}, "required": ["question"]},
    },
]


async def ask_billing_specialist(args, ctx):
    result = await runner.run(billing_specialist, f"Answer: {args['question']}")
    return ToolResult(content=result.final_output)


async def ask_technical_specialist(args, ctx):
    result = await runner.run(technical_specialist, f"Answer: {args['question']}")
    return ToolResult(content=result.final_output)


tool_handlers = {
    "ask_billing_specialist": ask_billing_specialist,
    "ask_technical_specialist": ask_technical_specialist,
}


async def main():
    triage_prompt = "\n".join(
        [
            "A customer asked:",
            '"I\'m getting a 401 Unauthorized error when calling the /api/chat endpoint."',
            "You MUST forward this to a specialist by calling a tool.",
            "After the specialist responds, relay their answer.",
        ]
    )

    await runner.run(
        triage_agent,
        triage_prompt,
        tool_handlers=tool_handlers,
        client_tools=triage_tools,
        max_tool_roundtrips=6,
        callbacks={
            "on_tool_call": lambda name, args: print(f"Triage chose: {name}"),
            "on_tool_result": lambda name, result, is_error: print("Specialist responded."),
            "on_token": lambda delta: print(delta, end="", flush=True),
        },
    )


asyncio.run(main())
```

> **Tip:** Keep the triage agent's instructions short — put the routing rules in the task prompt. If routing rules are in the `instructions` field, the server's Auto-mode system prompt may interpret them as a prompt-injection attempt rather than operating instructions.

---

## 19. Suggested Tasks / Recipes {#suggested-tasks-recipes}

### Recipe 1: CLI agent with device-flow auth and streaming {#recipe-1-cli-agent-with-device-flow-auth-and-streaming}

A complete interactive CLI agent with device-flow auth, streaming, and multi-turn sessions.

```python
import asyncio
from pathlib import Path

from matilda_agent_sdk import MatildaClient, Runner, create_session
from matilda_client import create_file_token_store


async def main():
    token_store = create_file_token_store(Path.home() / ".matilda" / "tokens.json")

    runner = Runner(MatildaClient(base_url="https://matilda.com/api"))

    # Try to restore persisted tokens, fall back to interactive login
    restored = await runner.auth.restore(
        client_id="matilda-code",
        token_store=token_store.store,
        token_lock=token_store.lock,
    )
    if not restored:
        print("Starting device flow authentication...")
        await runner.auth.login_with_device_flow(
            client_id="matilda-code",
            token_store=token_store.store,
            token_lock=token_store.lock,
        )
        print("Authenticated!")

    session = create_session(
        {"name": "cli-assistant", "purpose": "general", "instructions": "Be helpful, concise, and friendly."},
        runner=runner,
    )

    while True:
        user_input = await asyncio.to_thread(input, "\nYou: ")
        if not user_input.strip() or user_input.lower() == "exit":
            break

        print("Agent: ", end="", flush=True)
        async for event in session.stream(user_input):
            if event.type == "response.output_text.delta":
                print(event.delta, end="", flush=True)
        print()


asyncio.run(main())
```

### Recipe 2: Client tools (weather + calculator) {#recipe-2-client-tools-weather-calculator}

An agent that uses client tools to answer questions requiring external data.

```python
import asyncio
import json

from matilda_agent_sdk import MatildaClient, Runner, ToolResult


async def get_weather(args, ctx):
    city = args.get("city", "unknown")
    # In reality, call a weather API
    return ToolResult(content=json.dumps({"city": city, "temp": 22, "condition": "sunny"}))


async def calculate(args, ctx):
    try:
        # NOTE: eval is a stand-in for a real expression parser — never eval
        # untrusted input in production.
        result = eval(str(args["expression"]), {"__builtins__": {}}, {})
        return ToolResult(content=str(result))
    except Exception:
        return ToolResult(content="Invalid expression", is_error=True)


client_tools = [
    {"name": "get_weather", "description": "Get current weather for a city", "parameters": {"type": "object"}},
    {"name": "calculate", "description": "Evaluate a math expression", "parameters": {"type": "object"}},
]

tool_handlers = {"get_weather": get_weather, "calculate": calculate}


async def main():
    runner = Runner(MatildaClient(base_url="https://matilda.com/api"))

    if not await runner.auth.get_tokens():
        await runner.auth.login_with_device_flow(client_id="matilda-code")

    async for event in runner.stream(
        {"name": "assistant", "instructions": "Use the available tools to answer."},
        "What is the weather in Sydney, and what is 15 * 23?",
        tool_handlers=tool_handlers,
        client_tools=client_tools,
    ):
        if event.type == "client.tool.requested":
            print(f"→ {event.name}({json.dumps(event.args)})")
        if event.type == "client.tool.result":
            print(f"← {event.result}")
        if event.type == "response.output_text.delta":
            print(event.delta, end="", flush=True)


asyncio.run(main())
```

### Recipe 3: Multi-agent code review pipeline {#recipe-3-multi-agent-code-review-pipeline}

Sequential pipeline: security review → performance review → synthesis.

````python
import asyncio

from matilda_agent_sdk import Agent, MatildaClient, Runner


async def main():
    runner = Runner(MatildaClient(base_url="https://matilda.com/api"))

    if not await runner.auth.get_tokens():
        await runner.auth.login_with_device_flow(client_id="matilda-code")

    security = Agent(
        name="security",
        purpose="analysis",
        instructions="Review for vulnerabilities. Be specific.",
    )

    performance = Agent(
        name="performance",
        purpose="analysis",
        instructions="Review for efficiency. Be specific.",
    )

    synthesiser = Agent(
        name="synthesiser",
        purpose="analysis",
        instructions="Merge reviews into a prioritised action list. Use 🔴 🟡 🟢 priority.",
    )

    code = 'def get_user_data(user_id, db): query = "SELECT * FROM users WHERE id = " + user_id'

    sec, perf = await asyncio.gather(
        runner.run(security, f"Review:\n```python\n{code}\n```"),
        runner.run(performance, f"Review:\n```python\n{code}\n```"),
    )

    combined = f"## Security\n{sec.final_output}\n\n## Performance\n{perf.final_output}"
    result = await runner.run(synthesiser, combined)
    print(result.final_output)


asyncio.run(main())
````

### Recipe 4: Dynamic instructions with metadata {#recipe-4-dynamic-instructions-with-metadata}

An agent whose instructions adapt based on runtime metadata.

```python
import asyncio

from matilda_agent_sdk import Agent, run


def reviewer_instructions(ctx):
    lang = ctx["metadata"].get("language", "auto-detect")
    strictness = ctx["metadata"].get("strictness", "normal")
    return "\n".join(
        [
            "Review the following code.",
            f"Language: {lang}",
            f"Strictness: {strictness}",
            "Focus on: correctness, security, and readability.",
            "Cite line numbers when possible.",
        ]
    )


agent = Agent(name="code-reviewer", purpose="code", instructions=reviewer_instructions)


async def main():
    result = await run(
        agent,
        "def add(a, b): return a + b",
        metadata={"language": "Python", "strictness": "strict"},
    )

    print(result.final_output)


asyncio.run(main())
```

### Recipe 5: Stream resume with disconnect recovery {#recipe-5-stream-resume-with-disconnect-recovery}

Start a stream, simulate a disconnect, and resume from the last cursor.

```python
import asyncio

from matilda_agent_sdk import resume_agent_stream, stream


async def main():
    stream_id = None
    last_event_id = None
    received_text = ""

    print("Starting stream...")
    try:
        async for event in stream({"name": "writer"}, "Write a very long essay about Australia."):
            if event.type == "stream.started":
                stream_id = event.stream_id
            if event.type == "response.cursor":
                last_event_id = event.last_event_id
            if event.type == "response.output_text.delta":
                received_text += event.delta
                # Simulate disconnect after 500 chars
                if len(received_text) > 500:
                    print("\n--- Simulated disconnect ---")
                    break
    except Exception as err:
        print(f"Disconnected: {err}")

    print(f"Received {len(received_text)} chars before disconnect.")

    # Resume from the last cursor
    if stream_id:
        print("\n--- Resuming ---")
        result = await resume_agent_stream(
            stream_id,
            last_event_id,
            on_event=lambda event: (
                print(event.delta, end="", flush=True)
                if event.type == "response.output_text.delta"
                else None
            ),
        )
        print(f"\nTotal output: {len(result.final_output)} chars")


asyncio.run(main())
```

### Recipe 6: Custom Runner with file token store {#recipe-6-custom-runner-with-file-token-store}

A standalone Runner with persistent auth for CLI or long-running service use.

```python
import asyncio
from pathlib import Path

from matilda_agent_sdk import MatildaClient, Runner
from matilda_client import create_file_token_store


async def main():
    token_path = Path.home() / ".matilda" / "tokens.json"
    token_store = create_file_token_store(token_path)

    runner = Runner(MatildaClient(base_url="https://matilda.com/api"))

    # Restore persisted tokens or login interactively
    restored = await runner.auth.restore(
        client_id="matilda-code",
        token_store=token_store.store,
        token_lock=token_store.lock,
    )
    if not restored:
        await runner.auth.login_with_device_flow(
            client_id="matilda-code",
            token_store=token_store.store,
            token_lock=token_store.lock,
        )

    # Runner is ready — tokens auto-refresh on 401
    result = await runner.run(
        {"name": "helper", "instructions": "Be concise."},
        "What is the capital of Australia?",
    )
    print(result.final_output)

    # Later: logout clears the token store
    # await runner.auth.logout()


asyncio.run(main())
```

---

## 20. Exports Reference {#exports-reference}

### Classes {#classes}

| Export | Description |
|---|---|
| `Agent` | Named persona with static or dynamic instructions. |
| `Runner` | Main execution class. Holds `auth`, `files`, `conversations`, `feedback` resources. |
| `Session` | Multi-turn conversation wrapper with auto-managed `conversation_id`. |
| `AgentAuth` | Managed auth: login, restore, token refresh, logout (via `runner.auth`). |
| `MatildaAgentRunError` | HTTP-level failure (`status`, `response_text`). |
| `MatildaAgentStreamError` | SSE stream error (`code`, `errors`, partial `result`). |
| `MatildaClient` | Underlying API client (re-exported from `matilda-client`). |
| `MatildaConfig` | Client config dataclass (re-exported). |
| `FilesResource` | File upload/retrieve resource (re-exported; also at `runner.files`). |
| `ConversationsResource` | Conversation list/retrieve/update/feedback resource (re-exported). |
| `FeedbackResource` | Bug-report resource, agent-stamped (`report_bug`; also at `runner.feedback`). |

### Errors (re-exported) {#errors-re-exported}

| Export | Description |
|---|---|
| `MatildaError` | Shared base class. |
| `MatildaAPIError` | Non-2xx HTTP response. |
| `QuotaExceededError` | 429 quota error with structured attributes. |
| `MatildaStreamError` | Terminal stream `error` event (`code`). |
| `MatildaContentTypeError` | Streaming endpoint answered 2xx with a non-SSE body (proxy/gateway misconfiguration). |
| `SafetyReplaceError` | Safety filter replacement (`categories`). |
| `MatildaObjectParseError` | Structured output parse/validation failure (`raw`, `cause`). |
| `UploadError` | File upload failure. |

### Functions {#functions}

| Export | Description |
|---|---|
| `run(agent, prompt, **options)` | Run an agent turn via the shared default runner. Returns `AgentRunResult`. |
| `stream(agent, prompt, **options)` | Stream agent events. Returns an async generator of `AgentEvent`. |
| `stream_text(agent, prompt, **options)` | Stream text deltas. Returns an async generator of `str`. |
| `run_text(agent, prompt, **options)` | Run and return just the text. Returns `str`. |
| `stream_object(agent, prompt, schema, **options)` | Stream with structured output. Async generator of `AgentObjectEvent`. |
| `run_object(agent, prompt, schema, **options)` | Run with structured output. Returns `AgentObjectResult`. |
| `create_session(agent, **options)` | Create a `Session`. |
| `build_agent_chat_request(**options)` | Build a chat request dict without executing. Async. |
| `resume_agent_stream(stream_id, last_event_id=None, **options)` | Resume a detached durable stream. Returns `AgentRunResult`. |
| `configure_default_client(client)` | Install the shared default `MatildaClient`. |

### Constants {#constants}

| Export | Value | Description |
|---|---|---|
| `DEFAULT_AGENT_PURPOSE` | `'code'` | Default agent purpose. |
| `DEFAULT_AGENT_STALL_TIMEOUT_MS` | `45_000` | Default SSE stall watchdog timeout. |
| `DEFAULT_MAX_TOOL_ROUNDTRIPS` | `25` | Default maximum tool roundtrips. |
| `DEFAULT_MAX_RETRIES` | `0` | Default maximum retries. |
| `AGENT_SDK_NAME` / `AGENT_SDK_LANGUAGE` | — | SDK provenance constants stamped on requests. |
| `DSML_TOOL_CALL_OPEN` / `DSML_TOOL_CALL_CLOSE` | — | DSML tool-call tag constants. |

### Types {#types}

| Export | Description |
|---|---|
| `AgentPurpose` | `Literal['code', 'analysis', 'general']` |
| `AgentInstructions` | `str \| Callable[[dict], str \| Awaitable[str]]` |
| `AgentEvent` | Union of the 21 agent stream event dataclasses. |
| `AgentObjectEvent` | `AgentEvent` plus the terminal `ObjectEvent`. |
| `ObjectEvent[T]` | Terminal structured-output event (`type="object"`; re-exported from `matilda-client`). |
| `AgentRunResult` | Result of an agent run (dataclass). |
| `AgentObjectResult[T]` | `AgentRunResult` plus the validated `object`. |
| `ToolHandler` | `(args, ctx) -> Awaitable[ToolResult]` |
| `ToolHandlers` | `dict[str, ToolHandler]` |
| `ToolExecutionContext` | Context passed to tool handlers (`tool_call_id`). |
| `ToolResult` | `{content, is_error}` dataclass. |
| `StreamErrorDetail` | Alias of `matilda_client.StreamErrorInfo` — `{code, message}` pair on `AgentRunResult.errors`. Shared class, so `isinstance()` holds across both packages. |
| `UsageSummary` | Alias of `matilda_client.Usage` — token usage summary on `AgentRunResult`. Shared class, so `isinstance()` holds across both packages. |
| `UsageEvent` | `response.usage` event (re-exported from `matilda-client`). |
| `EventCallback` / `ObjectEventCallback` | Catch-all event callback types. |
| `MatildaClientProtocol` | Protocol a `Runner` client's object must satisfy. |
| `RunStarted` / `StreamStarted` | Agent lifecycle events. |
| `ClientToolRequested` / `ClientToolExecuting` / `ClientToolResult` / `ClientToolRoundtrip` | Client-tool events. |
| `TurnRetrying` | Retry notice event. |
| `OutputTextDelta` / `OutputTextReplace` / `StatusEvent` / `QueuedEvent` / `ToolCallStarted` / `ToolCallProgress` / `ToolCallCompleted` / `GenerationStatus` / `CursorEvent` / `Truncated` / `Completed` / `ResponseError` | Chat events re-exported from `matilda-client` so consumers dispatch on one set of types. |

### Auth imports {#auth-imports}

Auth types (`AuthError`, `StorageAdapter`, `TokenManager`, `TokenSet`, `MemoryStorage` / `memory_storage`, `AuthServerMetadata`, login-flow functions, `create_file_token_store`, `DEFAULT_SUCCESS_REDIRECT`) live in `matilda_client` / `matilda_client.auth` / `matilda_client.auth_local` — import them from there. See the [client SDK exports](https://help.maincode.com/products/matilda/v1/en/sdks/python-client#s-exports-reference).

---

*Built and maintained by [Maincode](https://maincode.com) — Australia's sovereign AI company.*