# Agents > Agent runtime: decorators, context, LLM clients, serving, and the Kafka runtime. Source: https://docs.pyrula.io/reference/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)), ) ```python APIKeyAuth(keys: Mapping[str, str] | str) ``` --- ## `AgentClient` Async HTTP client for Pyrula agents ```python AgentClient(base_url: str, *, prefix: str = '/agents', timeout: float = 30.0, headers: Optional[dict[str, str]] = None, transport: Optional[httpx.AsyncBaseTransport] = None) ``` ### `submit` ```python 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` ```python 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` ```python 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` ```python replay(self, agent_name: str, turn_id: str, *, headers: Optional[dict[str, str]] = None) -> AsyncIterator[Event] ``` ### `stream_with_replay_fallback` ```python 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` ```python 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` ```python 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` ```python 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` ```python 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. ```python AgentContext(run_id: str = '', metadata: dict[str, Any] = , deps: Optional[DepsT] = None, principal: Any = None, state: Optional[TurnState] = None, llm: Any = None) ``` **Fields:** - `run_id: str = ''` - `metadata: dict[str, Any] = ` - `llm: Any = None` - `state: Optional[TurnState] = None` - `cancelled: bool` - `turn_id: str` - `memory: Any` - `thread: Optional[ThreadHandle]` ### `mcp_tools` ```python 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` ```python emit_status(self, text: str) -> TurnEvent ``` ### `call_agent` ```python call_agent(self, agent_fn: Any, *, llm: Optional[Any] = None, id: Optional[str] = None, **kwargs: Any) -> Any ``` ### `gather_subagents` ```python gather_subagents(self, *calls: tuple[Any, dict[str, Any]], id: Optional[str] = None, poll_interval: float = 0.5) -> list[Any] ``` ### `sleep` ```python sleep(self, duration: timedelta | float) -> None ``` ### `sleep_until` ```python sleep_until(self, dt: datetime) -> None ``` ### `wait_for_signal` ```python 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` ```python 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` ```python 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. ```python 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] = , 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] = ` - `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). ```python 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` ```python 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. ```python 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. ```python 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` ```python 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` ```python 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. ```python 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. ```python 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` ```python 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` ```python 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``). ```python LiteLLM(model: str, *, api_base: Optional[str] = None, api_key: Optional[str] = None, extra: Optional[dict[str, Any]] = None, **options: Any) ``` ### `complete` ```python 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` ```python put(self, namespace: tuple[str, ...], key: str, value: dict[str, Any], *, source: MemorySource, written_by_run_id: Optional[str]) -> None ``` ### `get` ```python get(self, namespace: tuple[str, ...], key: str) -> Optional[MemoryItem] ``` ### `delete` ```python delete(self, namespace: tuple[str, ...], key: str) -> None ``` ### `list` ```python list(self, namespace_prefix: tuple[str, ...], *, key_prefix: str = '', limit: int = 100) -> MemoryItems ``` ### `search` ```python search(self, namespace_prefix: tuple[str, ...], query: str, *, limit: int = 10) -> MemoryItems ``` --- ## `NestedDurableOperationError` Durable orchestration attempted inside an atomic step or tool. ```python 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.). ```python 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` ```python complete(self, messages: list[dict[str, Any]], system: Optional[str] = None) -> CompletionResult ``` --- ## `PoisonOptions` ```python 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. ```python 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` ```python run(self) -> None ``` ### `run_async` ```python run_async(self) -> None ``` ### `stop` ```python 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` ```python 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). ```python 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. ```python 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. ```python 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. ```python 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` ```python 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` ```python 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. ```python UnjournaledSurface(surface: str, journaled_path: str) ``` **Fields:** - `surface: str` - `journaled_path: str` --- ## `ValkeyStore` ```python 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``. ```python 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. ```python 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 = , 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 = ` - `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](https://docs.pyrula.io/agents/integrations/guarantees.md) ## Functions ### `agent` ```python 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] ``` Register `fn` as a Pyrula agent, usable bare or with keyword options. `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` ```python 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` ```python 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` ```python registered_agents() -> list[AgentCallable] ``` ### `step` ```python 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` ```python 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] ``` Register `fn` as a tool callable from an agent's LLM loop. `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.