Runtime¶
Auto-generated from source docstrings (mkdocstrings). The complete public contract is agentmaker.__all__ plus each subpackage __all__.
agentmaker.runtime: the agent runtime machinery (orchestrated / used cross-cuttingly by agents).
Houses the harness (the cross-cutting choke point for LLM / tool calls) plus 6 cross-cutting
capabilities: guardrails / hitl / hooks / execution / sessions / observability. They depend only on
the foundation layer (core / retrieval / tools) and together form the middle layer of the
foundation -> runtime -> agents stack. The harness is the coordinator among them (not a container):
it orchestrates these capabilities at runtime, but the others are used directly by agents as well,
so they live in runtime as peers of the harness.
Harness(llm: LLMClient, *, tool_registry: Optional[ToolRegistry] = None, confirm: Optional[ConfirmCallback] = None, tracer=None, permissions=None, hooks=None, compactor=None, tool_retriever=None, context_builder=None, sources=None, reducer=None, window_budget=None, prompts=None, token_counter: TokenCounter = count_tokens)
The Agent's cross-cutting services: acall_llm / aexec_tool / trace, shared across paradigms.
tool_registry: Tool registry (for tool execution); pure-reasoning paradigms may omit it.
confirm: High-risk tool confirmation callback, forwarded to ToolRegistry; if omitted, the safe choice is to deny (registry default, so a headless server does not hang). CLI scenarios may pass cli_confirm explicitly.
tracer: Optional tracer (must have an emit(event) method); if omitted, trace is a no-op (zero overhead).
permissions: Optional tool permissions (ToolPermissions, allow/deny lists); when attached, exec_tool runs the
permission gate before HITL approval: a tool that is denied / not in allow is rejected outright, without even
asking for approval (deny = hard ban). If not attached, there is no restriction.
hooks: Optional lifecycle hook list (Hook, observe-only); when attached, call_llm fires before/after_model and
exec_tool fires before/after_tool (run-level events are fired by the Agent). Zero overhead if not attached.
compactor: Optional history compactor (HistoryCompactor); compacts overlong history during assemble to prevent token buildup.
tool_retriever: Optional Tool-RAG (ToolRetriever); tools_for takes only the relevant tool subset by query.
context_builder: Optional context assembler (ContextBuilder); once attached together with sources, assemble
retrieves memory/RAG by query and stitches it into a single guardrailed system block for injection (history still stays as role-tagged messages).
sources: Optional list of context sources (CallableSource wrapping memory.search / rag.retrieve, etc.); requires context_builder.
reducer: Optional trajectory-reduction knobs (ReducerConfig); reduce() uses it to decide how many verbatim steps to keep at the tail of each paradigm.
The trajectory's token budget comes from the window budget (see window_budget), not configured here. If omitted, ReducerConfig() defaults are used.
window_budget: Optional window-allocation knobs (WindowBudgetConfig); context_block and reduce use it to divide the
whole window across output reserve / fixed overhead (system + tool schema) / retrieval block / trajectory in one
accounting, replacing the earlier "two half-windows each guessing on their own".
If omitted, WindowBudgetConfig() defaults are used; when the window is unknown (llm.context_window is None), it falls back to "no cap / no reduction".
Half-set assembly fails loud: context_builder and sources must come as a pair. Giving only one means the retrieval block silently has no effect (sources are consumed only by context_block), so raise instead of letting someone believe memory/RAG is wired up. The other dependencies are independent; missing one simply means that capability is off, not a half-set.
async
¶
acall_llm(messages: list[dict], *, tools: Optional[list[dict]] = None, **kwargs) -> LLMResponse
Call the LLM (include tools only when given, to match a direct call) + emit a trace. Returns an LLMResponse.
Timing and the assembled trace event (model/usage/latency/whether there are tool calls) happen only when a tracer is attached; otherwise there is zero extra overhead. When hooks are attached, before_model / after_model fire around the call (timing brackets only the LLM call, not the hook time). When a RunPolicy is attached, check_limits runs before the call and record_llm after (over-limit / cancelled raises and aborts this turn).
async
¶
Stream the LLM: yield text piece by piece; finish cleanly whether it drains normally / the consumer interrupts early / the stream raises mid-flow: emit one trace event (streaming usage is usually None, so only model/latency are recorded) and call record_llm.
When hooks are attached, before_model fires before the stream starts. A plain text stream has no single response object, so after_model does not fire; when tools are passed the stream ends with a terminal LLMResponse (see LLMClient.stream) and after_model fires for it, matching acall_llm. When a RunPolicy is attached, check_limits runs before the start and record_llm after (streaming usage is usually None, only the call count is counted). Wrap-up goes in finally: a generator only runs to its next yield, so if the consumer breaks/closes early (GeneratorExit) or the stream raises mid-flow, statements after the loop are skipped. Only finally guarantees that an already-initiated streaming call is always accounted for and that before_model has a paired wrap-up event.
async
¶
Make the model output a structured result per a pydantic model: inject the schema, call the LLM, pydantic-validate, feed failures back and retry, return the instance. The cross-cutting implementation (single async copy).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
messages
|
The unified message list. |
required | |
schema_model
|
A pydantic BaseModel subclass: it both fixes the output shape and performs validation. |
required | |
retries
|
How many times to retry after a validation failure (default 1). |
_DEFAULT_STRUCTURED_RETRIES
|
|
**kwargs
|
Passed through to the LLM (temperature, etc.). |
{}
|
Returns:
| Type | Description |
|---|---|
|
An instance of schema_model (already validated). |
Notes
The schema goes through each provider's native path via output_schema (json_schema / output_config / responseSchema) and is also injected into the system prompt: native providers get double insurance, json_object/none providers rely on the prompt. Everything is finally pydantic-validated; failures feed the error back and retry. After retries attempts it still raises LLMError if unresolved.
async
¶
aexec_tool(name: str, parameters: dict, *, call_id: Optional[str] = None, decisions: Optional[dict] = None) -> ToolResponse
Execute a tool (with permission gate + high-risk confirmation, the confirmation callback held by this harness) + emit a trace. Returns a ToolResponse. The cross-cutting implementation (single async copy); the permission gate / HITL approval gate are pure-logic helpers _permission_gate / _approval_gate.
Permissions: when permissions are attached, the permission gate runs first: a tool that is denied / not in allow returns a rejection result outright (before HITL). HITL: passing decisions (this turn's decision table {call_id: bool}) means HITL mode. For a high-risk tool, if that call_id has no decision it raises ApprovalRequired (a suspend signal captured by the policy loop); if already rejected it returns a rejection result. Not passing decisions (the default) goes through the synchronous confirm path. If no confirm is passed the safe choice is to deny (no stdin dependency); a blocking confirm like cli_confirm is dispatched to a thread pool (so it does not stall the event loop), but server scenarios should still use HITL (checkpoint_store + decisions) rather than interactive confirm. Timing and the assembled trace event (tool name/params/status/result/latency; secrets and sensitive paths redacted by the Tracer) happen only when a tracer is attached.
async
¶
Assemble the history messages sent to the LLM and, as needed, inject one memory/RAG system context block at the front. The cross-cutting implementation (single async copy).
- When a compactor is attached: overlong history is summary-compacted first (to prevent token buildup; via acompact, with the summary going through _asummarize so it is governed). When compaction actually happens (acompact returns a new list, not the original history object), emit an EVENT_CONTEXT_COMPACT (symmetric with trajectory reduction's EVENT_CONTEXT_REDUCE: it lets the app see whether compaction actually fired and how many old turns were summarized away, no longer leaving saved tokens untraceable).
- When context_builder + sources are attached and query is non-empty: retrieve memory/RAG by query and stitch it into a guardrailed system message inserted at the front (the conversation history still stays as role-tagged messages, not flattened). Retrieval is synchronous IO, dispatched to a thread pool.
history is list[Message]; query is this turn's user input (for memory/RAG retrieval, no block injected if omitted); scope runs through from run and is passed to retrieval sources (each source decides which dimensions to use, see CallableSource). Returns list[dict].
memory/RAG to a single guardrailed system context block of text; returns an empty string if builder/sources are not configured or query is empty.
A public method: the single-loop Agent inserts it as a system message at the front automatically via aassemble; the Plan / Reflection recipes call it directly and stitch it into each stage's prompt. The retrieval block's token budget comes from the window budget (WindowBudget.rag_budget), same source as the trajectory budget, no longer each taking half the window. scope runs through from run and is passed to build_block to source.fetch (each source decides which dimensions to use).
async
¶
The async implementation of context_block (used by aassemble): budget / guardrail / trace match the sync version, but multiple sources run concurrently via abuild_block's gather.
async
¶
Loss-aware reduction of a paradigm's own trajectory (acts only when over budget; emits a trace if reduction happens). kind: agent / plan / reflection. The cross-cutting implementation (single async copy): the reduction functions (REDUCERS) are natively async, and the summary callback is _asummarize (completes within the same event loop, with no thread / loop hopping, so governance counting and limit pass-through naturally hold).
The budget comes from the window budget WindowBudget.trajectory_budget (same-source accounting as the retrieval block; when the window is unknown, no reduction and returned as-is); each paradigm's preserve-policy is in context.reducer, keeping the lifeline (Agent's protected region + recent steps, Reflection's latest answer + critique points, Plan's key numbers). data's shape is determined by kind (agent=messages / plan=list[str] / reflection=list[{kind,text}]); returned as-is means no reduction happened this time.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
**kw
|
The paradigm's private reduction parameters, passed through to the reduction function as-is (e.g. Agent's turn_start=this turn's start index). |
{}
|
async
¶
The tool schema to expose to the LLM this turn: when a tool_retriever is attached, take the relevant subset by query (Tool-RAG), otherwise the full set; finally filter out banned tools by permissions (deny=hard ban, not even the schema is sent: saves tokens + does not leak the full registry to the model). The cross-cutting implementation (single async copy); retrieval is synchronous IO, dispatched to a thread pool to avoid blocking the event loop.
Emit a "gate-blocked / invalid params" tool_call event (shared by aexec_tool's denied/rejected and the unified Agent's invalid_args).
Calls denied by the permission gate, rejected by HITL, or failing parameter parsing previously did not get traced, so an audit could not see "the AI tried to call X but was blocked", exactly the thing that should be recorded. No latency (nothing was really executed); status is "denied" / "rejected" / "invalid_args". A no-op with zero overhead when no tracer is attached.
Emit one structured trace event; a no-op with zero overhead when no tracer is attached.
CallableGuardrail(fn: Callable[[str], Union[bool, GuardrailResult]], *, message: str = 'guardrail triggered')
Bases: Guardrail
Wrap any callable into a guardrail (mirroring CallableSource): fn(text) returns a bool or a GuardrailResult.
When fn returns a bool, False is a tripwire and uses the message given at construction; fn may also return a GuardrailResult directly, carrying its own message. Example: CallableGuardrail(lambda t: len(t) < 4000, message="input too long").
Both sync and async fn are accepted: a pure-computation fn returns a bool / GuardrailResult directly; a fn
that returns an awaitable (an async def, a lambda wrapping an async call, or an object with an async
__call__) is awaited via acheck (the framework execution layer goes through acheck). The synchronous
check is a pure-sync path and fails loud on an awaitable (to avoid bool(coroutine) being always truthy and
silently letting the text through).
message: The block explanation used when fn returns a bool that is False.
check(text: str) -> GuardrailResult
Call fn synchronously: wrap a bool result into a GuardrailResult, pass a GuardrailResult through unchanged. An async fn fails loud on this path.
async
¶
acheck(text: str) -> GuardrailResult
Call fn asynchronously: await an async fn directly; call a sync-signature fn inline, and if it returns an awaitable, await that before normalizing.
Why the second isawaitable fallback: _is_async (iscoroutinefunction) only recognizes functions defined
with async def, not a lambda wrapping an async call or an object with an async call: their signature
is synchronous yet the call returns a coroutine. Without this fallback the un-awaited coroutine would be
judged truthy by _coerce's bool(), silently letting through a guardrail that should have blocked (such as
an async LLM moderation check).
Bases: ABC
Guardrail interface: check a piece of text and return a GuardrailResult. Input guardrails check user input; output guardrails check the model's reply.
Two tracks: synchronous implementations override check (pure-computation guardrails); the framework execution layer (the Agent.run template) calls acheck. By default acheck inlines a direct call to check (most guardrails are pure computation like length / regex checks, not worth dispatching to a thread pool). Override acheck if the guardrail does blocking I/O or wants to call an LLM to moderate.
abstractmethod
¶
check(text: str) -> GuardrailResult
Check text and return a GuardrailResult (passed plus a human-readable message).
async
¶
acheck(text: str) -> GuardrailResult
Async check (called by the framework execution layer).
Defaults to a direct inline call to the synchronous check; guardrails with blocking I/O or LLM moderation override this method.
dataclass
¶
The result of a guardrail check.
Attributes:
| Name | Type | Description |
|---|---|---|
passed |
bool
|
True lets the text through; False signals a tripwire. |
message |
str
|
A human-readable explanation of the block, shown to the user on a tripwire (may be empty when passed=True). |
ApprovalRequired(pending: PendingAction)
Bases: Exception
An internal control-flow signal (not an error): harness.exec_tool raises it in HITL mode when it hits a high-risk tool with no decision. The policy loop catches it, packages the state, and turns it into an Interrupt. Analogous to the OpenAI Agents SDK's RunToolApprovalItem.
dataclass
¶
Interrupt(pendings: list, scope: Optional[Scope] = None)
The suspend result of run / resume: the pending actions plus the credentials needed to resume.
Receiving it means "this turn is suspended, awaiting your approval": read pendings (there may be more than
one), show them to a human, and resume once decided.
- Single action: agent.resume(True, scope=interrupt.scope) (True approves / False rejects; pass scope by keyword).
- Multiple actions (one turn requested several high-risk tools, or parallel sub-agents each suspended):
agent.resume({call_id: True/False, ...}, scope=...) approves / rejects per call_id; you may also pass a
single bool to uniformly approve / reject all pending actions.
Attributes:
| Name | Type | Description |
|---|---|---|
pendings |
list
|
The list of actions awaiting approval (one suspend may contain several). |
scope |
Optional[Scope]
|
The resume credential; resume uses it to load the suspended state back from the CheckpointStore (required for multi-session, may be empty for single-session). |
Convenience: .pending returns the first pending action (single-action case), or None if there are none.
property
¶
pending: Optional[PendingAction]
The first pending action (convenient access for the single-action case); None if there are none.
dataclass
¶
A high-risk action awaiting human approval.
Attributes:
| Name | Type | Description |
|---|---|---|
tool_name |
str
|
The name of the tool to execute. |
arguments |
dict
|
The call arguments. |
call_id |
str
|
The unique identifier of this tool call (the function-calling tool_call id); resume matches decisions against it. |
Lifecycle hook base class (observe-only). Subclass it, override only the events you care about; the rest default to no-ops.
All method return values are ignored (pure side effects); an exception raised inside a method propagates upward (fail loud). A hook is the app's own code, so wrap risky I/O yourself. For interception / modification use Guardrail / Permissions / HITL, not a hook.
on_run_start(input_text: str, *, scope: Optional[Scope] = None)
Fires when a run begins (before the input guardrails). input_text is this turn's user input.
Fires before each LLM call; messages is the message list about to be sent (also fires for streaming calls).
after_model(response: LLMResponse)
Fires after each (non-streaming) LLM call; response is an LLMResponse. Streaming has no single response object, so it does not fire.
Fires just before a tool actually executes (already past the permission gate / HITL approval gate; rejected or suspended tools do not fire); name / parameters are the tool name and its arguments.
after_tool(name: str, parameters: dict, result: ToolResponse)
Fires after a tool executes; result is a ToolResponse.
Fires on a guardrail tripwire; stage is "input" / "output", message is the human-readable block explanation.
on_interrupt(pendings: list[PendingAction], *, scope: Optional[Scope] = None)
Fires on a HITL suspend; pendings is the list of PendingAction awaiting approval (more than one when a turn has several high-risk actions, or parallel sub-agents each suspend).
Fires just before a non-guardrail exception propagates upward; error is that exception (guardrail tripwires go through on_guardrail_trip and do not also fire this event).
ConversationSearch(store: SessionStore, retriever, *, index_sync: Optional[IndexSync] = None)
Bases: SessionStore
A semantically searchable session store: wraps any SessionStore (the source of truth) + a retrieval backbone (the derived index).
Write path: append / append_many stamp each message with a stable message_id (stored in message.metadata, persisted along with the source of truth); after landing in the source of truth, they are fed into the index best-effort via IndexSync.index (an index failure does not affect the source of truth and is marked pending, the same "source of truth is authoritative, eventually consistent" semantics as Memory). On clear, it first fetches all message ids of the session and drops them from the index too.
retriever: The shared retrieval backbone (HybridRetriever); may share the same instance as memory / rag (isolated by base="conversation").
index_sync: Optional index-sync seam; if omitted, SyncIndexSync(retriever) is used (in-process bookkeeping; to persist bookkeeping, pass an instance with SqliteBookkeeping).
Append one message: stamp id, land in the source of truth, feed the index (best-effort).
Atomically append multiple: stamp ids, land in the source of truth in a single transaction, feed the whole batch to the index (best-effort).
Read session messages in insertion order (pure delegation to the source of truth).
clear(*, scope: Optional[Scope] = None, all_scopes: bool = False) -> None
Clear the session: first drop the index by the message_ids in the source of truth (drop is best-effort), then clear the source of truth.
Old messages predating this class have no message_id (not in the index), so just skip them.
list_scopes(*, along: str = 'session', scope: Optional[Scope] = None) -> List[ScopeSummary]
Enumerate a dimension's values in the source of truth (pure delegation; the index is not involved). See SessionStore.list_scopes.
search(query: str, *, top_k: int = 5, scope: Optional[Scope] = None) -> List[RetrievalResult]
Semantically search past conversation messages, returning the most relevant top_k (content like "user: ...", metadata with role / time).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
query
|
str
|
The query text. |
required |
top_k
|
int
|
Number of results to return. |
5
|
scope
|
Optional[Scope]
|
Session placement (the same scope as at write time); isolated retrieval in the shared backbone by base="conversation". |
None
|
Returns:
| Type | Description |
|---|---|
List[RetrievalResult]
|
List[RetrievalResult]: from most to least relevant. |
async
¶
asearch(query: str, **kwargs) -> List[RetrievalResult]
Async version of search (to_thread).
pending_reindex(*, scope: Optional[Scope] = None) -> set
The set of message ids the index sync marked pending (index writes that failed and have not converged), for app monitoring (same semantics as Memory.pending_reindex).
Close the source of truth (the backbone is closed uniformly by whoever shares it).
Wraps conversation retrieval as a tool: the model can proactively "search what we discussed before" (agentic episodic recall, aligned with Letta conversation_search).
Same shape as MemoryTool / RAGTool (based on the Tool base class); read-only, no confirmation gate. scope is bound at construction (matching the served Agent's session scope) and not exposed to the model: the model only supplies the query, never touching the tenant boundary.
dataclass
¶
ScopeSummary(dimension: str, value: str, message_count: int, first_at: Optional[datetime], last_at: Optional[datetime])
A session overview of one value (one "bucket") along a scope dimension: who, how many messages, time span. The element type returned by list_scopes.
Bases: ABC
Conversation-history storage interface: append / read / clear conversation messages by Scope.
Isolation semantics: load / clear match all scope dimensions exactly (an empty scope only hits the "all-empty" default
session bucket, never crossing into other sessions); a truly global operation across all sessions requires an explicit
all_scopes=True (unlike the B semantics of memory / rag retrieval: misreading / mis-deleting a session is far more
dangerous than a retrieval miss, so the default is tightened and the dangerous path made explicit).
abstractmethod
¶
Append one message to the session identified by scope (append-only).
Atomically append multiple messages (a turn's user+assistant land together, avoiding a half-turn of history).
The default implementation appends one by one (non-atomic, a fallback only); a backend with transaction support should override it as a single transaction (see SqliteSessionStore).
abstractmethod
¶
Read all messages of the scope's session in insertion order; an empty session returns [].
By default it matches all scope dimensions exactly (an empty scope reads only the "all-empty" default bucket, not crossing into other sessions);
all_scopes=True reads across all sessions (for ops / export, use with care).
abstractmethod
¶
clear(*, scope: Optional[Scope] = None, all_scopes: bool = False) -> None
Clear all messages of the scope's session (matching all scope dimensions exactly); all_scopes=True clears all sessions (use with care).
async
¶
Async version of append (defaults to to_thread).
async
¶
Async version of append_many (defaults to to_thread).
async
¶
Async version of load (defaults to to_thread).
async
¶
aclear(*, scope: Optional[Scope] = None, all_scopes: bool = False) -> None
Async version of clear (defaults to to_thread).
list_scopes(*, along: str = 'session', scope: Optional[Scope] = None) -> List[ScopeSummary]
Enumerate which values exist along the along dimension, giving each bucket a message count and first/last time (e.g. list "which sessions exist" for the app to build a conversation list).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
along
|
str
|
The dimension to enumerate (base/user/agent/session/app), default "session". |
'session'
|
scope
|
Optional[Scope]
|
Uses non-empty dimensions (B semantics) to constrain the "other dimensions": e.g. Scope(user="alice") lists
only sessions under alice; passing None / Scope() enumerates globally. This is a read-only discovery operation
that returns no message bodies and deletes no data, so it uses B semantics (unlike load/clear's exact match of all
dimensions). The enumerated |
None
|
Returns:
| Type | Description |
|---|---|
List[ScopeSummary]
|
List[ScopeSummary]: descending by most recent activity (last_at); |
Raises NotImplementedError by default: not every backend can DISTINCT + aggregate efficiently; SqliteSessionStore implements it.
async
¶
alist_scopes(*, along: str = 'session', scope: Optional[Scope] = None) -> List[ScopeSummary]
Async version of list_scopes (defaults to to_thread).
Bases: SessionStore
Conversation history stored in SQLite: one row per message (scope columns + role / content / time / metadata), ordered by rowid.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
db_path
|
str
|
SQLite file path; the default ":memory:" is for self-tests only, use a file path in production to persist (may share the database with memory / the vector store by passing the same file path). |
':memory:'
|
Append one message (append-only: only INSERT, never modify old rows). metadata is stored as JSON, and time as a UTC-normalized isoformat.
Write multiple messages in a single transaction (all succeed or all fail), avoiding a half-turn of history where the user write succeeds but the assistant write fails.
Read all messages of the scope's session by rowid (i.e. insertion order).
By default it matches all scope dimensions exactly (an empty scope reads only the "all-empty" default bucket); all_scopes=True reads across all sessions.
clear(*, scope: Optional[Scope] = None, all_scopes: bool = False) -> None
Delete all messages of the scope's session (by default matching all scope dimensions exactly); all_scopes=True clears all sessions (use with care).
prune(*, scope: Optional[Scope] = None, all_scopes: bool = False, keep_last: Optional[int] = None, before: Optional[datetime] = None) -> int
Truncate a session's history and return the number of messages deleted: keep_last=N keeps only the most recent N of this scope, before=time deletes messages earlier than that time.
Sessions accumulated over the long term make each turn's load full-table scan and grow slower; the app can prune periodically to control single-session length / clean up stale sessions.
At least one of keep_last and before must be given (giving neither raises, to guard against an accidental wipe; to clear the whole session use clear); if both are given, it first deletes messages earlier than
before, then truncates the remainder by keep_last. By default it matches all scope dimensions exactly (like load / clear); all_scopes=True spans all sessions
(in which case keep_last means "the most recent N across all sessions combined", mixing across users, use with care).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
scope
|
Optional[Scope]
|
Target session; defaults to the "all-empty" bucket. |
None
|
all_scopes
|
bool
|
Span all sessions (use with care). |
False
|
keep_last
|
Optional[int]
|
Keep only this many most recent messages (must be >= 0). |
None
|
before
|
Optional[datetime]
|
Delete messages earlier than this time (naive is treated as UTC). |
None
|
list_scopes(*, along: str = 'session', scope: Optional[Scope] = None) -> List[ScopeSummary]
DISTINCT + aggregate along the along dimension to get each bucket's message count and first/last time (B semantics constrain the other dimensions). See the base class list_scopes.
Bases: ABC
Checkpoint storage interface: save / load / clear the current execution-state checkpoint by Scope.
abstractmethod
¶
save(state_json: str, *, scope: Optional[Scope] = None) -> None
Save (overwrite) the current checkpoint for scope; state_json is a serialized ExecutionState JSON string.
abstractmethod
¶
load(*, scope: Optional[Scope] = None) -> Optional[str]
Read the current checkpoint for scope; return None if absent.
abstractmethod
¶
clear(*, scope: Optional[Scope] = None) -> None
Clear the checkpoint for scope (called after a run completes normally or a resume succeeds).
async
¶
asave(state_json: str, *, scope: Optional[Scope] = None) -> None
Async version of save (defaults to to_thread).
dataclass
¶
ExecutionState(messages: list[dict], input_text: str, remaining: int = 0, decisions: dict = dict(), pending: list = list(), meta: dict = dict(), run_id: Optional[str] = None, step: int = 0, completed: bool = False)
The unified execution state of one run.
Unified fields (shared by all paradigms, the only ones cross-cutting concerns recognize): messages: The execution trajectory (OpenAI dicts: system / user / assistant[with tool_calls] / tool). input_text: This run's user input (stored into conversation history on completion). remaining: How many more LLM calls are allowed (guards against exceeding the iteration limit). decisions: The HITL decision table {call_id: bool}. pending: The LIST of HITL actions currently suspended awaiting approval (shown to the caller); an empty list when not suspended. One suspend can contain multiple (multiple high-risk tools requested in one turn, or parallel sub-agents each suspending), and resume injects decisions by call_id. completed: Whether this run has finished (during finalization, mark first, then clear the checkpoint). A leftover checkpoint with completed=True means "finished, only cleanup remains"; _guard_pending / _load_execution_state use this to clear it directly rather than re-run, resolving the double-execution / double-bookkeeping window from "persisting history and clearing the checkpoint are not atomic". Paradigm-specific: meta: Each paradigm's own resume state (e.g. fc's pending_calls, ReAct's step index, Plan's steps/cursor). This is where the "per-paradigm snapshot/restore hook" lands: the paradigm-private bit of state beyond the unified fields. Trace correlation: run_id: This run's trace correlation id; saved on suspend, restored on resume (continuing the same run_id across suspend). step: The trace step number this run has reached; same, used for resume continuation (see execution/run_context).
Serialize to a JSON string (stored into CheckpointStore); pending is flattened into plain fields.
The unified fields are all JSON-friendly types; the paradigm-private meta / messages / pending.arguments are filled by each paradigm, and if they store a non-serializable object, this wraps the bare TypeError into a clear SessionError (pointing to where to look), making it easy to locate which paradigm wrote dirty state.
classmethod
¶
from_json(raw: str) -> ExecutionState
Restore from a JSON string (used when CheckpointStore loads it back).
dataclass
¶
RunPolicy(max_llm_calls: Optional[int] = None, max_tool_calls: Optional[int] = None, max_tokens: Optional[int] = None, deadline_seconds: Optional[float] = None, cancel: Optional[Callable[[], bool]] = None)
A single run's limits and cancellation (attached to an Agent, counted independently per run; a None field = that item is unlimited).
Nested semantics: limits are counted against the OUTERMOST run globally. A nested child Agent's own RunPolicy (AgentTool delegation / Plan/Reflection child executor) does not take effect inside the parent run (it warns). To limit subtasks, set the global limits on the parent Agent's run_policy.
Fields
max_llm_calls: Maximum LLM calls in a run (including streaming, including nesting such as Plan child executors); must be >= 1 (a run must call the LLM at least once).
max_tool_calls: Maximum tools ACTUALLY executed in a run (those blocked by permission / confirmation are not counted); must be >= 0 (0 = tools disabled this run:
the LLM can still be called, and the run aborts the moment the model wants to execute a tool: a hard limit for a "read-only / safe mode").
max_tokens: Cumulative token limit for a run (the sum of usage.total_tokens across LLM responses); must be >= 1.
deadline_seconds: A run's wall-time limit (measured from the start of the run; a resume is a new run and re-times); must be > 0.
cancel: Cooperative cancellation hook () -> bool, checked before each LLM / tool call, returning True aborts. Must be fast and non-blocking;
when one instance serves multiple sessions, the callback can call current_run_id() to identify which run it is and decide accordingly (cancel per run).
Bases: CheckpointStore
Checkpoints persisted to SQLite: at most one row per scope (full-dimension unique index on scope + upsert overwrite, always keeping only the latest).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
db_path
|
str
|
SQLite file path; defaults to ":memory:" for self-tests only. In production pass a file path to persist (can share a database with session / memory by passing the same file path). |
':memory:'
|
save(state_json: str, *, scope: Optional[Scope] = None) -> None
Overwrite-save: use the full-dimension unique index on scope to upsert (INSERT OR REPLACE): exactly one row per scope, a single atomic statement.
More robust than the old "DELETE+INSERT as two statements": a single statement has no "deleted but not yet inserted" intermediate state, and the unique index means concurrent saves of the same scope only replace and never produce duplicate rows (whereas the two interleaved statements could leave two rows).
load(*, scope: Optional[Scope] = None) -> Optional[str]
Read the current checkpoint for scope (exact match on all scope dimensions; the unique index guarantees at most one row); return None if absent.
Bases: TraceExporter
Appends each event as one JSON line to a file (JSON Lines): convenient for streaming appends, line-by-line reads, and feeding into log pipelines.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
path
|
str
|
Output file path; opened in append mode. Call close before the process exits (or close the whole Tracer). |
required |
Bases: TraceExporter
Collects events into an in-memory list (Tracer's default backend; summary / str / tests read it). Lost on restart; for in-process observation / debugging only.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
max_events
|
Optional[int]
|
How many events to keep at most (ring buffer: drops the oldest when exceeded), default 2048 (aligned with OTel convention). Prevents unbounded memory growth for a long-running process with tracing on, and prevents summary / str full scans from getting slower. Pass None explicitly to retain everything, or attach a JsonlExporter / SqliteExporter. |
2048
|
Append the event to the in-memory list; drop the oldest when exceeding max_events (ring buffer, keeps only the most recent).
Bases: TraceExporter
Maps events to OpenTelemetry spans, connecting to standard backends like Jaeger / Grafana / Datadog.
Lazily imports opentelemetry (raises a clear error at construction if not installed, without dragging down
other exporters). One span per event:
- Real duration: the event already carries latency_ms, so the span start is rewound to now - latency and
the end set to now, giving Jaeger's waterfall a real width (rather than a zero-width isolated point).
Events without latency get an instantaneous span.
- Parent span attribution: with the default carrier_provider=None, no context is passed, so OTel uses the
current active context (if the app wraps await agent.arun(...) in
with tracer.start_as_current_span(...), the AB span naturally attaches; otherwise each becomes a root).
Pass a carrier_provider (e.g. current_trace_carrier) to explicitly extract and attach the parent
context from an upstream W3C carrier, for the case where the app holds a traceparent but has not set it as
the current OTel context (e.g. only a header was passed across processes). The carrier is obtained
out-of-band via a callback (not part of the event dict), which both avoids Tracer redaction accidentally
masking traceparent's 32-hex segment and avoids changing the event schema.
- run_id is always attached as a span attribute, so backends can filter by it to aggregate per run.
Event fields become span attributes (non-primitive types are JSON-encoded, None is skipped: OTel attributes
only accept primitive types).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
tracer_name
|
str
|
OTel tracer name (instrumentation scope). Span export / sampling is decided by the app-configured OTel SDK TracerProvider (this class only produces spans, not bound to a specific backend). |
'agentmaker'
|
carrier_provider
|
Optional zero-arg callback |
None
|
Produce a span with real duration (parent context per the class docstring), event fields as attributes; run_id as an attribute so backends can aggregate per run.
Bases: TraceExporter
Inserts each event as one SQLite row (type + full event JSON + created_at): can share a DB with sessions / memory, and is SQL-queryable for auditing.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
db_path
|
str
|
SQLite file path; defaults to ":memory:" for self-test only. Provide a file path in production to persist. |
':memory:'
|
Tracer(*, redact: bool = True, max_value_len: int = 200, exporters: Optional[list[TraceExporter]] = None, strict: bool = False, extra_secret_keys: Optional[Iterable[str]] = None, extra_secret_patterns: Optional[Iterable[Any]] = None)
Minimal tracer that collects structured events; on emit it redacts + truncates long values, then fans out to each exporter.
The Harness sends events via emit(event); tracer.events reads all of them from memory (already redacted, kept
by the default MemoryExporter), str(tracer) shows the timeline, tracer.summary() shows event count / usage /
latency totals, and tracer.close() closes file / DB backends.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
redact
|
bool
|
Whether to redact (default True; §8 red line, normally leave it on: turn it off only in tests that are known to carry no sensitive data). Only controls "secret / PII masking", and does not affect truncation: with redaction off, long values are still truncated to max_value_len. |
True
|
max_value_len
|
int
|
Maximum characters kept for a single string value, truncated beyond that (prevents the trace from being blown up by long text / tool results); always in effect. |
200
|
exporters
|
Optional[list[TraceExporter]]
|
Where events go (a list of TraceExporter); defaults to |
None
|
strict
|
bool
|
Whether to re-raise when a single exporter throws (default False = fault-tolerant, swallow). A trace is side-channel observation; one sink failing (disk full / DB lock / OTel collector unreachable) should not take down the main flow; pass True when debugging / testing wants fail-loud. |
False
|
extra_secret_keys
|
Optional[Iterable[str]]
|
Additional secret key names contributed by the app (substring match, case-insensitive):
if a key name contains any of these, its value is masked to |
None
|
extra_secret_patterns
|
Optional[Iterable[Any]]
|
Additional secret-value regexes contributed by the app (str or a compiled
re.Pattern): matched fragments are masked to |
None
|
property
¶
Convenience read of events collected in memory: returns the list of the first MemoryExporter (the default config has one); empty list if none.
Receive one event: after redaction (optional) + long-value truncation (always), fan out to each exporter. Called by the Harness.
A single exporter throwing is swallowed by default (fault tolerance: side-channel observation does not take down the main flow); with strict=True it re-raises (fail-loud). Cleaning itself throwing (e.g. an odd object's str() raising) also does not bubble up and kill the run: the event is dropped and counted (only re-raised when strict).
Statistics: total event count, count by type, cumulative token usage, cumulative latency (ms).
Clear the events collected in memory (only affects MemoryExporter; file / DB sinks are untouched).
Close all exporters (release file / DB handles); call before the process exits.
Bases: ABC
Trace export backend interface: export is called once per (already redacted) event; close releases resources (files / connections).
Command-line high-risk confirmation battery: prints the action, asks y/n, returns whether it was approved.
An explicit battery, not the default: CLI / teaching scenarios pass confirm=cli_confirm explicitly to use it. The
framework no longer treats it as a fallback default: with no confirm passed, the safe choice is to deny (registry
semantics), avoiding a headless server input() that hangs forever / raises EOFError.
The current run's run_id; None if not inside a run context. An app / hook can read it to correlate its own logs.
The current run's scope (Scope); None if not inside a run context.
Lets a tool fetch "the parent Agent's scope for the current run" during execution. The typical case is AgentTool delegating to a child Agent isolating history by the parent session, avoiding "multiple sessions sharing one tool instance -> child Agent history leaking across sessions". The Tool.run(parameters) signature cannot obtain scope, so it is passed through via this contextvar.
The step number the current run has reached (for persistence and resume continuation); 0 if not in a context.
The current run's upstream W3C trace carrier ({"traceparent": ...}); None if not in a run context or not passed.
Typical use: OTelExporter(carrier_provider=current_trace_carrier). Export and run are in the same async context
(emit is an inline synchronous call), so what is read is the current run's carrier, making each AB span a child of
the app's request span (rather than each becoming its own root).
async
¶
Governed llm.chat (async): check_limits -> await chat -> record_llm (count + tokens) -> optional trace -> hard token limit.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
llm
|
LLMClient (or an object of the same shape). |
required | |
messages
|
Message list, passed to chat as-is. |
required | |
tracer
|
Optional tracer (duck-typed emit); zero overhead if not attached. |
None
|
|
origin
|
str
|
Origin label for the event (e.g. "memory.summary" / "rag.mqe"), to distinguish bypass calls in the trace. |
''
|
**kwargs
|
Passed through to llm.chat. |
{}
|
Returns:
| Type | Description |
|---|---|
|
LLMResponse. |