Runtime¶

Long-lived resource container. Create once at startup, share across an application. Holds providers, tools, budget, and trace configuration.

The methods listed below are part of the v1.0.0 stable surface (stability spec §1.2). Other public attributes on Runtime (e.g. make_llm_node, connect_mcp) work today but are not stability-promised.

`Runtime`¶

Runtime ¶

Arcana Agent Runtime -- create once, use many times.

Holds long-lived resources: - Provider connections (gateway registry, with automatic fallback chain) - Tool registry + gateway - Trace backend - Default budget policy - Default engine config

Provider fallback behavior

When multiple providers are registered (e.g. providers={"deepseek": "sk-xxx", "openai": "sk-yyy"}), the runtime automatically builds a fallback chain based on registration order (dict key order). The first provider is primary; subsequent providers serve as fallbacks. If the primary provider fails with a retryable error after exhausting retries, the request is automatically forwarded to the next provider in the chain.

Use runtime.fallback_order to inspect the resolved order.

Parameters:

Name	Type	Description	Default
`providers`	`dict[str, str \| dict[str, str]] \| None`	Mapping of provider name to API key. The first key becomes the default provider; remaining keys form the automatic fallback chain in insertion order.	`None`
`namespace`	`str \| None`	Optional namespace for tenant isolation. When set, memory and trace are partitioned so that multiple Runtimes sharing the same backing stores don't see each other's data. When `None` (the default), behavior is unchanged.	`None`

budget_remaining_usd `property` ¶

budget_remaining_usd: float | None

Remaining USD budget, or None if no budget limit is set.

budget_used_usd `property` ¶

budget_used_usd: float

Total USD spent across all runs.

tokens_used `property` ¶

tokens_used: int

Total tokens used across all runs.

tokens_remaining `property` ¶

tokens_remaining: int | None

Remaining token budget, or None if no token limit is set.

init ¶

__init__(*, providers: dict[str, str | dict[str, str]] | None = None, tools: list[Callable] | None = None, guardrails: list[Callable] | None = None, mcp_servers: list[MCPServerConfig] | None = None, budget: Budget | None = None, trace: bool = False, memory: bool = False, memory_budget_tokens: int = 800, config: RuntimeConfig | None = None, namespace: str | None = None, context_strategy: Any = None) -> None

run `async` ¶

run(goal: str, *, engine: str = 'conversation', max_turns: int | None = None, budget: Budget | None = None, tools: list[Callable] | None = None, skills: list[str] | None = None, auto_route: bool = True, response_format: type[BaseModel] | None = None, images: list[str] | None = None, input_handler: Callable | None = None, system: str | None = None, context: dict[str, Any] | str | None = None, provider: str | None = None, model: str | None = None, on_parse_error: Callable | None = None) -> RunResult

Run a task to completion.

Parameters:

Name	Type	Description	Default
`goal`	`str`	What to accomplish	required
`engine`	`str`	"conversation" (V2, default) or "adaptive" (V1)	`'conversation'`
`max_turns`	`int \| None`	Override default max turns	`None`
`budget`	`Budget \| None`	Override default budget for this run	`None`
`tools`	`list[Callable] \| None`	Additional tools for this run only	`None`
`skills`	`list[str] \| None`	Optional skill names to force into this run's working set when those skills exist in `RuntimeConfig.skill_paths`.	`None`
`auto_route`	`bool`	Enable intent-routing fast paths. When False, the conversation engine skips classifier-based direct-answer routing and enters the normal turn loop.	`True`
`response_format`	`type[BaseModel] \| None`	Pydantic model class for structured output. When provided, the LLM response is parsed and validated against this model. `result.output` will be an instance of the model rather than a plain string. Tools remain available — the agent uses tools during reasoning and returns structured output on the final turn.	`None`
`images`	`list[str] \| None`	Optional list of image inputs (URLs, file paths, or data URIs) to include in the initial user message.	`None`
`input_handler`	`Callable \| None`	Optional callback for the ask_user built-in tool. Can be sync or async. When None, the LLM receives a fallback message and proceeds with best judgment.	`None`
`system`	`str \| None`	System prompt for this run. Overrides `RuntimeConfig.system_prompt`. When None, falls back to the config value or the engine's default.	`None`
`context`	`dict[str, Any] \| str \| None`	Additional context for the agent. A dict is serialized as JSON; a string is used as-is. Injected into the goal as a `<context>` block so the agent can reference prior step outputs or external data.	`None`
`provider`	`str \| None`	Override the default provider for this run only.	`None`
`model`	`str \| None`	Override the default model for this run only.	`None`
`on_parse_error`	`Callable \| None`	Optional callback invoked when the LLM returns text that cannot be parsed into the `response_format` model. Receives `(raw_string, error)` where error is a `json.JSONDecodeError` or `pydantic.ValidationError`. Return a fixed `BaseModel` instance to recover, or `None` to preserve the failure. Supports async. Does NOT fire for provider-level rejections (e.g. the provider does not support `json_schema` mode) -- those surface as `ProviderError` and are handled by provider capability detection / auto-downgrade.	`None`

run_batch `async` ¶

run_batch(tasks: list[dict[str, Any]], *, concurrency: int = 5) -> BatchResult

Run multiple independent tasks concurrently.

Each task dict must have a "goal" key and may include any keyword arguments accepted by :meth:run (tools, system, provider, model, response_format, etc.).

Individual failures do not crash the batch -- the corresponding RunResult will have success=False.

Parameters:

Name	Type	Description	Default
`tasks`	`list[dict[str, Any]]`	List of task dicts, each with `"goal"` key and optional kwargs matching `run()` parameters.	required
`concurrency`	`int`	Maximum number of concurrent runs (default 5).	`5`

Returns:

Type	Description
`BatchResult`	BatchResult with all results preserving input order.

chat `async` ¶

chat(*, system_prompt: str | None = None, max_turns_per_message: int = 10, budget: Budget | None = None, input_handler: Callable | None = None, max_history: int | None = None) -> AsyncGenerator[ChatSession, None]

Create a multi-turn chat session.

Unlike run() (single goal -> result), chat() maintains conversation history across multiple user messages. Each send() is one conversation turn where the agent may use tools before responding.

Parameters:

Name	Type	Description	Default
`max_history`	`int \| None`	Maximum number of non-system messages to retain. When set, older non-system messages are trimmed after each `send()` / `stream()` call. System messages are always preserved. `None` (default) means unlimited -- no trimming.	`None`

Usage::

async with runtime.chat() as c:
    r = await c.send("Hello")
    r = await c.send("Tell me more about X")
    print(c.total_cost_usd)

chain `async` ¶

chain(steps: list[ChainStep | list[ChainStep]], *, input: str = '', budget: Budget | None = None) -> ChainResult

Run a pipeline of agent steps with optional parallel branches.

Each step's output is automatically passed as context to the next. Use nested lists for parallel execution::

steps=[
    ChainStep(name="filter", ...),
    [ChainStep(name="classify", ...), ChainStep(name="analyze", ...)],
    ChainStep(name="integrate", ...),
]

Parameters:

Name	Type	Description	Default
`steps`	`list[ChainStep \| list[ChainStep]]`	Ordered list. Each element is a `ChainStep` (sequential) or a `list[ChainStep]` (parallel branch).	required
`input`	`str`	Initial input text fed as context to the first step	`''`
`budget`	`Budget \| None`	Shared budget across all steps	`None`

collaborate ¶

collaborate(*, budget: Budget | None = None, cognitive_primitives: list[str] | None = None, channel_history_limit: int | None = None) -> AgentPool

Create a collaboration pool for multi-agent communication.

Returns an :class:AgentPool that provides shared communication primitives. The pool itself is an async context manager — enter it with async with to get automatic cleanup on exit. The user is responsible for orchestration; the framework does not impose topology, turn order, or stop conditions.

Parameters:

Name	Type	Description	Default
`budget`	`Budget \| None`	Optional pool-level budget; all agents in the pool share one `BudgetTracker`.	`None`
`cognitive_primitives`	`list[str] \| None`	Pool-level default for :ref:`cognitive primitives <cognitive-primitives>` (v0.7.0+). Each :meth:`AgentPool.add` call inherits this list unless it supplies its own `cognitive_primitives` argument. `None` (default) falls back to `RuntimeConfig.cognitive_primitives`.	`None`
`channel_history_limit`	`int \| None`	Optional bound on the pool `Channel`'s retained message history (v0.8.1+). `None` (default) keeps unbounded history -- fine for short-lived pools, but long-running pools should set a positive `int` to bound memory. `0` disables history retention. Negative values raise `ValueError`. Delivery queues are unaffected; only the read-only `channel.history` view is bounded.	`None`

Usage::

async with runtime.collaborate(cognitive_primitives=["pin"]) as pool:
    planner = pool.add("planner", system="You plan")
    # executor opts out even though the pool enables pin
    executor = pool.add("executor", system="You execute",
                        cognitive_primitives=[])

    plan = await planner.send("Make a plan for: ...")
    result = await executor.send(f"Execute: {plan.content}")

This is a synchronous factory. Do not await the return value.

session `async` ¶

session(*, engine: str = 'conversation', max_turns: int | None = None, budget: Budget | None = None, tools: list[Callable] | None = None, system: str | None = None) -> AsyncGenerator[Session, None]

Create a session for manual control.

Usage

async with runtime.session() as s: result = await s.run("Do something") print(s.state) print(s.trace_events)

close `async` ¶

close() -> None

Cleanup runtime resources.

on ¶

on(event: str, callback: EventCallback) -> Runtime

Subscribe to runtime events. Returns self for chaining.

Events

"run_start": (run_id: str, goal: str) "run_end": (run_id: str, result: RunResult) "error": (run_id: str, error: Exception)

off ¶

off(event: str, callback: EventCallback) -> Runtime

Unsubscribe from runtime events.

Runtime¶

Runtime¶

Runtime ¶

budget_remaining_usd property ¶

budget_used_usd property ¶

tokens_used property ¶

tokens_remaining property ¶

__init__ ¶

run async ¶

run_batch async ¶

chat async ¶

chain async ¶

collaborate ¶

session async ¶

close async ¶

on ¶

off ¶

`Runtime`¶

budget_remaining_usd `property` ¶

budget_used_usd `property` ¶

tokens_used `property` ¶

tokens_remaining `property` ¶

init ¶

run `async` ¶

run_batch `async` ¶

chat `async` ¶

chain `async` ¶

session `async` ¶

close `async` ¶