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

• 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

• 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

1. Installation
2. Quick Start
3. Configuration
4. Authentication
5. Agent
6. Runner — Run (Non-Streaming)
7. Runner — Stream (Full Events)
8. Runner — Text Helpers
9. Runner — Structured Output
10. Client Tools
11. Session (Multi-Turn)
12. Stream Resume
13. Files
14. Conversations
15. Feedback
16. Error Handling
17. Multi-Agent Patterns
18. Suggested Tasks / Recipes
19. Exports Reference

1. Installation

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):

pip install pydantic

Import

from matilda_agent_sdk import Agent, Runner, run, stream

2. Quick Start

Minimal: run a single agent turn

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

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

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

configure_default_client(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.

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 for the full field list.

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.

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

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

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

Environment URLs

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.

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

4. 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

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

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):

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:

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(...)

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.

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

Parameters

Returns TokenSet.

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.

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

Returns TokenSet.

runner.auth.restore(...)

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_*.

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

Returns TokenSet | None.

runner.auth.get_tokens()

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

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

Returns TokenSet | None.

runner.auth.logout()

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

await runner.auth.logout()

Returns None.

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:

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

Imported from matilda_client:

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

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

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

AuthError class

Imported from matilda_client:

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

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

# 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

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

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

AgentPurpose

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

Dynamic instructions

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

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 = str | Callable[[dict], str | Awaitable[str]]

The context dict passed to the callable:

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

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:

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(...)

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

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

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

6. Runner — Run (Non-Streaming)

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

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

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

All options are keyword-only.

callbacks

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

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"),
    },
)

AgentRunResult

throw_on_stream_error=False

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:

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(agent, prompt, **options)

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

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

AgentEvent

A union of 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"

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

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

StreamStarted — type="stream.started"

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

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

OutputTextDelta — type="response.output_text.delta"

A text chunk from the assistant.

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

StatusEvent — type="response.status"

Stream lifecycle status change.

@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.

@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.

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

ToolCallProgress — type="response.tool_call.progress"

Progress update from a running server-side tool.

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

ToolCallCompleted — type="response.tool_call.completed"

A server-side tool invocation finished.

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

ClientToolRequested — type="client.tool.requested"

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

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

ClientToolExecuting — type="client.tool.executing"

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

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

ClientToolResult — type="client.tool.result"

A client tool handler returned a result.

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

ClientToolRoundtrip — type="client.tool.roundtrip"

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

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

TurnRetrying — type="turn.retrying"

A retryable error occurred and the turn is being retried.

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

GenerationStatus — type="response.generation_status"

Generation phase update.

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

OutputTextReplace — type="response.output_text.replace"

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

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

UsageEvent — type="response.usage"

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

@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 (event ID). Persist this to resume from this point.

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

Truncated — type="response.truncated"

The response was cut short.

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

Completed — type="response.completed"

The stream finished successfully.

Completed()

ResponseError — type="response.error"

An error occurred during the stream.

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

8. 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)

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

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)

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

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

SafetyReplaceError

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

Convenience functions

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

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

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)

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.

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:

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

runner.run_object(agent, prompt, schema, **options)

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.

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

Extends AgentRunResult with one additional field:

Convenience functions

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

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

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

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.

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

┌─────────────────────────────────────────────────────────┐
│  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

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 = Callable[[dict, ToolExecutionContext], Awaitable[ToolResult]]

ToolExecutionContext

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

ToolResult

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

ToolHandlers

ToolHandlers = dict[str, ToolHandler]

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.

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

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:

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

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

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.

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)

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)

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.run(prompt, **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)

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

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

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

session.turns

A list[AgentRunResult] — one per completed turn.

session.last_turn

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

Session options

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.

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

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

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

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, ...)

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

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

Returns AgentRunResult.

401 auto-refresh

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

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

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, ...)

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

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

Returns FileComplete:

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

runner.files.upload_many(files, *, file_concurrency=3, ...)

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.

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)

Retrieves metadata for a previously uploaded file.

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

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

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

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(...)

Lists conversations with pagination.

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']})")

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

runner.conversations.retrieve(conversation_id)

Retrieves a full conversation thread with all messages.

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=...)

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

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

Returns None.

runner.conversations.set_message_feedback(conversation_id, message_id, feedback)

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

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

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

15. Feedback

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.

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

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

17. Error Handling

MatildaAgentRunError

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

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

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.

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

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

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.

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

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). 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.

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

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)

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

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

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:

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

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

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

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

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

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

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

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.

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

Recipe 1: CLI agent with device-flow auth and streaming

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

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)

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

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

Sequential pipeline: security review → performance review → synthesis.

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

An agent whose instructions adapt based on runtime metadata.

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

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

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

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

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

Classes

Errors (re-exported)

Functions

Constants

Types

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.

Built and maintained by Maincode — Australia's sovereign AI company.