Matilda Python client SDK

Version 0.4.0

A small, self-contained public Python SDK for building Matilda clients. Async-first (built on httpx), fully typed, with typed stream events and zero runtime dependencies beyond httpx and the standard library. Requires Python ≥ 3.12.

The SDK follows the OpenAI client shape where it helps: constructor config, resource groups, request options, typed API errors, and async-iterable streaming. It does not expose model or provider selection — Matilda core owns routing, safety, resumable SSE, server-side tool execution, and policy.

What's included

• Chat — non-streaming, full-event streaming, text-only streaming, schema-constrained structured output, and durable stream resume
• Conversations — list, retrieve, rename, and set message feedback
• Files — upload (single and parallel), retrieve metadata
• Feedback — report harmful content and submit response feedback
• System One — typed state evaluation with the matilda-jev structured-decision model
• Auth — managed PKCE browser login, RFC 8628 device flow, token refresh, and persistent token storage
• API Keys — create, list, and revoke mc_live_ API keys via SDK methods or the matilda-key CLI

What's NOT included

• Local tool-execution loop — for client-side tool execution (client_tools, local approval/sandbox loops, tool-result continuation), use the Matilda agent SDK
• Session class — multi-turn conversations are managed via conversation_id (see §16. Multi-Turn Conversations)
• Devices — push-notification device management is not yet exposed on the Python client surface

Table of Contents

1. Installation
2. Quick Start
3. Configuration
4. Authentication
5. Chat — Non-Streaming
6. Chat — Streaming (Full Events)
7. Chat — Text-Only Helpers
8. Chat — Structured Output
9. Chat — Durable Streaming (Resume)
10. Conversations (History)
11. Files
12. Feedback
13. System One (matilda-jev)
14. Error Handling
15. Multi-Turn Conversations
16. Suggested Tasks / Recipes

1. Installation

pip install matilda-client==0.4.0
# or: uv add matilda-client

Requires Python ≥ 3.12.

pydantic is an optional dependency — install it alongside the SDK if you want schema-validated structured output (§8). The SDK detects pydantic models by duck-typing, so any version works:

pip install pydantic

Import

from matilda_client import MatildaClient

Auth modules

The standalone auth helpers live in two submodules — the transport-agnostic OAuth protocol core, and the local-machine adapters (loopback receiver, file token store, managed login flows):

from matilda_client.auth import (
    create_token_manager,
    fetch_auth_server_metadata,
    memory_storage,
)
from matilda_client.auth_local import (
    create_file_token_store,
    create_loopback_receiver,
    run_device_login_flow,
    run_loopback_login_flow,
)

2. Quick Start

Minimal: send a message

import asyncio
import os

from matilda_client import MatildaClient


async def main():
    async with MatildaClient(token=os.environ["MATILDA_ACCESS_TOKEN"]) as client:
        response = await client.chat.create(input="Summarize this thread.")
        print(response.output_text)


asyncio.run(main())

Note: The token option above is fine for quick testing, but for production use we recommend the managed auth flows (login_with_browser or login_with_device_flow) which auto-wire a TokenManager with automatic token refresh. See §4. Authentication.

Minimal: streaming

import os

from matilda_client import MatildaClient


async def main():
    async with MatildaClient(token=os.environ["MATILDA_ACCESS_TOKEN"]) as client:
        async for event in client.chat.stream(input="Write a short plan."):
            if event.type == "response.output_text.delta":
                print(event.delta, end="", flush=True)

Authenticated: device flow + chat

import asyncio

from matilda_client import MatildaClient


async def main():
    async with MatildaClient() as client:
        # Authenticate via RFC 8628 device flow — prints a code to stderr
        await client.auth.login_with_device_flow(client_id="matilda-code")

        # Token is now managed automatically — no manual header wiring
        response = await client.chat.create(input="Hello, Matilda!")
        print(response.output_text)


asyncio.run(main())

3. Configuration

MatildaClient

client = MatildaClient(
    token="",                     # static Bearer JWT
    # keyword-only below:
    base_url=DEFAULT_BASE_URL,    # https://matilda.com/api
    get_token=None,               # async token provider
    config=None,                  # or pass a pre-built MatildaConfig
)

MatildaConfig

A frozen dataclass — create a new client to change configuration.

Environment URLs

Instance isolation

Each MatildaClient instance holds its own independent config. Multiple instances in the same process are fully isolated — constructor options are scoped to that instance.

local = MatildaClient(base_url="http://localhost:3000/api")
prod = MatildaClient(base_url="https://matilda.com/api")

print(local.config.base_url)  # http://localhost:3000/api
print(prod.config.base_url)   # https://matilda.com/api

config (attribute)

Returns the current MatildaConfig.

cfg = client.config
print(cfg.base_url, cfg.token)

origin (config property)

The scheme+host of base_url — the origin where core-auth endpoints live. Auth flows resolve endpoints against config.origin, so base_url must be an absolute URL:

client.config.origin  # e.g. "https://matilda.com"

4. Authentication

The SDK provides a managed auth tier: on successful login, a TokenManager is auto-configured as the client's token provider. Every subsequent request automatically carries a managed access token with single-flight, skew-aware auto-refresh — and a 401 response triggers one force_refresh + retry before the failure surfaces.

OAuth client IDs

The following client IDs are registered in the Matilda OAuth client registry:

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.

async def main():
    async with MatildaClient() as client:
        tokens = await client.auth.login_with_browser(
            client_id="matilda-code",
            open_browser=lambda url: print(f"Open: {url}"),
        )
        print(tokens.access_token)

Parameters

Returns

TokenSet — the token set from the login flow. The TokenManager is also auto-configured on the client instance.

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.

async def main():
    async with MatildaClient() as client:
        tokens = await client.auth.login_with_device_flow(
            client_id="matilda-code",
            on_event=lambda e: (
                print(f"Visit {e.verification_uri} and enter code: {e.user_code}")
                if e.type == "user_code"
                else None
            ),
        )

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

auth.refresh_token(refresh_token, client_id)

Manually refresh an access token using a refresh token. This bypasses the TokenManager — use it only when you need raw token exchange.

tokens = await client.auth.refresh_token(old_refresh_token, "matilda-code")

Returns TokenSet.

auth.get_tokens()

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

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

Returns TokenSet | None.

auth.logout()

Clears the token store, destroys the TokenManager, and disconnects the client's token provider.

await client.auth.logout()

Returns None.

Token persistence

By default, tokens are stored in memory (MemoryStorage()). For cross-process persistence (e.g. CLI sessions), use create_file_token_store:

from pathlib import Path

from matilda_client import MatildaClient, create_file_token_store


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

    async with MatildaClient() as client:
        await client.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

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]: ...

Implement this (sync or async) to store tokens in a database, keychain, or any custom backend.

TokenManager

manager = TokenManager(
    token_endpoint="https://matilda.com/api/auth/oauth/token",
    client_id="matilda-code",
    store=...,
)

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

Also constructible via the create_token_manager(**kwargs) factory. The SDK auto-creates one on login. Refresh is single-flight per instance, evicts the session on invalid_grant, and re-reads the store inside the optional TokenLock so a process that lost a cross-process refresh race sees the winner's fresh token.

TokenSet

@dataclass(frozen=True)
class TokenSet:
    access_token: str
    expires_at: int              # epoch milliseconds
    refresh_token: str | None
    id_token: str | None         # OIDC id_token when 'openid' scope is granted

TokenSet.to_json_dict() / TokenSet.from_json_dict(...) serialise with camelCase keys — the on-disk JSON contract is shared with the TypeScript SDK, so token stores can be exchanged between the two.

AuthError class

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

Three frozen dataclasses, each 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

Standalone auth helpers

The matilda_client.auth module provides the raw PKCE and device-flow protocol helpers without the managed TokenManager. The managed client.auth.* methods cover most integrations, but these remain available for integrators who need direct protocol access:

The matilda_client.auth_local module adds the local-machine adapters on top of the protocol core:

API Key Management

The SDK provides methods for managing mc_live_ API keys. These hit the same JWT-protected endpoints the developer dashboard uses — the session JWT from a prior login_with_browser() or login_with_device_flow() call is carried automatically by the TokenManager.

api_keys.create(...)

Mints a new API key. The secret is returned only at creation time — store it immediately.

async def main():
    async with MatildaClient() as client:
        await client.auth.login_with_device_flow(client_id="matilda-code")

        key = await client.api_keys.create(
            name="ci-runner",
            # scopes=["api:code"],                  # omit → server default
            # expires_at="2026-12-31T23:59:59Z",    # omit → never expires
        )

        print(key["secret"])      # mc_live_...  — shown only once
        print(key["keyPrefix"])   # mc_live_abcd
        print(key["id"])          # UUID for revocation

Parameters

Returns a dict — the key metadata plus the one-time secret.

api_keys.list()

Lists all non-revoked API keys for the authenticated user. Secrets are never included — only the keyPrefix for identification.

keys = await client.api_keys.list()
for key in keys:
    status = "REVOKED" if key.get("revokedAt") else "ACTIVE"
    print(f"{key['keyPrefix']}  {key['name']}  {status}")

Returns list[dict].

api_keys.revoke(key_id)

Revokes a key by ID. The key immediately stops working for authentication.

await client.api_keys.revoke(key["id"])

Returns a dict (the revoked key with revokedAt set).

Response shape

API key dicts use the server's camelCase wire keys:

# ApiKey dict:
{
    "id": "8f3a2b1c-...",
    "name": "ci-runner",
    "keyPrefix": "mc_live_abcd1234",
    "scopes": ["api:code"],
    "createdAt": "2026-08-18T10:30:00.000Z",
    "lastUsedAt": None,
    "revokedAt": None,
    "expiresAt": None,
    # "secret": "mc_live_..."  — create() only, shown exactly once
}

CLI — matilda-key

The package ships a matilda-key CLI that mints API keys via device-flow login. This is the easiest way for social-login (Google/Apple) users to get an API key without writing code.

Installation

The console script is installed with the package (on the venv's PATH), or run without installing via uvx:

matilda-key create-api-key --name "my-key"
# or, without installing:
uvx --from matilda-client matilda-key create-api-key --name "my-key"

Usage

matilda-key create-api-key --name <key-name> [--scopes ...] [--expires-at ...] [--api-base-url ...] [--client-id ...]

The MATILDA_API_BASE_URL environment variable is also honoured as a fallback for --api-base-url.

What happens when you run it

1. Device-flow login — a verification URL and code are printed to stderr. Open the URL, sign in with Google or Apple, enter the code.
2. Key minting — once authenticated, an API key is created from the session JWT.
3. Output — key metadata (ID, name, prefix, scopes, expiry) is printed to stderr. The secret is printed to stdout.

The stdout/stderr separation is deliberate: the secret on stdout is clean and pipeable, while the login flow and metadata remain visible on the terminal via stderr.

Piping the secret

# Capture into an env var (login flow still visible on terminal):
MATILDA_API_KEY=$(matilda-key create-api-key --name "ci-runner")

# Pipe to a file:
matilda-key create-api-key --name "ci-runner" > /tmp/key.txt

# Use in CI:
export MATILDA_API_KEY="$(cat /tmp/key.txt)"

Full example

$ matilda-key create-api-key --name "ci-runner" --scopes api:code

Starting device-flow login...

  Open https://matilda.com/device and enter code: ABCD-1234

Login successful!

Minting API key "ci-runner"...

API key created successfully.
  ID:          8f3a2b1c-...
  Name:        ci-runner
  Prefix:      mc_live_abcd1234
  Scopes:      api:code
  Expires:     never
  Created at:  2026-08-18T10:30:00.000Z

Secret printed to stdout. Store it securely — it won't be shown again.

The secret (mc_live_...) is on stdout; everything else is on stderr.

5. Chat — Non-Streaming

chat.create(...)

Sends a chat message and returns the complete response. Internally this runs the stream and collects all events.

async def main():
    async with MatildaClient(token="...") as client:
        response = await client.chat.create(
            input="What is the capital of Australia?",
            conversation_id="conv-123",
            response_mode="instant",
        )

        print(response.output_text)
        print(response.usage)

Parameters

MatildaChatResponse

Response modes

• 'auto' — Server decides the optimal response depth.
• 'instant' — Optimised for low latency.
• 'deep' — Optimised for thoroughness.

6. Chat — Streaming (Full Events)

chat.stream(...)

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

async def main():
    async with MatildaClient(token="...") as client:
        async for event in client.chat.stream(input="Explain quantum computing."):
            if event.type == "response.created":
                print(f"Stream started: {event.stream_id}")
            elif event.type == "response.output_text.delta":
                print(event.delta, end="", flush=True)
            elif event.type == "response.tool_call.started":
                print(f"\nTool: {event.tool}")
            elif event.type == "response.usage":
                print(f"\nTokens: {event.usage.output_tokens}")
            elif event.type == "response.completed":
                print("\n--- Done ---")
            elif event.type == "response.error":
                print(f"Error: {event.code} — {event.message}")

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

ChatEvent

A union of 15 event dataclasses:

ResponseCreated — type="response.created"

Emitted once at stream start with the durable stream ID.

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

OutputTextDelta — type="response.output_text.delta"

A text chunk from the assistant.

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

OutputTextReplace — type="response.output_text.replace"

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

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

StatusEvent — type="response.status"

Stream lifecycle status change.

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

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

ToolCallCompleted — type="response.tool_call.completed"

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

GenerationStatus — type="response.generation_status"

Generation phase update.

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

UsageEvent — type="response.usage"

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

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

CursorEvent — type="response.cursor"

Durable stream cursor (Redis stream entry ID). Persist this to resume from this point.

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

ArtifactEvent — type="response.artifact"

The server generated a file (rendered PDF, chart image, spreadsheet) and published a download card for it. The payload only carries client-safe fields — never internal storage paths. chat.create() also collects these into MatildaChatResponse.artifacts.

@dataclass(frozen=True)
class Artifact:
    artifact_id: str
    url: str  # authed BFF download route, never a presigned URL
    filename: str | None = None
    content_type: str | None = None
    size_bytes: int | None = None


@dataclass(frozen=True)
class ArtifactEvent:
    artifact: Artifact
    # type = "response.artifact"

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"

7. Chat — Text-Only 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.

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

from matilda_client import MatildaClient, SafetyReplaceError


async def main():
    async with MatildaClient(token="...") as client:
        try:
            async for chunk in client.chat.stream_text(input="Write a haiku."):
                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.

chat.create_text(...)

Non-streaming convenience that returns just the final output text. Safety replace is handled naturally — the replacement text is returned. Raises MatildaStreamError if the stream produced any error events.

text = await client.chat.create_text(input="What is 2 + 2?")
print(text)  # "4"

SafetyReplaceError

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

8. Chat — Structured Output

Structured output constrains the model's response to a JSON Schema, server-side (grammar-constrained decoding), and then validates it client-side against your pydantic model. Pass a pydantic model class (or a plain JSON-schema dict), receive a validated object — no prompt engineering, no brittle JSON extraction. The SDK duck-types on model_json_schema, so any pydantic v2 version works and pydantic stays an optional dependency.

chat.stream_object(schema, *, input=...)

Streams exactly like chat.stream() — you receive every ChatEvent — plus one final ObjectEvent with the parsed, schema-validated object.

from pydantic import BaseModel


class Recipe(BaseModel):
    name: str
    prep_time_minutes: float
    ingredients: list[str]


async def main():
    async with MatildaClient(token="...") as client:
        async for event in client.chat.stream_object(
            Recipe,
            input="Give me a recipe for pavlova.",
        ):
            if event.type == "response.output_text.delta":
                print(event.delta, end="", flush=True)  # raw JSON streaming in
            if event.type == "object":
                recipe = event.object  # a validated Recipe instance
                print(f"\nValidated: {recipe.name}")

The final event:

@dataclass(frozen=True)
class ObjectEvent[T]:
    object: T  # your pydantic model (or parsed JSON for dict schemas)
    # type = "object"

chat.create_object(schema, *, input=...)

Non-streaming convenience. Returns a MatildaObjectResponse — everything chat.create() returns, plus the validated object.

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


async def main():
    async with MatildaClient(token="...") as client:
        response = await client.chat.create_object(
            Invoice,
            input="Extract the invoice total: $1,250.00 AUD due 30 Sep.",
            conversation_id=conversation_id,
        )

        print(response.object.total)      # 1250.0 (float)
        print(response.object.currency)   # "AUD" (str)
        print(response.output_text)       # raw JSON text as returned

MatildaObjectResponse

Extends MatildaChatResponse with one additional field:

Raw JSON Schema via response_schema

If you don't want pydantic validation, pass a stringified JSON Schema directly as response_schema on any chat call:

import json


async def main():
    async with MatildaClient(token="...") as client:
        response = await client.chat.create(
            input="List three Australian birds.",
            response_schema=json.dumps({
                "type": "object",
                "properties": {"birds": {"type": "array", "items": {"type": "string"}}},
                "required": ["birds"],
                "additionalProperties": False,
            }),
        )
        data = json.loads(response.output_text)  # guaranteed valid, schema-conforming JSON

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

Local $ref/$defs (pydantic's nested-model output) are inlined client-side before the schema is sent — the server's grammar compiler does not resolve #/$defs/... pointers. A recursive shape cannot be inlined: for a pydantic model schema it degrades to unconstrained at the recursion point (client-side model_validate still guards output); a recursive dict schema raises ValueError, since a dict has no client-side validation to catch the unconstrained recursion point.

OpenAI-compatible endpoint

The OpenAI-compatible endpoint (POST /api/v1/chat/completions) also honours structured output via the standard response_format parameter, so the OpenAI Python SDK's structured-output option works against Matilda as-is:

• { "type": "json_schema", "json_schema": { "name": "...", "schema": {...} } } — grammar-constrained to your schema (the schema is applied with strict: true server-side; the strict and name fields you supply are re-wrapped downstream).
• { "type": "json_object" } — guarantees valid JSON output without a schema (OpenAI JSON mode).

Safety replace and structured output. If the server replaces the output mid-stream (safety filter), stream_object and create_object raise SafetyReplaceError — the replacement text is in str(err) and the triggering categories in .categories. Deltas already yielded to your consumer are not rolled back; if you render streamed JSON, handle response.output_text.replace events (or choose non-streaming create_object) to avoid showing half-rendered output that is later discarded.

Truncation throws. If the stream is truncated before the JSON completes, both helpers raise MatildaObjectParseError with the partial text in .raw. See §15. Error Handling.

Stream errors throw. If the server emits an error event mid-stream, both helpers raise MatildaStreamError with the server's error code and message (f"{code}: {message}").

9. Chat — Durable Streaming (Resume)

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

Durable streaming lifecycle

1. Start a stream — chat.stream() emits a ResponseCreated event with a stream_id.
2. Persist the stream_id and conversation_id immediately.
3. If disconnected, call chat.active_stream(conversation_id) to check if the stream is still live.
4. Call chat.resume(stream_id, last_event_id=...) to replay buffered events from last_event_id onwards.

chat.resume(stream_id, *, last_event_id=None, ...)

Resumes a previously detached stream by replaying buffered events from last_event_id. Returns an async generator of ChatEvent. Raises MatildaAPIError (404) when the stream buffer has expired.

async for event in client.chat.resume(saved_stream_id, last_event_id=saved_last_event_id):
    if event.type == "response.output_text.delta":
        print(event.delta, end="", flush=True)

chat.active_stream(conversation_id)

Checks whether a conversation has an active stream.

result = await client.chat.active_stream("conv-123")
# {"stream_id": "abc-123" | None, "status": "active" | "done" | "error" | None}

Returns a dict:

# ActiveStreamLookup:
{"stream_id": "abc-123" or None, "status": "active" | "done" | "error" | None}

chat.notify_on_completion(stream_id, enabled=True)

Request a push notification when a backgrounded stream completes.

await client.chat.notify_on_completion(stream_id, enabled=True)

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

Full resume example

import asyncio
import os

from matilda_client import MatildaClient


async def main():
    async with MatildaClient(token=os.environ["MATILDA_ACCESS_TOKEN"]) as client:
        stream_id = None
        last_event_id = None

        # Start streaming
        async for event in client.chat.stream(
            input="Write a long essay about Australia.",
            conversation_id="conv-123",
        ):
            if event.type == "response.created":
                stream_id = event.stream_id
            if event.type == "response.cursor":
                last_event_id = event.last_event_id
            if event.type == "response.output_text.delta":
                print(event.delta, end="", flush=True)

        # Later — check if the stream is still active, then resume
        active = await client.chat.active_stream("conv-123")
        if active["status"] == "active" and stream_id:
            print("\n--- Resuming ---")
            async for event in client.chat.resume(stream_id, last_event_id=last_event_id):
                if event.type == "response.output_text.delta":
                    print(event.delta, end="", flush=True)


asyncio.run(main())

10. Conversations (History)

Conversation resources return plain dict objects using the server's camelCase wire keys.

conversations.list(...)

Lists conversations with pagination.

async def main():
    async with MatildaClient(token="...") as client:
        result = await client.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": [
        {
            "id": "conv-123",
            "userId": "user-1",
            "title": "My chat",
            "createdAt": "2026-08-18T10:30:00.000Z",
            "updatedAt": "2026-08-18T11:00:00.000Z",
        }
    ],
    "total": 1,
    "limit": 20,
    "offset": 0,
}

conversations.retrieve(conversation_id)

Retrieves a full conversation thread with all messages.

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

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

conversations.update(conversation_id, *, title=...)

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

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

Returns None.

conversations.set_message_feedback(conversation_id, message_id, feedback)

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

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

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

11. Files

files.upload(file, ...)

Uploads a single file by path. Files at or above the chunked threshold (MatildaConfig.chunked_threshold, 16 MiB default) use the parallel multipart protocol; smaller files use single-shot upload.

async def main():
    async with MatildaClient(token="...") as client:
        result = await client.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'

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

Uploads multiple files in parallel (auto-selecting single-shot vs chunked per file). One file's failure does not abort the others — per-file outcomes arrive in the result list, in input order, as either FileComplete or UploadError.

async def main():
    async with MatildaClient(token="...") as client:
        results = await client.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].

files.retrieve(file_id)

Retrieves metadata for a previously uploaded file.

file = await client.files.retrieve("file-abc123")
print(f"{file['filename']} — {file['status']} ({file['sizeBytes']} bytes)")
if file.get("extractedText"):
    print(f"Extracted: {file['extractedText'][:100]}...")

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

files.list_artifacts(conversation_id, *, limit=None, offset=None)

Lists generated artifacts for a conversation — used to re-hydrate artifact history across sessions.

async def main():
    async with MatildaClient(token="...") as client:
        result = await client.files.list_artifacts("conv-123", limit=50)
        for item in result.artifacts:
            print(f"{item.filename} ({item.content_type}, {item.size_bytes} bytes)")
        print(f"Total: {result.total}")

• conversation_id (positional, required) is sent as the conversationId query parameter — the server returns 400 when it is missing.
• limit (1–200, default 50) and offset (default 0) page the list.

Returns an ArtifactListResult:

@dataclass
class ArtifactListItem:
    artifact_id: str
    filename: str
    content_type: str
    size_bytes: int
    conversation_id: str | None
    message_id: str | None
    created_at: str

@dataclass
class ArtifactListResult:
    artifacts: list[ArtifactListItem]
    total: int

An empty conversation returns 200 with an empty list (never a 404).

files.download_artifact(artifact_id, filename=None)

Downloads a generated artifact's bytes (GET /files/{artifact_id}/content).

async def main():
    from pathlib import Path

    async with MatildaClient(token="...") as client:
        artifact = await client.files.download_artifact("art-123")
        print(f"{artifact.suggested_filename} ({artifact.content_type})")
        Path(artifact.suggested_filename).write_bytes(artifact.content)

• artifact_id (positional, required) is URL-encoded onto the path.
• filename (optional) is the fallback suggestion used when the server's Content-Disposition header carries no filename.

Returns an ArtifactDownload — the caller owns the bytes and chooses how to surface them (write to disk, stream elsewhere, etc.):

@dataclass
class ArtifactDownload:
    content: bytes
    content_type: str
    suggested_filename: str

suggested_filename falls back to the passed filename, then to the artifact_id, when Content-Disposition is absent. An unknown or foreign artifact returns a 404, which raises MatildaAPIError.

Using files in chat

Upload a file, then reference its file_id in a chat message:

async def main():
    async with MatildaClient(token="...") as client:
        upload = await client.files.upload("report.pdf", content_type="application/pdf")

        response = await client.chat.create(
            input="Summarise this report.",
            file_ids=[upload.file_id],
        )
        print(response.output_text)

Server-advertised upload settings (runtime config)

On its first upload, the client fetches the unauthenticated GET /config/public once and caches the result for the client's lifetime:

• Sizing. A chunk_size / chunked_threshold you set explicitly (any value other than the MatildaConfig defaults) always wins; otherwise the server-advertised uploadChunkSizeBytes / uploadChunkedThresholdBytes apply.
• Direct mode. When the server advertises uploadDirectEnabled, parts bypass core-api: the client mints a short-TTL presigned URL per part (POST /files/uploads/:uploadId/part-url?partNumber=N, body {"contentLength": N}) and PUTs the bytes straight to the uploads host. The proxied add-part route is disabled on such deployments (it answers 409), so there is no client-side opt-out. Presigned destinations are pinned to the server-advertised uploadsHost (https, exact hostname match) — anything else fails the upload with UploadError("Upload destination is not permitted."). The object-store ETag (quotes stripped) is collected per part and sent to the completion call, exactly as with the relay path.
• Fail-open. An unreachable or erroring /config/public yields the defaults (chunked on, direct off), so a config outage never switches transport.

client.fetch_public_runtime_config()

Fetches the same runtime config directly. Raises MatildaAPIError on a non-2xx response — unlike the upload path, which fails open.

config = await client.fetch_public_runtime_config()
print(config.upload_direct_enabled, config.uploads_host)

Returns PublicRuntimeConfig:

@dataclass
class PublicRuntimeConfig:
    upload_chunk_size_bytes: int | None = None
    upload_chunked_threshold_bytes: int | None = None
    upload_chunked_enabled: bool = True
    upload_direct_enabled: bool = False
    uploads_host: str | None = None

12. Feedback

feedback.report(...)

Reports a message for harmful, inaccurate, off-topic, or privacy-violating content.

await client.feedback.report(
    message_id="msg-456",
    conversation_id="conv-123",
    reason="inaccurate",
    comment="The capital of Australia is Canberra, not Sydney.",
)

Parameters

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

feedback.response(...)

Submits general response feedback (positive/negative sentiment with platform context).

await client.feedback.response(
    message_id="msg-456",
    conversation_id="conv-123",
    comment="Great answer!",
    platform="web",
    app_version="1.0.0",
)

Parameters

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

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-client), 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 client.feedback.report_bug(
    title="stream() silently closes on 429",
    description="The stream closes without error when the API returns 429.",
    reproduction="Call chat.stream() in a loop until rate-limited; observe the close event.",
)

Parameters

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

13. System One (matilda-jev)

System One is Matilda's structured-decision service, powered by the matilda-jev model. You send a state object plus a map of typed questions; the API returns an answer for each question. Schema validation lives in jev itself, so new question shapes arrive without an SDK release.

async with MatildaClient(token="mc_live_...") as client:
    decision = await client.systemone.evaluate(
        state={"ticket": "Charged twice", "customer_tier": "gold"},
        questions={"urgent": {"type": "noul", "instructions": "Needs a same-day response?"}},
    )

print(decision.answers["urgent"])  # {'type': 'noul', 'noul': 0.96}
print(decision.model_version)      # 'matilda-decision-v1'

Missing or expired credentials raise MatildaAPIError with status 401, and an otherwise-valid key without the api:systemone scope raises 403; create a key with the api:systemone scope to call it (see §4. API Key Management). Caller-fixable jev errors (400 / 413 / 422 / 429) pass through with jev's response body in error.response_text.

systemone.evaluate(*, state, questions, access_token=None, fingerprint=None)

Evaluates one state against typed questions.

Returns a SystemOneDecision:

# SystemOneDecision:
#   model: str                    — public model id ("matilda-jev")
#   model_version: str | None     — concrete jev build, from the x-matilda-model-version header
#   answers: dict[str, Any]       — answer per question id
#   usage: dict[str, Any] | None  — token accounting, when the deployment reports it

15. Error Handling

Every error raised by this SDK derives from MatildaError.

MatildaAPIError

Raised on non-2xx HTTP responses from the API.

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

QuotaExceededError

A MatildaAPIError subclass for HTTP 429 with a structured quota_exceeded body — the daily quota is spent. Carries the quota payload as attributes:

class QuotaExceededError(MatildaAPIError):
    tier: str | None
    messages_used: int | None
    messages_limit: int | None
    tokens_used: int | None
    tokens_limit: int | None
    resets_at: str | None
    upgrade_action: str | None

MatildaStreamError

A terminal error event on the chat SSE stream, surfaced by the text/object helpers. str(err) is f"{code}: {message}".

class MatildaStreamError(MatildaError):
    code: str  # a ChatErrorCode

MatildaContentTypeError

Raised by the streaming entry points (chat.stream(), chat.stream_object(), chat.resume()) when a streaming endpoint returns a 2xx response whose Content-Type is not text/event-stream — e.g. an HTML error page from a proxy or a JSON error body from a misconfigured gateway. Raised before parsing begins so an unconsumable body fails loudly instead of silently yielding no events.

class MatildaContentTypeError(MatildaError): ...
# str(err) names the offending Content-Type (or its absence)

SafetyReplaceError

Raised by chat.stream_text(), chat.stream_object(), and chat.create_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 that triggered the replacement.

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

MatildaObjectParseError

Raised by chat.stream_object() and chat.create_object() when the response cannot be parsed as JSON or fails pydantic validation — e.g. when a stream is truncated. raw contains the full response text; cause is the underlying json.loads or pydantic error. See §8. Chat — Structured Output.

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

UploadError

Raised when a file upload fails after retries (also used for missing-token upload attempts).

AuthError

Raised by auth flows. The code field is an OAuth error code (e.g. 'invalid_grant', 'authorization_pending', 'expired_token', 'access_denied'). The retryable field distinguishes transient failures (5xx, network) from permanent ones (revoked refresh token).

class AuthError(MatildaError):
    code: str
    retryable: bool

Chat error codes (ChatErrorCode)

These codes are emitted via the ResponseError stream event and appear in MatildaChatResponse.errors. Also available as the CHAT_ERROR_CODES frozenset:

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_STALL_TIMEOUT_MS — 3× the server's 15-second keep-alive ping cadence)
• Disable: Pass stall_timeout_ms=0 (not recommended — mobile loses background-to-foreground hung-stream recovery)

Error handling example

from matilda_client import (
    MatildaAPIError,
    MatildaClient,
    QuotaExceededError,
    SafetyReplaceError,
)


async def main():
    async with MatildaClient(token="...") as client:
        try:
            text = await client.chat.create_text(input="Hello!")
            print(text)
        except QuotaExceededError as err:
            print(f"Daily quota exceeded — resets at {err.resets_at}.")
        except MatildaAPIError 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)}")

16. Multi-Turn Conversations

The client SDK does not have a Session class. Multi-turn conversations are managed by passing a conversation_id to each chat call. The server reconstructs the full conversation history server-side from the session store.

Pattern

1. Generate a conversation ID (any unique string, e.g. a UUID).
2. Pass it to every chat.create() or chat.stream() call.
3. The server maintains the conversation history — you only send the latest message.

Important: Omitting conversation_id auto-creates a new conversation per call, and the response never returns that auto-created ID. Mint your own ID (e.g. uuid.uuid4()) and thread it through — it's the only way to continue a conversation.

import asyncio
from uuid import uuid4

from matilda_client import MatildaClient


async def main():
    async with MatildaClient() as client:
        # Authenticate with device flow — token refresh is handled automatically
        if not await client.auth.get_tokens():
            await client.auth.login_with_device_flow(client_id="matilda-code")

        conversation_id = str(uuid4())

        # Turn 1
        r1 = await client.chat.create(
            input="What is the capital of France?", conversation_id=conversation_id
        )
        print(r1.output_text)  # "Paris"

        # Turn 2 — server remembers the previous turn
        r2 = await client.chat.create(
            input="What about Germany?", conversation_id=conversation_id
        )
        print(r2.output_text)  # "Berlin"

        # Turn 3
        r3 = await client.chat.create(
            input="And Italy?", conversation_id=conversation_id
        )
        print(r3.output_text)  # "Rome"


asyncio.run(main())

Contrasting with the agent SDK

The Matilda agent SDK provides a Session class that wraps an Agent with auto-managed conversation_id, a turns list, and session-level defaults. If you need client-side tool execution, approval loops, or session state management, consider the agent SDK. For simple chatbot integrations, the client SDK's conversation_id pattern is sufficient.

Retrieving conversation history

# List all conversations
result = await client.conversations.list(limit=50)

# Retrieve a specific conversation with full message history
conv = await client.conversations.retrieve(conversation_id)
for msg in conv["messages"]:
    print(f"[{msg['role']}] {msg['content']}")

17. Suggested Tasks / Recipes

Recipe 1: CLI chatbot

A complete interactive CLI chatbot with device-flow auth and streaming.

import asyncio
from pathlib import Path
from uuid import uuid4

from matilda_client import MatildaClient, create_file_token_store


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

    async with MatildaClient() as client:
        # Authenticate if needed
        if not await client.auth.get_tokens():
            print("Starting device flow authentication...")
            await client.auth.login_with_device_flow(
                client_id="matilda-code",
                token_store=token_store.store,
                token_lock=token_store.lock,
            )
            print("Authenticated!")

        conversation_id = str(uuid4())

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

            print("Matilda: ", end="", flush=True)
            async for chunk in client.chat.stream_text(
                input=user_input, conversation_id=conversation_id
            ):
                print(chunk, end="", flush=True)
            print()


asyncio.run(main())

Recipe 2: File Q&A

Upload a document and ask questions about it.

import asyncio
from uuid import uuid4

from matilda_client import MatildaClient


async def main():
    async with MatildaClient() as client:
        # Authenticate with device flow — token refresh is handled automatically
        if not await client.auth.get_tokens():
            await client.auth.login_with_device_flow(client_id="matilda-code")

        # Upload a file
        upload = await client.files.upload(
            "report.pdf",
            content_type="application/pdf",
            on_progress=lambda pct: print(f"\rUploading: {pct}%", end="", flush=True),
        )
        print(f"\nUploaded: {upload.file_id} ({upload.status})")

        # A conversation_id is required for follow-ups to share history — omitting it
        # auto-creates a new conversation per call.
        conversation_id = str(uuid4())

        # Ask a question about it
        response = await client.chat.create(
            input="Summarise the key findings in this report.",
            file_ids=[upload.file_id],
            conversation_id=conversation_id,
        )
        print(response.output_text)

        # Follow-up question in the same conversation
        follow_up = await client.chat.create(
            input="What are the recommendations?",
            file_ids=[upload.file_id],
            conversation_id=conversation_id,
        )
        print(follow_up.output_text)


asyncio.run(main())

Recipe 3: Conversation history browser

List, paginate, and inspect conversation history.

import asyncio

from matilda_client import MatildaClient


async def main():
    async with MatildaClient() as client:
        # Authenticate with device flow — token refresh is handled automatically
        if not await client.auth.get_tokens():
            await client.auth.login_with_device_flow(client_id="matilda-code")

        # List first page
        offset = 0
        limit = 10
        page = await client.conversations.list(limit=limit, offset=offset)

        print(f"Total conversations: {page['total']}\n")

        for conv in page["conversations"]:
            print(f"[{conv['id']}] {conv['title']}")
            print(f"  Updated: {conv['updatedAt']}\n")

        # Load next page
        offset += limit
        if offset < page["total"]:
            page = await client.conversations.list(limit=limit, offset=offset)
            for conv in page["conversations"]:
                print(f"[{conv['id']}] {conv['title']}")

        # Retrieve a full conversation
        if page["conversations"]:
            full = await client.conversations.retrieve(page["conversations"][0]["id"])
            print(f"\n--- {full['title']} ---")
            for msg in full["messages"]:
                print(f"\n[{msg['role'].upper()}]")
                print(msg["content"])
                if msg.get("feedback"):
                    print(f"  Feedback: {msg['feedback']}")


asyncio.run(main())

Recipe 4: Durable stream recovery

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

import asyncio
from uuid import uuid4

from matilda_client import MatildaClient


async def main():
    async with MatildaClient() as client:
        # Authenticate with device flow — token refresh is handled automatically
        if not await client.auth.get_tokens():
            await client.auth.login_with_device_flow(client_id="matilda-code")

        conversation_id = str(uuid4())
        stream_id = None
        last_event_id = None
        received_text = ""

        # Start streaming — simulate disconnect after a few events
        print("Starting stream...")
        try:
            async for event in client.chat.stream(
                input="Write a very long, detailed essay about the history of computing.",
                conversation_id=conversation_id,
            ):
                if event.type == "response.created":
                    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
                if event.type == "response.completed":
                    print("Stream completed naturally.")
        except Exception as err:
            print(f"Disconnected: {err}")

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

        # Check if the stream is still active
        if stream_id:
            active = await client.chat.active_stream(conversation_id)
            print(f"Stream status: {active['status']}")

            if active["status"] == "active":
                print("\n--- Resuming ---")
                async for event in client.chat.resume(
                    stream_id, last_event_id=last_event_id
                ):
                    if event.type == "response.output_text.delta":
                        received_text += event.delta
                        print(event.delta, end="", flush=True)
                    if event.type == "response.completed":
                        print("\n\n--- Resume complete ---")
                        print(f"Total received: {len(received_text)} chars")


asyncio.run(main())

Recipe 5: Multi-environment setup

Run a local dev backend and production clients in the same process. Each instance manages its own auth independently.

import asyncio

from matilda_client import MatildaClient


async def main():
    async with (
        MatildaClient(base_url="http://localhost:3000/api") as local,
        MatildaClient(base_url="https://matilda.com/api") as production,
    ):
        # Authenticate each instance independently — device flow or browser login
        if not await local.auth.get_tokens():
            await local.auth.login_with_device_flow(client_id="matilda-code")
        if not await production.auth.get_tokens():
            await production.auth.login_with_browser(client_id="matilda-code")

        # Run the same prompt against both environments
        local_response, prod_response = await asyncio.gather(
            local.chat.create_text(input="Explain quantum entanglement."),
            production.chat.create_text(input="Explain quantum entanglement."),
        )

        print("Local:", local_response)
        print("Production:", prod_response)

        # Instances are fully isolated — each manages its own token lifecycle,
        # and config is immutable after construction, so one instance's setup
        # can never leak into another.


asyncio.run(main())

Recipe 6: Custom token store

Implement StorageAdapter to store tokens in a database or other custom backend.

import asyncio

from matilda_client import MatildaClient


# Example: a database-backed token store
class DatabaseTokenStore:
    def __init__(self, db):
        self._db = db

    async def get(self, key: str) -> str | None:
        row = await self._db.fetchone("SELECT value FROM tokens WHERE key = ?", (key,))
        return row["value"] if row else None

    async def set(self, key: str, value: str) -> None:
        await self._db.execute(
            "INSERT INTO tokens (key, value) VALUES (?, ?) "
            "ON CONFLICT (key) DO UPDATE SET value = excluded.value",
            (key, value),
        )

    async def remove(self, key: str) -> None:
        await self._db.execute("DELETE FROM tokens WHERE key = ?", (key,))


async def main():
    token_store = DatabaseTokenStore(my_database)

    async with MatildaClient() as client:
        await client.auth.login_with_device_flow(
            client_id="matilda-code",
            token_store=token_store,
            # No cross-process lock needed — the database handles concurrency
        )

        # Tokens are now persisted in the database and survive process restarts
        response = await client.chat.create(input="Hello!")
        print(response.output_text)


asyncio.run(main())

Exports Reference

Client

Chat responses and events

Errors

Auth

Files and feedback types

Constants and version helpers

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