Executor

Running tool calls: retries, plans and the tool loop. Concepts: Executor, Executor policies.

agent_adapter.exec.integration is a seam for flow compilers and framework bridges; most code never needs it. Checkpoint stores for crash-safe runs: Durable runs.

agent_adapter.exec

Executor: runs tools with guards, hooks, approval, cache, retries, batches, plans and loops.

OnError = Literal['collect', 'fail_fast'] module-attribute

AgentDepthExceeded

Bases: RunLimitExceeded

An agent tool was called deeper than its max_depth (agents calling agents).

Backoff dataclass

Exponential backoff with jitter; delays in seconds.

delay(attempt, rng)

Seconds to wait after failed attempt attempt (1-based).

Raises:

Type	Description
ToolUsageError	attempt is below 1.

HandoffModel

Wraps a model: once a handoff tool has answered, the loop ends with that answer.

cache_key_fields()

The wrapped model's cache key fields.

generate(request) async

The handoff's answer if one just arrived, else the wrapped model's reply.

stream(request) async

The handoff's answer as one chunk if one just arrived, else the wrapped stream.

LoopAgent dataclass

A run_tool_loop agent: its model, the tools it may use and its instructions.

first_messages(task)

The history a fresh run starts from: the instructions, then the task.

SyncTwins

run_sync, run_call_sync, run_many_sync and resume_sync for the Executor.

Each blocks until done on one long-lived loop shared by the process (a runtime.LoopRunner), so loop-bound state such as clients and semaphores stays valid across calls. In async code, await the async method instead.

resume_sync(approval_id, decision=None, ctx=None, *, decided_by='', reason='', timeout=None)

resume from sync code: decide a parked call and run it once.

run_call_sync(call, ctx=None, *, registry=None, timeout=None)

run_call from sync code; failures are in the ToolResult.

run_many_sync(calls, ctx=None, *, concurrency=None, on_error='collect', registry=None, timeout=None)

run_many from sync code; results keep input order.

run_sync(tool, args=None, ctx=None, *, call_id=None, timeout=None)

run from sync code; returns the value and raises like run.

Raises:

Type	Description
TimeoutError	past timeout seconds.
ToolUsageError	called from a tool already running on the shared loop.

ToolLoopResult dataclass

Where a tool loop ended: the full history, the last reply and every tool result.

agent_depth()

How many agent tools are running around the current call (0 outside any).

agent_tool(agent, *, name, description, executor=None, max_depth=DEFAULT_MAX_DEPTH, timeout=None, confirm=False, tags=())

A tool (kind "agent") that runs agent on the model's task inside the calling run.

default_executor()

The default Executor, created on first use, on the default registry.

handoff(to, *, name, description, executor=None, max_depth=DEFAULT_MAX_DEPTH, timeout=None)

An agent tool tagged "handoff": under a HandoffModel, its answer is the loop's answer.

is_control_flow(exc)

True for CancelledError, a registered type, a class named GraphBubbleUp in the MRO, or an exception group holding any of them.

recover_tool_loop(model, registry, run_id, *, checkpoints, checkpoint_key=None, rerun_in_doubt=False, max_steps=8, executor=None, ctx=None, output_schema=None, tool_choice='auto', concurrency=None, stage='tool_loop', request_options=None, compression=None, on_text=None, on_event=None) async

Continue a durable tool loop from its last checkpoint; finished tool calls replay.

A call that started but logged no result runs again only if its tool is idempotent=True, or with rerun_in_doubt=True after you checked it. max_steps counts the model calls of the whole run, before the crash too; tool_results and usage cover this call only.

Raises:

Type	Description
CheckpointNotFound	nothing is stored for run_id.
CheckpointConflict	another worker took the run, it ended, or it is not this loop.
CheckpointStoreCorrupt	the stored checkpoint or its signature does not check out.
RecoveryNeedsDecision	a call may have run and its tool is not idempotent.
(ToolUsageError, ConfigError, ModelError)	as run_tool_loop.

register_control_flow(*types)

Treat types (and subclasses) as control flow in the executor and the approval ports.

Raises:

Type	Description
ToolUsageError	no types given, or one is not a BaseException subclass.

reset_default_executor()

Drop the default Executor and its default sinks; the next use starts fresh.

runner_for(executor)

executor when given, else the default Executor.

Also exported here, documented under the canonical path: Executor, RunLimits, Step, acall, call, configure, run_tool_loop, set_default_executor.

agent_adapter.exec.checkpoints

Checkpoint stores for crash-safe runs: the port, and memory, SQLite, Redis and SQL backends.

CallState = Literal['started', 'done', 'pending', 'unrecordable'] module-attribute

CheckpointKind = Literal['flow', 'loop'] module-attribute

CheckpointStatus = Literal['running', 'paused', 'done', 'denied', 'failed'] module-attribute

CallRecord

Bases: BaseModel

One tool call of a durable run: started, done (result kept), pending or unrecordable.

Checkpoint

Bases: BaseModel

One saved position of a run: what it is, its status, and the sealed document.

CheckpointStore

Bases: Protocol

Where durable runs keep their latest checkpoint and the tool calls of the current step.

save is a compare-and-set on the step: step 0 only for a new run, else exactly one past the stored step; anything else raises CheckpointConflict, so two workers never both continue a run. A save that moves to a new segment drops the older segments' call logs. record_call refuses a segment that is not the run's current one (CheckpointConflict).

MemoryCheckpointStore

A CheckpointStore in a dict; it survives a failed run, not a process exit.

calls(run_id, segment) async

The call records of one segment, in the order first written.

delete(run_id) async

Forget the run and its call log; False when nothing was stored.

latest(run_id) async

The run's latest checkpoint, or None.

list_unfinished(*, name=None, limit=100) async

Runs whose latest checkpoint is running, oldest first, optionally of one name.

Raises:

Type	Description
ConfigError	a negative limit.

record_call(run_id, segment, call) async

Keep a call record in the run's current segment (latest record per key wins).

Raises:

Type	Description
CheckpointConflict	segment is not the run's current one.

save(run_id, step, checkpoint) async

Store the run's next checkpoint (compare-and-set on step).

Raises:

Type	Description
CheckpointConflict	step is not 0 for a new run or one past the stored step.
ToolUsageError	a bad run id, step or checkpoint.

RedisCheckpointStore

CheckpointStore on Redis (agent-adapter[redis]), shared by every worker and host.

client_factory() returns a redis.asyncio.Redis (one per event loop), e.g. lambda: Redis.from_url(url). save, record_call and delete are Lua scripts, so a step's compare-and-set is atomic. A run's keys expire ttl_s after its last write (None: never), except while it is paused: a paused run waits for a person and never expires. On Redis Cluster put a hash tag in prefix ("{checkpoints}:"), since the scripts touch a run's keys and the shared index together.

Raises:

Type	Description
ConfigError	prefix is not a string ending in ":", or ttl_s is not a finite number > 0 or None.

aclose() async

Close the running loop's client.

calls(run_id, segment) async

The call records of one segment, in the order first written.

Raises:

Type	Description
CheckpointStoreCorrupt	a stored record is unreadable or has another key.

delete(run_id) async

Forget the run and its call log; False when nothing was stored.

latest(run_id) async

The run's latest checkpoint, or None.

Raises:

Type	Description
CheckpointStoreCorrupt	the stored hash is unreadable or disagrees with itself.

list_unfinished(*, name=None, limit=100) async

Runs whose latest checkpoint is running, oldest first, optionally of one name.

Raises:

Type	Description
ConfigError	a negative limit.

record_call(run_id, segment, call) async

Keep a call record in the run's current segment (latest record per key wins).

Raises:

Type	Description
CheckpointConflict	segment is not the run's current one.

save(run_id, step, checkpoint) async

Store the run's next checkpoint (compare-and-set on step).

Raises:

Type	Description
CheckpointConflict	step is not 0 for a new run or one past the stored step.
CheckpointStoreCorrupt	the stored step is not an integer.
ToolUsageError	a bad run id, step or checkpoint.

SQLAlchemyCheckpointStore

CheckpointStore on a SQLAlchemy async engine (agent-adapter[sqlalchemy]), every host.

engine is a URL ("postgresql+asyncpg://...") or a callable returning an AsyncEngine; one engine is made per event loop. A step's compare-and-set is one INSERT ... ON CONFLICT DO NOTHING (step 0) or UPDATE ... WHERE step = n - 1. metadata holds both tables, for Alembic. Backend errors propagate (fail closed).

Raises:

Type	Description
ConfigError	a bad engine or table; on first use, another dialect or a table without the store's columns.
MissingDependency	SQLAlchemy is not installed.

aclose() async

Dispose of the running loop's engine.

calls(run_id, segment) async

The call records of one segment, in the order first written.

Raises:

Type	Description
CheckpointStoreCorrupt	a stored record is unreadable or has another key.

delete(run_id) async

Forget the run and its call log; False when nothing was stored.

latest(run_id) async

The run's latest checkpoint, or None.

Raises:

Type	Description
CheckpointStoreCorrupt	the stored row is unreadable or disagrees with itself.

list_unfinished(*, name=None, limit=100) async

Runs whose latest checkpoint is running, oldest first, optionally of one name.

Raises:

Type	Description
ConfigError	a negative limit.

record_call(run_id, segment, call) async

Keep a call record in the run's current segment (latest record per key wins).

Raises:

Type	Description
CheckpointConflict	segment is not the run's current one.

save(run_id, step, checkpoint) async

Store the run's next checkpoint (compare-and-set on step).

Raises:

Type	Description
CheckpointConflict	step is not 0 for a new run or one past the stored step.
ToolUsageError	a bad run id, step or checkpoint.

SQLiteCheckpointStore

CheckpointStore in a SQLite file (WAL), safe across threads, loops and processes.

Each call runs in a worker thread on its own connection; writes hold BEGIN IMMEDIATE, so a step's compare-and-set is atomic for every process; timeout is the lock wait (s).

Raises:

Type	Description
ConfigError	a bad path or timeout (here); on first use, a database without WAL support, foreign tables with these names, or another schema version.

calls(run_id, segment) async

The call records of one segment, in the order first written.

Raises:

Type	Description
CheckpointStoreCorrupt	a stored record is unreadable or has another key.

delete(run_id) async

Forget the run and its call log; False when nothing was stored.

latest(run_id) async

The run's latest checkpoint, or None.

Raises:

Type	Description
CheckpointStoreCorrupt	the stored row is unreadable or disagrees with itself.

list_unfinished(*, name=None, limit=100) async

Runs whose latest checkpoint is running, oldest first, optionally of one name.

Raises:

Type	Description
ConfigError	a negative limit.

record_call(run_id, segment, call) async

Keep a call record in the run's current segment (latest record per key wins).

Raises:

Type	Description
CheckpointConflict	segment is not the run's current one.

save(run_id, step, checkpoint) async

Store the run's next checkpoint (compare-and-set on step).

Raises:

Type	Description
CheckpointConflict	step is not 0 for a new run or one past the stored step.
ToolUsageError	a bad run id, step or checkpoint.

StoredCheckpoint

Bases: BaseModel

A run's latest checkpoint and its step; the next save must use step + 1.

idempotency_key()

A key unique to the running tool call of a durable run, the same after a recovery.

Pass it to an external API that accepts idempotency keys (payments, mail), so a call that runs again after a crash, or on a worker that was cut off, is not applied twice.

agent_adapter.exec.control

agent_adapter.exec.control re-exports, documented under the canonical path: is_control_flow, register_control_flow.

agent_adapter.exec.integration

Semi-public executor seam for framework integrations (LangGraph, ADK); not in exec.__all__.

Integrations that drive the executor's approval, journal and resume internals import them from here, never from the private modules, so a refactor there keeps this surface stable.

AnswerEntry = tuple[ApprovalRequest, ApprovalDecision, bool] module-attribute

JournalData = dict[str, Any] module-attribute

Pending = tuple[ApprovalRequest, ApprovalStore | None] module-attribute

RAN_UNKEYED = 'ran, arguments not sealable' module-attribute

gather_unwrapped = _gather_unwrapped module-attribute

Run factories concurrently in input order; the first escaping exception cancels the rest.

make_answer = _make_answer module-attribute

The ResumeAnswer for a visit from its AnswerEntrys.

make_journal = _make_journal module-attribute

A Journal for one node visit (replays finished calls when the node re-runs).

settle_own = _settle_own module-attribute

Decide and consume a resumed answer's record in the executor's own store, if it is there.

visit_scope = _visit_scope module-attribute

Context manager installing a resume answer and a journal for one node visit.

Journal

The finished executor calls of one node visit, replayed when a resumed node re-runs.

A call is keyed by tool, the canonical digest of its validated arguments, and how many calls with that key came before it in the visit. Child journals (a Map item, a nested flow's node) are keyed by name and occurrence. Pending results are never recorded; a call that ran but cannot be recorded leaves a marker, so a pause fails instead of re-running it.

durable property

Whether a durable run logs this journal's calls.

child(name, seed=None, *, calls=True)

The next child journal named name (recorded one, else a new one from seed).

claim(tool, arguments, returns=None)

The next slot for this call (None when it is not journaled).

Parameters:

Name	Type	Description	Default
returns	Any	The tool's return annotation; the classes it reaches may be replayed.	None

Raises:

Type	Description
ToolUsageError	the call already ran in this visit with a result that could not be recorded (it must not run again).

close()

Record and replay nothing from now on (children too).

data(drop=())

Everything recorded, as JSON data; children named in drop are left out.

record(slot, result)

Keep a finished result for replay, or a marker when it cannot be kept.

settle(slot, result) async

Record a call's result, then tell the observer what was kept for its replay.

started(slot) async

Tell the observer a call is about to run, or that it replayed.

NotSealable

Bases: ToolUsageError

A value has no canonical encoding; path names where, type_name what.

approval_store(port)

The approval store behind a port (port.store), if it has one.

check_plain(value, path='$')

Refuse anything but exact JSON types (encoded data read back from elsewhere).

Raises:

Type	Description
NotSealable	a value is not plain JSON.

decode(data, decoder=None, known=None)

The value encode made data from; only call it on data whose MAC was checked.

Parameters:

Name	Type	Description	Default
known	Mapping[str, type] | None	The classes a $model / $enum may name, by type_name (reachable_types); any other name goes to decoder, else is refused before anything is built.	None

Raises:

Type	Description
NotSealable	unknown tag, a class not allowed, or a value the decoder refused.

encode(value, encoder=None, path='$')

value in the tagged canonical form (JSON data).

Raises:

Type	Description
NotSealable	a value of an unknown type and no encoder (or the encoder failed).

same_decision(stored, decision)

True when a stored decision approves (or denies) the same way with the same edits.

unsealable_calls(data, where='')

Calls in journal data that ran but could not be recorded (children included).