Agents

zeroth.runtime.agents

Agent runtime foundation layered on the governed runtime primitives.

This package provides everything needed to run AI agents: configuration, prompt assembly, provider adapters, tool attachments, output validation, retry logic, and thread state management. Import the classes you need directly from this package.

CascadingProviderAdapter

CascadingProviderAdapter(
    inner: ProviderAdapter, *, cheap_model: str
)

Cheap-first cascade: try cheap_model; escalate to the incumbent on hard failure.

AgentContentBlockedError

AgentContentBlockedError(
    message: str,
    *,
    direction: str,
    findings: list[str],
    audit_record: dict | None = None,
)

Bases: AgentRuntimeError

Raised when a content-safety guardrail blocks an agent's input or output.

Carries a structured audit_record mapping so the orchestrator's failure-audit path (_record_failed_execution_audit) persists a rejected NodeAuditRecord with the findings — i.e. a blocked run is auditable, not just an exception (SAFE-03).

AgentInputValidationError

Bases: AgentRuntimeError

Raised when the data you pass into an agent does not match the expected format.

AgentOutputValidationError

Bases: AgentRuntimeError

Raised when the AI model's response does not match the expected output format.

AgentProviderError

Bases: AgentRuntimeError

Raised when the AI model provider fails to produce a response.

AgentRetryExhaustedError

AgentRetryExhaustedError(
    *, attempts: int, last_error: Exception
)

Bases: AgentRuntimeError

Raised when the agent has retried the maximum number of times and still failed.

AgentRuntimeError

Bases: Exception

Base error for all agent runtime failures.

Catch this if you want to handle any error from the agent runtime in one place. All other errors in this module inherit from this.

AgentTimeoutError

Bases: AgentProviderError

Raised when the AI model takes too long to respond.

MCPServerConfig

Bases: BaseModel

Configuration for connecting to an MCP server via stdio transport.

AgentConfig

Bases: BaseModel

The main configuration for an agent.

This holds everything the runtime needs to know about an agent: its name, instruction, which AI model to use, what input/output shapes it expects, which tools it can use, retry behavior, and more.

declared_tool_refs property

declared_tool_refs: list[str]

Return the list of tool names this agent is allowed to use.

AgentRunResult

Bases: BaseModel

The result you get back after an agent finishes running.

Contains the input that was sent, the output that was produced, how many attempts it took, the prompt that was used, the raw provider response, and an audit trail of what happened.

ContentSafetyConfig

Bases: BaseModel

Content-safety policy for an agent's own input and output.

Detects PII and blocklisted terms in agent input/output and, depending on mode, flags (audit only), redacts, or blocks the run. Opt-in (enabled=False by default): unlike model-boundary sanitization, this imposes a content policy on the application's own typed data, which is high-blast-radius, so it is off until explicitly configured. See zeroth.governance.guardrails.content.

InMemoryThreadStateStore

InMemoryThreadStateStore()

A simple thread state store that keeps everything in memory.

Good for tests and local development. Data is lost when the process stops.

load async

load(thread_id: str) -> dict[str, Any] | None

Load the latest saved state for a thread, or None if nothing was saved.

checkpoint async

checkpoint(thread_id: str, state: dict[str, Any]) -> None

Save a snapshot of the thread state and keep it in the history.

latest

latest(thread_id: str) -> dict[str, Any] | None

Return the most recently saved state for a thread (sync version).

history

history(thread_id: str) -> list[dict[str, Any]]

Return all saved state snapshots for a thread, oldest first.

ModelParams

Bases: BaseModel

Per-node LLM parameters. None means use provider default.

PromptAssembly

Bases: BaseModel

The fully assembled prompt ready to be sent to the AI model.

Contains the list of messages, a rendered text version, and metadata that gets logged for auditing purposes.

PromptConfig

Bases: BaseModel

Settings that control what goes into the prompt sent to the AI model.

For example, you can choose whether to include the input/output schemas, thread state, tool references, or memory references in the prompt. You can also specify keys whose values should be hidden (redacted).

PromptMessage

Bases: BaseModel

A single message in a prompt (like one chat bubble).

Each message has a role (system, user, or assistant) and text content.

RetryPolicy

Bases: BaseModel

Controls how many times the agent should retry when something goes wrong.

You can configure whether to retry on validation errors, provider errors, or timeouts, and how long to wait between retries.

max_attempts property

max_attempts: int

Return the total number of attempts (initial try plus retries).

ToolOutputSafetyConfig

Bases: BaseModel

Model-boundary safety controls for untrusted tool/memory output.

Tool results and memory-sourced content are re-injected into the LLM, where they can carry prompt-injection payloads. These settings length-cap, screen, and provenance-wrap that untrusted content before the model sees it. Enabled by default (governance-first); set enabled=False to restore the prior raw re-injection. See zeroth.runtime.agents.sanitization.

AgentAuditSerializer

AgentAuditSerializer(
    *, redact_keys: set[str] | None = None
)

Creates audit-safe copies of prompts and responses.

Replaces sensitive values (like passwords or tokens) with placeholder text so the audit log does not contain secrets.

serialize_prompt

serialize_prompt(
    assembly: PromptAssembly,
) -> dict[str, Any]

Turn a prompt assembly into a redacted dictionary for logging.

serialize_response

serialize_response(response: Any) -> dict[str, Any]

Turn a provider response into a redacted dictionary for logging.

serialize_record

serialize_record(
    *,
    prompt: PromptAssembly,
    response: Any,
    extra: dict[str, Any],
) -> dict[str, Any]

Combine prompt, response, and extra info into one redacted audit record.

The nested prompt/response/extra sections are content and do not survive the audit capture boundary's metadata-only default. The two signals that are not content -- how many attempts the call took and whether the provider response came from the cache -- are therefore also promoted to flat, allowlisted keys, so the econ waste detectors still see them in a persisted record.

PromptAssembler

Builds the messages that get sent to the AI model.

Takes the agent config, the user's input data, and optional thread state, then constructs a system message and user message with all the relevant context the model needs to produce a good response.

assemble

assemble(
    config: AgentConfig,
    input_payload: BaseModel | Mapping[str, Any],
    *,
    thread_state: Mapping[str, Any] | None = None,
    runtime_context: Mapping[str, Any] | None = None,
) -> PromptAssembly

Build a complete prompt from the agent config and input data.

Combines the agent's instruction, schemas, tools, memory refs, input payload, and thread state into a structured set of messages. Sensitive fields are redacted based on the prompt config.

DeterministicProviderAdapter

DeterministicProviderAdapter(
    responses: Sequence[ProviderResponse | Any | Exception],
)

A fake provider adapter for tests that returns pre-set responses.

You give it a list of responses when you create it, and each call to ainvoke pops the next one off the list. Useful for testing agent behavior without calling a real AI model.

ainvoke async

ainvoke(request: ProviderRequest) -> ProviderResponse

Return the next queued response, or raise if the queue is empty.

LiteLLMProviderAdapter

LiteLLMProviderAdapter(
    *,
    default_timeout: float = 600.0,
    secret_provider: SecretProvider | None = None,
    tenant_id: str | None = None,
    allow_env_fallback: bool = True,
    llm_key_map: dict[str, str] | None = None,
    llm_base_url_map: dict[str, str] | None = None,
)

Universal LLM adapter using LangChain's ChatLiteLLM wrapper.

Routes to any LiteLLM-supported provider (OpenAI, Anthropic, 100+ others) based on the model string in ProviderRequest.model_name. Uses LangChain interface per D-01 for governed-runtime compatibility.

Model strings use LiteLLM format: openai/gpt-4o, anthropic/claude-sonnet-4-5-20250514, etc.

Secret isolation (WS-F)

When a :class:SecretProvider is supplied, the api_key for each model's provider is resolved through it (scoped to tenant_id) and injected into the ChatLiteLLM constructor, so the key never comes from a process-global environment variable. Clients are cached by (model, tenant_id, key_fingerprint) so a different tenant or a rotated key never reuses a stale client. With allow_env_fallback=False a missing key raises :class:SecretResolutionError instead of letting LiteLLM read the ambient env. When no provider is supplied (or fallback is allowed and the key is missing), LiteLLM's own env resolution is used, preserving the original behaviour.

ainvoke async

ainvoke(request: ProviderRequest) -> ProviderResponse

Send request to LLM via ChatLiteLLM and return normalized response.

When request.output_model is set, uses LangChain's with_structured_output() for provider-agnostic structured output. This handles schema generation, provider-specific formatting, and response parsing automatically, returning a typed Pydantic instance.

ProviderAdapter

Bases: Protocol

The interface that all provider adapters must follow.

Any class with an ainvoke method that takes a ProviderRequest and returns a ProviderResponse can be used as a provider adapter.

ainvoke async

ainvoke(request: ProviderRequest) -> ProviderResponse

Send a request to the AI model and return its response.

ProviderRequest

Bases: BaseModel

The request object sent to an AI model provider.

Contains the model name, the list of messages to send, and any extra metadata the provider might need.

ProviderResponse

Bases: BaseModel

The response received from an AI model provider.

Contains the text content, the raw provider-specific response, any tool calls the model wants to make, and extra metadata.

CachingProviderAdapter

CachingProviderAdapter(
    inner: ProviderAdapter,
    cache: ResponseCache | None = None,
    *,
    cache_tool_calls: bool = False,
)

Exact-match response cache over any ProviderAdapter (PRES-02).

The key covers everything that changes a response: model, messages, model params, tools, tool choice, and the output schema. Tool-call responses are not cached by default (replaying a stale tool decision is unsafe).

ainvoke async

ainvoke(request: ProviderRequest) -> ProviderResponse

Return a cached response on an exact match, else call the inner adapter.

FallbackProviderAdapter

FallbackProviderAdapter(
    targets: Sequence[ProviderTarget],
    *,
    should_fallback: Callable[[BaseException], bool]
    | None = None,
)

Tries an ordered list of provider targets, failing over on transient errors.

across_models classmethod

across_models(
    adapter: ProviderAdapter,
    models: Sequence[str],
    *,
    should_fallback: Callable[[BaseException], bool]
    | None = None,
) -> FallbackProviderAdapter

Build a chain over one adapter and an ordered list of model strings.

ainvoke async

ainvoke(request: ProviderRequest) -> ProviderResponse

Invoke targets in order; return the first success, else raise the last error.

InMemoryResponseCache

InMemoryResponseCache(
    *, maxsize: int = 1024, ttl: float | None = None
)

Process-local response cache backed by cachetools (LRU, or TTL when set).

Intended for single-process use; not shared across workers. Concurrent fan-out branches may double-compute a cold key (benign), never corrupt it.

get

get(key: str) -> ProviderResponse | None

Return the cached response for key, or None on miss/expiry.

set

set(key: str, response: ProviderResponse) -> None

Store response under key (evicting per the LRU/TTL policy).

ProviderTarget dataclass

ProviderTarget(
    adapter: ProviderAdapter, model_name: str | None = None
)

One step in a fallback chain: an adapter, with an optional model override.

When model_name is set, the request's model is rewritten to it before the adapter is called — the common case for falling over to a different model on the same (e.g. LiteLLM) adapter.

ResponseCache

Bases: Protocol

A keyed store of provider responses (pluggable: in-memory, Redis, semantic, ...).

get

get(key: str) -> ProviderResponse | None

Return the cached response for key, or None on miss.

set

set(key: str, response: ProviderResponse) -> None

Store response under key.

AgentRunner

AgentRunner(
    config: AgentConfig,
    provider: ProviderAdapter,
    *,
    prompt_assembler: PromptAssembler | None = None,
    output_validator: OutputValidator | None = None,
    audit_serializer: AgentAuditSerializer | None = None,
    thread_state_store: ThreadStateStore | None = None,
    tool_bridge: ToolAttachmentBridge | None = None,
    tool_executor: Any | None = None,
    granted_tool_permissions: list[str] | None = None,
    memory_resolver: MemoryConnectorResolver | None = None,
    budget_enforcer: Any | None = None,
    context_tracker: Any | None = None,
    tool_output_sanitizer: ToolOutputSanitizer
    | None = None,
    content_guardrail: ContentGuardrail | None = None,
)

Runs an agent end-to-end: prompt assembly, model call, output validation.

This is the main class you use to execute an agent. Give it a config and a provider, then call run() with your input data. It handles retries, tool calls, thread state, memory, and audit logging.

fork_for_dispatch

fork_for_dispatch() -> AgentRunner

Create an isolated runner while retaining safe service dependencies.

run async

run(
    input_payload: BaseModel | Mapping[str, Any],
    *,
    thread_id: str | None = None,
    runtime_context: Mapping[str, Any] | None = None,
    enforcement_context: Mapping[str, Any] | None = None,
) -> AgentRunResult

Execute the agent within an OBS tracing span; delegates to :meth:_run.

Kept as a thin wrapper so the public signature (inspected by the orchestrator for enforcement_context) is unchanged while every agent run produces one zeroth.agent span.

HeuristicInjectionScreener

HeuristicInjectionScreener(
    patterns: tuple[tuple[str, Pattern[str]], ...]
    | None = None,
)

Best-effort regex screener for common prompt-injection patterns.

Not a security guarantee — a curated set of high-signal heuristics whose job is to surface suspicious tool/memory output in the audit trail. Swap in a stronger screener (e.g. a classifier) via any object implementing InjectionScreener.

screen

screen(text: str) -> tuple[str, ...]

Return a sorted, de-duplicated tuple of matched flag names.

matching_spans

matching_spans(
    text: str,
) -> tuple[tuple[str, int, int], ...]

Return every heuristic match with its source-text span.

InjectionScreener

Bases: Protocol

Anything that can flag suspected prompt-injection patterns in untrusted text.

screen

screen(text: str) -> tuple[str, ...]

Return flag names for suspected injection patterns (empty tuple if clean).

SanitizedContent dataclass

SanitizedContent(
    text: str,
    original_length: int,
    truncated: bool = False,
    flags: tuple[str, ...] = (),
    blocked: bool = False,
)

The outcome of sanitizing one piece of untrusted content.

text is ready to hand to the model; the remaining fields describe what was done and are recorded in the audit trail.

as_audit

as_audit() -> dict[str, object]

Return a JSON-friendly summary for the tool/audit record.

ToolOutputSanitizer

ToolOutputSanitizer(
    *,
    max_output_chars: int = DEFAULT_MAX_TOOL_OUTPUT_CHARS,
    wrap_with_provenance: bool = True,
    screener: InjectionScreener | None = None,
    screening_mode: str = "flag",
)

Sanitizes untrusted tool/memory output before it re-enters the model.

Pipeline: length-cap -> screen (on the full content) -> optionally block -> provenance-wrap. Screening runs on the full content, before truncation, so a payload cannot hide past the cap.

sanitize

sanitize(
    content: str,
    *,
    source: str,
    max_output_chars: int | None = None,
) -> SanitizedContent

Sanitize one piece of untrusted content attributed to source.

max_output_chars overrides the sanitizer default for this call (used for per-tool caps); None inherits the default.

RepositoryThreadResolver dataclass

RepositoryThreadResolver(
    thread_repository: ThreadRepository,
)

Finds an existing thread or creates a new one in the database.

Use this when you need to look up a thread by ID, creating it automatically if it does not exist yet.

resolve async

resolve(
    thread_id: str | None,
    *,
    graph_version_ref: str,
    deployment_ref: str,
    participating_agent_refs: list[str] | None = None,
    state_snapshot_refs: list[str] | None = None,
    checkpoint_refs: list[str] | None = None,
    memory_bindings: list[ThreadMemoryBinding]
    | None = None,
    run_id: str | None = None,
    status: ThreadStatus | None = None,
) -> ThreadResolution

Look up a thread by ID, or create one if it does not exist yet.

resolve_optional async

resolve_optional(
    thread_id: str | None, **kwargs: Any
) -> ThreadResolution | None

Like resolve, but returns None if no thread ID is provided.

RepositoryThreadStateStore

RepositoryThreadStateStore(
    database: AsyncDatabase | None = None,
    *,
    tenant_id: str,
    workspace_id: str | None,
    run_repository: RunRepository | None = None,
    thread_repository: ThreadRepository | None = None,
)

Saves and loads thread state using the database.

Each state save creates a checkpoint record in the database, so you can trace the full history of a thread's state over time. This is the production-grade alternative to InMemoryThreadStateStore.

load async

load(thread_id: str) -> dict[str, Any] | None

Load the most recent state snapshot for a thread from the database.

load_optional async

load_optional(
    thread_id: str | None,
) -> dict[str, Any] | None

Like load, but returns None if no thread ID is provided.

checkpoint async

checkpoint(thread_id: str, state: dict[str, Any]) -> str

Save a state snapshot for the thread and return the checkpoint ID.

checkpoint_optional async

checkpoint_optional(
    thread_id: str | None, state: dict[str, Any]
) -> str | None

Like checkpoint, but does nothing and returns None if no thread ID is provided.

ThreadResolution

Bases: BaseModel

The result of looking up or creating a thread.

Tells you which thread was found (or created), whether it was newly created, and what state was restored from a previous run.

ToolAttachmentAction

Bases: StrEnum

The types of actions that can happen with a tool attachment.

Used for categorizing events in audit logs and event routing.

ToolAttachmentBinding

Bases: BaseModel

A resolved, ready-to-use version of a tool attachment.

Created from a ToolAttachmentManifest when the tool is actually needed at runtime. Contains the same information but represents a tool that has been looked up and is ready to execute.

from_manifest classmethod

from_manifest(
    manifest: ToolAttachmentManifest,
) -> ToolAttachmentBinding

Create a binding from a manifest, copying all its fields.

ToolAttachmentBridge

ToolAttachmentBridge(
    registry: ToolAttachmentRegistry | None = None,
)

High-level helper for validating tool calls and building audit records.

Sits between the agent runner and the tool registry. Checks that requested tools were declared, that the caller has the right permissions, and creates audit-friendly records of each tool call.

from_config classmethod

from_config(
    attachments: Sequence[ToolAttachmentManifest],
) -> ToolAttachmentBridge

Create a bridge with a registry pre-loaded from a list of manifests.

resolve_declared_tools

resolve_declared_tools(
    declared_tool_refs: Sequence[str],
) -> list[ToolAttachmentBinding]

Resolve a list of declared tool aliases into bindings.

ensure_declared_tools

ensure_declared_tools(
    requested_tool_refs: Sequence[str],
    declared_tool_refs: Sequence[str],
) -> list[ToolAttachmentBinding]

Verify that all requested tools were declared, then resolve them.

Raises UndeclaredToolError if any requested tool is not in the declared list.

validate_permissions

validate_permissions(
    binding: ToolAttachmentBinding | ToolAttachmentManifest,
    granted_permissions: Sequence[str] | None,
) -> None

Check that the caller has all the permissions a tool requires.

Raises ToolPermissionError if any required permission is missing.

check_capabilities

check_capabilities(
    binding: ToolAttachmentBinding | ToolAttachmentManifest,
    effective_capabilities: set[Capability],
    *,
    node_id: str,
) -> None

Gate a tool call on the agent's granted Capability set (WS-C).

The tool's required_capabilities (the target unit's declared capabilities) must be covered by effective_capabilities (what the policy guard granted the agent node). Raises :class:~zeroth.governance.policy.errors.CapabilityDeniedError otherwise.

Fail-closed: an empty granted set denies any tool that requires a capability. Only invoked when enforcement is active — the runner passes None (and does not call this) when the policy guard is not wired.

build_resolution_audit

build_resolution_audit(
    *,
    declared_tool_refs: Sequence[str],
    resolved_bindings: Sequence[ToolAttachmentBinding],
    requested_tool_refs: Sequence[str] | None = None,
) -> dict[str, Any]

Create an audit record showing which tools were declared, requested, and resolved.

build_call_audit

build_call_audit(
    *,
    binding: ToolAttachmentBinding | ToolAttachmentManifest,
    arguments: Mapping[str, Any],
    granted_permissions: Sequence[str] | None = None,
    outcome: Mapping[str, Any] | None = None,
    error: str | None = None,
    at_least_once: bool = False,
) -> dict[str, Any]

Create an audit record for a single tool call, including its arguments and result.

at_least_once marks a call that bypassed the side-effect operation boundary entirely -- today, an MCP tool, which is not a graph node and so never reaches RuntimeToolExecutor. Such a call has no operation record, no replay suppression and no reconciliation, and saying so is the point: an unmarked record would read as though the guarantee applied.

ToolAttachmentError

Bases: ValueError

Base error for anything that goes wrong with tool attachments.

ToolAttachmentManifest

Bases: BaseModel

Describes a tool that an agent is allowed to use.

Each manifest declares the tool's alias (short name), what code it points to, what permissions it needs, and whether it can cause side effects (like writing to a database).

to_openai_tool

to_openai_tool() -> dict[str, Any]

Convert to OpenAI function-calling tool format.

ToolAttachmentRegistry

ToolAttachmentRegistry(
    attachments: Sequence[ToolAttachmentManifest]
    | None = None,
)

A lookup table of all the tools an agent has declared.

Tools are stored by their alias (short name). You can register new tools, look them up, and resolve them into bindings ready for use.

register

register(
    manifest: ToolAttachmentManifest,
) -> ToolAttachmentManifest

Add a tool to the registry. Raises if a different tool with the same alias exists.

get

get(alias: str) -> ToolAttachmentManifest

Look up a tool by its alias. Raises KeyError if not found.

has

has(alias: str) -> bool

Check whether a tool with this alias is registered.

resolve

resolve(alias: str) -> ToolAttachmentBinding

Look up a tool by alias and return a ready-to-use binding.

resolve_many

resolve_many(
    aliases: Sequence[str],
) -> list[ToolAttachmentBinding]

Resolve multiple tool aliases into bindings at once.

declared_aliases

declared_aliases() -> list[str]

Return a sorted list of all registered tool aliases.

ToolPermissionError

Bases: ToolAttachmentError

Raised when a tool needs permissions that the agent does not have.

UndeclaredToolError

Bases: ToolAttachmentError

Raised when an agent tries to use a tool it was not configured to use.

OutputValidator

Checks that the AI model's response matches the expected output shape.

Takes the raw response from the provider and tries to parse it into the Pydantic model you defined. Raises AgentOutputValidationError if the response does not fit.

validate

validate(
    output_model: type[BaseModel],
    response: ProviderResponse | Any,
) -> BaseModel

Parse the provider response into the expected output model.

If the provider already returned a typed Pydantic model instance (via LangChain's with_structured_output), returns it directly when it matches the expected type. Otherwise extracts the payload from the response and validates it against the output model.

build_response_format

build_response_format(
    output_model: type[BaseModel],
) -> dict[str, Any] | None

Build OpenAI-style response_format from a Pydantic model.

Returns None if the model is a bare BaseModel (no custom fields), since that means no structured output constraint was intended.

wrap_untrusted

wrap_untrusted(
    text: str, *, source: str, flags: tuple[str, ...] = ()
) -> str

Frame untrusted content in explicit provenance delimiters.

Everything between the markers is to be treated as data, never as instructions. Forged delimiters inside text are defanged first so the untrusted content cannot pretend the block has ended.

normalize_declared_tool_refs

normalize_declared_tool_refs(
    declared_tool_refs: Sequence[str],
) -> list[str]

Normalize tool aliases into a stable, deduplicated order.