Skip to content
Pyrula
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

Section titled “Types”

APIKeyAuth

Section titled “APIKeyAuth”

FastAPI dependency that validates a static Bearer token.

Usage::

router = RunRouter(
    control=...,
    config=RunRouterConfig(auth_config=AuthConfig(provider=APIKeyAuth("sk-proj-xxx"))),
)

AgentClient

Section titled “AgentClient”

Async HTTP client for Pyrula agents

submit

Section titled “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

Section titled “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

Section titled “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

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

stream_with_replay_fallback

Section titled “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

Section titled “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.

Bounded the same way resolve_continuation is: a self-loop (continued_to pointing to the current turn itself) raises immediately, and any other chain - cyclic or merely runaway - raises after _CONTINUATION_MAX_HOPS reconnects. Both raise ContinuationChainError so a pathological chain fails loud instead of reconnecting forever.

cancel

Section titled “cancel”
cancel(self, agent_name: str, turn_id: str, headers: Optional[dict[str, str]] = None) -> None

status

Section titled “status”
status(self, agent_name: str, turn_id: str, headers: Optional[dict[str, str]] = None) -> TurnStatusInfo

AgentContext

Section titled “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.

Fields:

  • run_id: str = ''
  • metadata: dict[str, Any] = field(default_factory=dict)
  • llm: Any = None
  • state: Optional[TurnState] = None
  • cancelled: bool
  • turn_id: str
  • memory: Any
  • thread: Optional[ThreadHandle]

mcp_tools

Section titled “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). On replay with a dead server, returns stub callables named from the run’s recorded tool_use names instead. A named but unconnected server raises; server=None with no MCP configured returns [].

emit_status

Section titled “emit_status”
emit_status(self, text: str) -> TurnEvent

emit

Section titled “emit”
emit(self, kind: str, payload: dict[str, Any], id: Optional[str] = None) -> TurnEvent

call_agent

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

gather_subagents

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

spawn_workflow

Section titled “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

Section titled “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

Section titled “AgentLimits”

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

Fields:

  • max_tool_calls: Optional[int] = None
  • max_llm_calls: Optional[int] = DEFAULT_MAX_LLM_CALLS
  • max_agent_depth: Optional[int] = None
  • max_parallel_subagents: Optional[int] = None
  • max_event_count_per_turn: Optional[int] = None
  • 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

Section titled “AgentRegistrationError”

AgentRouter

Section titled “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).

Fields:

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

lifespan

Section titled “lifespan”
lifespan(self) -> AsyncGenerator[None, None]

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

apply_redaction_hook

Section titled “apply_redaction_hook”
apply_redaction_hook(store: Any, hook: Callable[[str, dict[str, Any]], dict[str, Any]]) -> None

Patch store.append_event_sequenced in-place to fire hook for every event.

Patches the canonical allocator (append_event delegates to it, and submit_run writes RUN_INIT through it directly), so every persisted event is redacted. Only intercepts appends through this store instance. Raises ValueError if a hook is already installed to prevent silent hook-chaining.

AgentRouterConfig

Section titled “AgentRouterConfig”

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

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

Section titled “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.

Fields:

  • provider = 'anthropic'
  • model = model
  • streaming_mode = streaming_mode
  • max_buffer_bytes = max_buffer_bytes
  • max_retries = max_retries
  • retry_delay = retry_delay
  • max_tokens = max_tokens
  • prompt_caching = prompt_caching
  • cache_ttl = cache_ttl
  • llm_call_timeout = llm_call_timeout
  • rate_limit = rate_limit

stream

Section titled “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

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

AppOptions

Section titled “AppOptions”

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

Fields:

  • prefix: str = '/agents'
  • auth: Any = UNSET
  • deployment_mode: Optional[str] = None
  • sse_keepalive_interval: float = 15.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
  • idempotency_ttl_s: float = 86400.0
  • cors_origins: Optional[list[str]] = None

ConfigurationError

Section titled “ConfigurationError”

ErrorCode

Section titled “ErrorCode”

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'
  • INTERRUPT_TIMEOUT = 'interrupt_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

InMemoryStore

Section titled “InMemoryStore”

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

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

Fields:

  • max_thread_turns = max_thread_turns

LiteLLM

Section titled “LiteLLM”

LiteLLM adapter - routes to any supported provider.

Model names follow LiteLLM format: "provider/model".

complete

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

NestedDurableOperationError

Section titled “NestedDurableOperationError”

Durable orchestration attempted inside an atomic step or tool.

Fields:

  • operation = operation
  • operation_name = operation_name
  • boundary_kind = boundary_kind
  • boundary_name = boundary_name

OpenAILLM

Section titled “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.).

Fields:

  • provider = 'openai'
  • base_url = base_url

complete

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

PoisonOptions

Section titled “PoisonOptions”

Fields:

  • max_retries: int = 3
  • retry_ttl_s: float = 3600.0

PyrulaError

Section titled “PyrulaError”

Fields:

  • code: Optional[ErrorCode] = None

PyrulaWorker

Section titled “PyrulaWorker”

Simplified entry point for running a Pyrula worker process.

Fields:

  • connector_state: Optional[str]

run

Section titled “run”
run(self) -> None

run_async

Section titled “run_async”
run_async(self) -> None

submit

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

StepContext

Section titled “StepContext”

Context available inside a decorated step.

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

Section titled “StepResultTypeError”

Step result does not match its declared return type.

Fields:

  • step_id = step_id
  • validation_error = validation_error

StoreUnavailable

Section titled “StoreUnavailable”

Store operation failed after exhausting the retry budget.

Fields:

  • code = ErrorCode.STORE_UNAVAILABLE

ToolContext

Section titled “ToolContext”

Fields:

  • http: httpx.AsyncClient
  • deps: Any = None
  • state: Optional[TurnState] = None
  • tool_name: Optional[str] = None
  • tool_use_id: 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:tool_use_id, deterministic across replay (both components are journaled). 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. None when the context is unbound (no turn / no tool_use_id).
  • cache: CacheProxy
  • cancelled: bool
  • turn_id: str

emit_status

Section titled “emit_status”
emit_status(self, text: str) -> TurnEvent

ValkeyStore

Section titled “ValkeyStore”

Fields:

  • max_thread_turns = max_thread_turns

WorkerOptions

Section titled “WorkerOptions”

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

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
  • poison: PoisonOptions = field(default_factory=PoisonOptions)
  • shard_lease_interval: float = 10.0

Constants

Section titled “Constants”

Functions

Section titled “Functions”

agent

Section titled “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: VersionBehavior = 'pinned') -> 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. See docs/agents for the full guide.

create_app

Section titled “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.

registered_agents

Section titled “registered_agents”
registered_agents() -> list[AgentCallable]

step

Section titled “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

Section titled “tool”
tool(fn: Optional[Callable[..., Any]] = None, retry: int = 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) -> 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 the tool as having side effects (branch replay mocks it instead of calling it live); cache_ttl caches results by input.