Agents

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

Types

`APIKeyAuth`

FastAPI dependency that validates a static Bearer token.

Usage::

router = AgentRouter(worker=..., config=AgentRouterConfig(auth_config=AuthConfig(provider=APIKeyAuth("sk-proj-xxx"))))

`AgentClient`

Async HTTP client for Pyrula agents.

`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.

`cancel`

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

`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 from pyrula.workflows BaseContext: step, sleep, signal, timers, determinism, checkpoint, emit, emit_status, interrupt.

Adds agent-only: call_agent, gather, memory, thread, llm.

Fields:

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: the worker pool’s tools for one server (or all when server is None). Replay with a dead server: stub callables named from the run’s recorded tool_use names, so tool_map matches the recorded run.

server=None means “all MCP tools”: the live pool’s tools, or on replay the recorded MCP tool_use names, or [] when no MCP is configured (asking for all tools is empty, not an error). A named, unconnected server raises.

`emit_status`

emit_status(self, text: str) -> TurnEvent

`emit`

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

`sleep_until`

sleep_until(self, dt: datetime) -> None

`wait_for_signal`

wait_for_signal(self, name: str) -> Signal

`interrupt`

interrupt(self, event_id: str, payload: dict[str, Any]) -> InterruptResponse

`call_agent`

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

`gather`

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

`spawn_workflow`

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

Submit a @workflow run; returns run_id immediately.

`wait_for_workflow`

wait_for_workflow(self, name: str, run_id: str, poll_interval: float = 0.5) -> Any

Poll the store until a spawned workflow completes.

Returns the workflow result on success. Raises RuntimeError if the workflow ended with an error event.

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

`AgentLimits`

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

`AgentRegistrationError`

`AgentRouter`

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

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.
draining: bool True after lifespan shutdown begins; used by readiness probes.

`ping_store`

ping_store(self) -> None

Ping the backing store. Raises StoreUnavailable if unreachable.

`attach`

attach(self, app: Any, prefix: str = '') -> None

Mount the router on a FastAPI app and compose lifespans.

`lifespan`

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

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

`apply_redaction_hook`

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

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

Only intercepts appends through this store instance. Raises ValueError if a hook is already installed to prevent silent hook-chaining.

`AgentRouterConfig`

Configuration for AgentRouter.

Grouped config replaces the 15+ individual constructor parameters. Individual params still work for backward compatibility.

Fields:

sse_keepalive_interval: float = 15.0
reporter: Any = None
cloud: bool = True
max_queue_depth: Optional[int] = None
idempotency_ttl_s: float = 86400.0
redaction_hook: Optional[Callable[[str, dict[str, Any]], dict[str, Any]]] = None
heartbeat_config: Optional[ReaderHeartbeatConfig] = None
auth_config: Optional[AuthConfig] = None
deps: Any = None
deps_factory: Optional[Callable[..., Any]] = None

`AnthropicLLM`

Anthropic Messages API adapter.

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

`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) -> str

`AppOptions`

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`

`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'

`LifecycleAdapter`

Run-neutral lifecycle + event-log view over a pyrula.agents turn-scoped store.

Implements both core ABCs in one object. The engine holds a single reference; mixin methods (cache_get, cache_set, set_thread_completed, get_sleeping_turns, wake_turn, …) pass through via __getattr__.

Idempotent: LifecycleAdapter(LifecycleAdapter(store)) returns the existing adapter (no double-wrap).

When the underlying store is a core store (with run_* naming), the run_* methods delegate directly; turn_* aliases still work by remapping to run_*.

Fields:

raw: Store

`submit_turn`

submit_turn(self, name: str, turn_id: str, params: dict[str, object], **kwargs: Any) -> str

`claim_next_turn`

claim_next_turn(self, name: str, worker_id: str) -> Optional[str]

`claim_run`

claim_run(self, name: str, run_id: str, worker_id: str) -> None

`release_turn`

release_turn(self, name: str, turn_id: str, **kwargs: Any) -> None

`requeue_turn`

requeue_turn(self, name: str, turn_id: str, **kwargs: Any) -> None

`expire_turn`

expire_turn(self, name: str, turn_id: str, **kwargs: Any) -> bool

`set_turn_owner`

set_turn_owner(self, name: str, turn_id: str, owner: str) -> None

`resume_turn`

resume_turn(self, name: str, turn_id: str, action: str, value: Any) -> None

`recover_pending_turns`

recover_pending_turns(self, name: str, now: Optional[float] = None) -> list[str]

`submit_run`

submit_run(self, name: str, run_id: str, params: dict[str, object], metadata: Optional[dict[str, Any]] = None, owner: Optional[str] = None, idempotency_key: Optional[str] = None, idempotency_body_hash: Optional[str] = None, idempotency_ttl: int = 86400) -> str

`claim_next_run`

claim_next_run(self, name: str, worker_id: str) -> Optional[str]

`release_run`

release_run(self, name: str, run_id: str, worker_id: Optional[str] = None) -> None

`requeue_run`

requeue_run(self, name: str, run_id: str, worker_id: Optional[str] = None) -> None

`recover_pending_runs`

recover_pending_runs(self, name: str, now: Optional[float] = None) -> list[str]

`get_pending_count`

get_pending_count(self, name: str) -> int

`heartbeat`

heartbeat(self, name: str, run_id: str, worker_id: str) -> None

`request_cancel`

request_cancel(self, name: str, run_id: str) -> str

`get_cancel_requested_at`

get_cancel_requested_at(self, name: str, run_id: str) -> Optional[str]

`request_pause`

request_pause(self, name: str, run_id: str) -> str

`get_pause_requested_at`

get_pause_requested_at(self, name: str, run_id: str) -> Optional[str]

`clear_pause`

clear_pause(self, name: str, run_id: str) -> None

`resume_paused_turn`

resume_paused_turn(self, name: str, run_id: str) -> None

`expire_run`

expire_run(self, name: str, run_id: str, ttl_seconds: int = 360) -> bool

`set_run_owner`

set_run_owner(self, name: str, run_id: str, owner: str) -> None

`get_run_owner`

get_run_owner(self, name: str, run_id: str) -> Optional[str]

`ping`

ping(self) -> None

`list_runs`

list_runs(self, name: str, limit: int = 50, cursor: Optional[str] = None, owner: Optional[str] = None) -> tuple[list[dict[str, Any]], Optional[str]]

`set_interrupted`

set_interrupted(self, name: str, run_id: str, event_id: str, payload: dict[str, Any]) -> None

`resume_run`

resume_run(self, name: str, run_id: str, action: str, value: Any) -> None

`get_interrupt_response`

get_interrupt_response(self, name: str, run_id: str) -> Optional[dict[str, Any]]

`complete_run`

complete_run(self, name: str, run_id: str) -> None

`append_event`

append_event(self, name: str, run_id: str, event: RunEvent) -> None

`append_event_sequenced`

append_event_sequenced(self, name: str, run_id: str, event: RunEvent) -> int

`load_run`

load_run(self, name: str, run_id: str, max_events: Optional[int] = None) -> list[RunEvent]

`replay_run`

replay_run(self, name: str, run_id: str, cursor: Optional[Any] = None, count: Optional[int] = None) -> tuple[list[RunEvent], Optional[Any]]

`get_last_event`

get_last_event(self, name: str, run_id: str) -> Optional[RunEvent]

`read_entries_from`

read_entries_from(self, name: str, run_id: str, cursor: Optional[str], count: int) -> tuple[list[tuple[str, RunEvent]], Optional[str]]

`LiteLLM`

LiteLLM adapter - routes to any supported provider.

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

`complete`

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

`OpenAI`

OpenAI Chat Completions adapter.

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

Fields:

provider = 'openai'
base_url = base_url

`complete`

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

`PoisonOptions`

Fields:

max_retries: int = 3
retry_ttl_s: float = 3600.0

`PyrulaError`

Base class for Pyrula exceptions.

Fields:

code: Optional[ErrorCode] = None

`PyrulaWorker`

Simplified entry point for running a Pyrula worker process.

Fields:

connector_state: Optional[str]

`run`

run(self) -> None

`run_async`

run_async(self) -> None

`submit`

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

`StoreUnavailable`

Fields:

code = ErrorCode.STORE_UNAVAILABLE

`ValkeyStore`

Fields:

SLEEPING_INDEX_KEY: str = 'sleeping:index'
SCHEDULES_KEY: str = 'schedule:index'
SCHEDULE_FIRE_TTL: int = 300
max_thread_turns = max_thread_turns

`WorkerOptions`

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)

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: str = CaptureLevel.lifecycle, mcp_servers: Optional[list[Any]] = None) -> AgentCallable | Callable[[Callable[..., Any]], AgentCallable]

`create_app`

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

`data_step`

data_step(ctx: BaseContext, step_id: str, op: Any, *args: Any, **kwargs: Any) -> Any

Durable data operation with codec-driven replay.

`registered_agents`

registered_agents() -> list[AgentCallable]