`agentkit.agents`¶

The Agent, Workflow, and Cognition machinery, plus the control primitives (signals, RunPolicy, Handoff, ActorBudget) and policies (RoundRobinPolicy, SelectorPolicy, PlanPolicy, LedgerPolicy).

See the Agents concept for the mental model.

Opinionated compositions built FROM the runtime + middlewares.

A single Agent is the foundational pattern: a leaf runs its own ReAct loop over its tools (or short-circuits to a single chat call when tools=None); a coordinator dispatches to its children according to a Policy. Workflow is the typed, developer-authored DAG counterpart — explicit control where Agent/Policy is emergent.

Orchestrator-Worker. A central coordinator decomposes a task and dispatches it to specialised workers, then reduces their outputs. In agentkit this is Agent(children=..., policy=PlanPolicy(planner=...)) (planned, named steps) or Agent(children=..., policy=LedgerPolicy(assessor=...)) (planner + ledger). The planner's "send this sub-task to that worker" verb IS a Handoff — a typed transfer of control with a named target and a reason — and the policy interprets it via route_by_handoff. The worker's reply lands back on the shared transcript / scratchpad, not in a parent-side return channel: coordination is via the blackboard.

Routing. SelectorPolicy is the seam: a Selector returns the next speaker by name. The pre-built selectors stack three control regimes — route_by_handoff for the typed Handoff verb; handoff_selector for the marker-only legacy path; and llm_selector for fully emergent "ask a model" routing.

Agent `dataclass` ¶

Agent(
    name: str,
    model: str | None = None,
    prompt: Prompt | str | None = None,
    request_builder: RequestBuilder | None = None,
    response_format: dict[str, Any] | None = None,
    temperature: float = 0.0,
    max_tokens: int | None = None,
    output: type | dict[str, Any] | None = None,
    parse: Callable[[str], Any] | None = None,
    max_repairs: int = 1,
    cognition: Cognition = SingleCallCognition(),
    memory: Any = None,
    policy: RunPolicy | None = None,
    channel: Any = None,
)

An agent with one Cognition Strategy.

Identity + chat-call configuration live here. The cognition owns the turn-taking regime + its own configuration (tools, children, iteration ceiling, termination, checkpointer, …).

Required

name

Required (any chat-using cognition): model

Prompt wiring (one of): prompt: a Prompt (versioned) OR a plain str (wrapped into a one-off Prompt with version="inline" so traces still attribute clearly). If neither prompt nor request_builder is given, an empty inline Prompt is used — this keeps the ergonomic Agent("a", "m") form working in tests and small examples without silently producing an un-traced run. request_builder: a fully wired RequestBuilder (prompt + grounding + compaction). When set, this overrides prompt entirely.

Output shaping

response_format, max_tokens, temperature: passed through on each ChatRequest. output: type | dict — Pydantic BaseModel, attrs class, dataclass, or raw JSON Schema dict. The adapter is built once in __post_init__; its JSON Schema lands in the cache-stable prefix via the RequestBuilder. parse: Callable[[str], T] that returns the typed object or raises on invalid. AgentResult.parsed carries the typed object. On invalid output, the error is reflected back to the model up to max_repairs times. max_repairs: reflect-and-retry budget for parse failures.

Cognition (the Strategy plug-in): cognition: the turn-taking strategy. Defaults to SingleCallCognition(). Pass ReActCognition(tools=…) for a tool loop, or CoordinatorCognition(children=…, policy=…) for multi-agent orchestration. Any Cognition Protocol impl works.

Memory (the external-reach seam): memory: a MemorySource the cognition queries to ground reasoning. Composable: pass a CompositeMemory to fan out across vector + journal + tool-wrapped sources, or a single VectorMemory for plain RAG. None (default) disables the memory hook. Results are stamped onto working_context.scratchpad["memory"] and the cognition decides how to fold them into the prompt (typically via the RequestBuilder's grounding hook).

run `async` ¶

run(
    task: str,
    ctx: Ctx,
    *,
    context: WorkingContext | None = None,
) -> AgentResult

Run the agent on task. Collects the stream into the final AgentResult.

stream `async` ¶

stream(
    task: str,
    ctx: Ctx,
    context: WorkingContext | None = None,
) -> AsyncIterator[StreamEvent]

Stream the agent's run — message_delta tokens, tool events, and a terminal final event.

The actual loop is owned by self.cognition; the Agent opens the invoke_agent span, threads the working context, and stamps terminal attributes on close. agentkit.agent.queue_wait_ms measures the gap between stream() being called and the span opening — meaningful for coordinator paths where a parent dispatched us under a bounded semaphore.

resume `async` ¶

resume(
    run_id: str, decisions: dict[str, str], ctx: Ctx
) -> AgentResult

Resume a suspended tool-loop run with human approval decisions per pending tool call.

Only supported when self.cognition is a ReActCognition — that's where suspend/resume + checkpoint state live. Calling resume on any other cognition is a contract violation; raised explicitly so the caller's bug surfaces immediately.

ActorBudget ¶

ActorBudget(
    *,
    max_tokens: int,
    max_cost_usd: float,
    max_steps: int,
    max_wall_seconds: float,
    clock: Callable[[], float] = _monotonic_seconds,
)

Per-agent four-axis budget with reservation accounting.

Construct with the agent's caps. The wall-clock axis is rooted at construction time (_started_at) — the budget knows when "now - started_at" exceeds max_wall_seconds even without explicit ticks.

Concurrency: not thread-safe; assumed owned by a single agent's loop. Cross-agent budget interactions (parent's spawn / settlement of children) happen at well-defined transitions and the framework runs those serially on the parent's event loop.

exhausted ¶

exhausted() -> bool

True if any axis has nothing left to give. The agent loop checks this at the top of every iteration.

charge ¶

charge(
    *,
    tokens: int = 0,
    cost_usd: float = 0.0,
    steps: int = 0,
) -> None

Record actual spend. Soft-exceeds the cap (lets the in-flight tool call complete) — the next exhausted() check trips the loop. Never raises so a recently-charged call doesn't surface as a crash; the loop checks exhausted() and stops cleanly.

can_spawn_child ¶

can_spawn_child(
    *,
    request_tokens: int,
    request_cost_usd: float,
    request_steps: int,
) -> bool

Precheck — would the requested slice fit?

Pure: no state mutation. Pair with reserve_for_child for the actual carve-out.

reserve_for_child ¶

reserve_for_child(
    *, tokens: int, cost_usd: float, steps: int
) -> None

Move a slice from "available" to "reserved".

Raises BudgetExhausted if the request doesn't fit (after a can_spawn_child precheck this shouldn't fire, but the explicit raise here is the enforcement seam for races between concurrent spawn attempts).

settle_child ¶

settle_child(
    *,
    reserved_tokens: int,
    reserved_cost_usd: float,
    reserved_steps: int,
    used_tokens: int,
    used_cost_usd: float,
    used_steps: int,
) -> None

Settle a child's reservation against its actual usage.

Releases the reserved slice and accumulates the child's actual spend into the parent's used_*. Capped at the reservation so an over-spending child can't silently exceed its slice on the parent's books — the parent observes "this child spent up to its allowance"; the over-spend stays on the child's own metrics.

Idempotent if reservations are tracked correctly upstream — but the framework calls this exactly once per child DoneSignal.

tighten ¶

tighten(
    *,
    new_max_tokens: int | None = None,
    new_max_cost_usd: float | None = None,
    new_max_steps: int | None = None,
    new_max_wall_seconds: float | None = None,
) -> None

Lower (never raise) the caps. The parent uses this when a BudgetReducedSignal lands on a child mid-flight.

Caps never go BELOW current used_* (that would retroactively exhaust). The function silently clamps to used_* if the proposed new cap is lower — better to truncate to "no further work" than to claim the agent has already over-spent.

Autonomy ¶

Bases: StrEnum

Run-wide autonomy tier. Carried on RunContext.autonomy and consulted by :func:should_gate to decide which actions pause for a human.

AUTO — gate only actions a tool/author explicitly required.
GATED — also gate "key" steps/nodes.
MANUAL — gate everything.

BlockedSignal `dataclass` ¶

BlockedSignal(
    correlation_id: str | None = None,
    causation_id: str | None = None,
    sender_id: str | None = None,
    timestamp_us: int = 0,
    reason: str = "",
)

Bases: DataSignal

Child has hit a wall with no proposed recovery. Same effect as EscalateSignal from the parent's POV, but the absence of options signals "I genuinely don't know what to do next" — typically leads to the parent retiring the child's task.

BudgetExhausted ¶

BudgetExhausted(axis: str, message: str)

Bases: RuntimeError

Raised when a budget axis would go negative.

Carries the axis name so callers can react differently (a token-exhaustion is a "wind down gracefully" signal; a wall-clock-exhaustion may warrant a hard cancel).

BudgetReducedSignal `dataclass` ¶

BudgetReducedSignal(
    correlation_id: str | None = None,
    causation_id: str | None = None,
    sender_id: str | None = None,
    timestamp_us: int = 0,
    constraints: dict[str, float] = dict(),
)

Bases: ControlSignal

Parent shrunk the child's remaining budget envelope.

constraints is an open dict-of-numerics rather than a fixed set of axes because different projects use different budget shapes (some care about tokens + cost only, some add wall-clock + steps). The child applies whichever keys it recognises and ignores the rest.

CancelSignal `dataclass` ¶

CancelSignal(
    correlation_id: str | None = None,
    causation_id: str | None = None,
    sender_id: str | None = None,
    timestamp_us: int = 0,
    reason: str = "",
)

Bases: ControlSignal

Hard stop. The child finishes its current tool call, ships its final delta upward, and exits. Distinct from MergeWithPeerSignal in that no successor takes over — the work is just terminated.

ContextUpdateSignal `dataclass` ¶

ContextUpdateSignal(
    correlation_id: str | None = None,
    causation_id: str | None = None,
    sender_id: str | None = None,
    timestamp_us: int = 0,
    mutations: list[MutationT] = list(),
)

Bases: ControlSignal, Generic[MutationT]

Parent broadcasts a useful new piece of context to siblings — typically a finding from sibling A that siblings B + C should fold into their own awareness (e.g. "A rejected contentfarm.com/x, don't waste a fetch on it").

The child applies mutations to its own context without claiming authorship — the originating agent_id on each mutation preserves the audit trail.

ControlSignal `dataclass` ¶

ControlSignal(
    correlation_id: str | None = None,
    causation_id: str | None = None,
    sender_id: str | None = None,
    timestamp_us: int = 0,
)

Bases: SignalEnvelope

Parent → child directive base. Subclass to participate in a project's control-plane dispatch. The framework dispatches by isinstance; no discriminator field is required.

DataSignal `dataclass` ¶

DataSignal(
    correlation_id: str | None = None,
    causation_id: str | None = None,
    sender_id: str | None = None,
    timestamp_us: int = 0,
)

Bases: SignalEnvelope

Child → parent result base. Subclass to participate in a project's data-plane dispatch.

DoneSignal `dataclass` ¶

DoneSignal(
    correlation_id: str | None = None,
    causation_id: str | None = None,
    sender_id: str | None = None,
    timestamp_us: int = 0,
    final_delta: list[MutationT] = list(),
    confidence: float = 0.0,
    metrics: dict[str, float] = dict(),
)

Bases: DataSignal, Generic[MutationT]

Terminal signal — the child finished its work.

Ships the FINAL delta (whatever the journal accumulated past the last watermark) plus the final confidence + optional metrics rollup (used_tokens, used_cost, used_steps — projects pick the keys that matter to them).

The parent absorbs final_delta into its own journal, drops the child registry entry, and refunds the unused budget slice.

EscalateSignal `dataclass` ¶

EscalateSignal(
    correlation_id: str | None = None,
    causation_id: str | None = None,
    sender_id: str | None = None,
    timestamp_us: int = 0,
    reason: str = "",
    options: list[str] = list(),
)

Bases: DataSignal

Child can't make progress and is asking the parent to pick a recovery. options is a structured menu so the parent (or its LLM planner) can choose deterministically — typical entries are project-defined verbs like "retry", "retire", "reformulate", "spawn_helper".

The child typically blocks (no new work) until the parent replies with a RedirectSignal / BudgetReducedSignal / CancelSignal / ContextUpdateSignal.

Handoff `dataclass` ¶

Handoff(target: str, reason: str = '', message: str = '')

A typed transfer of control from one agent to another in a coordinator Agent's roster.

Returned by an agent (via a tool call result OR an output marker) to say "I'm done; the next turn should be target, with this context." The coordinator's SelectorPolicy interprets it; the receiving agent inherits the transcript and the shared scratchpad.

target — the name of the receiving agent. Must exist in the coordinator's roster; an unknown target falls back to the selector's default (never stalls the run). reason — a short human-readable reason for the transfer. Goes into traces and the rendered marker so the receiving agent sees why it was handed control. message — optional text the orchestrator wants the receiving agent to see as the next user turn. If empty, the receiving agent just inherits the transcript with no fresh task framing. Multi-line is fine; it's wrapped inside the marker render.

render ¶

render() -> str

Render this Handoff as a marker string that parse_handoff can read back.

Format: HANDOFF:<target>[ <reason>]. The optional message is appended on its own line so multi-line context survives the round-trip without breaking the single-line marker contract handoff_selector already supports.

MergeInbox ¶

MergeInbox(*, buffer_size: int = 512)

Bases: Generic[DataT]

Multiplexed channel: many children → one read seam on the parent. Each enqueued item is (sender_id, signal) so the parent's dispatcher can attribute work back to a specific child without per-child polling.

Bounded by buffer_size. When full, producers (child emits) block via await put — that's the right backpressure shape (the parent is the bottleneck; making children wait for the parent to catch up keeps memory bounded).

put `async` ¶

put(item: tuple[str, DataT]) -> None

Block until there's room. Backpressure flows to the producing child.

put_nowait ¶

put_nowait(item: tuple[str, DataT]) -> None

Best-effort put. Raises asyncio.QueueFull if full — caller decides whether to drop or block.

get_nowait ¶

get_nowait() -> tuple[str, DataT]

Pull one item without blocking. Raises asyncio.QueueEmpty if empty — pair with empty().

qsize ¶

qsize() -> int

Approximate queue depth — diagnostic only (race-prone).

MergeWithPeerSignal `dataclass` ¶

MergeWithPeerSignal(
    correlation_id: str | None = None,
    causation_id: str | None = None,
    sender_id: str | None = None,
    timestamp_us: int = 0,
    survivor_agent_id: str = "",
    reason: str = "",
)

Bases: ControlSignal

Parent decided this child's task overlaps with a peer's; hand control over to the survivor. The child ships its final delta upward and exits. The parent typically renders a "merged_into" relationship on its visualisation surface.

PolicyVerdict `dataclass` ¶

PolicyVerdict(
    allowed: bool,
    capabilities: tuple[str, ...],
    reason: str = "",
)

Frozen — the return value of RunPolicy.check is a decision. Callers may stash the verdict and re-consult it, or log it into an audit trail alongside the run's other values. In-place mutation of .allowed between inspection and second read would silently flip the trifecta gate.

ProgressSignal `dataclass` ¶

ProgressSignal(
    correlation_id: str | None = None,
    causation_id: str | None = None,
    sender_id: str | None = None,
    timestamp_us: int = 0,
    mutations: list[MutationT] = list(),
    confidence: float | None = None,
)

Bases: DataSignal, Generic[MutationT]

Streamed delta — child made progress, here's what changed.

Used as an ACK-less stream: the child emits at every meaningful boundary and the parent absorbs without replying. The child advances its journal watermark only after the parent has absorbed (the runner orchestrates this via the merge loop).

RedirectSignal `dataclass` ¶

RedirectSignal(
    correlation_id: str | None = None,
    causation_id: str | None = None,
    sender_id: str | None = None,
    timestamp_us: int = 0,
    new_state: StateT | None = None,
    reason: str = "",
)

Bases: ControlSignal, Generic[StateT]

Parent retasked the child mid-flight — typically the parent refined a sub-task and is handing the child a new context.

The child replaces its read-only snapshot with new_state and typically clears any role-cached state derived from the old snapshot. Local journal stays — the child's authored history survives the retask.

RunPolicy ¶

RunPolicy(*, mode: PolicyMode = 'flag')

Pre-run trifecta gate. mode controls what happens when the tool set covers all three lethal caps:

"flag" → return a PolicyVerdict with allowed=False so the caller decides (audit-log it, gate the run behind a human, split the run).
"deny" → raise PermissionError immediately so the run never starts.

SignalChannel ¶

SignalChannel(
    *,
    agent_id: str,
    buffer_size: int = 256,
    merge_buffer_size: int | None = None,
    clock: Callable[[], int] = _monotonic_us,
    observer: Any = None,
)

Bases: Generic[ControlT, DataT]

Per-agent bidirectional channel. The single seam an agent uses to talk to its parent and to its children.

Concurrency model:

The owning agent reads inbox and merge_inbox; writes outbox via emit. No locking needed — one consumer per queue.
The parent agent writes inbox via the channel's send_to method.
Children write to this channel's merge_inbox via THEIR own channel's emit — wired by attach_parent.

The dual-write on emit (outbox + parent's merge_inbox) is what lets tests + replay read from outbox without a parent context, while production agents see signals via the merge inbox without polling each child.

Build a channel owned by agent_id.

merge_buffer_size defaults to 2 * buffer_size because the merge inbox aggregates from N children — a single child's outbox depth is the right baseline, but the merge side needs headroom for fan-in.

clock is injectable so tests can pin timestamps; default is monotonic microseconds.

observer is an optional ObserverPort-shaped sink (anything with an async emit(Observation)). When wired, every emit stamps a signal.emitted observation so the run's audit timeline sees the coordination cascade — same shape as observations from the LLM / tool / memory seams. Kept optional so unit tests + orphan channels (no runner attached) still work at zero-config.

attach_parent ¶

attach_parent(
    parent_merge_inbox: MergeInbox[DataT],
) -> None

Wire this channel's emit to also fan up to the parent's merge inbox. Called by the parent's spawn handler after constructing the child.

emit `async` ¶

emit(signal: DataT) -> None

Emit a data signal upward.

Stamps sender_id and timestamp_us on the envelope at emit time. Puts the signal on this agent's outbox AND on the parent's merge inbox (if attached).

Both writes await so backpressure flows correctly — if the parent is slow, the child blocks. Tests that read from a detached channel (no parent attached) only see the outbox write; production flows see both.

send_to `async` ¶

send_to(signal: ControlT) -> None

Deliver a control signal to THIS agent's inbox. Called by the parent (or the runner) to push down directives.

sender_id + timestamp_us get stamped here too — symmetric with emit, since the parent is the "sender" from the child's POV when it reads the inbox.

try_send_to ¶

try_send_to(signal: ControlT) -> bool

Non-blocking control-signal delivery. Returns False if the inbox is full — used by best-effort sibling broadcast where dropping a hint is preferable to blocking the parent.

SignalEnvelope `dataclass` ¶

SignalEnvelope(
    correlation_id: str | None = None,
    causation_id: str | None = None,
    sender_id: str | None = None,
    timestamp_us: int = 0,
)

Common stamp every signal carries.

Fields:

correlation_id — the originating signal that started this cascade. Lets the audit timeline collapse a fan-out cascade back to its trigger.
causation_id — the immediate predecessor of THIS signal in the cascade graph. Distinct from the agent's parent_id (which is structural) — causation tracks the message DAG.
sender_id — id of the agent that emitted the signal. Stamped at emit time by the channel; user code typically leaves it unset at construction.
timestamp_us — monotonic microseconds since channel start. The channel stamps this at emit time too.

LedgerPolicy `dataclass` ¶

LedgerPolicy(
    assessor: Any = None,
    planner: Any = None,
    max_rounds: int = 20,
    max_replans: int = 3,
    name: str = "ledger",
)

Stall-aware supervisor: plan → assess progress → route to next_speaker, or re-plan on stall — bounded by max_replans and the max_rounds ceiling.

PlanPolicy `dataclass` ¶

PlanPolicy(
    planner: Planner | None = None,
    best_effort: bool = False,
    name: str = "plan",
)

Dispatches a plan of named-child steps. The plan comes from planner (required when steps= is not supplied at execute time).

best_effort=False (default) fail-fast: a failing step cancels its group and raises. best_effort=True: each slot is a result OR a :class:~agentkit.kernel.errors.Failure wrapping the raised exception, so partial progress survives — failures land in AgentResult.evals['errors'] as (child_name, failure) tuples (the Failure carries the originating exception on .cause).

Human-gate: a Step.gate("name") step suspends the plan at its group and checkpoints to ctx.store; resume via :meth:resume. Mirrors Workflow.human_gate — the coordinator-level counterpart of the same primitive.

resume `async` ¶

resume(
    coordinator: Agent, decisions: dict[str, str], ctx: Ctx
) -> AgentResult

Resume a plan suspended at a human-gate step.

Loads the checkpoint keyed by ctx.correlation_id from ctx.store. If the gate's decision in decisions is "approve", the checkpoint is deleted and execution continues from the group AFTER the gate. Any other value — including a missing key — is treated as a rejection: the checkpoint is deleted (so a stale run can't be re-resumed) and an AgentResult with stop_reason="rejected" is returned; no further groups run.

Takes coordinator as its first argument (mirroring :meth:execute) because the children roster isn't part of the checkpoint — the caller re-supplies the coordinator on resume, matching how Workflow.resume re-supplies the graph.

RoundRobinPolicy `dataclass` ¶

RoundRobinPolicy(
    name: str = "roundrobin",
    max_turns: int = 50,
    note_parser: Callable[[str], dict[str, Any]]
    | None = None,
    compactor: Any = None,
    compact_every: int = 0,
)

Children speak in fixed rotation over a shared transcript. The coordinator Agent's termination is the smart stop on top of max_turns (the never-hang ceiling). When the coordinator carries no termination, the default MaxTurns(len(children)) makes the loop visit each child exactly once per turn.

Optional knobs

max_turns: hard ceiling on the loop. Defaults to 50 — the never-hang backstop on top of the coordinator's smart termination. note_parser: Callable[[str], dict] — agent-side scratchpad write channel (e.g. marker_notes()). Extracts notes from each child's reply onto the shared blackboard scratchpad. compactor / compact_every: compact the blackboard's transcript every N turns. Only active when context is a shared blackboard (else there is no transcript-as-blackboard to compact).

SelectorPolicy `dataclass` ¶

SelectorPolicy(
    selector: Selector,
    name: str = "selector",
    max_turns: int = 50,
    note_parser: Callable[[str], dict[str, Any]]
    | None = None,
    compactor: Any = None,
    compact_every: int = 0,
)

A Selector picks the next child each turn. Sync OR async; optionally ctx-aware (a third arg). Returns the next child's name or None to fall back to round-robin.

Optional knobs

max_turns: hard ceiling on the loop. Defaults to 50. note_parser: Callable[[str], dict] — agent-side scratchpad write channel. compactor / compact_every: same as RoundRobinPolicy.

Suspended `dataclass` ¶

Suspended(
    run_id: str,
    pending: tuple[ToolCall, ...] | tuple[str, ...],
    reason: str = "awaiting_approval",
)

Carried in AgentResult.evals['suspended'] when the loop pauses for human approval.

pending is a tuple, not a list — the operator UI renders the pending items and the resume path threads them back verbatim; a mutable list inside a frozen shell would let a stray suspended.pending.append(...) desync the operator's rendered view from what actually resumes. The frozen tuple pins the handshake at both ends.

pending is narrowed to a tuple of ToolCall OR a tuple of str (gate-name identifiers, emitted by Workflow when a human_gate node suspends) — the two suspend surfaces produce different-shaped identifiers, and the union catches drift from a third caller passing arbitrary objects.

WorkflowResult `dataclass` ¶

WorkflowResult(
    outputs: dict[str, Any],
    usage: Usage,
    steps: int,
    stop_reason: WorkflowStopReason,
    suspended: Suspended | None = None,
)

Terminal result of a Workflow run. Carries every node's latest output, the merged usage, the number of node executions (incl. re-runs from loop-back), the stop reason (complete | suspended | max_steps | deadlock), and a Suspended when the run paused on a human-gate.

Workflow ¶

Workflow(name: str = 'workflow', *, max_steps: int = 100)

A typed node/edge graph with a deterministic, concurrent, bounded engine.

fn ¶

fn(
    name: str, f: Callable[..., Any], *, after: Any = ()
) -> str

A pure function node. f is sync or async, called f(inputs) or f(inputs, goal) by arity.

coordinator ¶

coordinator(
    name: str,
    coordinator: Any,
    *,
    after: Any = (),
    prompt: Callable[[dict[str, Any], str], str]
    | None = None,
) -> str

A coordinator Agent (or any runnable with async run(task, ctx)) as a graph node — emergent inside explicit. Output is the coordinator's last transcript message; usage is merged.

human_gate ¶

human_gate(name: str, *, after: Any = ()) -> str

A node that suspends for a human decision; its output is the decision passed to resume.

route ¶

route(
    from_: str, *, when: Callable[[Any], bool], to: str
) -> Workflow

Conditional edge: after from_ runs, if when(output) is true, (re)activate to. A route to an ancestor is a bounded loop-back (guarded by max_steps).

run `async` ¶

run(
    goal: str,
    ctx: Ctx,
    *,
    decisions: dict[str, Any] | None = None,
    on_existing: OnExisting = "start_fresh",
) -> WorkflowResult

Execute the workflow to completion or a human-gate suspend.

on_existing controls what happens when a checkpoint already exists under ctx.correlation_id on ctx.checkpointer:

"start_fresh" (default, preserved historical behaviour) — ignore any existing checkpoint and run from step 1. Callers who want silent overwrite (the old default) keep the default.
"resume" — consult ctx.checkpointer.resume(run_id) (which itself filters terminal DONE/FAILED snapshots by default). If a resumable checkpoint exists, replay from it; otherwise start fresh. The checkpoint's state is expected to carry {"goal", "done", "steps"} (the shape the Workflow itself writes at a human-gate suspend), so a checkpoint written by a previous run / resume picks up cleanly.
"fail" — consult the checkpointer for ANY snapshot (terminal or not) and raise CheckpointerError if one exists. This is the idempotency-guard mode: use when the caller must not silently re-run a job that already has persisted state.

When ctx.checkpointer is None, the non-start_fresh modes degrade cleanly: "fail" cannot detect a prior run so it proceeds; "resume" has nowhere to resume from and starts fresh.

agentkit.agents¶

Agent dataclass ¶

run async ¶

stream async ¶

resume async ¶

ActorBudget ¶

exhausted ¶

charge ¶

can_spawn_child ¶

reserve_for_child ¶

settle_child ¶

tighten ¶

Autonomy ¶

BlockedSignal dataclass ¶

BudgetExhausted ¶

BudgetReducedSignal dataclass ¶

CancelSignal dataclass ¶

ContextUpdateSignal dataclass ¶

ControlSignal dataclass ¶

DataSignal dataclass ¶

DoneSignal dataclass ¶

EscalateSignal dataclass ¶

Handoff dataclass ¶

render ¶

MergeInbox ¶

put async ¶

put_nowait ¶

get_nowait ¶

qsize ¶

MergeWithPeerSignal dataclass ¶

PolicyVerdict dataclass ¶

ProgressSignal dataclass ¶

RedirectSignal dataclass ¶

RunPolicy ¶

SignalChannel ¶

attach_parent ¶

emit async ¶

send_to async ¶

try_send_to ¶

SignalEnvelope dataclass ¶

LedgerPolicy dataclass ¶

PlanPolicy dataclass ¶

resume async ¶

RoundRobinPolicy dataclass ¶

SelectorPolicy dataclass ¶

Suspended dataclass ¶

WorkflowResult dataclass ¶

Workflow ¶

fn ¶

coordinator ¶

human_gate ¶

route ¶

run async ¶

`agentkit.agents`¶

Agent `dataclass` ¶

run `async` ¶

stream `async` ¶

resume `async` ¶

BlockedSignal `dataclass` ¶

BudgetReducedSignal `dataclass` ¶

CancelSignal `dataclass` ¶

ContextUpdateSignal `dataclass` ¶

ControlSignal `dataclass` ¶

DataSignal `dataclass` ¶

DoneSignal `dataclass` ¶

EscalateSignal `dataclass` ¶

Handoff `dataclass` ¶

put `async` ¶

MergeWithPeerSignal `dataclass` ¶

PolicyVerdict `dataclass` ¶

ProgressSignal `dataclass` ¶

RedirectSignal `dataclass` ¶

emit `async` ¶

send_to `async` ¶

SignalEnvelope `dataclass` ¶

LedgerPolicy `dataclass` ¶

PlanPolicy `dataclass` ¶

resume `async` ¶

RoundRobinPolicy `dataclass` ¶

SelectorPolicy `dataclass` ¶

Suspended `dataclass` ¶

WorkflowResult `dataclass` ¶

run `async` ¶