matilda-client

Public Python client SDK for the Matilda API — the Python port of @maincode-ai/matilda-client-sdk. OAuth login (PKCE + device flow) with managed token refresh, streaming chat with typed events, grammar-constrained structured output, robust chunked parallel file uploads, conversations, feedback, and API-key management. Async-first, minimal dependencies: httpx + stdlib only (per ADR-025 SDK-boundary rule).

Install

pip install matilda-client

Authentication

Two managed login flows (core-auth is an OAuth facade over FusionAuth; endpoints are discovered via RFC 8414 metadata). On success the client is auto-wired with a TokenManager — every request carries a managed access token, refreshed automatically (single-flight) before expiry and once on a 401:

async with MatildaClient(base_url="https://matilda.maincode.com/api") as client:
    # Headless (servers, CI, SSH): prints a URL + user code, polls until approved
    tokens = await client.auth.login_with_device_flow(client_id="matilda-code")

    # Desktop: opens the browser, receives the callback on 127.0.0.1 (RFC 8252)
    tokens = await client.auth.login_with_browser(
        client_id="matilda-code", open_browser=lambda url: webbrowser.open(url)
    )

# Persist the session across runs (0600 JSON file + cross-process refresh lock):
from matilda_client import create_file_token_store
store = create_file_token_store("~/.matilda/tokens.json")
await client.auth.login_with_device_flow(
    client_id="matilda-code", token_store=store.store, token_lock=store.lock
)

await client.auth.get_tokens()  # current TokenSet or None
await client.auth.logout()      # clears the session + detaches the provider

Or skip the flows and bring your own token:

MatildaClient(token="eyJ...")                      # static JWT
MatildaClient(get_token=my_async_token_provider)   # provider, 401→refresh→retry

The lower-level protocol is exported too (fetch_auth_server_metadata, create_pkce_pair, begin_login/complete_login for web BFF redirects, request_device_code/poll_device_token, TokenManager, StorageAdapter) — mirror of the TS SDK's auth-core/auth-node split.

matilda-key CLI

matilda-key create-api-key --name "ci-runner" [--scopes api:code,api:chat] [--expires-at 2026-12-31T23:59:59Z]

Runs a device-flow login, mints an API key, and prints the secret to stdout (pipeable) with all metadata on stderr.

Quick start

import asyncio
from matilda_client import MatildaClient

async def main():
    # Token comes from the matilda-code OAuth/PKCE flow (a FusionAuth Bearer JWT).
    async with MatildaClient(token="eyJ...", base_url="https://matilda.maincode.com/api") as client:
        response = await client.chat.create(input="G'day")
        print(response.output_text)

asyncio.run(main())

Chat

stream yields the full typed event stream; create accumulates one turn; stream_text / create_text are the text-only conveniences.

# Full typed events
async for event in client.chat.stream(input="Summarise this", file_ids=["f_..."]):
    if event.type == "response.output_text.delta":
        print(event.delta, end="")

# Just the text, raising on stream errors
text = await client.chat.create_text(input="Hello")

# Multi-turn: thread the conversation id you want to continue. Omitting it
# starts a NEW conversation, and the response does not return the auto-created
# id — so hold the id yourself (from conversations.list or your own store).
await client.chat.create(input="Hi", conversation_id="conv_...")
await client.chat.create(input="And a follow-up", conversation_id="conv_...")

Events mirror the TS SDK union: ResponseCreated, OutputTextDelta, OutputTextReplace, StatusEvent / QueuedEvent, ToolCall*, GenerationStatus, UsageEvent, CursorEvent, Truncated, Completed, ResponseError.

Structured output

Grammar-constrained generation from a JSON-schema dict or a pydantic model class (duck-typed — pydantic is not a dependency; local $ref/$defs are inlined before sending since the server's grammar compiler does not resolve pointers):

schema = {
    "type": "object",
    "properties": {"name": {"type": "string"}, "age": {"type": "integer"}},
    "required": ["name"],
}
result = await client.chat.create_object(schema, input="Extract a person from 'Ada, 36'")
print(result.object)  # {'name': 'Ada', 'age': 36} — MatildaObjectParseError on failure

Resume a detached stream

active = await client.chat.active_stream(conversation_id)
if active.get("streamId"):
    async for event in client.chat.resume(active["streamId"], last_event_id=cursor):
        ...

Files

Chunked parallel uploads (OpenAI-shaped, S3-multipart-backed), auto-selecting single-shot vs chunked per file.

result = await client.files.upload("dataset.jsonl", on_progress=lambda pct: print(f"{pct}%"))
results = await client.files.upload_many(["a.jsonl", "b.jsonl"], file_concurrency=3)

• Small files (< chunked_threshold, default 16 MiB): single-shot POST /files/upload.
• Large files: create session → upload parts in parallel (default 4 concurrent) with per-part exponential-backoff retry (default 3) → complete (assembles S3 object + runs scan/extract). A dropped chunk just retries.

Conversations, feedback, API keys

conversations = await client.conversations.list(limit=20)
await client.conversations.update(conv_id, title="Quarterly report")
await client.conversations.set_message_feedback(conv_id, msg_id, "positive")

await client.feedback.report(message_id=msg_id, conversation_id=conv_id, reason="inaccurate")
await client.feedback.report_bug(title="SDK crash on resume", description="...")

key = await client.api_keys.create(name="ci")   # secret returned exactly once
await client.api_keys.revoke(key_id)

api_keys hits core-auth at {origin}/api/auth/api-keys — the origin is derived from base_url, so a base_url of https://host/api works as-is.

Errors

MatildaError is the base class. Non-2xx responses raise MatildaAPIError (QuotaExceededError on 429 with a structured quota body). Stream errors are ResponseError events on stream/create, and raise as MatildaStreamError on the text/object paths; SafetyReplaceError fires there when the server replaces in-flight output. Uploads raise UploadError after retries are exhausted.

If config.get_token (an async token provider) is configured, a 401 triggers one automatic retry with a force-refreshed token.

Configuration

Development

python -m venv .venv && .venv/bin/pip install -e ".[dev]"
.venv/bin/python -m pytest tests   # includes a contract-drift guard against @matilda/contracts
.venv/bin/ruff check src tests