Flows

Flow graphs, nodes and the runner. Concepts: Flows, Flow pause and resume.

agent_adapter.flow.visit is a seam for flow compilers; most code never needs it.

agent_adapter.flow

Portable graph flows: a builder, merge rules, sub-flows, a node cache, a graph, a runner.

NodeFn = Callable[[S], Any] | Callable[[S, ToolContext], Any] module-attribute

keyed = Merge.keyed module-attribute

Call dataclass

Run one tool through the executor; args may be a function of the state.

FlowCheckpoint dataclass

Where a paused flow stopped; pass it to FlowRunner.resume (JSON: dump_checkpoint).

FlowGraph dataclass

A validated flow: nodes in declaration order, static edges, routes, ends and merge rules.

predecessors(node)

Every node that can lead to node, in declaration order.

successors(node)

Every node node can lead to: static edges, then route targets.

FlowResult dataclass

Bases: Generic[S]

How a run ended: the state, the status, supersteps run and visits per node.

Join dataclass

Fan-in: runs once nothing else is pending; fn is an optional reducer step.

Map dataclass

Fan out over the list field over: one tool call (or fn call) per item.

Merge

The built-in merge rules; annotate a state field with one (Annotated[T, Merge.sum]).

keyed(by) staticmethod

A list of mappings or models; a later item with the same by key replaces earlier.

Raises:

Type	Description
FlowDefinitionError	by is empty.

MergeRule

Bases: Protocol

Combines the updates one superstep makes to a field; updates are in deterministic order.

A rule that adds a write onto the current value (like append or sum) should also set additive = True; sub-flows then start the field empty and refuse to map it in and back.

NodeCache dataclass

Reuse a node's updates for equal inputs: the state, its reads fields, or key(state).

NodeSpec dataclass

One node: its action and run options.

Route dataclass

Bases: Generic[S]

Pick the next node(s): fn(state) returns a label or labels found in targets.

RouteSpec dataclass

A conditional exit: fn(state) returns labels; targets maps label -> node.

SubFlow dataclass

Run another Flow as one node: parent state in through input, its result out by into.

input: child field -> parent field, or fn(parent_state) -> the child's initial values; None copies fields of the same name (additive append / sum fields start empty). into: a parent field for the child's final state, parent field -> child field, or fn(child_state) -> the parent update; None writes back same-named fields it changed.

apply_updates(state, updates, rules)

Merge updates (in order) into a copy of state and re-validate it.

Raises:

Type	Description
FlowError	unknown keys, conflicting replace writes, a failing rule, or an invalid result.

merge_rules(model)

The merge rule of every field of model, from Annotated metadata (default replace).

Raises:

Type	Description
FlowDefinitionError	a field carries more than one merge rule.

node_model(ctx)

The flow runner's model, instrumented by its executor and bound to this node, or None.

node_prompt(ctx)

The running node's prompt, or None.

Also exported here, documented under the canonical path: Flow, FlowRunner.

agent_adapter.flow.runner

The local flow runner: LangGraph-style supersteps on the executor, with pause and resume.

FlowStatus = Literal['done', 'paused', 'denied'] module-attribute

Also exported here, documented under the canonical path: FlowCheckpoint, FlowResult, FlowRunner.

agent_adapter.flow.visit

Semi-public flow seam for flow compilers (LangGraph, ADK); not in flow.__all__.

The compilers run node visits, gates, routes, resumes and seals exactly like the local runner. They import those internals from here, never from the private modules, so a refactor there keeps this surface stable.

ItemOutcome = tuple[str, Any] module-attribute

NOT_SEALABLE = 'state-not-sealable' module-attribute

NO_PORT = 'no approval port configured' module-attribute

SPENT = SpentCheckpoints() module-attribute

SUPERSEDED = 'superseded' module-attribute

TAMPERED = 'checkpoint-tampered' module-attribute

ask_node_gate = _ask module-attribute

Ask the port for a gate: its decision, the pending request, or a denial without a port.

node_gate_request = _request module-attribute

The ApprovalRequest a gated node visit asks for (call id = the visit's id).

superseded = _superseded module-attribute

Pending records a resume never reuses: not the checkpoint's, a call id the flow did not set.

NodeEnv dataclass

What every node of one flow run shares.

NodeOutcome dataclass

A node's updates, or the approval it is waiting for.

allowed_classes(state_model)

The classes a checkpoint of this state model may hold (exec._canon.reachable_types).

apply_edits(graph, state, ready, decisions)

Apply each granted gate's edited_args as a state update, in node-name order.

call_route(flow, route, state) async

The labels route.fn(state) returns (sync or async), as a list.

Raises:

Type	Description
FlowError	the route function failed.

describe_error(exc)

An error's kind and message for AdapterErrors; only the class name otherwise.

frozen(data)

A read-only copy.

graph_digest(graph)

SHA-256 of the graph: names, kinds, actions, gates, edges, routes, start and ends.

Functions count by identity (module:qualname), so renaming or moving one changes it; a changed body under the same name does not. A Call / Map counts by its tool, static arguments and options; a SubFlow by its child graph's digest and its field mapping.

mac(key, data)

HMAC-SHA256 hex of the canonical JSON of a payload.

merge_outcomes(graph, state, outcomes, step)

Merge every node's updates: node-name order (LangGraph task order), then Map item order.

next_frontier(graph, frontier, ready, routed)

The next frontier in declaration order: unrun nodes, ready's edges and route picks.

node_call_id(flow, node, visit, scope='')

The id of one node visit (node events and the node's context); scope for sub-flows.

not_sealable(flow, exc)

The FlowError for a value that cannot be sealed.

pick_targets(flow, route, labels, visits_before, loops)

The nodes labels name; picking the source or an earlier-visited node counts a loop.

Raises:

Type	Description
FlowError	a label the route does not know.
FlowLoopExceeded	max_loops passed and no on_exhausted.

resume_entry(pending, decision)

The answer for a paused tool approval: exact call id for Call / Map calls.

route_state(graph, state, outcome)

What a route sees: the superstep's input state plus its own node's updates only.

route_targets(flow, route, state, visits_before, loops) async

The nodes a route picks; picking the source or an earlier-visited node counts a loop.

run_node(env, node, state, ctx, *, visit, approval=None, answer=None, items=None, journal=None, child=None) async

Run one node visit; failures raise FlowError, a pending approval is returned.

Parameters:

Name	Type	Description	Default
answer	ResumeAnswer | None	Answers the tool approvals this visit resumes; spent when the visit ends.	None
items	Mapping[int, ItemOutcome] | None	Map items finished before the pause (by index); they do not run again.	None
journal	JournalData | None	Executor calls this visit finished before the pause; they replay.	None
child	FlowCheckpoint | None	Where a SubFlow node's child run paused; it continues from there.	None

runner_key(key)

The checkpoint key a runner signs with.

Raises:

Type	Description
ToolUsageError	key is not bytes of at least 16 bytes.

spend_once(pending, decision, port) async

Spend a checkpoint's pending approval: settle it in the store first, then remember it.

Raises:

Type	Description
ApprovalAlreadyDecided	the checkpoint was already resumed (here or via the store).
ToolUsageError	the store record belongs to another request.

withdraw(flow, ctx, pendings, reason=NOT_SEALABLE) async

Deny and consume each pending record; a record no store reaches is logged.

Parameters:

Name	Type	Description	Default
reason	str	The denial reason: NOT_SEALABLE (the pause failed; ERROR when a record cannot be withdrawn) or SUPERSEDED (the pause stands; WARNING).	NOT_SEALABLE