Agent SDK · Running agents

Run.

Run an agent turn to completion and get the full result.

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

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

Python

from matilda_agent_sdk import Runner

runner = Runner()

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

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

Run options

All options are keyword-only.

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

Note

Prefer runner.run_object / runner.stream_object over response_schema — they convert a pydantic model for you; see Structured output. Client tool handlers are wired up in Client tools.

`callbacks`

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

Python

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

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

`AgentRunResult`

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

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

Python

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

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