matilda-agent-sdk

Python port of @maincode-ai/matilda-agent-sdk. Layered on top of matilda-client, the public Python SDK for the Matilda API — adds agent lifecycle, streaming, client tool execution with human-in-the-loop composition, and multi-agent pipelines.

Async-first. Dependencies: matilda-client + httpx (pulled in transitively).

Install

pip install matilda-agent-sdk

Requires Python 3.12+.

Quick start

import asyncio
from matilda_agent_sdk import Agent, AgentAuth, MatildaClient, Runner

async def main():
    async with MatildaClient() as client:
        await AgentAuth(client).login_with_device_flow(client_id="matilda-code")
        runner = Runner(client)
        result = await runner.run(
            Agent(name="reviewer", instructions="Be terse."),
            "Summarise the latest commit.",
        )
        print(result.final_output)

asyncio.run(main())

Agents

An Agent is a named, purposed system prompt. purpose selects the model routing and response mode — code (default) gets the routing-intent block and responseMode "auto"; analysis maps to responseMode "deep".

Agent(
    name="summariser",
    purpose="general",
    instructions="You write tight, high-signal summaries.",
    context="Assume the reader has no prior context.",
    metadata={"owner": "growth", "tier": "foundational"},
)

instructions may be a string or a sync/async callable of a context dict (agent_name, input, purpose, metadata) resolved per turn. Name is required and trimmed; a blank name raises ValueError. The effective handler merges the callable result with context (joined as Context: on a new line).

Running agents

At its simplest, Runner.run() blocks until the turn finishes:

from matilda_agent_sdk import Agent, Runner, configure_default_client

runner = Runner()  # default: bare MatildaClient; override with configure_default_client(...)

result = await runner.run(
    Agent(name="researcher", purpose="analysis"),
    "What changed in v1.1.0 of the SDK?",
)
print(result.final_output)   # str
print(result.usage)          # UsageSummary(output_tokens=182, context_pct=None, ...)
print(result.errors)         # [StreamErrorDetail(...)] — only if the stream errored and you swallowed it

Runner.stream() is the async-generator variant — yields every agent event (see below), still raises on stream errors unless you opt out.

Structured output

run_object() / stream_object() feed a JSON schema (dict or pydantic model class — pydantic is not a dependency) into the server-side grammar compiler and validate the accumulated text before returning:

schema = {
    "type": "object",
    "properties": {
        "name": {"type": "string"},
        "age": {"type": "integer"},
    },
    "required": ["name"],
}

obj = await runner.run_object(
    agent,
    "Extract a person from 'Ada, 36'",
    schema=schema,
)
print(obj.object)  # {'name': 'Ada', 'age': 36}

run_object raises MatildaObjectParseError if the model's text is not valid JSON or fails schema validation (local $ref/$defs are inlined first since the server cannot resolve pointers).

Client tools (human-in-the-loop)

Declare a tool shape once and hand the handler a ToolHandlers dict; the agent may then request a tool execution mid-turn and you decide what to allow:

from matilda_agent_sdk import ClientToolRequested, ToolHandlers

async def confirm(args, ctx):
    # Async prompt the human — `ctx.tool_call_id` ties this request to the stream.
    approved = await asyncio.to_thread(input, f"run {args['command']}? [y/n] ")
    if approved.strip().lower() != "y":
        return ToolResult(
            content="Human declined the tool execution.",
            is_error=True,
        )
    return ToolResult(content="approved")

async def git_status(args, ctx):
    if args.get("command", "").strip() != "git status":
        return ToolResult(content="Only `git status` is allowed.", is_error=True)
    proc = await asyncio.create_subprocess_exec("git", "status", "--porcelain", stdout=PIPE)
    out, _ = await proc.communicate()
    return ToolResult(content=out.decode() or "(clean working tree)")

handlers: ToolHandlers = {
    "confirm": confirm,
    "git_status": git_status,
}

result = await runner.run(
    Agent(name="triage", purpose="code"),
    "Check the working tree state before committing.",
    client_tools=[
        {
            "name": "git_status",
            "description": "Show the git working-tree state.",
            "parameters": {"type": "object", "properties": {"command": {"type": "string"}},
                            "required": ["command"]},
        },
    ],
    tool_handlers=handlers,
)

Tool roundtrips are capped (max_tool_roundtrips, default 25) so a buggy model cannot loop forever; the cap is enforced client-side and the runner raises MatildaAgentStreamError when it trips.

The DSML interceptor rewires <｜DSML｜tool_call> text tokens into real tool events — the model sometimes emits tool calls as literal text, and the interceptor terminates, parses, and re-emits them as first-class events (ClientToolRequested etc.).

Streaming text

For a pure text stream (e.g. piping to stdout), stream_text() yields deltas and raises on the safety line:

async for delta in runner.stream_text(agent, "Draft the release notes."):
    print(delta, end="")

run_text() is the same idea for a single string, and raises SafetyReplaceError when the server replaces in-flight output.

Sessions (multi-turn threads)

A Session threads turns together with a single conversation_id and replays the accumulated transcript ahead of each turn — the server's agent traffic is persist: false, so mentioning "the previous turn" only works when the client resends the history:

from matilda_agent_sdk import create_session

session = create_session(agent)  # or Session(agent)

await session.run("Draft a 1-line commit message.")
reply = await session.run("Now make it sound more like a human wrote it.")
print(session.turns[-1].final_output)

Session exposes conversation_id, turns, and last_turn for iteration or persistence. Pass a runner=/client= pair to bind it to a specific client config; otherwise it uses the shared default runner.

Parallel fan-out

import asyncio
from matilda_agent_sdk import run_text

results = await asyncio.gather(
    run_text({"name": "bug-triage"}, "Triage crash #1234"),
    run_text({"name": "bug-triage"}, "Triage crash #1235"),
    run_text({"name": "bug-triage"}, "Triage crash #1236"),
)

For load or auth isolation, give each runner its own client:

runner_a = Runner(client_a)
runner_b = Runner(client_b)
await asyncio.gather(runner_a.run(...), runner_b.run(...))

Runner(client) stamps agent provenance per request on its own traffic only — the supplied client (and shared singleton) is never mutated, so parallel runners and later direct client calls each report the right SDK.

Module-level conveniences

Once you've installed a client into the default runner (via configure_default_client(...)), the module-level functions mirror the TS SDK:

from matilda_agent_sdk import configure_default_client, run, run_text, run_object, stream_text

configure_default_client(MatildaClient(token="eyJ..."))

result = await run({"name": "informer"}, "Status?")

configure_default_client(client) installs the default once; subsequent calls reuse it unless you reset the process.

Auth

The SDK reuses the shared client for auth — a Runner with no client materialises a bare MatildaClient. For managed login, use AgentAuth (loopback PKCE / device flow):

from matilda_agent_sdk import AgentAuth, Runner

token_manager = await AgentAuth().login_with_device_flow(client_id="matilda-code")

AgentAuth snapshots the client's previous get_token provider before installing the TokenManager, and restores it on logout() — the client's auth state is never blanked when it is shared with other callers.

See the matilda-client README for token stores, persistence, and matilda-key.

Client resources

Runner proxies the underlying client resources so agent code doesn't have to hold a separate client reference: runner.files, runner.conversations, and runner.feedback map directly onto client.files, client.conversations, and client.feedback (the agent-layer FeedbackResource stamps the agent SDK as the reporting package on report_bug).

await runner.files.upload("dataset.jsonl")
await runner.conversations.list(limit=20)
await runner.feedback.report_bug(title="SDK crash on resume", description="...")

Resume a detached stream

resume_agent_stream(stream_id, last_event_id=...) replays a detached stream from the cursor and accumulates it into a AgentRunResult — the same shape the normal run() returns. It is resilient to a 401 by force-refreshing the managed token once before giving up.

result = await resume_agent_stream("str_...", last_event_id="ev_42")
print(result.final_output)

Errors

• MatildaError — base.
• MatildaAgentRunError — hard HTTP failure surfaced by the agent path.
• MatildaAgentStreamError(result) — the run finished in an errored stream state; result carries the accumulated events, errors, and partial output.
• MatildaObjectParseError — structured-output text was not valid JSON or failed the provided schema.
• SafetyReplaceError — the server replaced in-flight output (only raised by run_text / stream_text / run_object; on run / stream it's recorded on the result's safety_replace field).

Development

python -m venv .venv && .venv/bin/pip install -e ".[dev]"
.venv/bin/python -m pytest tests
.venv/bin/ruff check src tests