Agents#

wreath.agents connects model providers, MCP tools, durable jobs, ChatOps identities, approvals, memory, artifacts, and recording without a vendor SDK or a second runtime. Nothing runs on the ordinary request path unless an application creates an agent.

A bounded runtime#

Create the outbound client through the application so Wreath owns its pool, DNS and destination policy for the lifespan. The adapter streams response bytes directly into the selected backplane.

agent-runtime.pypython

import os

from wreath import Wreath
from wreath.agents import (
    AgentCatalog,
    AgentProfile,
    AgentRuntime,
    HTTPClientTransport,
    OpenAIResponsesBackplane,
)

app = Wreath()
http = app.http_client("models", base_url="https://api.openai.com")
transport = HTTPClientTransport(http, base_url="https://api.openai.com/v1")
models = OpenAIResponsesBackplane(
    api_key=os.environ["OPENAI_API_KEY"],
    transport=transport,
)
profile = AgentProfile(
    name="support",
    backplane=models,
    model="gpt-5",
    max_turns=6,
    max_tool_calls=8,
    max_output_tokens=2_000,
    max_total_tokens=12_000,
    timeout=90,
)
runtime = AgentRuntime(AgentCatalog((profile,), default="support"))

The same normalized contract is implemented by AnthropicMessagesBackplane, GeminiGenerateContentBackplane, OpenAICompatibleBackplane, and AzureOpenAIBackplane. Azure accepts either an API key or an asynchronous Entra token provider and fetches a fresh token for each attempt. A catalog may map tenants to different profiles. Retryable provider failure may cross to a declared fallback only before the provider emits an event; partial output and tool effects are never replayed through another model.

For deployments that select a provider by capability, residency and live health, put a RoutedBackplane behind the profile. ModelCandidate declares immutable capability and region facts; ModelRoutePolicy fixes each tenant's ordered candidate set. Construction refuses a tenant with no statically eligible candidate. Request-time routing may narrow, but never widen, those constraints, and never fails over after a provider event.

AgentRuntime.execute(prompt, context=...) yields normalized text, tool-call, usage, and completion events. AgentRuntime.run(...) also implements ChatOps' vendor-neutral AgentBackend protocol.

Governed tools without protocol loopback#

MCPToolCatalog selects a fixed subset of an existing MCP registry. Selection compiles schemas once and refuses unknown, duplicate, over-limit, sampling, and elicitation tools before requests begin.

agent-tools.pypython

from typing import Any

from wreath import Request
from wreath.agents import (
    AgentCatalog,
    AgentProfile,
    AgentRuntime,
    MCPToolCatalog,
)
from wreath.mcp import MCP


def support_agent(models: Any) -> AgentRuntime:
    mcp = MCP(name="support-tools", version="1")

    @mcp.tool(
        description="Read one support case.",
        action="Case::read",
        resource='Case::"requested"',
    )
    async def read_case(request: Request, case_id: str) -> dict[str, Any]:
        return await request.state.cases.read(request.identity, case_id)

    profile = AgentProfile(
        name="support",
        backplane=models,
        model="gpt-5",
        tools=("read_case",),
        delegation_scope=frozenset({"Case::read"}),
    )
    return AgentRuntime(
        AgentCatalog((profile,)),
        tools=MCPToolCatalog(mcp),
    )

Invocation calls the tool directly. It still uses the MCP binding schema, Cedar and role/permission/second-factor checks, tenant-and-principal rate limits, progress, error normalization, and recording. Every call receives an unambiguous effect ID. Operations that require an interactive MCP client session refuse at profile construction instead of failing halfway through a turn.

RemoteMCPClient speaks MCP Streamable HTTP directly to an HTTPS endpoint. It owns initialize/initialized, protocol and session headers, bounded paginated discovery, schema drift detection, JSON and SSE responses, cancellation, close, token lookup and effect-aware audit records. RemoteMCPToolCatalog combines connected clients and refuses tool-name collisions. MCPHTTPClientTransport binds it to an application-owned HTTPClient without enabling redirects or an off-origin credential path. A remote tool with an unknown outcome is never retried.

FederatedToolCatalog combines local and remote catalogs under explicit namespace prefixes. It compiles each participating child once, preserves declared tool order and invokes the owning child directly. Qualified names make equal tool names from different servers unambiguous; malformed names, catalog drift and collisions refuse before an invocation begins.

ChatOps and federated authority#

chat-agent.pypython

from wreath.agents import AgentRuntime
from wreath.chat import ChatOps


def register_support_agent(chat: ChatOps, runtime: AgentRuntime) -> None:
    chat.agent(
        "ask",
        runtime,
        description="Ask the support agent",
        action="Support::ask",
        resource="conversation",
    )

ChatOps.agent requires the normal durable ChatOps jobs and transactional inbox owners. Slack, Teams, and Discord identity resolution happens before command authorization. The resolved Wreath principal is carried into the agent invocation; a profile with delegation_scope calls the existing Principal.narrow(...) owner, so an agent can only lose authority. The direct MCP executor evaluates that narrowed identity again for every tool.

Safety-biased durable effects and approval#

DurableAgent adapts a model completion and effect executor to JobRunner. Turn and tool-call IDs are stable, jobs have automatic retries disabled, and tenant, principal, job key, attempt, and fence are checked before model work. A tool effect is checkpointed only after it succeeds. A checkpoint observed before execution is skipped; recovery attempts are refused because a local checkpoint cannot prove whether a suspended remote effect happened. An unknown outcome is surfaced for an operator instead of guessed safe. Use a stable tool-call ID as the remote system's idempotency key when that system supports one. End-to-end recovery requires a store and effect sink that atomically claim the effect or preserve an explicit unknown state.

ApprovalStore is the persistence contract for human approval. The included InMemoryApprovalStore is bounded, expiring, single-use, tenant-and-principal bound, and may require authentication newer than the approval request. Production deployments can implement the same contract over their durable store.

ChatApprovalFlow(chat, approvals) registers bounded dynamic action prefixes and renders native Slack Block Kit buttons, Teams Adaptive Card actions, or ephemeral Discord components. Approval and denial use only the verified, server-resolved tenant and identity. The action payload carries an opaque approval ID, never authority; the store owns the action, resource, expiry and fresh-auth requirement.

Memory and artifacts#

ContextAssembler performs one bounded store read, checks every returned record's tenant, principal, and conversation, and selects deterministically under item and character ceilings. A memory store must declare positive retention and implement erasure. Records retain explicit trust and provenance labels; Wreath does not silently promote external text to system instructions.

AgentArtifactManager writes through an existing ObjectStore. Byte and artifact-count ceilings are checked before an over-limit chunk reaches storage. Streaming writes hash and forward each chunk in one pass, and returned metadata includes ownership, media type, trust, digest, object stat, and a Provenance envelope.

Recording#

Pass AgentObservability to AgentRuntime to record model and tool boundaries:

agent-observability.pypython

from typing import Any

from wreath.agents import (
    AgentCatalog,
    AgentObservability,
    AgentProfile,
    AgentRuntime,
    MCPToolCatalog,
)
from wreath.mcp import MCP


def observed_agent(
    profile: AgentProfile,
    mcp: MCP,
    deployment_observer: Any,
) -> AgentRuntime:
    return AgentRuntime(
        AgentCatalog((profile,)),
        tools=MCPToolCatalog(mcp),
        observability=AgentObservability(observer=deployment_observer),
    )

The default record contains identity, correlation, provider/model or tool/call ID, duration, outcome, usage, and fallback status—not prompts, arguments, or results. Payload capture requires an explicit bounded RedactionPolicy; structured capture also requires an application redactor. Recorder failures are counted and cannot turn a completed tool effect into a retry.

`wreath.agents`#

Provider-neutral model agents with governed tools and bounded execution.

`AgentBudgetExceeded`#

`AgentArtifact`#

AgentArtifact(key: 'str', artifact_id: 'str', ordinal: 'int', tenant: 'str', principal_id: 'str', conversation: 'str', digest: 'str', media_type: 'str', trust: 'ArtifactTrust', provenance: 'Provenance', stat: 'ObjectStat')

AgentArtifact(key: str, artifact_id: str, ordinal: int, tenant: str, principal_id: str, conversation: str, digest: str, media_type: str, trust: ArtifactTrust, provenance: Provenance, stat: ObjectStat) -> None

`AgentArtifactManager`#

AgentArtifactManager(store: ObjectStore, *, max_bytes: int, max_artifacts: int) -> None

`erase`#

async erase(context: Any, artifact_id: str, *, ordinal: int) -> None

`key`#

key(context: Any, artifact_id: str, *, ordinal: int) -> str

`write`#

async write(context: Any, *, artifact_id: str, ordinal: int, body: bytes | bytearray | memoryview, media_type: str = 'application/octet-stream', trust: ArtifactTrust = 'model') -> AgentArtifact

`write_stream`#

async write_stream(context: Any, *, artifact_id: str, ordinal: int, chunks: AsyncIterable[bytes], media_type: str = 'application/octet-stream', trust: ArtifactTrust = 'model') -> AgentArtifact

`AgentCapturePolicy`#

AgentCapturePolicy(redaction: 'RedactionPolicy', fields: 'frozenset[str]', redact: 'Callable[[str, Any], str] | None' = None)

AgentCapturePolicy(redaction: RedactionPolicy, fields: frozenset[str], redact: Callable[[str, Any], str] | None = None) -> None

`AgentCatalog`#

AgentCatalog(profiles: tuple[AgentProfile, ...], *, default: str | None = None, tenants: Mapping[str, str] | None = None) -> None

`profiles` (property)#

profiles: tuple[AgentProfile, ...]

`select`#

select(tenant: str, *, requested: str | None = None) -> AgentProfile

`AgentConfigurationError`#

`AgentInvocationContext`#

AgentInvocationContext(tenant: 'str', principal: 'Any', conversation: 'str', correlation_id: 'str | None' = None, delegation: 'Any' = None, metadata: 'Mapping[str, Any]' = <factory>)

AgentInvocationContext(tenant: str, principal: Any, conversation: str, correlation_id: str | None = None, delegation: Any = None, metadata: Mapping[str, Any] = <factory>) -> None

`AgentObservation`#

AgentObservation(kind: "Literal['model', 'tool']", tenant: 'str', principal_id: 'str', conversation: 'str', correlation_id: 'str | None', duration: 'float', outcome: 'AgentOutcome', provider: 'str | None' = None, model: 'str | None' = None, request_id: 'str | None' = None, tool: 'str | None' = None, call_id: 'str | None' = None, usage: 'AgentUsage | None' = None, fallback: 'bool' = False, payload: 'tuple[CapturedAgentPayload, ...]' = ())

AgentObservation(kind: "Literal['model', 'tool']", tenant: str, principal_id: str, conversation: str, correlation_id: str | None, duration: float, outcome: AgentOutcome, provider: str | None = None, model: str | None = None, request_id: str | None = None, tool: str | None = None, call_id: str | None = None, usage: AgentUsage | None = None, fallback: bool = False, payload: tuple[CapturedAgentPayload, ...] = ()) -> None

`AgentObservability`#

AgentObservability(*, observer: AgentObserver | None = None, capture: AgentCapturePolicy | None = None) -> None

`model`#

async model(context: Any, *, provider: str, model: str, request_id: str | None, duration: float, outcome: AgentOutcome, usage: AgentUsage | None = None, fallback: bool = False, payloads: Mapping[str, Any] | None = None) -> None

`tool`#

async tool(context: Any, *, tool: str, call_id: str, duration: float, outcome: AgentOutcome, payloads: Mapping[str, Any] | None = None) -> None

`AgentObserver`#

AgentObserver(*args, **kwargs)

`record`#

async record(event: AgentObservation) -> None

`AgentProfile`#

AgentProfile(name: 'str', backplane: 'ModelBackplane', model: 'str', system_prompt: 'str' = '', tools: 'tuple[str, ...]' = (), fallbacks: 'tuple[ModelTarget, ...]' = (), max_turns: 'int' = 8, max_tool_calls: 'int' = 16, max_output_tokens: 'int | None' = None, max_total_tokens: 'int | None' = None, max_prompt_bytes: 'int' = 262144, max_tool_argument_bytes: 'int' = 1048576, max_tool_result_bytes: 'int' = 1048576, timeout: 'float | None' = 120.0, delegation_scope: 'frozenset[str]' = frozenset(), delegation_ttl: 'float' = 300.0)

AgentProfile(name: str, backplane: ModelBackplane, model: str, system_prompt: str = '', tools: tuple[str, ...] = (), fallbacks: tuple[ModelTarget, ...] = (), max_turns: int = 8, max_tool_calls: int = 16, max_output_tokens: int | None = None, max_total_tokens: int | None = None, max_prompt_bytes: int = 262144, max_tool_argument_bytes: int = 1048576, max_tool_result_bytes: int = 1048576, timeout: float | None = 120.0, delegation_scope: frozenset[str] = frozenset(), delegation_ttl: float = 300.0) -> None

`targets` (property)#

targets: tuple[ModelTarget, ...]

`AgentRuntime`#

AgentRuntime(catalog: AgentCatalog, *, tools: _ToolCatalog | None = None, observability: Any = None) -> None

`execute`#

execute(prompt: str, *, context: AgentInvocationContext, profile: str | None = None) -> AsyncIterator[ModelResponseEvent]

`run`#

run(request: AgentRequest) -> AsyncIterator[AgentEvent]

`AgentUsage`#

AgentUsage(input_tokens: 'int' = 0, output_tokens: 'int' = 0, cached_input_tokens: 'int' = 0)

AgentUsage(input_tokens: int = 0, output_tokens: int = 0, cached_input_tokens: int = 0) -> None

`total_tokens` (property)#

total_tokens: int

`AnthropicMessagesBackplane`#

AnthropicMessagesBackplane(*, api_key: str, transport: Transport, model: str | None = None, base_url: str = 'https://api.anthropic.com/v1', api_version: str = '2023-06-01', streaming: bool = True, default_max_output_tokens: int = 4096, max_retries: int = 2, max_response_bytes: int = 16777216) -> None

`stream` (inherited from `_Backplane`)#

stream(request: ModelRequest) -> AsyncIterator[ModelResponseEvent]

`ApprovalDenied`#

`ApprovalExpired`#

`ApprovalGrant`#

ApprovalGrant(approval_id: 'str', tenant: 'str', principal_id: 'str', action: 'str', resource: 'str | None', approved_at: 'float')

ApprovalGrant(approval_id: str, tenant: str, principal_id: str, action: str, resource: str | None, approved_at: float) -> None

`ApprovalMismatch`#

`ApprovalRequest`#

ApprovalRequest(approval_id: 'str', tenant: 'str', principal_id: 'str', action: 'str', resource: 'str | None', issued_at: 'float', expires_at: 'float', require_fresh_auth: 'bool' = False, state: "Literal['pending', 'denied', 'used']" = 'pending')

ApprovalRequest(approval_id: str, tenant: str, principal_id: str, action: str, resource: str | None, issued_at: float, expires_at: float, require_fresh_auth: bool = False, state: "Literal['pending', 'denied', 'used']" = 'pending') -> None

`ApprovalStore`#

ApprovalStore(*args, **kwargs)

`claim`#

async claim(approval_id: str, *, tenant: str, principal_id: str, authenticated_at: float | None = None) -> ApprovalGrant

`deny`#

async deny(approval_id: str, *, tenant: str, principal_id: str) -> None

`issue`#

async issue(*, approval_id: str, tenant: str, principal_id: str, action: str, resource: str | None = None, ttl: float, require_fresh_auth: bool = False, issued_at: float | None = None) -> ApprovalRequest

`ArtifactLimitExceeded`#

`AssembledContext`#

AssembledContext(records: 'tuple[MemoryRecord, ...]', characters: 'int')

AssembledContext(records: tuple[MemoryRecord, ...], characters: int) -> None

`AzureOpenAIBackplane`#

AzureOpenAIBackplane(*, endpoint: str, deployment: str, transport: Transport, api_key: str | None = None, token_provider: TokenProvider | None = None, api_version: str | None = None, streaming: bool = True, max_retries: int = 2, max_response_bytes: int = 16777216) -> None

`stream`#

stream(request: ModelRequest) -> AsyncIterator[ModelResponseEvent]

`BackplaneError`#

BackplaneError(message: str, *, retryable: bool = False, status: int | None = None, request_id: str | None = None, output_started: bool = False) -> None

`ContextAssembler`#

ContextAssembler(store: MemoryStore, *, max_items: int = 32, max_chars: int = 16000) -> None

`assemble`#

async assemble(*, tenant: str, principal_id: str, conversation: str) -> AssembledContext

`erase`#

async erase(*, tenant: str, principal_id: str, conversation: str) -> None

`remember`#

async remember(record: MemoryRecord) -> None

`retention_seconds` (property)#

retention_seconds: float

`CapturedAgentPayload`#

CapturedAgentPayload(field: 'str', length: 'int', value: 'str | None' = None, truncated: 'bool' = False)

CapturedAgentPayload(field: str, length: int, value: str | None = None, truncated: bool = False) -> None

`ChatApprovalFlow`#

ChatApprovalFlow(chat: ChatOps, store: ApprovalStore) -> None

`issue`#

async issue(context: ChatContext, *, approval_id: str, action: str, resource: str | None = None, ttl: float, require_fresh_auth: bool = False) -> ChatReply

`on_approved`#

on_approved(action: str, handler: Callable[[ChatContext, ApprovalGrant], Awaitable[ChatReply]]) -> None

`schema_owners` (property)#

schema_owners: tuple[ApprovalStore, ...]

`DurableAgent`#

DurableAgent(*, name: str, jobs: Any, backend: DurableModelBackend, tools: DurableToolExecutor, checkpoints: EffectCheckpointStore, principal_id: str | None = None) -> None

`enqueue`#

async enqueue(invocation: Any, *, prompt: str, message_id: str) -> int | None

`DurableEffectContext`#

DurableEffectContext(turn_id: 'str', tenant: 'str', principal_id: 'str', correlation_id: 'str | None', job_context: 'Any')

DurableEffectContext(turn_id: str, tenant: str, principal_id: str, correlation_id: str | None, job_context: Any) -> None

`fence` (property)#

fence: int

`DurableModelBackend`#

DurableModelBackend(*args, **kwargs)

`complete`#

async complete(turn: DurableTurn) -> ModelResult

`DurableToolExecutor`#

DurableToolExecutor(*args, **kwargs)

`execute`#

async execute(call: ToolCall, *, context: DurableEffectContext) -> Any

`DurableTurn`#

DurableTurn(turn_id: 'str', tenant: 'str', principal_id: 'str', conversation: 'str', prompt: 'str', correlation_id: 'str | None' = None)

DurableTurn(turn_id: str, tenant: str, principal_id: str, conversation: str, prompt: str, correlation_id: str | None = None) -> None

`as_payload`#

as_payload() -> dict[str, str | None]

`from_invocation` (classmethod)#

from_invocation(invocation: Any, *, prompt: str, message_id: str) -> DurableTurn

`from_payload` (classmethod)#

from_payload(payload: Mapping[str, Any]) -> DurableTurn

`EffectCheckpoint`#

EffectCheckpoint(turn_id: 'str', call_id: 'str', tenant: 'str', principal_id: 'str', fence: 'int')

EffectCheckpoint(turn_id: str, call_id: str, tenant: str, principal_id: str, fence: int) -> None

`EffectCheckpointStore`#

EffectCheckpointStore(*args, **kwargs)

`complete`#

async complete(checkpoint: EffectCheckpoint) -> bool

`completed`#

async completed(turn_id: str, call_id: str, *, tenant: str, principal_id: str) -> EffectCheckpoint | None

`validate_fence`#

async validate_fence(turn_id: str, *, tenant: str, principal_id: str, fence: int) -> None

`GeminiGenerateContentBackplane`#

GeminiGenerateContentBackplane(*, api_key: str, transport: Transport, model: str | None = None, base_url: str = 'https://generativelanguage.googleapis.com/v1beta', streaming: bool = True, max_retries: int = 2, max_response_bytes: int = 16777216) -> None

`stream` (inherited from `_Backplane`)#

stream(request: ModelRequest) -> AsyncIterator[ModelResponseEvent]

`FederatedToolCatalog`#

FederatedToolCatalog(namespaces: Mapping[str, Any], *, separator: str = '__', max_tools: int = 32) -> None

`select`#

select(names: tuple[str, ...]) -> _FederatedToolSet

`HTTPClientTransport`#

HTTPClientTransport(client: HTTPClient, *, base_url: str) -> None

`InMemoryApprovalStore`#

InMemoryApprovalStore(*, max_entries: int = 4096, clock: Callable[[], float] = <built-in function time>) -> None

`claim`#

async claim(approval_id: str, *, tenant: str, principal_id: str, authenticated_at: float | None = None) -> ApprovalGrant

`deny`#

async deny(approval_id: str, *, tenant: str, principal_id: str) -> None

`issue`#

async issue(*, approval_id: str, tenant: str, principal_id: str, action: str, resource: str | None = None, ttl: float, require_fresh_auth: bool = False, issued_at: float | None = None) -> ApprovalRequest

`MemoryRecord`#

MemoryRecord(memory_id: 'str', tenant: 'str', principal_id: 'str', conversation: 'str', content: 'str', trust: 'MemoryTrust', provenance: 'str', created_at: 'float')

MemoryRecord(memory_id: str, tenant: str, principal_id: str, conversation: str, content: str, trust: MemoryTrust, provenance: str, created_at: float) -> None

`MemoryStore`#

MemoryStore(*args, **kwargs)

`append`#

async append(record: MemoryRecord) -> None

`erase`#

async erase(*, tenant: str, principal_id: str, conversation: str) -> None

`recent`#

async recent(*, tenant: str, principal_id: str, conversation: str, limit: int) -> Sequence[MemoryRecord]

`MCPToolCatalog`#

MCPToolCatalog(mcp: Any, *, max_tools: int = 32) -> None

`select`#

select(names: tuple[str, ...]) -> _SelectedMCPTools

`MCPHTTPResponse`#

MCPHTTPResponse(url: 'str', status: 'int', headers: 'Mapping[str, str]', body: 'bytes')

MCPHTTPResponse(url: str, status: int, headers: Mapping[str, str], body: bytes) -> None

`MCPHTTPClientTransport`#

MCPHTTPClientTransport(client: HTTPClient, *, endpoint: str) -> None

`close`#

async close() -> None

`request`#

async request(method: str, url: str, *, headers: Mapping[str, str], body: bytes | None, max_response_bytes: int) -> MCPHTTPResponse

`MCPHTTPTransport`#

MCPHTTPTransport(*args, **kwargs)

`close`#

async close() -> None

`request`#

async request(method: str, url: str, *, headers: Mapping[str, str], body: bytes | None, max_response_bytes: int) -> MCPHTTPResponse

`MCPInvocationAudit`#

MCPInvocationAudit(client: 'str', tool: 'str', call_id: 'str', effect_id: 'str', tenant: 'str', principal_id: 'str', conversation: 'str', correlation_id: 'str | None', outcome: 'str')

MCPInvocationAudit(client: str, tool: str, call_id: str, effect_id: str, tenant: str, principal_id: str, conversation: str, correlation_id: str | None, outcome: str) -> None

`MCPInvocationObserver`#

MCPInvocationObserver(*args, **kwargs)

`record`#

async record(event: MCPInvocationAudit) -> None

`MCPRemoteError`#

`MCPRemoteScopeError`#

`MCPTokenProvider`#

MCPTokenProvider(*args, **kwargs)

`token`#

async token(origin: str) -> str

`MCPToolCatalogDrift`#

`MCPToolLimitExceeded`#

`ModelBackplane`#

ModelBackplane(*args, **kwargs)

`stream`#

stream(request: ModelRequest) -> AsyncIterator[ModelResponseEvent]

`ModelCandidate`#

ModelCandidate(name: 'str', target: 'ModelTarget', capabilities: 'frozenset[str]', regions: 'frozenset[str]')

ModelCandidate(name: str, target: ModelTarget, capabilities: frozenset[str], regions: frozenset[str]) -> None

`ModelMessage`#

ModelMessage(role: "Literal['system', 'user', 'assistant', 'tool']", content: 'str', name: 'str | None' = None, call_id: 'str | None' = None)

ModelMessage(role: "Literal['system', 'user', 'assistant', 'tool']", content: str, name: str | None = None, call_id: str | None = None) -> None

`ModelRequest`#

ModelRequest(model: 'str', messages: 'tuple[ModelMessage, ...]', tools: 'tuple[ToolSpecification, ...]' = (), max_output_tokens: 'int | None' = None, temperature: 'float | None' = None, metadata: 'Mapping[str, Any]' = <factory>)

ModelRequest(model: str, messages: tuple[ModelMessage, ...], tools: tuple[ToolSpecification, ...] = (), max_output_tokens: int | None = None, temperature: float | None = None, metadata: Mapping[str, Any] = <factory>) -> None

`ModelResult`#

ModelResult(output: 'str' = '', tool_calls: 'tuple[ToolCall, ...]' = ())

ModelResult(output: str = '', tool_calls: tuple[ToolCall, ...] = ()) -> None

`ModelResponseEvent`#

ModelResponseEvent(kind: "Literal['text', 'tool_call', 'usage', 'completed']", text: str | None = None, tool_name: str | None = None, tool_call_id: str | None = None, arguments: Mapping[str, Any] | None = None, usage: ModelUsage | None = None, provider_request_id: str | None = None) -> None

`completed` (classmethod)#

completed(*, request_id: str | None = None) -> ModelResponseEvent

`text_delta` (classmethod)#

text_delta(text: str, *, request_id: str | None = None) -> ModelResponseEvent

`tool_call` (classmethod)#

tool_call(name: str, call_id: str, arguments: Mapping[str, Any], *, request_id: str | None = None) -> ModelResponseEvent

`usage_report` (classmethod)#

usage_report(usage: ModelUsage, *, request_id: str | None = None) -> ModelResponseEvent

`ModelRoutePolicy`#

ModelRoutePolicy(tenant: 'str', candidates: 'tuple[str, ...]', required_capabilities: 'frozenset[str]', allowed_regions: 'frozenset[str]')

ModelRoutePolicy(tenant: str, candidates: tuple[str, ...], required_capabilities: frozenset[str], allowed_regions: frozenset[str]) -> None

`ModelTarget`#

ModelTarget(backplane: 'ModelBackplane', model: 'str')

ModelTarget(backplane: ModelBackplane, model: str) -> None

`ModelUsage`#

ModelUsage(input_tokens: 'int' = 0, output_tokens: 'int' = 0, cached_input_tokens: 'int' = 0)

ModelUsage(input_tokens: int = 0, output_tokens: int = 0, cached_input_tokens: int = 0) -> None

`total_tokens` (property)#

total_tokens: int

`OpenAICompatibleBackplane`#

OpenAICompatibleBackplane(*, base_url: str, transport: Transport, api_key: str | None = None, model: str | None = None, streaming: bool = True, max_retries: int = 2, max_response_bytes: int = 16777216) -> None

`stream` (inherited from `_Backplane`)#

stream(request: ModelRequest) -> AsyncIterator[ModelResponseEvent]

`OpenAIResponsesBackplane`#

OpenAIResponsesBackplane(*, api_key: str, transport: Transport, model: str | None = None, base_url: str = 'https://api.openai.com/v1', streaming: bool = True, max_retries: int = 2, max_response_bytes: int = 16777216) -> None

`stream` (inherited from `_Backplane`)#

stream(request: ModelRequest) -> AsyncIterator[ModelResponseEvent]

`PostgresApprovalStore`#

PostgresApprovalStore(session_factory: Callable[[], Any], *, schema: str = 'wreath', clock: Callable[[], float] = <built-in function time>) -> None

`claim`#

async claim(approval_id: str, *, tenant: str, principal_id: str, authenticated_at: float | None = None) -> ApprovalGrant

`component`#

component() -> Component

`deny`#

async deny(approval_id: str, *, tenant: str, principal_id: str) -> None

`issue`#

async issue(*, approval_id: str, tenant: str, principal_id: str, action: str, resource: str | None = None, ttl: float, require_fresh_auth: bool = False, issued_at: float | None = None) -> ApprovalRequest

`purge`#

async purge(*, limit: int = 1000) -> int

`schema_sql`#

schema_sql() -> str

`RemoteMCPClient`#

RemoteMCPClient(endpoint: str, *, transport: MCPHTTPTransport, name: str = 'remote', token_provider: MCPTokenProvider | None = None, audit: MCPInvocationObserver | None = None, max_tools: int = 128, max_pages: int = 16, max_schema_bytes: int = 65536, max_response_bytes: int = 4194304, max_sse_events: int = 256) -> None

`close`#

async close() -> None

`connect`#

async connect() -> None

`invoke`#

async invoke(name: str, arguments: Mapping[str, Any], *, call_id: str, context: Any) -> Mapping[str, Any]

`protocol_version` (property)#

protocol_version: str | None

`session_id` (property)#

session_id: str | None

`specifications` (property)#

specifications: tuple[Any, ...]

`RemoteMCPToolCatalog`#

RemoteMCPToolCatalog(clients: Sequence[RemoteMCPClient], *, max_clients: int = 16) -> None

`connect`#

async connect() -> None

`select`#

select(names: tuple[str, ...]) -> RemoteMCPToolSet

`RemoteMCPToolSet`#

RemoteMCPToolSet(owners: Mapping[str, RemoteMCPClient], names: tuple[str, ...]) -> None

`invoke`#

async invoke(name: str, arguments: Mapping[str, Any], *, call_id: str, context: Any) -> Mapping[str, Any]

`RoutedBackplane`#

RoutedBackplane(candidates: Iterable[ModelCandidate], policies: Iterable[ModelRoutePolicy], *, health: Callable[[ModelCandidate], bool] | None = None) -> None

`stream`#

stream(request: ModelRequest) -> AsyncIterator[ModelResponseEvent]

`StaleAgentFence`#

StaleAgentFence(turn_id: str, fence: int) -> None

`ToolCall`#

ToolCall(name: 'str', arguments: 'Mapping[str, Any]', call_id: 'str | None' = None)

ToolCall(name: str, arguments: Mapping[str, Any], call_id: str | None = None) -> None

`ToolSpecification`#

ToolSpecification(name: 'str', description: 'str', input_schema: 'Mapping[str, Any]')

ToolSpecification(name: str, description: str, input_schema: Mapping[str, Any]) -> None

`UnknownModelOutcome`#

`UnknownToolOutcome`#

`stable_tool_call_id`#

stable_tool_call_id(turn_id: str, index: int, name: str, arguments: Mapping[str, Any]) -> str

`stable_turn_id`#

stable_turn_id(tenant: str, principal_id: str, conversation: str, message_id: str) -> str

Agents#

A bounded runtime#

Governed tools without protocol loopback#

ChatOps and federated authority#

Safety-biased durable effects and approval#

Memory and artifacts#

Recording#

wreath.agents#

AgentBudgetExceeded#

AgentArtifact#

AgentArtifactManager#

erase#

key#

write#

write_stream#

AgentCapturePolicy#

AgentCatalog#

profiles (property)#

select#

AgentConfigurationError#

AgentInvocationContext#

AgentObservation#

AgentObservability#

model#

tool#

AgentObserver#

record#

AgentProfile#

targets (property)#

AgentRuntime#

execute#

run#

AgentUsage#

total_tokens (property)#

AnthropicMessagesBackplane#

stream (inherited from _Backplane)#

ApprovalDenied#

ApprovalExpired#

ApprovalGrant#

ApprovalMismatch#

ApprovalRequest#

ApprovalStore#

claim#

deny#

issue#

ArtifactLimitExceeded#

AssembledContext#

AzureOpenAIBackplane#

stream#

BackplaneError#

ContextAssembler#

assemble#

erase#

remember#

retention_seconds (property)#

CapturedAgentPayload#

ChatApprovalFlow#

issue#

on_approved#

schema_owners (property)#

DurableAgent#

enqueue#

DurableEffectContext#

fence (property)#

DurableModelBackend#

complete#

DurableToolExecutor#

execute#

DurableTurn#

as_payload#

from_invocation (classmethod)#

from_payload (classmethod)#

EffectCheckpoint#

EffectCheckpointStore#

complete#

completed#

validate_fence#

GeminiGenerateContentBackplane#

stream (inherited from _Backplane)#

FederatedToolCatalog#

`wreath.agents`#

`AgentBudgetExceeded`#

`AgentArtifact`#

`AgentArtifactManager`#

`erase`#

`key`#

`write`#

`write_stream`#

`AgentCapturePolicy`#

`AgentCatalog`#

`profiles` (property)#

`select`#

`AgentConfigurationError`#

`AgentInvocationContext`#

`AgentObservation`#

`AgentObservability`#

`model`#

`tool`#

`AgentObserver`#

`record`#

`AgentProfile`#

`targets` (property)#

`AgentRuntime`#

`execute`#

`run`#

`AgentUsage`#

`total_tokens` (property)#

`AnthropicMessagesBackplane`#

`stream` (inherited from `_Backplane`)#

`ApprovalDenied`#

`ApprovalExpired`#

`ApprovalGrant`#

`ApprovalMismatch`#

`ApprovalRequest`#

`ApprovalStore`#

`claim`#

`deny`#

`issue`#

`ArtifactLimitExceeded`#

`AssembledContext`#

`AzureOpenAIBackplane`#

`stream`#

`BackplaneError`#

`ContextAssembler`#

`assemble`#

`erase`#

`remember`#

`retention_seconds` (property)#

`CapturedAgentPayload`#

`ChatApprovalFlow`#

`issue`#

`on_approved`#

`schema_owners` (property)#

`DurableAgent`#

`enqueue`#

`DurableEffectContext`#

`fence` (property)#

`DurableModelBackend`#

`complete`#

`DurableToolExecutor`#

`execute`#

`DurableTurn`#

`as_payload`#

`from_invocation` (classmethod)#

`from_payload` (classmethod)#

`EffectCheckpoint`#

`EffectCheckpointStore`#

`complete`#

`completed`#

`validate_fence`#

`GeminiGenerateContentBackplane`#

`stream` (inherited from `_Backplane`)#

`FederatedToolCatalog`#

`select`#

`HTTPClientTransport`#

`InMemoryApprovalStore`#

`claim`#

`deny`#

`issue`#

`MemoryRecord`#