Skip to content

API Reference

Auto-generated from docstrings. For the concepts and conventions behind each primitive, see Core Concepts and Running Agents; this page is the exhaustive signature-level reference.

Agent

Agent

Agent(
    *,
    name: str = _UNSET,
    instructions: str | Callable[..., Any] | None = _UNSET,
    model: str | Model = _UNSET,
    model_settings: ModelSettings | None = _UNSET,
    tools: list[FunctionTool] | None = _UNSET,
    subagents: SubagentsList | None = _UNSET,
    guardrails: GuardrailsList | None = _UNSET,
    mcp: list[Any] | None = _UNSET,
    output_type: type | None = _UNSET,
    hooks: Hooks[Any] | None = _UNSET,
    memory: MemoryLike
    | Literal["auto", "llm"]
    | None = _UNSET,
    knowledge: KnowledgeLike
    | Literal["auto", "llm"]
    | None = _UNSET,
    compact: bool | Compactor | None = _UNSET,
    max_turns: int = _UNSET,
    max_tokens: int | None = _UNSET,
    timeout: float | None = _UNSET,
)

An Agent whose config comes from class attributes instead of init args.

usage accumulates token usage across every run/run_sync/run_streamed call made on this instance; last_usage holds just the most recent call's usage. Both are read from RunContext.usage, which the turn loop populates regardless of hooks.

One run is bounded three ways, each None/unset meaning "no ceiling of that kind": max_turns (model calls, default 10), max_tokens (total tokens the run may spend), and timeout (wall-clock seconds). Hitting any of them ends the run as Run(status="error"), the same as every other RunaError. max_tokens is the run's whole budget and is not the same knob as ModelSettings(max_tokens=...), which caps one response's length.

An instance carries per-run state (history, usage), so it is not safe to share across concurrent runs; see run for the rule and the two ways to satisfy it.

Build config from class attributes and wire up any subagents and guardrails.

These parameters are the closed set of settings (_AGENT_FIELDS is read off this signature): each one can be declared on the subclass or passed here, the argument winning, and an unset one falls back to the class attribute of the same name -- in one pass over that tuple (_settings), so adding one here resolves it. A name that is neither is a TypeError from this signature, and a UserError naming the setting it probably meant when it is declared in a class body instead (see _reject_unknown_settings).

mcp lists the agent's MCP servers; self.mcp_servers is what it resolved to.

memory opts this agent into long-term memory, one of: - "auto" (or a Memory(...) instance, or any object shaped like runa.memory's MemoryLike): the turn loop retrieves relevant memories before each run and persists new ones after -- no manual memory.search/.remember calls. - "llm": the model gets a search_memory tool and decides itself when to call it; no automatic retrieval/persistence. - None (the default): the agent behaves exactly as if runa.memory didn't exist. "llm" always uses a default Memory(); pass your own instance for "auto" mode if you need a non-default db_path/model/store -- or a wholesale custom MemoryLike object to replace embeddings-based retrieval entirely, not just its storage backend.

knowledge opts this agent into retrieval from application/domain documents, one of "auto", "llm", a Knowledge(...) (or KnowledgeLike-shaped) instance, or None (the default) -- the same four shapes as memory, with the same meaning: "auto"/an instance searches automatically before every turn (no tools=[...] wiring needed); "llm" gives the model a search_knowledge tool it calls itself; None leaves the agent unaffected. "llm" always uses a default Knowledge(); pass your own instance for "auto" mode if you need a non-default directory/db_path/model/store, or a custom KnowledgeLike object for a retrieval pipeline of your own.

compact keeps history/session from growing without bound, one of: - True: runa.compact.default_compactor -- past DEFAULT_COMPACTION_TOKENS, keep only the most recent exchange. A rolling window, not a summary. - a runa.compact.Compactor (any (items, usage_tokens) -> items | None callable): your own strategy -- a different threshold, an LLM summary, whatever you return. - False (the default): off. Resolved to a Compactor (or None) here, as self.compactor, the way memory=/ knowledge= are; anything else is a UserError at construction.

max_turns caps how many model calls one run may make before MaxTurnsExceeded; max_tokens caps the tokens they may spend before MaxTokensExceeded; timeout caps the run's wall-clock seconds before RunTimeout. All three are optional ceilings, and all three end the run as Run(status="error").

Methods:

Name Description
as_tool

Wrap this agent as a tool another agent can call; see runa.handoff.agent_as_tool.

run

Run a turn asynchronously, appending it to the conversation history.

run_sync

Synchronous run, for callers not already inside an event loop.

run_streamed

Run a turn as a stream of events: the same run as run, with the same arguments.

evaluate

Run every case in dataset through this agent and grade it: see runa.eval.

Attributes:

Name Type Description
graph Digraph

Render this agent, and its tools/subagents, as a Graphviz diagram.

graph property

graph: Digraph

Render this agent, and its tools/subagents, as a Graphviz diagram.

Displays inline in Jupyter; call .render(filename) to save it, or .source for the raw DOT text. Actually rendering an image needs the system Graphviz dot binary.

as_tool

as_tool(
    tool_name: str | None, tool_description: str | None
) -> FunctionTool

Wrap this agent as a tool another agent can call; see runa.handoff.agent_as_tool.

run async

run(
    message: MessageContent | RunState,
    context: Any = None,
    hooks: Hooks[Any] | None = None,
    session: Session | str | None = None,
) -> Run

Run a turn asynchronously, appending it to the conversation history.

message is plain text, or a list for a multimodal message: a bare string is text, or an image when it is an image URL or a data:image/... URI; a local image is a Path("cat.jpg"), never a bare string, since the list may carry a user's own words (see content.parts). Build a part explicitly with content.text(...)/content.image(...) when a string doesn't have a recognizable image extension. It can also be the RunState of a paused Run, once its interruptions are approved or rejected, to resume that run.

It is always one user turn, never a transcript: a list of {"role": ...} messages raises TypeError. Start from an earlier conversation by setting self.history directly, or by seeding the session with add_items before the first run.

context is available to a single-argument instructions callable (and to tools, guardrails, etc.) as-is; it is never sent to the model. hooks receives lifecycle callbacks (on_agent_start, on_tool_end, etc.); it defaults to LoggingHooks.

Pass a session to persist conversation history there instead of on self.history; the session supplies prior turns automatically, so only the new message is sent as input, and self.history is left untouched. Resume a paused run with the same session it started with.

A session is normally the conversation's id: session="user-42" persists to whichever backend RUNA_DATABASE_URL names, so an app moves to Postgres without naming one here. Pass a Session instead when the run needs more than the id -- db.session(id, user_id=...) to scope automatic memory, or your own store.

Returns a Run exposing .output, .status, .interruptions, .trace, .usage and .error. A tool call needing approval pauses the run (status="paused"). Nothing a run fails on propagates: a guardrail tripwire, MaxTurnsExceeded, a model error, and equally an exception out of application code the run called -- a Hooks method, a Session, a Compactor -- all come back as status="error" with the message on .error (UnexpectedError wraps the last kind, and its traceback is logged). The two that do raise are asyncio.CancelledError, because a caller cancelling a run wants it stopped, and OperatorError, which is not a run failure at all (see runa.exceptions). self.history only changes once a turn completes.

Token usage for this call is recorded to self.last_usage and accumulated into self.usage, regardless of session, hooks, or whether the run errored.

An Agent instance holds the state of the conversation it is running, so one instance runs one conversation at a time. Two overlapping session-less runs on the same instance raise UserError rather than quietly interleaving their histories; give each run its own session, or its own Agent. Under a web server, build the agent inside the request handler (runa serve does exactly this).

run_sync

run_sync(
    message: MessageContent | RunState,
    context: Any = None,
    hooks: Hooks[Any] | None = None,
    session: Session | str | None = None,
) -> Run

Synchronous run, for callers not already inside an event loop.

The event loop is this call's own, and an MCP server's session cannot outlive the loop it was opened on (see runa.mcp), so each is closed before that loop goes: a runa chat turn shuts its .stdio servers down instead of leaving a subprocess per turn behind. await agent.run(...) on an app's own loop keeps them open for the agent's lifetime.

A shutdown that fails is logged, never raised: the turn is over, its Run is the answer this call owes, and one server refusing to go must not take the others' shutdown with it.

run_streamed

run_streamed(
    message: MessageContent | RunState,
    context: Any = None,
    hooks: Hooks[Any] | None = None,
    session: Session | str | None = None,
) -> RunStream

Run a turn as a stream of events: the same run as run, with the same arguments.

Iterate the returned RunStream for StreamEvents (raw_response_event, run_item_stream_event, agent_updated_stream_event) as they arrive; once it ends, its .run holds the Run that run would have returned, paused, completed or errored. History and usage are recorded only once the stream is fully consumed, so a caller that stops iterating early leaves them unchanged.

One instance still runs one conversation at a time: a stream shares run's latch, so starting to consume a second session-less stream on the same instance while one is in flight raises UserError rather than interleaving the two histories.

evaluate async

evaluate(
    dataset: Iterable[Case],
    *,
    judge: str | None = None,
    threshold: float | None = None,
    thresholds: dict[str, float] | None = None,
    concurrency: int = 8,
) -> Report

Run every case in dataset through this agent and grade it: see runa.eval.

Deterministic checks and judge-graded semantic metrics (task completion, answer correctness/relevance, faithfulness, tool correctness) are chosen automatically per case based on what evidence it supplies; no metric configuration is required. judge overrides the model semantic metrics grade with, defaulting to this agent's own model. Up to concurrency cases run at once.

AgentLike

Bases: Protocol

Every attribute Runa reads off an agent: the other side of the seam, as types.

Attributes:

Name Type Description
name str

What this agent is called, everywhere one agent has to be told from another.

instructions str | Callable[..., Any] | None

Its system prompt: a string, a callable the run resolves, or none at all.

model str | Model

The model name a run resolves through its provider, or a Model built directly.

model_settings ModelSettings

How that model is sampled.

tools list[FunctionTool]

The tools it offers the model, its delegates among them (see runa.handoff).

mcp_servers list[Any]

The MCP servers whose tools join tools once a run connects to them.

handoffs list[Any]

Its handoff targets, as Handoff.by_tool_name reads them (runa.handoff).

bound_guardrails dict[Phase, list[BoundGuardrail]]

Its guardrails by the phase each one runs in: the resolved form of guardrails.

output_type type | None

The type its final answer is parsed into, or None for text.

memory Any

Its long-term memory: a MemoryLike or None once resolved (see this module).

knowledge Any

Its document knowledge: a KnowledgeLike or None once resolved.

compactor Compactor | None

What trims a long conversation, if anything: the resolved form of compact.

hooks Hooks[Any] | None

The lifecycle callbacks that belong to this agent rather than to a run.

name property

name: str

What this agent is called, everywhere one agent has to be told from another.

instructions property

instructions: str | Callable[..., Any] | None

Its system prompt: a string, a callable the run resolves, or none at all.

model property

model: str | Model

The model name a run resolves through its provider, or a Model built directly.

model_settings property

model_settings: ModelSettings

How that model is sampled.

tools property

tools: list[FunctionTool]

The tools it offers the model, its delegates among them (see runa.handoff).

mcp_servers property

mcp_servers: list[Any]

The MCP servers whose tools join tools once a run connects to them.

handoffs property

handoffs: list[Any]

Its handoff targets, as Handoff.by_tool_name reads them (runa.handoff).

bound_guardrails property

bound_guardrails: dict[Phase, list[BoundGuardrail]]

Its guardrails by the phase each one runs in: the resolved form of guardrails.

output_type property

output_type: type | None

The type its final answer is parsed into, or None for text.

memory property

memory: Any

Its long-term memory: a MemoryLike or None once resolved (see this module).

knowledge property

knowledge: Any

Its document knowledge: a KnowledgeLike or None once resolved.

compactor property

compactor: Compactor | None

What trims a long conversation, if anything: the resolved form of compact.

hooks property

hooks: Hooks[Any] | None

The lifecycle callbacks that belong to this agent rather than to a run.

Tool

tool

tool(func: Callable[..., Any]) -> FunctionTool
tool(
    func: None = None,
    *,
    name_override: str | None = None,
    description_override: str | None = None,
    guardrails: GuardrailsList | None = None,
    needs_approval: _NeedsApproval = False,
) -> Callable[[Callable[..., Any]], FunctionTool]
tool(
    func: Callable[..., Any] | None = None,
    *,
    name_override: str | None = None,
    description_override: str | None = None,
    guardrails: GuardrailsList | None = None,
    needs_approval: _NeedsApproval = False,
) -> (
    FunctionTool
    | Callable[[Callable[..., Any]], FunctionTool]
)

Wrap a plain function as a FunctionTool, deriving its schema from its signature.

guardrails=[...] (or {...}) wires @guardrail predicates, bare (wired as both sides) or bound via .input/.output, against the tool call's parsed arguments and its return value, respectively.

FunctionTool dataclass

FunctionTool(
    name: str,
    description: str,
    params_json_schema: dict[str, Any],
    on_invoke_tool: Callable[
        [RunContext, _Governance, str, str], Awaitable[Any]
    ],
    guardrails: dict[Phase, list[BoundGuardrail]] = dict(),
    needs_approval: _NeedsApproval = False,
    delegate: Any = None,
)

A tool the model can call: its schema, and the callable that actually runs it.

Attributes:

Name Type Description
on_invoke_tool Callable[[RunContext, _Governance, str, str], Awaitable[Any]]

Called with (ctx, governance, arguments_json, call_id); returns what the tool does.

guardrails dict[Phase, list[BoundGuardrail]]

This tool's guardrails by Phase, checked against the call's arguments

delegate Any

The sub-Agent behind a .delegate tool (set by agent_as_tool): traced as a

on_invoke_tool instance-attribute

on_invoke_tool: Callable[
    [RunContext, _Governance, str, str], Awaitable[Any]
]

Called with (ctx, governance, arguments_json, call_id); returns what the tool does.

governance is the framework's, for the one tool that starts a run of its own -- a delegate, which shares it with its caller (see runa._governance). A function wrapped by @tool never sees it: its ctx is the RunContext alone.

guardrails class-attribute instance-attribute

guardrails: dict[Phase, list[BoundGuardrail]] = field(
    default_factory=dict
)

This tool's guardrails by Phase, checked against the call's arguments (Phase.TOOL_INPUT) and its return value (Phase.TOOL_OUTPUT).

delegate class-attribute instance-attribute

delegate: Any = None

The sub-Agent behind a .delegate tool (set by agent_as_tool): traced as a "delegate" span, not a plain "tool", and resolvable by name when resuming a paused run.

Guardrail

guardrail

guardrail(func: _Predicate) -> Guardrail

Turn a (value) -> bool predicate into a Guardrail; bind it via .input/.output.

Guardrail

Guardrail(func: _Predicate)

A predicate bound to neither side yet; .input/.output (or .i/.o) picks which.

@guardrail wraps a plain (value) -> bool predicate (tripping the guardrail on True) into this, the same way @tool wraps a plain function into a FunctionTool. Read .input to bind it to the input side, .output for the output side; .i/.o are the exact same binding under a shorter name, there is no third spelling. The same bound object works in both places it's listed:

  • In an Agent.guardrails list, it's bound to Phase.INPUT/Phase.OUTPUT: the predicate sees the latest user message as plain text (regardless of whether the run was passed a string or the running list of input items) on .input, or the agent's final output on .output.
  • In a @tool(guardrails=[...]) list, the same object is rebound to Phase.TOOL_INPUT/ Phase.TOOL_OUTPUT: the predicate sees the tool call's arguments (parsed from JSON into a dict) on .input, or the tool's raw return value on .output.

The predicate's docstring becomes output_info. Listed bare (no .input/.output), it's wired as both sides of whichever pair applies.

Store the predicate to bind on .input/.output access.

Attributes:

Name Type Description
input BoundGuardrail

Bind this predicate to the input side.

output BoundGuardrail

Bind this predicate to the output side.

i BoundGuardrail

Shorthand for .input.

o BoundGuardrail

Shorthand for .output.

input property

input: BoundGuardrail

Bind this predicate to the input side.

output property

output: BoundGuardrail

Bind this predicate to the output side.

i property

i: BoundGuardrail

Shorthand for .input.

o property

o: BoundGuardrail

Shorthand for .output.

Phase

Bases: StrEnum

Where a guardrail runs: agent-or-tool by input-or-output, the 2x2 named once.

The phase decides everything the four used to spell out separately: which list an entry belongs to, what the one value its function is handed means (the agent's own input or final output, a tool call's arguments or return value), which key its results are recorded under, and how a tripwire names it.

Attributes:

Name Type Description
on_tool bool

Whether this phase guards one tool call rather than the agent's own turn.

on_output bool

Whether this phase checks what came back rather than what is about to go in.

title str

How a tripwire message names this phase, e.g. "Tool input guardrail".

on_tool property

on_tool: bool

Whether this phase guards one tool call rather than the agent's own turn.

on_output property

on_output: bool

Whether this phase checks what came back rather than what is about to go in.

title property

title: str

How a tripwire message names this phase, e.g. "Tool input guardrail".

GuardrailVerdict dataclass

GuardrailVerdict(output_info: Any, tripped: bool)

What a guardrail function returns: the predicate's docstring, plus trip/no-trip.

output_info is the @guardrail predicate's __doc__ (or None), never anything it computed: a predicate returns a bool, so what it checks for is all the audit trail can say.

One shape for all four phases, because tripped is the only question anyone asks a verdict. Whether the guardrail watched an agent's turn or one tool call is Phase's to say -- the wrapper reads phase to know what the one value it was handed is, GuardrailResult.phase records it for the audit trail -- so the verdict doesn't re-encode it as a second type.

GuardrailResult dataclass

GuardrailResult(
    guardrail: Any,
    output: GuardrailVerdict,
    tripped: bool,
    phase: Phase,
)

A guardrail plus the verdict it returned and the Phase it returned it in.

Every guardrail run this run is recorded (see GuardrailResults), not just the one that stopped the run; tripped distinguishes the two.

GuardrailResults dataclass

GuardrailResults(
    by_phase: dict[Phase, list[GuardrailResult]] = dict(),
)

A run's guardrail audit trail: every GuardrailResult it produced, keyed by Phase.

One mapping rather than a list per phase, so the four names RUNA.md documents are written once, on GuardrailAudit, and a phase nothing ran in simply has no key.

Methods:

Name Description
record

Append result under its own phase, starting that phase's list on first use.

__getitem__

Everything that ran in phase, an empty list if nothing did.

snapshot

A copy, so a finished Run isn't still growing as a shared context records more.

record

record(result: GuardrailResult) -> None

Append result under its own phase, starting that phase's list on first use.

__getitem__

__getitem__(phase: Phase) -> list[GuardrailResult]

Everything that ran in phase, an empty list if nothing did.

snapshot

snapshot() -> GuardrailResults

A copy, so a finished Run isn't still growing as a shared context records more.

Human Approval

approval

approval(
    func: Callable[..., bool | Awaitable[bool]],
) -> _NeedsApproval

Turn a predicate into a @tool(needs_approval=...) callable.

Write the predicate like the tool it guards: parameters are looked up by name from the tool call's parsed arguments. Name a parameter ctx or call_id to receive the run context or call id instead of a tool argument. Works on def, async def, and lambda.

Session

SQLiteSession

SQLiteSession(
    session_id: str,
    db_path: str | Path | None = None,
    *,
    user_id: str | None = None,
)

Bases: Session

Conversation history for one session_id, persisted to the local db/runa.db.

Every method here is _off_loop: a plain synchronous body, handed to a worker thread to satisfy Session's async def. Each opens its own short-lived connection, so no sqlite3 object is ever touched from two threads.

Store session_id and where in runa.db its history lives.

db_path defaults to runa.db.sqlite_path(), so this history lands in the same local file as every other concern's -- including when RUNA_DATABASE_URL relocates it. Pass one to point at some other file.

user_id scopes this session's automatic memory, if its agent has any; see Session.

Methods:

Name Description
get_items

Return this session's items, oldest first, capped at the latest limit if given.

add_items

Append items, creating the session row on first write.

set_items

Replace this session's entire history with items, in one transaction.

pop_item

Remove and return this session's most recent item, or None if it has none.

clear_session

Delete this session and all of its items.

get_items

get_items(
    limit: int | None = None,
) -> list[ConversationItem]

Return this session's items, oldest first, capped at the latest limit if given.

add_items

add_items(items: list[ConversationItem]) -> None

Append items, creating the session row on first write.

set_items

set_items(items: list[ConversationItem]) -> None

Replace this session's entire history with items, in one transaction.

pop_item

pop_item() -> ConversationItem | None

Remove and return this session's most recent item, or None if it has none.

clear_session

clear_session() -> None

Delete this session and all of its items.

Session

Bases: ABC

What run_internal needs to persist and replay conversation history across turns.

A base class, where the read side (SessionStore) and Cache are protocols a backend satisfies by shape alone: inheritance is what lets set_items below hand a default implementation out, so a custom backend writes four methods and gets Agent(compact=...) without a fifth.

user_id is optional and unrelated to history: run_internal reads it (when set) to scope automatic Agent.memory retrieval/persistence to one user, so app code doesn't have to thread a user_id through Memory.remember/.search itself. None means memory that goes through this session lands in its own user-less scope, not everyone's.

Methods:

Name Description
get_items

Return this session's items, oldest first, capped at the latest limit if given.

add_items

Append items to this session's history.

pop_item

Remove and return this session's most recent item, or None if it has none.

clear_session

Delete this session and all of its items.

set_items

Replace this session's entire history with items, Agent(compact=...)'s hook.

get_items abstractmethod async

get_items(
    limit: int | None = None,
) -> list[ConversationItem]

Return this session's items, oldest first, capped at the latest limit if given.

add_items abstractmethod async

add_items(items: list[ConversationItem]) -> None

Append items to this session's history.

pop_item abstractmethod async

pop_item() -> ConversationItem | None

Remove and return this session's most recent item, or None if it has none.

clear_session abstractmethod async

clear_session() -> None

Delete this session and all of its items.

set_items async

set_items(items: list[ConversationItem]) -> None

Replace this session's entire history with items, Agent(compact=...)'s hook.

A concrete default built from clear_session/add_items, not @abstractmethod: an existing custom Session gets this for free, with no new method it's forced to implement. Override for a single-transaction replace if that matters for your store, the way SQLiteSession does.

Memory

Memory

Memory(
    *,
    model: str = DEFAULT_EMBEDDING_MODEL,
    dimensions: int | None = None,
    store: MemoryStore | None = None,
)

Durable semantic facts, scoped by user_id: remember/search/forget, nothing lower.

Memory() means OpenAI's text-embedding-3-small over whichever database runa.db resolves: embeddings and vector storage are entirely internal, so callers only ever see text in and MemoryMatches out. store= swaps that for a custom MemoryStore, unrelated to embeddings.

Configure the embedding model and, unless store is given, this deployment's store.

dimensions only needs setting for a model not in Runa's built-in size table.

Methods:

Name Description
remember

Embed text and store it for user_id, returning its new memory id.

search

Return user_id's k memories closest in meaning to query, nearest first.

forget

Delete user_id's memory memory_id, if it exists.

remember_from_conversation

Ask model what's durably worth remembering from conversation, and store it.

remember async

remember(
    text: str,
    *,
    user_id: str | None = None,
    metadata: dict[str, Any] | None = None,
) -> int

Embed text and store it for user_id, returning its new memory id.

Skips storing, returning the existing id instead, when user_id already has a near-duplicate memory: repeated auto-extraction of the same restated fact across conversations doesn't pile up duplicates. metadata is dropped, not merged, on a skip.

search async

search(
    query: str, *, user_id: str | None = None, k: int = 5
) -> list[MemoryMatch]

Return user_id's k memories closest in meaning to query, nearest first.

forget async

forget(
    memory_id: int, *, user_id: str | None = None
) -> None

Delete user_id's memory memory_id, if it exists.

remember_from_conversation async

remember_from_conversation(
    conversation: str, *, user_id: str | None, model: Any
) -> list[str]

Ask model what's durably worth remembering from conversation, and store it.

Called by run_internal.run_loop._run_async after a run when agent.memory is set, not meant to be called directly by app code, but part of MemoryLike, the contract a custom memory= object must implement alongside search. Returns the texts it stored, empty if none were worth it.

A reply this can't read JSON out of stores nothing and logs a warning, the way a failed retrieval degrades in run_loop._retrieve: "nothing worth remembering" and "the model answered with prose" are the same empty list to a caller, and an agent on memory="auto" that has quietly stopped learning is a failure nobody would otherwise report.

MemoryMatch dataclass

MemoryMatch(
    id: int,
    text: str,
    metadata: dict[str, Any] | None,
    distance: float,
)

One Memory.search result: its id, stored text, metadata, and distance to the query.

Knowledge

Knowledge

Knowledge(
    directory: str | Path = DEFAULT_KNOWLEDGE_DIR,
    *,
    model: str = DEFAULT_EMBEDDING_MODEL,
    dimensions: int | None = None,
    store: KnowledgeStore | None = None,
)

Application/domain documents, retrieved by meaning: search/ingest, nothing lower.

Knowledge() means app/knowledge/, OpenAI's text-embedding-3-small, and whichever database runa.db resolves: discovery, chunking, embeddings and vector storage are entirely internal, so callers only ever see a source directory in and KnowledgeMatches out. Attach an instance to Agent(knowledge=...) and run/run_sync search it automatically before every turn; no tools=[...] wiring needed.

Configure the source directory, embedding model, and, unless store is given, storage.

dimensions only needs setting for a model not in Runa's built-in size table.

A relative directory is resolved against the project runa.project.use named, the same way the store's own file is, so an app served from outside its own directory reads its corpus and writes its vectors in one place. Absolute paths are left alone, which is how app code that means a directory somewhere else says so.

Methods:

Name Description
ingest

Bring the store up to date with self.directory, returning how many chunks it wrote.

search

Return the k chunks closest in meaning to query, nearest first.

ingest async

ingest() -> int

Bring the store up to date with self.directory, returning how many chunks it wrote.

Zero means there was nothing to write: the store already held this corpus, or the corpus is empty. Idempotent on purpose, because every caller wants the same thing -- the store holding what the directory says -- and no caller knows whether the last request, the last process or another replica got there first. So this is the one ingest verb: what search calls before a query, and what runa.serve calls at startup to keep the cold-start cost off a user's first turn. A cheap "if it's stale" path beside an expensive "rebuild now" one would be two ways to ask one question, and the expensive one is the answer a deployment would reach for by name and then pay for on every replica's every restart.

When a rebuild is called for it is a full one: discovers supported files, extracts their text, chunks it, embeds every chunk, empties whatever was stored before, and stores the fresh set under this corpus's fingerprint, so editing or removing a source file never leaves stale chunks behind, with no file-tracking table to keep in sync. Safe to call when self.directory doesn't exist or has no supported files: stores nothing rather than raising, and an empty corpus is still a fingerprinted one.

The question asked is "does the store hold what I would write", not "have I ingested yet": the answer is the store's, so the second Knowledge over an already-ingested corpus searches it instead of rebuilding it -- which is what a per-request Agent (runa.serve) and every replica after the first are. The version is read twice, once before taking the lock and once under it. The first read is the common case, where the corpus is current and a caller pays only for that one question; the second is what keeps a queue of concurrent turns against a cold store from each rebuilding what the turn ahead of it just finished.

search async

search(query: str, *, k: int = 5) -> list[KnowledgeMatch]

Return the k chunks closest in meaning to query, nearest first.

Ingests first when the store doesn't already hold this corpus, so the common case (Knowledge() attached to an Agent) needs no manual .ingest() call, and an edited source file is picked up on the next search rather than on the next manual rebuild. That call is the safety net rather than the plan: a served deployment has already ingested at startup (serve.create_app), so what a turn usually pays here is one stat per file and one version read.

KnowledgeMatch dataclass

KnowledgeMatch(
    id: int, text: str, source: str, distance: float
)

One Knowledge.search result: its id, stored text, source file, and distance.

Cache

Cache

Bases: Protocol

What application code needs from a cache: get/set/delete/clear, nothing lower.

No inheritance required, every backend satisfies this by matching shape, the same escape hatch as MemoryStore/SessionStore's custom-backend story. The two local backends expire entries lazily: an expired key is only evicted (and reported as a miss) the next time get looks it up, not by a background sweep.

Methods:

Name Description
get

Return the value stored for key, or None if it's missing or expired.

set

Store value for key, replacing whatever was there.

delete

Remove key, if present.

clear

Remove every entry.

get async

get(key: str) -> Any

Return the value stored for key, or None if it's missing or expired.

set async

set(key: str, value: Any, ttl: float | None = None) -> None

Store value for key, replacing whatever was there.

ttl is seconds until the entry expires; None (the default) never expires.

delete async

delete(key: str) -> None

Remove key, if present.

clear async

clear() -> None

Remove every entry.

MemoryCache

MemoryCache()

In-process Cache: a plain dict, gone when the process exits.

Start empty.

Methods:

Name Description
get

Return the value stored for key, or None if it's missing or expired.

set

Store value for key, replacing whatever was there.

delete

Remove key, if present.

clear

Remove every entry.

get async

get(key: str) -> Any

Return the value stored for key, or None if it's missing or expired.

set async

set(key: str, value: Any, ttl: float | None = None) -> None

Store value for key, replacing whatever was there.

ttl is seconds until the entry expires; None (the default) never expires.

delete async

delete(key: str) -> None

Remove key, if present.

clear async

clear() -> None

Remove every entry.

SQLiteCache

SQLiteCache(db_path: str | Path | None = None)

Persistent Cache: the cache_entries table inside db/runa.db, surviving restarts.

Store where this cache's table lives, defaulting to db.sqlite_path().

The table is created on first use.

Methods:

Name Description
get

Return the value stored for key, or None if it's missing or expired.

set

Store value for key, replacing whatever was there.

delete

Remove key, if present.

clear

Remove every entry.

get async

get(key: str) -> Any

Return the value stored for key, or None if it's missing or expired.

A lookup reads (db/sqlite.py's read), so a miss on a project with no db/runa.db yet is a miss, not a new file: the first set is what creates one.

set async

set(key: str, value: Any, ttl: float | None = None) -> None

Store value for key, replacing whatever was there.

ttl is seconds until the entry expires; None (the default) never expires.

delete async

delete(key: str) -> None

Remove key, if present.

clear async

clear() -> None

Remove every entry.

MCP Server

MCPServer

MCPServer(
    *,
    name: str | None = None,
    needs_approval: _NeedsApproval = False,
    guardrails: GuardrailsList | None = None,
)

The one way to build a server: shared config here, finalized by picking a transport.

What's given here -- the server's name, and the approval/guardrail policy its tools carry -- is common to every transport; http/stdio take the params specific to that transport, a URL vs. a command to spawn. Every parameter is spelled out on all three, rather than collected as **kwargs and forwarded, so that a misspelled or unsupported option is a TypeError naming it at the call site -- the way a mistyped Agent attribute is a UserError naming it at construction -- instead of a server that quietly ignores it and misbehaves on connect.

The two transport classes are deliberately absent from __all__: MCPServer is the whole public surface, and a third transport is written by subclassing _MCPServerBase, not by reaching for MCPServerStdio directly.

Hold config shared by every transport, applied when http/stdio is called.

needs_approval (a bool, or an @approval predicate over the remote tool's arguments) and guardrails (@guardrail predicates, bound with .input/.output against a call's arguments and its result) apply to every tool this server exposes -- @tool's two parameters, asked of the server because that's where a remote tool's trust is decided.

Methods:

Name Description
http

Connect over streamable HTTP.

stdio

Spawn a local process over stdio.

http

http(url: str) -> MCPServerStreamableHttp

Connect over streamable HTTP.

stdio

stdio(
    command: str,
    args: list[str] | None = None,
    *,
    env: dict[str, str] | None = None,
    cwd: str | None = None,
) -> MCPServerStdio

Spawn a local process over stdio.

Running

Run dataclass

Run(
    output: Any,
    trace: Trace,
    usage: Usage,
    status: Status = "completed",
    error: str | None = None,
    interruptions: list[Interruption] = list(),
    guardrail_results: GuardrailResults = GuardrailResults(),
    _state: RunState | None = None,
    _governance: _Governance | None = None,
    _transcript: Transcript = Transcript(),
)

Bases: GuardrailAudit

The outcome of a single Agent.run()/run_sync() call.

output is the agent's final output (parsed into Agent.output_type when it declares one), None unless status is "completed". trace is the hierarchical Trace (agent/LLM/tool/ handoff/guardrail spans) captured for this call, see runa.tracing; always present, whatever the status, because the trace is opened before anything a run can fail on. usage is this call's token usage, same value as Agent.last_usage after the call.

status is "paused" when a tool call needs human approval: resolve each of interruptions on to_state(), then pass that state back to run/run_sync in place of a message. It is "error" when a RunaError (a guardrail tripwire, MaxTurnsExceeded, a model error, ...) stopped the run; error then holds that exception's message.

guardrail_results is the guardrail audit trail: every guardrail that ran (tripped or not, a delegate's included), keyed by the Phase it ran in, whatever the status. The four *_guardrail_results lists GuardrailAudit reads off it are the documented way in. The same trail is on a paused RunState.

to_payload() is how a transport renders all of this as JSON -- one wire shape, owned here rather than re-derived per transport.

Methods:

Name Description
to_state

The paused run's RunState: approve or reject its interruptions, then resume.

to_payload

This run as plain JSON-compatible data: what a Run looks like over the wire.

to_state

to_state() -> RunState

The paused run's RunState: approve or reject its interruptions, then resume.

to_payload

to_payload() -> dict[str, Any]

This run as plain JSON-compatible data: what a Run looks like over the wire.

A transport (runa serve's routes, runa.web, whatever ships next) renders a run by calling this, not by re-listing the fields itself. The shape is a fact about Run, so it belongs where Run is defined: the alternative is each transport re-deriving it, which is how serve came to report a tool_name that Interruption never had.

status is the Run's own, so a caller distinguishes "the agent answered" from "it needs an approval" from "it failed" without parsing prose. trace_id is the handle for runa traces show, which is what makes a production incident debuggable. A paused run carries state, RunState.to_json()'s blob: the one thing that can resolve the pause, handed back through RunState.from_json to resume (see runa.run_state). It is None for any other status.

One-way, unlike RunState.to_json(): a Run holds live objects (an Agent, a Trace's spans) and there is no from_payload. The guardrail audit trail is left out for the same reason RunState omits it -- every guardrail that ran is already a span in the trace trace_id names, tripped or not. Plain dicts only, no transport imported here, so this stays a runtime concern.

RunStream

RunStream(events: AsyncIterator[StreamEvent])

What Agent.run_streamed() returns: iterate it for StreamEvents.

Once iteration ends, run holds the same Run that run() would have returned, paused, completed, or errored.

Wrap events; run is set by the agent once they are exhausted.

Methods:

Name Description
__aiter__

Iterate the run's StreamEvents.

__aiter__

__aiter__() -> AsyncIterator[StreamEvent]

Iterate the run's StreamEvents.

ModelSettings dataclass

ModelSettings(
    temperature: float | None = None,
    top_p: float | None = None,
    max_tokens: int | None = None,
    tool_choice: ToolChoice = None,
    parallel_tool_calls: bool | None = None,
    reasoning_effort: ReasoningEffort = None,
    max_retries: int | None = None,
)

Per-call model parameters; a Model implementation uses whichever of these its API takes.

"Whichever" is a closed set, not a courtesy: every field here reaches both backends' request body, or is named in tests/test_models.py's closure test as one that deliberately doesn't. A setting that configures nothing is the failure check_model_name exists to prevent, one field over -- the bill and the caller's mental model disagree, with no signal anywhere.

Attributes:

Name Type Description
max_retries int | None

How many times to retry a failed request (connection errors, 408/409/429/5xx), on top of

max_retries class-attribute instance-attribute

max_retries: int | None = None

How many times to retry a failed request (connection errors, 408/409/429/5xx), on top of the first attempt. None uses the backend default of 2; 0 disables retrying outright, for an app that does its own. Honored identically by both backends.

Usage dataclass

Usage(
    requests: int = 0,
    input_tokens: int = 0,
    output_tokens: int = 0,
    total_tokens: int = 0,
    input_tokens_details: InputTokensDetails = InputTokensDetails(),
    output_tokens_details: OutputTokensDetails = OutputTokensDetails(),
)

Token usage for one or more model calls.

Agent.run/run_sync accumulate every call's Usage into Agent.usage via .add(), and record the latest one on Agent.last_usage.

Methods:

Name Description
add

Accumulate other's counts into this Usage, in place.

add

add(other: Usage) -> None

Accumulate other's counts into this Usage, in place.

RunState dataclass

RunState(
    agent: AgentLike,
    transcript: Transcript,
    ready_results: list[ConversationItem],
    pending: list[Interruption],
    run_context: RunContext,
    _governance: _Governance,
    trace: Trace,
    approvals: dict[str, bool] = dict(),
    rejection_messages: dict[str, str] = dict(),
    guardrail_results: GuardrailResults = GuardrailResults(),
)

Bases: GuardrailAudit

Enough of a paused run to resume it once its interruptions are approved or rejected.

transcript is the conversation as it stood at the pause (see Transcript): its items end with the assistant message that requested the paused calls. ready_results holds results already computed this turn for calls in that same message that didn't need approval: they're carried forward rather than re-executed on resume.

run_context is the paused run's RunContext, its context and usage so far. How the run governs itself -- the approval ledger, the paused delegates -- is _governance, which a caller reaches only through approve/reject, never directly (see runa._governance).

to_json()/to_string()/from_json()/from_string() let a paused run survive a process restart -- see their docstrings for what is (and isn't) preserved.

Methods:

Name Description
approve

Mark interruption approved; its tool runs when the run is resumed.

reject

Mark interruption rejected; its tool is skipped when the run is resumed.

to_json

Serialize this paused run to a JSON-compatible dict, to persist and resume later.

to_string

As to_json(), but as a JSON string.

from_json

Rebuild a paused RunState from to_json()'s output.

from_string

As from_json(), but from a JSON string produced by to_string().

Attributes:

Name Type Description
paused_message ConversationItem

The assistant message whose calls paused the run: the transcript's last, always.

paused_message property

paused_message: ConversationItem

The assistant message whose calls paused the run: the transcript's last, always.

Where a resumed run picks the turn back up -- it finishes this one message's calls before calling the model again, with approvals/rejection_messages as the decisions about them and ready_results as the results it already has. The turn loop reads the four from here rather than from a resumed-turn shape of its own.

approve

approve(
    interruption: Interruption, *, always: bool = False
) -> None

Mark interruption approved; its tool runs when the run is resumed.

always=True also records a sticky decision on the run's approval ledger: every future call to this tool (by name), in this run or a nested delegate call sharing it (see runa._governance), skips the approval prompt entirely.

reject

reject(
    interruption: Interruption,
    *,
    always: bool = False,
    rejection_message: str | None = None,
) -> None

Mark interruption rejected; its tool is skipped when the run is resumed.

rejection_message, if given, is fed back to the model instead of ApprovalLedger.DEFAULT_REJECTION. always=True sticks the rejection (and message, if any) for every future call to this tool, the same way approve(always=True) does.

to_json

to_json() -> dict[str, Any]

Serialize this paused run to a JSON-compatible dict, to persist and resume later.

Not included: agent and Interruption.agent are recorded by name and re-resolved by from_json/from_string against a fresh agent instance, since a live Agent can't round-trip through JSON; the guardrail audit trail and trace spans aren't included either (see RunState's docstring and _RunStateSchema's).

to_string

to_string() -> str

As to_json(), but as a JSON string.

from_json classmethod

from_json(
    initial_agent: AgentLike, state_json: dict[str, Any]
) -> RunState

Rebuild a paused RunState from to_json()'s output.

initial_agent is a fresh instance of the agent the run started with; the current agent (possibly switched by a handoff before the pause) and each pending interruption's agent are re-resolved from it by name -- a live Agent can't round-trip through JSON, so it was never serialized as an object in the first place. Reading a blob is pure: the tools themselves are resolved by the resumed run, off the shape it builds anyway, so rebuilding a state never reaches out to an agent's MCP servers.

A context that was a dataclass comes back as a plain dict, not reinstantiated as its original class -- there's no type registry to reverse that with. Raises UserError for an unknown schema version, a malformed field, or an agent name that can no longer be found -- never a raw pydantic.ValidationError.

from_string classmethod

from_string(
    initial_agent: AgentLike, state_string: str
) -> RunState

As from_json(), but from a JSON string produced by to_string().

Interruption dataclass

Interruption(
    name: str,
    arguments: str,
    call_id: str,
    agent: AgentLike,
)

One tool call paused on needs_approval, surfaced to the caller to resolve.

name is the tool's own name, which is all a decision needs: the sticky ledger is keyed by it, and the resumed run looks the tool itself up again off the agent's shape. Carrying the FunctionTool here as well would mean from_json had to rebuild one from a name, which for an MCP tool means connecting to the server just to read a deserialized blob.

Which run a paused call belongs to isn't here either: a call surfaced by a .delegate subagent appears in its caller's interruptions as the very same Interruption, so it is the stash of paused delegates that knows whose it is (runa.paused_delegates), not the interruption and not whichever RunState happens to be holding it.

Exceptions

RunaError

RunaError(message: str)

Bases: Exception

Base class for every exception a Runa agent run can end with.

One message, carried as both the exception's args and its .message, and nothing else: what a failed run accumulated is reported on the Run the turn loop returns in its place (run_internal/run_loop._errored), not hung off the exception on its way up. Every subclass is that plus its docstring: the ones built from structured fields (GuardrailTripwireTriggered, DuplicateToolCallError) keep their own __init__ to compose the message, then hand it here like any other.

Store message as both the exception's args and its .message.

MaxTurnsExceeded

MaxTurnsExceeded(message: str)

Bases: RunaError

Raised when a run reaches max_turns without producing a final output.

MaxTokensExceeded

MaxTokensExceeded(message: str)

Bases: RunaError

Raised when a run's accumulated token usage passes Agent.max_tokens.

The spend ceiling to MaxTurnsExceeded's call ceiling: max_turns bounds how many times a run may call the model, this bounds how much those calls may cost. Checked after each model call, so the run stops at the first call that crosses the line rather than before it.

RunTimeout

RunTimeout(message: str)

Bases: RunaError

Raised when a run passes Agent.timeout seconds without finishing.

The wall-clock ceiling max_turns/max_tokens can't give: a single tool call or model response that hangs would otherwise stall a run indefinitely. Like every RunaError this comes back as Run(status="error") rather than propagating, and the partial trace is still exported, so a timed-out run is as inspectable as a failed one.

ModelBehaviorError

ModelBehaviorError(message: str)

Bases: RunaError

Raised when the model does something a Model implementation can't make sense of.

For example: calling a tool that isn't in the request, or returning malformed tool-call JSON.

UserError

UserError(message: str)

Bases: RunaError

Raised when the caller has misused the framework itself (bad config, missing credentials).

UnexpectedError

UnexpectedError(exc: BaseException)

Bases: RunaError

Raised in place of an exception a run ended with that is none of Runa's own kinds.

A Session that can't read its database, a Hooks method with a bug in it, a Compactor that divides by zero: the extension points a run calls are application code, and the run boundary owes them the same contract it gives a tripped guardrail (Run(status="error"), with the trace around the failure) rather than one that holds only for failures Runa itself declared. The message names the original's type as well as its text, since neither alone identifies it, and the original is logged with its traceback where it was caught, so nothing about it is lost by coming back as a result.

Not raised for OperatorError (a sibling root, deliberately let through) nor for anything deriving from BaseException alone -- asyncio.CancelledError above all, which a caller asking a run to stop must get back as itself.

Name exc's type and message, since neither alone identifies what went wrong.

Serving

create_app

create_app(root: Path, *, api_key: str | None) -> FastAPI

Build the app serving the agents of the project at root.

api_key is the bearer token every route but /health requires; None disables the check entirely (runa serve --no-auth). create_app takes it as a parameter rather than reading the environment itself, the same way this takes root instead of assuming cwd, so a test (or an app embedding this) can be explicit: resolve_api_key is the conventional way to fill it in.

root also goes to project.use, which is how this server says it reads that app: its state resolves against root and its main.py is the one imported, so an embedded app served from outside its own directory keeps everything together -- a session route and the memory an agent writes during that same request resolve to one file, not two.

resolve_api_key

resolve_api_key(*, no_auth: bool) -> str | None

The token a server should require, read from RUNA_API_KEY, or None to serve open.

Refuses to start unauthenticated by accident: an agent endpoint costs money per call, so "nobody set the variable" has to be an error rather than a silently open door. no_auth=True (runa serve --no-auth, or the argument in a generated asgi.py) makes the same choice explicit and is then perfectly fine for local use, or behind something that already authenticates.

create_app takes the resolved token rather than calling this itself, so the environment is read in exactly one place a deployment can see, and a test can pass a token outright.

RunRequest

Bases: BaseModel

One turn for an agent to run: a new message, or a paused run and the decisions resolving it.

Exactly one of message/state -- the two are the same argument of Agent.run, which takes a message or the RunState of a paused run, so a body carrying both (or neither) is a 422 rather than a guess about which one was meant.

Decision

Bases: BaseModel

What a human decided about one Interruption of a paused run.

call_id is the one the paused payload reported, and the only handle a client needs: the tool, its arguments and the agent it belongs to are the server's to look back up off the state. always and rejection_message are RunState.approve/.reject's own keywords, named the same here on purpose -- the wire shape of an approval is the Python one.

Hooks

Hooks

Lifecycle callbacks for a run; every method is a no-op unless overridden.

Scope is where an instance is put, not which class it is:

  • Agent.run(hooks=...) fires for every agent in that one run, the ones reached by handoff/delegate included. Use it for cross-cutting concerns -- metrics, a single audit log for the whole run.
  • An Agent subclass's hooks attribute fires only for that agent, in every run it takes part in. Use it for a concern that belongs to one agent's identity.

Both scopes fire for the same event, run-scoped first, and read the same arguments. on_handoff is the one event an agent-scoped instance doesn't see for every turn of its own agent: it notifies the target, the agent the run was handed to, and never the sender.

LoggingHooks (below) is what a run with no hooks= gets.

Methods:

Name Description
on_agent_start

Called right before agent starts (or resumes, after a handoff) running.

on_agent_end

Called once agent has produced its final output for the run.

on_handoff

Called when from_agent hands the run off to to_agent.

on_tool_start

Called right before tool runs.

on_tool_end

Called once tool has returned (or raised) result.

on_llm_start

Called right before agent calls the model.

on_llm_end

Called once the model call has returned response.

on_agent_start async

on_agent_start(
    context: RunContext[TContext], agent: AgentLike
) -> None

Called right before agent starts (or resumes, after a handoff) running.

on_agent_end async

on_agent_end(
    context: RunContext[TContext],
    agent: AgentLike,
    output: Any,
) -> None

Called once agent has produced its final output for the run.

on_handoff async

on_handoff(
    context: RunContext[TContext],
    from_agent: AgentLike,
    to_agent: AgentLike,
) -> None

Called when from_agent hands the run off to to_agent.

on_tool_start async

on_tool_start(
    context: RunContext[TContext],
    agent: AgentLike,
    tool: FunctionTool,
) -> None

Called right before tool runs.

on_tool_end async

on_tool_end(
    context: RunContext[TContext],
    agent: AgentLike,
    tool: FunctionTool,
    result: object,
) -> None

Called once tool has returned (or raised) result.

on_llm_start async

on_llm_start(
    context: RunContext[TContext],
    agent: AgentLike,
    system_prompt: str | None,
    input_items: list[ConversationItem],
) -> None

Called right before agent calls the model.

on_llm_end async

on_llm_end(
    context: RunContext[TContext],
    agent: AgentLike,
    response: Any,
) -> None

Called once the model call has returned response.

LoggingHooks

Bases: Hooks[Any]

Logs each lifecycle event through the standard logging module, under the runa logger.

Agent.run/run_sync use an instance of this as the default hooks, so every run is logged without the caller having to ask; passing an explicit hooks overrides it. Assigned to an Agent subclass's hooks instead, it logs that one agent's events.

Methods:

Name Description
on_agent_start

Log that agent is about to run.

on_agent_end

Log that agent finished: its name at INFO, its output at DEBUG if policy allows.

on_handoff

Log a handoff as from_agent -> to_agent.

on_tool_start

Log that tool is about to run.

on_tool_end

Log that tool finished; its result only at DEBUG, under the tracing policy.

on_llm_start

Log that agent is about to call the model.

on_llm_end

Log that agent's model call returned.

on_agent_start async

on_agent_start(
    context: RunContext[Any], agent: AgentLike
) -> None

Log that agent is about to run.

on_agent_end async

on_agent_end(
    context: RunContext[Any], agent: AgentLike, output: Any
) -> None

Log that agent finished: its name at INFO, its output at DEBUG if policy allows.

on_handoff async

on_handoff(
    context: RunContext[Any],
    from_agent: AgentLike,
    to_agent: AgentLike,
) -> None

Log a handoff as from_agent -> to_agent.

on_tool_start async

on_tool_start(
    context: RunContext[Any],
    agent: AgentLike,
    tool: FunctionTool,
) -> None

Log that tool is about to run.

on_tool_end async

on_tool_end(
    context: RunContext[Any],
    agent: AgentLike,
    tool: FunctionTool,
    result: object,
) -> None

Log that tool finished; its result only at DEBUG, under the tracing policy.

on_llm_start async

on_llm_start(
    context: RunContext[Any],
    agent: AgentLike,
    system_prompt: str | None,
    input_items: list[ConversationItem],
) -> None

Log that agent is about to call the model.

on_llm_end async

on_llm_end(
    context: RunContext[Any],
    agent: AgentLike,
    response: Any,
) -> None

Log that agent's model call returned.

Streaming

StreamEvent module-attribute

StreamEvent = (
    RawResponsesStreamEvent
    | RunItemStreamEvent
    | AgentUpdatedStreamEvent
)

AgentUpdatedStreamEvent dataclass

AgentUpdatedStreamEvent(
    new_agent: AgentLike,
    type: Literal[
        "agent_updated_stream_event"
    ] = "agent_updated_stream_event",
)

A handoff switched the agent running this turn.

RawResponsesStreamEvent dataclass

RawResponsesStreamEvent(
    data: StreamDelta,
    type: Literal[
        "raw_response_event"
    ] = "raw_response_event",
)

A raw, provider-shaped fragment of a streamed response, passed through as-is.

RunItemStreamEvent dataclass

RunItemStreamEvent(
    name: Literal[
        "message_output_created",
        "tool_called",
        "tool_output",
        "handoff_occured",
    ],
    item: ConversationItem,
    type: Literal[
        "run_item_stream_event"
    ] = "run_item_stream_event",
)

One completed item produced mid-stream: a message, a tool call, a tool's output, ...

Evaluation

Case dataclass

Case(
    input: str,
    expected: str | None = None,
    expected_tool: str | None = None,
    context: list[str] | None = None,
    metadata: dict[str, Any] | None = None,
)

One eval case: an input, plus optional evidence of what a good run looks like.

expected enables answer-correctness grading; expected_tool enables the deterministic "was it called" (tool-correctness) check; context (retrieval passages the agent should be grounded in) enables faithfulness grading. Any combination is valid, including none of them, in which case only task completion and answer relevance run.

CaseReport dataclass

CaseReport(
    index: int,
    case: Case,
    run: Run,
    results: list[EvaluationResult] = list(),
)

One Case's Run plus every EvaluationResult graded against it.

Attributes:

Name Type Description
id str

A short, stable identifier for this case, e.g. case_23.

passed bool

Whether every result passed (SKIPPED doesn't count against it).

failure_reason str | None

The first non-passing result's reason, or None if the case passed.

id property

id: str

A short, stable identifier for this case, e.g. case_23.

passed property

passed: bool

Whether every result passed (SKIPPED doesn't count against it).

failure_reason property

failure_reason: str | None

The first non-passing result's reason, or None if the case passed.

Dataset

Dataset(cases: Iterable[Case])

An ordered, iterable collection of Case.

Wrap cases in a list, preserving order.

Methods:

Name Description
__iter__

Iterate over cases in order.

__len__

Return the number of cases.

__getitem__

Return the case at index.

from_jsonl

Load one Case per non-blank line of a JSONL file.

__iter__

__iter__() -> Iterator[Case]

Iterate over cases in order.

__len__

__len__() -> int

Return the number of cases.

__getitem__

__getitem__(index: int) -> Case

Return the case at index.

from_jsonl classmethod

from_jsonl(path: str | Path) -> Dataset

Load one Case per non-blank line of a JSONL file.

Each line's JSON object is passed as keyword arguments to Case, so a line only needs the fields it uses, e.g. {"input": "...", "expected": "..."}. A bare JSON string is shorthand for an input-only case: "..." is {"input": "..."}.

EvaluationResult dataclass

EvaluationResult(
    metric: str,
    status: Status,
    reason: str,
    score: float | None = None,
)

One metric's verdict on one case: deterministic or semantic, same shape either way.

Attributes:

Name Type Description
passed bool

Whether this result counts toward a case passing (SKIPPED doesn't count against it).

passed property

passed: bool

Whether this result counts toward a case passing (SKIPPED doesn't count against it).

Report dataclass

Report(
    agent_name: str,
    cases: list[CaseReport],
    baseline: dict[str, bool] | None = None,
)

The aggregated result of agent.evaluate(dataset), under the agent's declared name.

Methods:

Name Description
__str__

Render the report roughly as report.score/report.metrics/report.failures show.

Attributes:

Name Type Description
baseline dict[str, bool] | None

Each input of the agent's previous run mapped to whether it passed, None if first run.

passed list[CaseReport]

Every case that passed.

failed list[CaseReport]

Every case that didn't pass.

failures list[CaseReport]

Alias for failed, matching the failure-listing use case.

regressions list[CaseReport]

Every failed case whose input passed in the previous run (see baseline).

pass_rate float

The fraction of cases that passed, or 0.0 for an empty dataset.

metrics dict[str, float]

The mean score of each metric, across the cases where it wasn't skipped or errored.

score float

The headline score: mean task completion, falling back to the overall pass rate.

baseline class-attribute instance-attribute

baseline: dict[str, bool] | None = None

Each input of the agent's previous run mapped to whether it passed, None if first run.

passed property

passed: list[CaseReport]

Every case that passed.

failed property

failed: list[CaseReport]

Every case that didn't pass.

failures property

failures: list[CaseReport]

Alias for failed, matching the failure-listing use case.

regressions property

regressions: list[CaseReport]

Every failed case whose input passed in the previous run (see baseline).

pass_rate property

pass_rate: float

The fraction of cases that passed, or 0.0 for an empty dataset.

metrics property

metrics: dict[str, float]

The mean score of each metric, across the cases where it wasn't skipped or errored.

score property

score: float

The headline score: mean task completion, falling back to the overall pass rate.

__str__

__str__() -> str

Render the report roughly as report.score/report.metrics/report.failures show.

Every metric the cases scored is shown, labeled and ordered by its declaration in eval/evaluation/metrics.py: there is no list here to forget a metric from, which is how a metric could decide whether cases passed while never appearing above their failures.

Status

Bases: StrEnum

A metric's outcome for one case.

SKIPPED means the metric didn't apply (e.g. faithfulness with no retrieval context); ERROR means the check itself failed to run (e.g. the judge model call raised). Neither is a score of 0 or 1: a PASS/ FAIL verdict claims to have actually measured something.

Tracing

observe

observe(
    *,
    capture_inputs: bool | None = None,
    capture_outputs: bool | None = None,
    redact: list[str] | None = None,
    redactor: Callable[[Any], Any] | None = None,
    max_input_bytes: int | None = None,
    max_output_bytes: int | None = None,
    max_tool_result_bytes: int | None = None,
    exporter: TraceExporter
    | list[TraceExporter]
    | None = None,
)

Configure what tracing captures: this call for the process, .scoped(...) for one task.

observe(capture_inputs=False) applies the setting to the process-wide default and leaves it there: it is what an application configures itself with, once, at startup. with observe.scoped(capture_inputs=False): applies it to the current task only, and restores whatever that task saw before on exit.

The two scopes are two call sites rather than one name that infers which was meant from whether a with follows it, the way Guardrail.input/.output and Agent.handoff/ .delegate name the choice they make. The inference needed a constructor that wrote the process-wide default for __enter__ to then undo, which made the object single-use (it captured the default it would restore) and the write itself invisible to anyone reading the call site. Naming the scope instead means nothing is captured and nothing is undone.

The scopes differ in reach, deliberately. A block that changed the policy process-wide would mean one of runa.serve's concurrent requests deciding what every other in-flight request captures, so .scoped is a ContextVar. That costs the opposite: a task does not see an override installed by a task that started after it, so the plain call -- not a block held open around the whole application -- is what configures a whole deployment.

Apply the given overrides to the process-wide tracing policy, permanently.

There is nothing to hold on to afterwards: the call is the configuration. An instance is not a context manager, so a with observe(...) written out of habit is a TypeError at that line -- pointing at observe.scoped(...) -- rather than a block whose policy silently outlives it.

Methods:

Name Description
scoped

A with block that applies the given overrides to the current task only.

scoped staticmethod

scoped(
    *,
    capture_inputs: bool | None = None,
    capture_outputs: bool | None = None,
    redact: list[str] | None = None,
    redactor: Callable[[Any], Any] | None = None,
    max_input_bytes: int | None = None,
    max_output_bytes: int | None = None,
    max_tool_result_bytes: int | None = None,
    exporter: TraceExporter
    | list[TraceExporter]
    | None = None,
) -> _ScopedPolicy

A with block that applies the given overrides to the current task only.

Nothing is applied until the block is entered, and everything is restored when it exits. An inner block overrides what it names and inherits the rest from the block around it.

Span dataclass

Span(
    id: str,
    trace_id: str,
    parent_id: str | None,
    name: str,
    type: SpanType,
    start_time: float,
    end_time: float | None = None,
    status: SpanStatus = "ok",
    attributes: dict[str, Any] = dict(),
    input: Any = None,
    output: Any = None,
    error: str | None = None,
)

One traceable operation: an agent turn, an LLM call, a tool call, a handoff, ...

input/output/error stay None when that data isn't available or was withheld by the active privacy policy (see tracing.config.observe); missing data is never fabricated.

Attributes:

Name Type Description
duration float | None

Seconds between start_time and end_time, or None while the span is still open.

duration property

duration: float | None

Seconds between start_time and end_time, or None while the span is still open.

SpanRow dataclass

SpanRow(
    span: Span,
    label: str,
    name: str,
    duration: str,
    tokens: str | None,
    children: tuple[SpanRow, ...],
)

One span as a person reads it: the tree's shape and its labels, no characters chosen yet.

What Trace.walk yields. It carries everything a renderer needs to draw a row and nothing about how to draw it, so runa traces show can turn these into box-drawing lines and runa ui into nested <div>s without either re-deriving which spans are whose children, what a span type is called, or what a handoff is named once the model's prefix comes off.

span is the raw Span for anything a renderer wants that isn't person-readable yet: its input/output/attributes/status/error.

Attributes:

Name Type Description
hands_off bool

Whether every later sibling is the work of the agent this row handed off to.

hands_off property

hands_off: bool

Whether every later sibling is the work of the agent this row handed off to.

run_loop.py parents a handoff's aftermath to the same turn-level span as the handoff itself, so a row's position in the sibling sequence is the only place that boundary exists. A renderer that groups siblings visually needs it; one that only lists them can ignore it.

Trace dataclass

Trace(
    id: str,
    name: str,
    start_time: float,
    end_time: float | None = None,
    session_id: str | None = None,
    spans: list[Span] = list(),
    metadata: dict[str, Any] = dict(),
)

One logical agent execution: a start/end time, metadata, and its Span tree.

spans is a flat list: each Span.parent_id (or None, for a root span) is what gives it shape; __str__ walks that structure to render the tree shown in the "Human-readable trace representation" section of the design.

Methods:

Name Description
walk

This trace's root spans as SpanRows, each carrying its own ordered children.

__str__

Render this trace as a box-drawing tree of its spans, roughly matching the design doc.

Attributes:

Name Type Description
duration float | None

Seconds between start_time and end_time, or None while the trace is still open.

status SpanStatus

"error" if any span in this trace errored, "ok" otherwise.

errors list[Span]

Every span in this trace whose status is "error".

elapsed str

duration as a person reads it: "1.23s", or "..." while the trace is still open.

duration property

duration: float | None

Seconds between start_time and end_time, or None while the trace is still open.

status property

status: SpanStatus

"error" if any span in this trace errored, "ok" otherwise.

errors property

errors: list[Span]

Every span in this trace whose status is "error".

elapsed property

elapsed: str

duration as a person reads it: "1.23s", or "..." while the trace is still open.

walk

walk() -> tuple[SpanRow, ...]

This trace's root spans as SpanRows, each carrying its own ordered children.

The only traversal of the span tree. spans is flat and shaped solely by parent_id, so recovering the tree -- and ordering siblings by start time -- is a fact about traces, and it is learned here rather than again in every surface that shows one.

__str__

__str__() -> str

Render this trace as a box-drawing tree of its spans, roughly matching the design doc.

TraceExporter

Bases: Protocol

Exports a finished Trace somewhere: SQLite, the console, or a caller's own backend.

export is synchronous, and blocking in it is expected: an exporter is plain code doing plain I/O -- a file write, an HTTP POST, db/pool.run_sync over an async driver as PostgresTraceStore does -- and which thread it runs on is this module's decision rather than the exporter's. A finishing run awaits export_trace_off_loop, so the write happens on tracing's own thread while the loop goes on serving every other request; a caller that is synchronous all the way down (a tracing.trace block, runa traces, runa ui) calls export_trace inline on the thread it already has. TraceStore is synchronous for the same reason.

Methods:

Name Description
export

Export trace. Exceptions are caught by the caller; tracing must never fail open.

export

export(trace: Trace) -> None

Export trace. Exceptions are caught by the caller; tracing must never fail open.

ConsoleExporter

Prints every finished trace's human-readable tree (Trace.__str__) to stdout.

Methods:

Name Description
export

Print trace.

export

export(trace: Trace) -> None

Print trace.

SQLiteExporter

SQLiteExporter(path: Path | None = None)

Bases: StoreExporter

The local exporter: persists every finished trace to db/runa.db (tracing/sqlite.py).

Export to path, or to wherever runa.db says this deployment's local file is.

StoreExporter

StoreExporter(store: TraceStore)

A TraceExporter that writes every finished trace to a TraceStore.

The whole implementation, for every persistent exporter: which store it holds is decided once, when the exporter is built, instead of per write. SQLiteExporter and PostgresExporter are this class with a store already chosen, and _local_exporter builds one over whatever runa.db.traces() resolved.

Deciding once is a fix, not a tidy-up. The old SQLiteExporter.export called a module-level save_trace that asked runa.db for the backend all over again, so the exporter a deployment had installed and the store its traces landed in were two separate answers to one question.

Store where finished traces are written.

Methods:

Name Description
export

Persist trace. Exceptions are caught by the caller; tracing never fails a run.

export

export(trace: Trace) -> None

Persist trace. Exceptions are caught by the caller; tracing never fails a run.

TraceStore

Bases: Protocol

One deployment's trace history: write one, read one, read the recent ones.

No inheritance required, every adapter satisfies this by matching shape. list is the only query: get_errors-style convenience readers were a second shape over the same filter, so callers pass status="error" instead.

Methods:

Name Description
save

Persist trace and every span in it, replacing any existing one with the same id.

get

Look up one trace by id, with every span it has, or None if this store has none.

list

Return the most recent limit traces, newest first, optionally filtered.

save

save(trace: Trace) -> None

Persist trace and every span in it, replacing any existing one with the same id.

get

get(trace_id: str) -> Trace | None

Look up one trace by id, with every span it has, or None if this store has none.

list

list(
    *,
    limit: int = 50,
    agent: str | None = None,
    status: str | None = None,
    session_id: str | None = None,
) -> list[Trace]

Return the most recent limit traces, newest first, optionally filtered.

agent matches Trace.name (the workflow name Agent.run sets to the agent's class name); session_id matches the Session a session-backed run was passed, when it was passed one; status is "ok" or "error".

Ties on start_time break by trace id descending, in every adapter, so two traces that started in the same instant list in one order whatever the deployment.

Storage

traces

traces() -> TraceStore

This deployment's TraceStore: where finished traces are written and read back.

sessions

sessions() -> SessionStore

This deployment's SessionStore: the read side of its conversation history.

evals

evals() -> EvalStore

This deployment's EvalStore: every agent.evaluate() run and its baseline.

session

session(
    session_id: str, *, user_id: str | None = None
) -> Session

This deployment's session store for session_id.

user_id scopes the session's automatic memory, if its agent has any; see Session.

cache

cache() -> Cache

This deployment's persistent Cache: a table in whichever database it already has.

No second variable and no second service: the cache follows RUNA_DATABASE_URL like everything else, which under memory:// means the same MemoryCache you can also ask for by name. MemoryCache() stays the explicit choice for a cache that should die with the process while everything else persists, and RedisCache(url) for a deployment that wants its hot keys off the query path.

shared_url

shared_url() -> str | None

The Postgres URL this deployment shares, or None when its state is not shared.

None is the answer for an unset RUNA_DATABASE_URL and for an explicit sqlite:// or memory:// one, so a caller only ever has to branch on "shared or not", never on which of the local forms said so. Raises InvalidDatabaseURL for a URL that names none of the three: a typo in a deployment's database URL should stop the app, not silently route a replica's history to its own disk.

sqlite_path

sqlite_path() -> Path

Where the local SQLite file lives: RUNA_DATABASE_URL's path, or db/runa.db.

Three slashes is a relative path and four is absolute, the form SQLAlchemy and every DATABASE_URL convention already use, so sqlite:///data/runa.db relocates the file without a db_path= argument on five different constructors.

A relative path is taken relative to the project project.use named, so under project.use(Path("../other-app")) this is ../other-app/db/runa.db -- the same file that app's own runa chat writes to. Absolute paths and an unset project (an app running in its own directory) are unaffected.

ephemeral

ephemeral() -> bool

Whether this deployment keeps its state in this process only (memory://).

The honest answer to "where does state live" for a test suite: one monkeypatch.setenv swaps all six concerns onto their in-process adapters at once, with no live Postgres and no temporary file, and runa.db stays the only thing that decided.

SessionStore

Bases: Protocol

The read side of one deployment's conversation history.

Two questions, which is every question runa sessions, runa chat and the Sessions pages ask: which sessions are there, and what was said in one of them. No inheritance required, every adapter satisfies this by matching shape, the same escape hatch as Cache's.

Methods:

Name Description
listing

Return this deployment's sessions, most recently updated first.

messages

Return session_id's messages, oldest first.

listing

listing(
    *, agent: str | None = None
) -> list[SessionSummary]

Return this deployment's sessions, most recently updated first.

agent keeps only that agent's sessions, matching a session id that is either exactly agent (the old, pre-session-per-chat default) or starts with f"{agent}-" (cli/chat.py's current scheme), so --continue/--resume find history under either.

Ties on updated_at break by session id descending, in every adapter. Two sessions written in the same second used to list in a different order depending on the backend, which made runa chat --list and the Sessions page disagree between a local deployment and a shared one.

messages

messages(session_id: str) -> list[SessionMessage]

Return session_id's messages, oldest first.

Raises SessionNotFound if this deployment has no session row for session_id, rather than returning an empty list: "no such session" and "a session with nothing in it" are different answers, and runa chat --show prints different things for them.

EvalStore

Bases: Protocol

One deployment's eval history: every agent.evaluate() run and the cases it graded.

No inheritance required, every adapter satisfies this by matching shape.

Methods:

Name Description
save

Persist report as one run plus one row per case, returning the new run's id.

get

Look up one run by id, with every case it graded, or None if it doesn't exist.

list

Return the most recent limit runs, newest first, without their cases.

baseline

Map each input of agent_name's latest run to whether it passed.

save

save(report: Report) -> int

Persist report as one run plus one row per case, returning the new run's id.

get

get(run_id: int) -> EvalRun | None

Look up one run by id, with every case it graded, or None if it doesn't exist.

list

list(*, limit: int = 50) -> list[EvalRun]

Return the most recent limit runs, newest first, without their cases.

baseline

baseline(
    agent_name: str, *, before: int | None = None
) -> dict[str, bool] | None

Map each input of agent_name's latest run to whether it passed.

before looks at the latest run older than that run id instead, the baseline a past run was compared against -- strictly older, so before the newest run skips it. None when there's no such run. Keyed by input rather than index, so reordering, adding, or removing cases between runs still lines the rest up.

Exceptions

RunaError

RunaError(message: str)

Bases: Exception

Base class for every exception a Runa agent run can end with.

One message, carried as both the exception's args and its .message, and nothing else: what a failed run accumulated is reported on the Run the turn loop returns in its place (run_internal/run_loop._errored), not hung off the exception on its way up. Every subclass is that plus its docstring: the ones built from structured fields (GuardrailTripwireTriggered, DuplicateToolCallError) keep their own __init__ to compose the message, then hand it here like any other.

Store message as both the exception's args and its .message.

MaxTurnsExceeded

MaxTurnsExceeded(message: str)

Bases: RunaError

Raised when a run reaches max_turns without producing a final output.

ModelBehaviorError

ModelBehaviorError(message: str)

Bases: RunaError

Raised when the model does something a Model implementation can't make sense of.

For example: calling a tool that isn't in the request, or returning malformed tool-call JSON.

UserError

UserError(message: str)

Bases: RunaError

Raised when the caller has misused the framework itself (bad config, missing credentials).

UnexpectedError

UnexpectedError(exc: BaseException)

Bases: RunaError

Raised in place of an exception a run ended with that is none of Runa's own kinds.

A Session that can't read its database, a Hooks method with a bug in it, a Compactor that divides by zero: the extension points a run calls are application code, and the run boundary owes them the same contract it gives a tripped guardrail (Run(status="error"), with the trace around the failure) rather than one that holds only for failures Runa itself declared. The message names the original's type as well as its text, since neither alone identifies it, and the original is logged with its traceback where it was caught, so nothing about it is lost by coming back as a result.

Not raised for OperatorError (a sibling root, deliberately let through) nor for anything deriving from BaseException alone -- asyncio.CancelledError above all, which a caller asking a run to stop must get back as itself.

Name exc's type and message, since neither alone identifies what went wrong.

GuardrailTripwireTriggered

GuardrailTripwireTriggered(
    guardrail_result: GuardrailResult,
)

Bases: RunaError

Raised when a guardrail's tripwire trips, whichever Phase it ran in.

One exception for the 2x2 rather than four classes that differed only in which phase they named in their message: phase says where it tripped, and guardrail/output are the entry and the verdict guardrail_result pairs up.

Store the triggering result, and name its phase and guardrail in the message.

DuplicateToolCallError

DuplicateToolCallError(call_id: str, tool_name: str)

Bases: RunaError

Raised when a tool-call id that already executed once is submitted for execution again.

Guards against silently re-running a tool -- e.g. resuming the same RunState twice, or a model retry that reuses a call id.

Store the offending call_id/tool_name and build a message from them.