# Run

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

Source: https://maincode.com/docs/python-agent-sdk-run
Section: Agent SDK · Matilda documentation

---

## `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](https://maincode.com/docs/python-agent-sdk-structured-output). Client tool handlers are wired up in [Client tools](https://maincode.com/docs/python-agent-sdk-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)")
```