Pre-release: the packages are not on PyPI yet, so pip install does not resolve. Build from source to try it today.

Agents

Generated from the type stubs and docstrings. Do not edit by hand.

Types

`APIKeyAuth`

FastAPI dependency that checks a static Bearer token against named keys.

The principal is the matching key’s name, never the key: the name is what the router stores as the run owner. Each name owns its own runs, and replacing a key’s value under the same name keeps them. A single string is shorthand for {"api-key": key}.

Usage::

auth = APIKeyAuth({"ci": os.environ["CI_KEY"], "ops": os.environ["OPS_KEY"]})
router = RunRouter(
    control=...,
    config=RunRouterConfig(auth_config=AuthConfig(provider=auth)),
)

APIKeyAuth(keys: Mapping[str, str] | str)

`AgentClient`

Async HTTP client for Pyrula agents

AgentClient(base_url: str, *, prefix: str = '/agents', timeout: float = 30.0, headers: Optional[dict[str, str]] = None, transport: Optional[httpx.AsyncBaseTransport] = None)

`submit`

submit(self, agent_name: str, *, metadata: Optional[dict[str, Any]] = None, idempotency_key: Optional[str] = None, headers: Optional[dict[str, str]] = None, **params: Any) -> TurnHandle

`stream`

stream(self, agent_name: str, turn_id: str, last_event_id: Optional[str] = None, headers: Optional[dict[str, str]] = None, idle_timeout_seconds: Optional[float] = None) -> AsyncIterator[Event]

`stream_from_path`

stream_from_path(self, stream_path: str, *, last_event_id: Optional[str] = None, headers: Optional[dict[str, str]] = None, idle_timeout_seconds: Optional[float] = None) -> AsyncIterator[Event]

`replay`

replay(self, agent_name: str, turn_id: str, *, headers: Optional[dict[str, str]] = None) -> AsyncIterator[Event]

`stream_with_replay_fallback`

stream_with_replay_fallback(self, agent_name: str, turn_id: str, *, last_event_id: Optional[str] = None, headers: Optional[dict[str, str]] = None, include_overflow_event: bool = False) -> AsyncIterator[Event]

Stream live events, falling back to replay when the live cursor is stale.

This is useful when reconnecting with a saved Last-Event-ID that may have fallen behind the retained live stream window. In that case the live stream emits stream:overflow; this helper can switch to replay() automatically.

`run`

run(self, agent_name: str, *, metadata: Optional[dict[str, Any]] = None, idempotency_key: Optional[str] = None, headers: Optional[dict[str, str]] = None, **params: Any) -> list[Event]

Submit a turn and collect all events until the terminal event.

Follows ctx.continue_as_new chains transparently: a run:continued event means the turn handed off to a successor run in the same thread (one logical turn - see pyrula.workflows.continuation). The successor’s stream is a separate SSE connection (a different turn_id across physical runs), so reconnect and keep collecting into the same event list rather than returning the chain’s RUN_CONTINUED hop as if it were the final result.

There is no hop limit: a chain may be any length. Each run:continued carries the successor’s chain_hop, which must grow by at least one per reconnect, so a cycle (including a self-loop) raises ContinuationChainError instead of reconnecting forever.

`cancel`

cancel(self, agent_name: str, turn_id: str, *, reason: Optional[str] = None, headers: Optional[dict[str, str]] = None) -> None

Ask the turn to stop at its next step boundary. reason (at most 500 characters) is recorded on the cancel. Needs a worker to finish the turn; with none, use terminate.

`terminate`

terminate(self, agent_name: str, turn_id: str, *, reason: Optional[str] = None, headers: Optional[dict[str, str]] = None) -> TerminateInfo

End the turn now, without waiting for its worker.

No further step runs and cleanup code is skipped; a step already running may still finish its side effects, and its result is discarded. Open child runs follow their parent_close policy. Repeating the call is safe. reason is at most 500 characters.

`status`

status(self, agent_name: str, turn_id: str, *, headers: Optional[dict[str, str]] = None) -> TurnStatusInfo

`AgentContext`

Agent execution context extending BaseContext with LLM/stream/tool-loop.

Inherits step, sleep, signal, timers, determinism, checkpoint, emit, emit_status and interrupt from BaseContext; adds call_agent, gather_subagents, memory, thread and llm. run_id/metadata default to ""/{} so the context can be constructed from a bound TurnState alone; __post_init__ back-fills run_id from state when omitted.

AgentContext(run_id: str = '', metadata: dict[str, Any] = <factory>, deps: Optional[DepsT] = None, principal: Any = None, state: Optional[TurnState] = None, llm: Any = None)

Fields:

run_id: str = ''
metadata: dict[str, Any] = <factory>
llm: Any = None
state: Optional[TurnState] = None
cancelled: bool
turn_id: str
memory: Any
thread: Optional[ThreadHandle]

`mcp_tools`

mcp_tools(self, server: Optional[str] = None) -> list[Any]

Discovered MCP tool callables to splat into ctx.llm.stream(tools=[…]).

Live, returns the worker pool’s tools for one server (or all when server is None). The definitions handed out (name, description, input schema) are recorded as a data step, so a replay with the server down gets stubs that carry the same definitions and the run’s recorded model calls still match. A named but unconnected server raises; server=None with no MCP configured returns [].

`emit_status`

emit_status(self, text: str) -> TurnEvent

`call_agent`

call_agent(self, agent_fn: Any, *, llm: Optional[Any] = None, id: Optional[str] = None, **kwargs: Any) -> Any

`gather_subagents`

gather_subagents(self, *calls: tuple[Any, dict[str, Any]], id: Optional[str] = None, poll_interval: float = 0.5) -> list[Any]

`sleep`

sleep(self, duration: timedelta | float) -> None

`sleep_until`

sleep_until(self, dt: datetime) -> None

`wait_for_signal`

wait_for_signal(self, name: str, *, timeout: timedelta | float | None = None, deadline: Optional[datetime] = None) -> Signal

Suspend until a signal named name arrives, as BaseContext.wait_for_signal.

Embedded mode refuses timeout and deadline with ConfigurationError: it runs no timer loop to end the wait. A wait without either works there, since the router delivers the signal.

`spawn_workflow`

spawn_workflow(self, name: str, params: Optional[dict[str, Any]] = None) -> str

Submit a @workflow run; returns run_id immediately (fire-and-forget).

The run_id is replay-stable (ctx.uuid) and the submit is idempotent, so a crash-and-replay re-submits the SAME run rather than spawning a duplicate. To await a child’s result durably, use ctx.run_child instead.

`wait_for_workflow`

wait_for_workflow(self, name: str, run_id: str, poll_interval: float = 0.5, timeout: Optional[float] = None) -> Any

Poll the store until a spawned workflow completes.

Returns the workflow result on success, raises PyrulaError if the workflow ended with an error event, and WaitTimeoutError if timeout seconds elapse first (None waits indefinitely).

Recognizes run:complete/run:error event kinds so it works regardless of which runner executes the spawned workflow.

`AgentLimits`

Per-agent resource and context-window limits, passed to @agent(limits=...).

max_total_tokens and max_cost_usd cap what one run spends on live model calls, sub-agents included. Tokens are input plus output plus cache-creation tokens as billed; cache reads are excluded (price them with max_cost_usd). Both are checked before each live call, so a run stops once the total has reached the cap and may overshoot by one call per concurrent caller. Replayed calls are never checked. Compat drop-in and durable_chat_model calls inside the run count too, and max_llm_calls includes them. max_cost_usd prices usage by model id from prices, then from the dated pyrula.pricing.DEFAULT_PRICES, trying the id the call asked for before its canonical id (pyrula.contracts.model_ids); a model in neither stops the run with reason unpriced_model. A deployment id whose model only the reply names is not checked before its first call. A FakeLLM left at its default model costs nothing.

max_tool_result_bytes (default 100,000) cuts the text of one tool result that the model reads, at a UTF-8 character boundary, and appends a marker saying how much was shown. The journal keeps the full result. None sends results whole. Changing it changes the messages sent to the model, so replays of earlier runs see their input as changed.

AgentLimits(max_tool_calls: Optional[int] = None, max_llm_calls: Optional[int] = DEFAULT_MAX_LLM_CALLS, max_total_tokens: Optional[int] = None, max_cost_usd: Optional[float] = None, prices: Mapping[str, ModelPrice] = <factory>, max_agent_depth: Optional[int] = None, max_parallel_subagents: Optional[int] = None, max_tool_result_bytes: Optional[int] = DEFAULT_MAX_TOOL_RESULT_BYTES, max_event_count_per_turn: Optional[int] = 50000, max_event_payload_bytes: Optional[int] = None, max_turn_wall_seconds: Optional[float] = None, context_window: Optional[int] = None, reserved_output_tokens: Optional[int] = None, compaction_fraction: Optional[float] = None, history_compactor: Any = UNSET, estimate_tokens: Optional[Any] = None)

Fields:

max_tool_calls: Optional[int] = None
max_llm_calls: Optional[int] = DEFAULT_MAX_LLM_CALLS
max_total_tokens: Optional[int] = None
max_cost_usd: Optional[float] = None
prices: Mapping[str, ModelPrice] = <factory>
max_agent_depth: Optional[int] = None
max_parallel_subagents: Optional[int] = None
max_tool_result_bytes: Optional[int] = DEFAULT_MAX_TOOL_RESULT_BYTES
max_event_count_per_turn: Optional[int] = 50000
max_event_payload_bytes: Optional[int] = None
max_turn_wall_seconds: Optional[float] = None
context_window: Optional[int] = None
reserved_output_tokens: Optional[int] = None
compaction_fraction: Optional[float] = None
history_compactor: Any = UNSET
estimate_tokens: Optional[Any] = None

`AgentRegistrationError`

`AgentRouter`

FastAPI router that exposes pyrula.agents over HTTP with SSE streaming.

Subclasses RunRouter to inherit generic durable-run infrastructure (attach, _classify_status, _setup_auth, ping_store, draining). Overrides lifespan (per-agent recovery), route registration (agents URL patterns), and auth checks (authorize_turn semantics).

AgentRouter(*, worker: WorkerProxy | InlineWorker, config: Optional[AgentRouterConfig] = None, **fastapi_kwargs: Any)

Fields:

EGRESS_DETAIL_FIELDS = ALL_DETAIL_FIELDS
get_stream
get_replay
get_status
delete_cancel
post_terminate
post_resume
post_signal
post_create_thread
get_thread
delete_thread
get_thread_turns
get_agents
get_runs
connector_state: Optional[str] Pyrula Cloud connector state, or None when cloud is not active.

`lifespan`

lifespan(self) -> AsyncGenerator[None, None]

Embedded-mode startup (per-agent recovery) and shutdown (drain + cancel).

`AgentRouterConfig`

Configuration for AgentRouter - extends RunRouterConfig with agent-only fields.

AgentRouterConfig(sse_keepalive_interval: float = 15.0, max_queue_depth: Optional[int] = None, idempotency_ttl_s: float = 86400.0, redaction_hook: Optional[Callable[[str, dict[str, Any]], dict[str, Any]]] = None, expose_error_details: bool = False, heartbeat_config: Optional[ReaderHeartbeatConfig] = None, auth_config: Optional[AuthConfig] = None, sse_max_streams: Optional[int] = 256, sse_max_streams_per_owner: Optional[int] = 16, sse_max_duration_s: Optional[float] = 3600.0, reporter: Any = None, cloud: bool = True, deps: Any = None, deps_factory: Optional[Callable[..., Any]] = None)

Fields:

reporter: Any = None
cloud: bool = True
deps: Any = None
deps_factory: Optional[Callable[..., Any]] = None
heartbeat_config: Optional[ReaderHeartbeatConfig] = None
auth_config: Optional[AuthConfig] = None

`AnthropicLLM`

Anthropic Messages API adapter (the native Pyrula LLM client for @agent).

To keep your existing anthropic SDK code and get durability by swapping one import, use the drop-in pyrula.agents.compat.anthropic.AsyncAnthropic instead.

AnthropicLLM(model: str, *, streaming_mode: StreamingMode = StreamingMode.SAFE, max_buffer_bytes: int = 5 * 1024 * 1024, max_retries: int = DEFAULT_MAX_RETRIES, api_key: Optional[str] = None, max_tokens: int = 8192, prompt_caching: bool = True, cache_ttl: str = '5m', llm_call_timeout: Optional[float] = None, rate_limit: Optional[LLMRate] = None)

Fields:

provider = 'anthropic'
model: str
streaming_mode: StreamingMode
max_buffer_bytes: int
max_retries: int
max_tokens: int
prompt_caching: bool
cache_ttl: str
llm_call_timeout: Optional[float]
rate_limit: Optional[LLMRate]

`stream`

stream(self, messages: list[dict[str, Any]], tools: Optional[list[ToolDefinition]] = None, system: Optional[str] = None, tool_choice: Optional[dict[str, Any]] = None) -> AsyncGenerator[ContentBlock, None]

`complete`

complete(self, messages: list[dict[str, Any]], system: Optional[str] = None) -> CompletionResult

`AppOptions`

Options for create_app: mount prefix, auth, health/ready routes, and the AgentRouter knobs (queue depth, redaction, idempotency, CORS).

deployment_mode is "embedded" (default) or "separate"; any other value raises ValueError. Embedded runs the turns in the web process. Separate submits and streams only, and needs pyrula worker processes to run them.

AppOptions(prefix: str = '/agents', auth: Any = UNSET, deployment_mode: Literal['embedded', 'separate'] = 'embedded', sse_keepalive_interval: float = 15.0, sse_max_streams: Optional[int] = 256, sse_max_streams_per_owner: Optional[int] = 16, sse_max_duration_s: Optional[float] = 3600.0, cancel_on_reader_disconnect: bool = False, include_health: bool = True, include_ready: bool = True, health_path: str = '/healthz', ready_path: str = '/readyz', reporter: Optional[Reporter] = None, max_queue_depth: Optional[int] = None, deps: Any = None, cloud: bool = True, redaction_hook: Optional[Callable[[str, dict[str, Any]], dict[str, Any]]] = None, expose_error_details: bool = False, idempotency_ttl_s: float = 86400.0, cors_origins: Optional[list[str]] = None, host: Optional[str] = None, allow_unauthenticated: bool = False)

Fields:

prefix: str = '/agents'
auth: Any = UNSET
deployment_mode: Literal['embedded', 'separate'] = 'embedded'
sse_keepalive_interval: float = 15.0
sse_max_streams: Optional[int] = 256
sse_max_streams_per_owner: Optional[int] = 16
sse_max_duration_s: Optional[float] = 3600.0
cancel_on_reader_disconnect: bool = False
include_health: bool = True
include_ready: bool = True
health_path: str = '/healthz'
ready_path: str = '/readyz'
reporter: Optional[Reporter] = None
max_queue_depth: Optional[int] = None
deps: Any = None
cloud: bool = True
redaction_hook: Optional[Callable[[str, dict[str, Any]], dict[str, Any]]] = None
expose_error_details: bool = False
idempotency_ttl_s: float = 86400.0
cors_origins: Optional[list[str]] = None
host: Optional[str] = None
allow_unauthenticated: bool = False

`ConfigurationError`

`ErrorCode`

Codes an agent turn reports. The set is open: a release may add codes, so compare by value, never match exhaustively, and keep a fallback for an unknown code.

Fields:

STORE_UNAVAILABLE = 'store_unavailable'
STREAM_TOO_LARGE = 'stream_too_large'
TURN_TIMEOUT = 'turn_timeout'
VERSION_SKEW = 'version_skew'
CANCELLED = 'cancelled'
PAUSED = 'paused'
TOOL_ERROR = 'tool_error'
BUDGET_EXCEEDED = 'budget_exceeded'
AGENT_ERROR = 'agent_error'
BUFFER_OVERFLOW = 'buffer_overflow'
LLM_ERROR = 'llm_error'
LLM_RATE_LIMITED = 'llm_rate_limited'
LLM_PROVIDER_UNAVAILABLE = 'llm_provider_unavailable'
POISON_QUARANTINE = 'poison_quarantine'
MAX_RETRIES = 'max_retries'
OUTPUT_VALIDATION = 'output_validation'
AUTH_MISSING = 'auth_missing'
AUTH_INVALID = 'auth_invalid'
AUTH_FORBIDDEN = 'auth_forbidden'
NOT_FOUND = 'not_found'
INVALID_REQUEST = 'invalid_request'
BACKPRESSURE_REJECTED = 'backpressure_rejected'
IDEMPOTENCY_CONFLICT = 'idempotency_conflict'
INTERNAL = 'internal'
WAIT_TIMEOUT = 'wait_timeout'
CONTEXT_WINDOW_EXCEEDED = 'context_window_exceeded'
AUTH_UNAVAILABLE = 'auth_unavailable'
WRITE_TOOL_BLOCKED = 'write_tool_blocked'
CONTINUE_AS_NEW_NOT_SUPPORTED = _ContractErrorCode.CONTINUE_AS_NEW_NOT_SUPPORTED.value
DUPLICATE_RUN = _ContractErrorCode.DUPLICATE_RUN.value
FATAL = _ContractErrorCode.FATAL.value
POISON = _ContractErrorCode.POISON.value
REPLAY_INPUT_CHANGED = _ContractErrorCode.REPLAY_INPUT_CHANGED.value
STREAM_CARRY_MISSING = _ContractErrorCode.STREAM_CARRY_MISSING.value
INTERRUPT_HANDLER_FAILED = _ContractErrorCode.INTERRUPT_HANDLER_FAILED.value

`InMemoryStore`

In-memory store for testing only - workflow MemoryStore + agent threads.

Does not persist across process restarts. Use ValkeyStore for production.

InMemoryStore(*, heartbeat_interval: float = 10.0, lock_ttl: float = 30.0, max_replay_entries: int = 200000, max_retries: int = 3, max_store_requeues: int = 20, max_thread_turns: int = 0)

Fields:

max_thread_turns: int

`LLMClient`

Abstract base for LLM provider adapters.

The abstract core is stream and complete, and it is frozen: a provider that implements only those keeps working. A keyword or method added later is optional and the engine passes it only after accepts_keyword finds it on the provider’s signature, so providers need not take **kwargs. tool_choice on stream is the one optional keyword today; a provider without it is offered a forced tool unforced.

Fields:

streaming_mode: StreamingMode = StreamingMode.SAFE
max_buffer_bytes: int = 5 * 1024 * 1024
provider: Optional[str] = None
max_retries: int = DEFAULT_MAX_RETRIES

`stream`

stream(self, messages: list[dict[str, Any]], tools: Optional[list[ToolDefinition]] = None, system: Optional[str] = None, tool_choice: Optional[dict[str, Any]] = None) -> AsyncGenerator[ContentBlock, None]

`complete`

complete(self, messages: list[dict[str, Any]], system: Optional[str] = None) -> CompletionResult

`LiteLLM`

LiteLLM adapter - routes to any supported provider.

Model names follow LiteLLM format: "provider/model". api_base and api_key are forwarded on every call; extra carries any other acompletion keyword (for example custom_llm_provider).

LiteLLM(model: str, *, api_base: Optional[str] = None, api_key: Optional[str] = None, extra: Optional[dict[str, Any]] = None, **options: Any)

`complete`

complete(self, messages: list[dict[str, Any]], system: Optional[str] = None) -> CompletionResult

`MemoryBackend`

Pluggable agent-memory store (synchronous; the AgentMemory facade bridges to async). namespace is a hierarchical tuple; list/search take a namespace PREFIX. source is caller-asserted: “agent” | “user” | “system”. Backends store the namespace natively (no mandated wire encoding).

The method set and signatures are frozen. search ranks however the backend defines relevance. New capabilities, vector search among them, arrive as separate optional protocols the facade finds by method presence, and a new keyword on an existing method is passed only after a signature probe.

`put`

put(self, namespace: tuple[str, ...], key: str, value: dict[str, Any], *, source: MemorySource, written_by_run_id: Optional[str]) -> None

`get`

get(self, namespace: tuple[str, ...], key: str) -> Optional[MemoryItem]

`delete`

delete(self, namespace: tuple[str, ...], key: str) -> None

`list`

list(self, namespace_prefix: tuple[str, ...], *, key_prefix: str = '', limit: int = 100) -> MemoryItems

`search`

search(self, namespace_prefix: tuple[str, ...], query: str, *, limit: int = 10) -> MemoryItems

`NestedDurableOperationError`

Durable orchestration attempted inside an atomic step or tool.

NestedDurableOperationError(operation: str, operation_name: str, boundary_kind: str, boundary_name: str)

Fields:

operation: str
operation_name: str
boundary_kind: str
boundary_name: str

`OpenAILLM`

OpenAI Chat Completions adapter (the native Pyrula LLM client for @agent).

Named OpenAILLM (matching AnthropicLLM). If you instead want to keep your existing openai SDK code and get durability by swapping one import, use the OpenAI-SDK-shaped drop-in pyrula.agents.compat.openai.AsyncOpenAI.

base_url can point to any OpenAI-compatible API (Ollama, Groq, Together AI, OpenRouter, Azure, etc.).

OpenAILLM(model: str, *, streaming_mode: StreamingMode = StreamingMode.SAFE, max_buffer_bytes: int = 5 * 1024 * 1024, max_retries: int = DEFAULT_MAX_RETRIES, api_key: Optional[str] = None, base_url: Optional[str] = None, max_tokens: int = 8192, temperature: Optional[float] = None, top_p: Optional[float] = None, seed: Optional[int] = None, parallel_tool_calls: Optional[bool] = None, llm_call_timeout: Optional[float] = None, rate_limit: Optional[LLMRate] = None)

Fields:

provider = 'openai'
base_url: Optional[str]

`complete`

complete(self, messages: list[dict[str, Any]], system: Optional[str] = None) -> CompletionResult

`PoisonOptions`

PoisonOptions(max_retries: int = 3, retry_ttl_s: float = 3600.0)

Fields:

max_retries: int = 3
retry_ttl_s: float = 3600.0

`ProviderStreamTruncated`

A provider stream ended without its terminal event, after the caller read it to the end.

A proxy or load balancer closing the connection cleanly looks like a normal end to the SDK. The failure is journaled like a provider error, so a replay raises it again at the call instead of handing the body a partial answer once and a complete one after a resume.

Fields:

code = ErrorCode.LLM_PROVIDER_UNAVAILABLE

`PyrulaError`

Fields:

code: Optional[ErrorCode] = None

`PyrulaWorker`

Simplified entry point for running a Pyrula worker process.

health_port (or PYRULA_WORKER_HEALTH_PORT) serves /livez and /readyz on health_host (default loopback); /readyz turns 503 when stop() is called.

PyrulaWorker(*, agents: Optional[list[Any]] = None, store: Any, llm: Any = None, concurrency: int = 50, completed_turn_ttl: int = 86400, reporter: Optional[Reporter] = None, cloud: bool = True, health_port: Optional[int] = None, health_host: str = '127.0.0.1')

Fields:

connector_state: Optional[str]
stop_requested: bool Whether stop() has been called.

`run`

run(self) -> None

`run_async`

run_async(self) -> None

`stop`

stop(self) -> None

Signal the worker to shut down gracefully.

The worker stops claiming, lets in-flight turns finish within its shutdown grace, cancels the rest, then stops the cloud connector; run_async() returns once that is done. Safe before run_async() starts, and on a second call. Thread-safe: called from a thread other than the worker’s event loop, it hands the stop to that loop.

`submit`

submit(self, agent_name: str, metadata: Optional[dict[str, Any]] = None, **params: object) -> str

`SQLiteStore`

Durable single-host agent store in one SQLite file: the workflow SQLiteStore plus agent threads. max_thread_turns keeps only a thread’s newest turns (0 keeps all).

SQLiteStore(path: str | os.PathLike[str], *, retention_seconds: Optional[int] = DEFAULT_RETENTION_SECONDS, synchronous: Literal['full', 'normal'] = 'full', busy_timeout_s: Optional[float] = None, lock_ttl: float = 30.0, held_ttl: float = 86400.0, heartbeat_interval: Optional[float] = None, max_replay_entries: int = 200000, max_retries: int = 3, max_store_requeues: int = 20, clock: Callable[[], float] = time.time, max_thread_turns: int = 0)

Fields:

max_thread_turns: int

`StepContext`

Context available inside a decorated step.

StepContext(run_id: str, run_name: str, step_id: str, call_index: int, attempt: int, metadata: Mapping[str, Any], deps: DepsT | None, principal: Any, idempotency_key: str, _is_cancelled: Callable[[], bool])

Fields:

run_id: str
run_name: str
step_id: str
call_index: int
attempt: int
metadata: Mapping[str, Any]
deps: DepsT | None
principal: Any
idempotency_key: str
cancelled: bool

`StepResultTypeError`

Step result does not match its declared return type.

StepResultTypeError(step_id: str, validation_error: Exception)

Fields:

step_id: str
validation_error: Exception

`StoreUnavailable`

Store operation failed after exhausting the retry budget.

Fields:

code = ErrorCode.STORE_UNAVAILABLE

`ToolCallRejected`

A reviewer, or the approval deadline, kept a direct call to a gated tool from running.

Raised from await tool(ctx, ...) for a tool declared requires_approval. reason is the text a reviewer gave with the rejection, when there was one; expired is true when the call was held until its approval_timeout and nobody answered.

ToolCallRejected(tool_name: str, *, reason: Optional[str] = None, expired: bool = False)

Fields:

code = ErrorCode.TOOL_ERROR
tool_name: str
reason: Optional[str]
expired: bool

`ToolContext`

ToolContext(http: httpx.AsyncClient, deps: Any = None, state: Optional[TurnState] = None, tool_name: Optional[str] = None, tool_use_id: Optional[str] = None, call_key: Optional[str] = None, cache_getter: Optional[Callable[[str], Optional[Any]]] = None, cache_setter: Optional[Callable[[str, Any, float], None]] = None)

Fields:

http: httpx.AsyncClient
deps: Any = None
state: Optional[TurnState] = None
tool_name: Optional[str] = None
tool_use_id: Optional[str] = None
call_key: Optional[str] = None
cache_getter: Optional[Callable[[str], Optional[Any]]] = None
cache_setter: Optional[Callable[[str, Any, float], None]] = None
idempotency_key: Optional[str] Stable per-tool-call key, turn_id:call_key:tool_name. Tool effects are at-least-once - a crash between executing a tool and journaling its result re-fires the tool on replay. Pass this key to the external system (e.g. a Stripe idempotency key, a dedup table, a conditional write) so the re-fire is a no-op. It is built from where the call sits in the turn (the LLM call and the tool’s position in its batch, or the direct-call index), never from the provider’s tool_use_id, which a rerun regenerates.

The key is stable across crash retries because the turn replays from its journal and the model response comes back unchanged. With step_boundary_durability off the response can be regenerated, and a regenerated batch may shift positions, so a non-idempotent tool should then dedupe on its own business key. None when the context is unbound (no turn / no call position).

cache: CacheProxy
cancelled: bool
turn_id: str

`emit_status`

emit_status(self, text: str) -> TurnEvent

`UnjournaledSurface`

A compat drop-in namespace with no journaling wrapper was touched inside a run.

Forwarding it would run live on every replay with no warning - the one thing a durable drop-in must not do quietly. Usually raised on attribute access, at the point the caller reaches for the surface rather than once they invoke it - but durable_chat_model’s synchronous invoke has no separate attribute to refuse at, so it raises this from inside the call itself instead.

Also an AttributeError so hasattr(client, "responses") and getattr(client, "responses", default) behave normally instead of blowing up - third-party probes like langsmith.wrap_openai do exactly this on attribute access and must see a plain False/default, not a crash.

UnjournaledSurface(surface: str, journaled_path: str)

Fields:

surface: str
journaled_path: str

`ValkeyStore`

ValkeyStore(*, url: str, completed_ttl: int = 86400, maxlen: int = 100000, replay_page_size: int = 1000, replay_page_bytes: int = 8 * 1024 * 1024, heartbeat_interval: float = 10.0, lock_ttl: float = 30.0, max_replay_entries: Optional[int] = None, retry_attempts: int = 3, retry_base: float = 0.1, max_retries: int = 3, max_store_requeues: int = 20, max_thread_turns: int = 0, socket_timeout: float = 5.0, socket_connect_timeout: float = 2.0, shard_count: int = 64, shard_lease_ttl: float = 30.0, cluster: Optional[bool] = None, namespace: Optional[str] = None, eviction_policy_check: Optional[str] = None)

Fields:

max_thread_turns: int

`WaitTimeout`

A wait_for_signal or interrupt reached its deadline with no answer.

An ordinary exception, unlike the suspension signals: the body is running again when it is raised and may catch it and carry on. Uncaught, the run ends with wait_timeout.

WaitTimeout(wait_kind: str, name: str, call_index: Optional[int] = None, deadline: Optional[datetime] = None, message: Optional[str] = None)

Fields:

code = ErrorCode.WAIT_TIMEOUT
wait_kind: str
name: str
call_index: Optional[int]
deadline: Optional[datetime]

`WorkerOptions`

Tuning for WorkerRuntime: concurrency, poll intervals, shutdown timing, poison-retry policy, and shard-lease renewal.

WorkerOptions(max_concurrent_turns: int = 50, orphan_scan_interval: float = 60.0, shutdown_grace: float = 30.0, cancel_timeout: float = 5.0, claim_poll_interval: float = 0.1, reporter: Optional[Reporter] = None, timer_poll_interval_s: float = 5.0, timer_max_batch: int = 100, schedule_poll_interval_s: float = 30.0, reconcile_interval: float = 60.0, projection_interval: float = 60.0, poison: PoisonOptions = <factory>, shard_lease_interval: float = 10.0, health_port: Optional[int] = None, health_host: str = '127.0.0.1')

Fields:

max_concurrent_turns: int = 50
orphan_scan_interval: float = 60.0
shutdown_grace: float = 30.0
cancel_timeout: float = 5.0
claim_poll_interval: float = 0.1
reporter: Optional[Reporter] = None
timer_poll_interval_s: float = 5.0
timer_max_batch: int = 100
schedule_poll_interval_s: float = 30.0
reconcile_interval: float = 60.0
projection_interval: float = 60.0
poison: PoisonOptions = <factory>
shard_lease_interval: float = 10.0
health_port: Optional[int] = None
health_host: str = '127.0.0.1'

Constants

GUARANTEES: dict[str, AdapterGuarantee] - see the adapter guarantee matrix

Functions

`agent`

agent(fn: Optional[Callable[..., Any]] = None, /, *, name: Optional[str] = None, timeout: float = 600.0, hide_thinking: bool = False, llm: Any = None, limits: Optional[AgentLimits] = None, output_type: Optional[Any] = None, deps_type: Optional[Any] = None, capture: Optional[str] = None, mcp_servers: Optional[list[Any]] = None, version: Optional[str] = None, version_behavior: Optional[VersionBehavior] = None) -> AgentCallable | Callable[[Callable[..., Any]], AgentCallable]

name defaults to the function name. limits is an AgentLimits for tool/LLM-call/depth/context-window caps. capture sets this agent’s cloud capture ceiling (a remote config update can only lower it, never raise it). version/version_behavior control replay compatibility across code changes, same as @workflow: without version the behavior defaults to auto_upgrade, with one to pinned. See docs/agents for the full guide.

`create_app`

create_app(*, agents: Optional[list[Callable[..., Any]]] = None, store: Optional[Store] = None, llm: Any, options: Optional[AppOptions] = None) -> FastAPI

Build a FastAPI app with an AgentRouter mounted at options.prefix.

Defaults agents to every @agent imported in this process, and store to an InMemoryStore (logged, non-durable). Also mounts /healthz and /readyz unless disabled in options. Returns the app; it is not started - run it with uvicorn or mount it into a larger app.

`open_agent_store`

open_agent_store(url: Optional[str] = None, *, default: Optional[str] = None, allow_memory: bool = False, **options: Any) -> Any

Open the agent store url names (when None, PYRULA_STORE_URL, else default): the workflow store of that kind plus agent threads. Same URL forms and refusals as open_store.

`registered_agents`

registered_agents() -> list[AgentCallable]

`step`

step(fn: Callable[..., Any] | None = None, /, *, name: str | None = None, effect: Effect = 'write', retry: RetryPolicy | None = None, timeout: float | None = None) -> Any

Define an atomic durable workflow step.

`tool`

tool(fn: Optional[Callable[..., Any]] = None, /, *, retry: int | RetryPolicy = 0, timeout: float = 30.0, name: Optional[str] = None, description: Optional[str] = None, schema: Optional[dict[str, Any]] = None, cache_ttl: Optional[float] = None, writes: bool = False, read_only: bool = False, requires_approval: bool | Callable[[Mapping[str, Any]], bool] = False, approval_timeout: timedelta | float | None = None) -> ToolCallable | Callable[[Callable[..., Any]], ToolCallable]

schema overrides the input JSON Schema that would otherwise be derived from fn’s signature and type hints; description likewise overrides the docstring. writes marks a tool that can have side effects; case runs refuse live calls to it unless explicitly allowed, while recorded results still replay. read_only declares the tool free of side effects: only such tools run live in a counterfactual branch. Any other tool replays its recorded result there, or the branch refuses it, so mark a lookup read_only=True to let a branch call it with new input. It cannot be combined with writes. cache_ttl caches results by input. retry is an int, the number of retries after the first attempt with no wait between them, or a RetryPolicy, which sets the attempts (max_attempts counts the first), the jittered backoff between them and which errors are retried. A sync tool that times out is not retried, because its thread cannot be stopped and a retry would run a second copy beside it; exceptions are retried, and an async def tool retries on timeout. requires_approval holds each live call for a reviewer before it starts: True asks for every call, a predicate over the call’s input asks only when it returns true (and asks if it raises). The predicate must be pure, since a resumed run evaluates it again. Only an explicit approve runs the tool; a rejection, an expired approval_timeout (seconds or a timedelta; none waits indefinitely) or any other answer denies the call and the model reads the denial. It combines with writes and read_only. A direct call from agent code asks the same question and raises ToolCallRejected on a denial; a call from inside another tool, a workflow body or a ctx.step cannot wait and raises NestedDurableOperationError when the condition holds.

View this page as Markdown