agentkit.core¶
Value types (ChatRequest, LLMResult, Message, Scope, Usage),
the Port protocols, the middleware contract, and the concurrency +
resilience primitives. (Budget is a runtime concern — see
api-reference/runtime.)
See the Kernel concept for the mental model.
L0 kernel — opinion-free primitives: value types, the 4 infra seams, resilience combinators, structured concurrency, and the one middleware contract. Nothing here encodes a policy about how to use an LLM or tool; higher layers (runtime, middlewares, patterns) supply the opinions.
dataclass
¶
Failure(
category: ErrorClass,
source: str,
message: str,
retriable: bool = False,
cause: BaseException | None = None,
partial_output: Any = None,
children: tuple[Failure, ...] = (),
)
A failure as data. category is TRANSIENT/PERMANENT/UNKNOWN; retriable defaults from it.
Frozen — the docstring promises "a value a parent can reason
about" and compose_failures aggregates children into an
immutable tuple. Mutation of a child Failure observed by two
parents would corrupt one parent's view under the other's edits.
classmethod
¶
of(
exc: BaseException,
*,
source: str,
partial_output: Any = None,
) -> Failure
Build a Failure from an exception, classifying it.
retriable is True for TRANSIENT and UNKNOWN.
run_with_resilience retries both classes; keeping
Failure.retriable aligned means a higher layer that reads
it directly makes the same decision the kernel's resilience
loop would. Conservative on UNKNOWN means "try again" (matches
classify's prose: 'UNKNOWN # conservative').
The ergonomic, class-based middleware — the transform / guard / observe layer.
Override only the phases you need. Each phase is async and may be a plain coroutine OR an async
generator — both are supported:
on_request(ctx) # before: mutate ctx.request; raise Blocked to refuse; (yield/emit events)
on_response(ctx, result) # after success: return (or yield) a transformed result; default passes through
on_error(ctx, exc) # on failure: return/yield a value to recover, else raise (default re-raises)
Emit a product-facing event either by await ctx.emit(...) or by yield-ing an Observation (the
async-generator form). on_response returning/yielding a non-Observation value sets the result.
as_middleware() compiles these to the (call, next) primitive, wrapping next in one try/except so
on_request is always paired with exactly one on_response OR on_error — a balanced invariant that
standalone phase hooks tend to break. This layer cannot re-invoke or rewrite-and-re-run next;
retry / fallback / memoize / circuit-breaking are a separate resilience/caching concern written as
raw (call, next) middlewares, not a BaseMiddleware.
Bases: Exception
Raised by a middleware's on_request to refuse a unit of work before it runs (a guard/policy stop) —
e.g. detected prompt injection. Propagates as a typed refusal, distinct from a model/tool fault.
The rich, ergonomic view a middleware sees — a thin facade over the Call + its RunContext, so a
middleware reads ctx.messages / ctx.prompt / ctx.budget / ctx.vector rather than digging through
call.request and call.ctx. The request is mutable (ctx.request = … rewrites it before the op).
async
¶
Emit a product-facing event on the run's observer stream (never raises into the run).
dataclass
¶
Checkpoint(
run_id: str,
version: int,
state: dict[str, Any],
created_at: float,
status: CheckpointStatus,
metadata: dict[str, Any] = dict(),
)
A serialized snapshot of run state at a meaningful transition.
state is opaque to the port — the application (a leaf Agent, a coordinator Agent,
an application-defined state snapshot) interprets it. version is monotonic: v1 is the
first save for a run, v2 the next, and so on. status is a CheckpointStatus and
lets the port surface "is this run resumable?" without re-parsing state.
created_at is a unix timestamp stamped by the producer's ClockPort (or
time.time() when no clock is wired in) so replay and audit lines up across
services.
Bases: Protocol
The durable-run-state seam. Distinct from StorePort (a generic KV) because the thing
being persisted — a versioned, status-tagged snapshot — has its own access pattern: latest,
at-version (time-travel), list-versions, and delete-all-for-run. A single producer is the
authority over a run_id; two producers sharing a run_id would collide on version.
Bases: StrEnum
Coarse durability gate for a Checkpoint. Auto-resume keys off
this — a suspended checkpoint says "waiting on a human";
running says "engine is in motion"; done / failed are
terminal. Producers map their domain phases (e.g., the api's
RunPhase) onto this taxonomy at snapshot time.
Emission map (grep-able index — keep aligned with call sites):
RUNNING— the default snapshot status at every mid-run durable transition (Checkpointer.snapshot(...)with no explicit status). Set by the ReAct cognition after each tool-loop iteration and by coordinator policies at each turn.SUSPENDED— emitted by the ReAct cognition's human-approval path when a gated tool call needs a human decision (agentkit/agents/cognition/react.py— the_save(..., status="suspended", pending=...)call site). The suspended snapshot carries thependingtool calls soresume()can apply per-call decisions.StrEnumequivalence means the string"suspended"at the emission site ISCheckpointStatus.SUSPENDEDwhen the adapter round-trips it back (seetests/source/unit/state/adapters/test_postgres_checkpoint.pyfor the wire-level round-trip). Coordinators + custom producers MUST use this status (notRUNNING) when persisting a human-gate wait — auto-resume relies on it to distinguish "engine is chugging" from "waiting on the world".DONE/FAILED— emitted by the producer at terminal transitions;Checkpointer.resumefilters these by default so a "resume if any checkpoint exists" wiring cannot silently re-run a finished job.
True if no further work is expected — auto-resume should
NOT fire on a done or failed checkpoint.
Bases: Protocol
Deterministic time seam — tests and replay need a fakeable wall clock. now() returns a
unix timestamp; sleep(s) waits without blocking the loop. Use this instead of time.time() /
datetime.now() / asyncio.sleep inside middlewares + retry/backoff loops.
dataclass
¶
FetchResponse(
url: str,
status: int,
headers: dict[str, str],
body: str,
content_type: str,
fetched_at: float,
)
A text HTTP response. Binary fetch is out of scope — callers that need bytes use a different
port. fetched_at is a unix timestamp from the implementation's ClockPort (real or fake).
dataclass
¶
SearchHit(
url: str,
title: str,
snippet: str,
score: float | None = None,
metadata: dict[str, Any] = dict(),
)
One web-search result — every SearchPort adapter normalizes to this shape so callers
(Researcher agents, retrieval middlewares) are provider-agnostic.
Bases: Protocol
The single KV seam that backs the former cache / idempotency / checkpoint / audit ports.
get_or_set is single-flight; a producer that raises is NOT stored (failures are never cached).
The last three methods are the coordination half. Without them the seam could express "create if absent" and "read one log back", and nothing else, so three ordinary shapes had to go around the port entirely:
- read-modify-write. Allocating a monotonic ordinal is "read max, write
max+1", and two writers race it.
get_or_setcannot express "replace only if unchanged". - a counter with an expiry — the shape of every rate limit. The observed consequence was an application writing a raw Lua script straight at Redis, which put the check outside everything the framework can test, trace or meter.
- enumerating a prefix. "Everything recorded for this run" was answerable only if every writer had also maintained an index by hand.
async
¶
Replace key with value only if it currently equals expected; report whether it applied.
Returns a bool rather than raising. Losing a compare-and-set is an
ordinary outcome that the caller retries — it is the expected half of
an optimistic loop, not a fault. Raising would force every
read-modify-write to wrap its own body in a try/except and to tell
"someone beat me" apart from "the store is down" by exception type,
which is precisely the distinction the return value already makes.
Comparison is by EQUALITY, never identity: three of the four backends round-trip the value through JSON, so the caller never holds the object that was written and an identity check would make CAS on any container permanently impossible while appearing to work in memory.
An absent key compares equal to expected=None, so the first
iteration of a read-modify-write loop behaves like every later one.
That deliberately collapses "absent" and "stored null" — but so does
get, which is where the caller's expected came from, so the
distinction was never observable at the call site anyway.
ttl applies only on the write that lands; a refused CAS leaves the
existing expiry alone.
async
¶
Atomically add by to the integer at key (absent ⇒ 0) and return the new total.
by may be negative — refunds and released reservations are
decrements, and a counter that only went up would need a second key to
track what came back.
ttl opens a window on the increment that finds no window; it never
slides one that already exists (Redis's EXPIRE key ttl NX). The
alternative — resetting on every increment — breaks the exact case the
primitive is for: under sustained traffic the counter is touched more
often than the window is long, so it never expires, and the limit never
resets. The rate limiter would jam shut precisely under load. The
counter and its window are set together, atomically, so a failure
cannot leave a counter with no window (an immortal one) or a window
with no counter.
A key holding a non-integer raises StoreValueError — the same type
and message on every backend, so the caller does not have to know
whether Redis, Postgres or a dict answered. A stored null counts as
a non-integer, NOT as absent, which is the one place this deliberately
disagrees with compare_and_set: compare_and_set's expected came
out of a get, which cannot tell null from absent, so it has to accept
both; increment is handed no such value and the durable backends
genuinely cannot add to a JSON null.
Yield the KV keys beginning with prefix — full keys, in no promised order.
Declared def returning an AsyncIterator, not async def, for
the same reason as LLMPort.stream: an async def that yields is a
function returning an iterator, not a coroutine, so the async def
spelling would reject every real implementation under a strict type
check. Call sites are identical — async for k in store.scan(p).
Keys, not values, and never the append-log namespace: list(key) reads
logs, and a scan that surfaced log keys would hand back keys get
answers None for. A key equal to the prefix is included — a summary
record stored at the prefix itself is part of "everything under it".
limit is a cap on how many keys the caller is willing to receive,
NOT a page cursor: with no ordering promised, which keys arrive under
a limit is backend-defined. 0 is a real cap of zero and a negative
limit raises ValueError.
Iteration is concurrent-safe in the weak sense that matters: it must not raise when writers land mid-scan. No snapshot is promised — only that keys present throughout appear, exactly once.
dataclass
¶
One LLM-call replay payload. span_id ties it to the trace so the
trace acts as the index and the store carries the bulk.
operation is the unit of work — "chat" for an LLM turn,
"execute_tool" for a tool execution — matching the OTel GenAI
operation taxonomy span names. request is the input shape
(messages, tools, response_format for chat; arguments for tools).
response is the assembled LLMResult or tool output; None
when the call is still in flight or errored before producing one.
Bases: Protocol
Optional side-channel store for full LLM call replay.
Implementations are expected to be best-effort: a failed write
must NOT break the run. Adapters wrap their writes in suppress
blocks at the call site (see middlewares/tracing.py). The
default NoopReplayStore drops writes silently — the canonical
zero-config path.
dataclass
¶
ChatRequest(
messages: list[Message],
model: str,
tools: list[ToolSchema] | None = None,
response_format: dict[str, Any] | None = None,
temperature: float = 0.0,
max_tokens: int | None = None,
cache_hint: Any = None,
)
Frozen value type. To rewrite a field mid-chain (fallback swaps
the model, compaction rewrites messages), build a new instance
via dataclasses.replace and reassign call.request — the
Call reference is mutable; the ChatRequest it points at
is not.
"Frozen" now includes the three CONTAINER fields — messages,
tools, response_format — not just the references to them. Before,
frozen=True stopped at the reference and request.messages.append(m)
went straight through, which is a rewrite of the unit of work the chain is
mid-way through executing: the retry middleware re-sends the request it
kept, the tracing middleware snapshots list(call.request.messages)
AFTER the call, and memoize keys on the transcript — an in-place append
desyncs all three against a request that already went out on the wire.
Compaction shows the intended shape: it reads list(ctx.request.messages)
and REPLACES the request rather than editing the list.
messages and tools are FrozenList, response_format is a
FrozenDict; all three are still list/dict subclasses, so
request.messages[-1].content, len(), iteration, slicing and
json.dumps are untouched. None stays None — the optional fields
are not materialised into empty containers, because tools is None
("advertise nothing") and tools == [] mean different things to the
provider adapters.
The freeze is DEEP, but a Message and a ToolSchema are dataclasses,
not containers, so deep_freeze passes each one through as a leaf: the
walk is O(number of messages) and flat in transcript BYTES. Measured, min
of five::
1 msg x 5 chars / x 5000 chars 3.24 us / 3.25 us
10 msgs, 1 tool, response_format 1.74 us -> 6.67 us
100 msgs 26.4 us
messages already frozen 0.12 us (idempotent short-circuit)
Once per turn, against an LLM round trip measured in hundreds of ms. The
honest caveat: an agent loop rebuilds the request from a transcript that
grows every turn, so this is O(turns) per turn and O(turns^2) over a run —
~26 us at turn 100. That is the same shape the transcript COPY the loop
already does has, and it is why deep_freeze short-circuits: a caller
that hands the same FrozenList back pays 0.12 us, not the walk.
Not frozen: cache_hint, which is annotated Any and is whatever the
provider wanted. Freezing it would mean rewriting a provider object we do
not own — the same line _frozen.deep_freeze draws at non-container
leaves. It is passed through, never read by the framework.
dataclass
¶
A retrievable unit for semantic memory / RAG (metadata carries source + scope tags).
Bases: StrEnum
Provenance for known monetary measurements; unknown cost is never zero.
dataclass
¶
Delta(
text: str = "",
model: str | None = None,
usage: Usage | None = None,
provider: str | None = None,
finish_reason: str | None = None,
tool_calls: tuple[ToolCall, ...] = (),
parsed: Any = None,
partial: Any = None,
)
One streamed increment of an LLM response — the item the single stream() primitive yields.
A text fragment and/or tool_calls; the terminal delta carries usage/finish_reason/
model/provider. assemble_deltas() reduces a sequence back to an LLMResult, so a full
chat result is just the collected stream.
dataclass
¶
Message(
role: str,
content: str = "",
name: str | None = None,
tool_call_id: str | None = None,
tool_calls: tuple[ToolCall, ...] = (),
content_blocks: tuple[dict[str, Any], ...] = (),
metadata: dict[str, Any] = dict(),
)
One turn in a chat transcript.
Bases: StrEnum
Public model execution modes, independent of reasoning strategies.
Bases: StrEnum
The kind of unit of work a middleware is intercepting — match on it in a middleware
(if ctx.operation == Operation.MODEL_CALL: …). A str enum, so == "model_call" also holds.
dataclass
¶
Tenant-isolation key. Threaded through every memory recall / cache key / meter / callback.
dataclass
¶
StreamEvent(
type: StreamEventType,
text: str = "",
tool_call: Any = None,
tool_result: Any = None,
result: Any = None,
usage: Usage | None = None,
partial_output: Any = None,
)
A pattern-level run event emitted by Agent.stream/*.stream — a higher-level signal than a
Delta: where Delta is one transport increment of a single LLM response, a StreamEvent marks a
step in the run (a token message_delta, a tool_call/tool_result, an interrupt, the final
result). type discriminates the payload.
dataclass
¶
A tool invocation the model requested. id is echoed back on the matching tool-result message.
arguments is IMMUTABLE: the ToolCall flows into the ReAct approval
snapshot AND into the idempotency-key hash AND into the audit trail — a
tool impl that did args.pop("token") would have desynced all three.
Callers that need to mutate copy first: dict(tc.arguments).
The immutability was originally a MappingProxyType and is now a
:class:~agentkit.core.serialization._immutable.FrozenDict. Same guarantee, without the
stdlib limitation the proxy dragged in: a mappingproxy is not a dict
subclass, so json refused it, isinstance(args, dict) was False, and
dataclasses.asdict — which deep-copies every leaf — blew up on a type
with no pickle protocol. Measured on ToolCall("c1", "search", {...})::
MappingProxyType FrozenDict
dataclasses.asdict(tc) TypeError: cannot pickle 'mappingproxy' ok
json.dumps(tc.arguments) TypeError: not JSON serializable ok
isinstance(_, dict) False True
The asdict line is the one that shipped broken, and it was contagious:
a ToolCall does not have to be the argument, only reachable from it, so
asdict of ANY dataclass holding a ToolCall raised. That is why
resilience.stable_hash walks dataclass fields by hand instead of
calling asdict (see the note at its _stable_default) and why five
other call sites carry a defensive dict(tc.arguments) unwrap
(anthropic, openai_compat, checkpointer.persistence,
context.tokens, middlewares.memoize), plus tools.function's
dict(args), which is a snapshot rather than an unwrap. The count is
spelled out because the previous "four" was wrong and nothing would have
caught it.
One consequence to know about: asdict rebuilds each nested container as
type(obj)(...), so it hands back a FrozenDict, not a plain one. The
freeze survives serialisation, which is what a durable record wants — but
asdict is therefore NOT the "give me an editable copy" escape hatch.
dict(tc.arguments) is, and always was.
deep_freeze is deep, so tc.arguments["filters"]["tags"].append(...)
is refused too — a shallow freeze leaves the same bug one level down, where
decoded provider JSON actually lives. It also COPIES, so a caller that
keeps editing the dict it passed in cannot reach the stored payload.
Cost is O(payload) and it is NOT free — measured, min of five runs, on a realistic 3-key/2-level arguments dict::
ToolCall(id, name, args) 1.21 us -> 3.97 us (+2.8 us)
ToolCall(id, name) empty 1.22 us -> 1.48 us (+0.3 us)
copy.deepcopy(tool_call) 7.96 us -> 12.5 us
pickle round trip 6.65 us -> 12.9 us
Paid once per tool invocation, immediately before the tool does network or
disk I/O, so it is ~3 us against a millisecond floor. The copy numbers grew
because a FrozenDict deep-copies through its own __reduce__/
__deepcopy__ rather than the C fast path (6.6 us vs 3.7 us for the same
payload as a plain dict) — that is _frozen's price, and the
checkpointer was already paying the 8 us.
arguments being an unhashable container is also what would have cost
every ToolCall — and every Message / PrefixContext /
FrozenContext reachable from one — its hashability. See __hash__;
that decision is unchanged by this migration, and deliberately so.
dataclass
¶
ToolRequest(
name: str,
arguments: dict[str, Any],
tool: Any,
side_effecting: bool = False,
url_arg: str | None = None,
)
The tool half of the unit of work — what the Invoker runs through
the middleware chain before the terminal calls tool.run.
arguments is frozen deeply at construction, for the same reason as
ToolCall.arguments one layer up but with a shorter fuse: this dict is
read by the chain BEFORE the tool sees it (egress_audit pulls the URL
out of r.arguments[url_arg], memoize folds the whole dict into the
idempotency key) and then handed to the tool impl as run(args, ctx). A
tool that did args.pop("token") on the way in would have made the
arguments that were AUTHORISED and the arguments that were EXECUTED two
different things, with the audit trail recording the first.
FrozenDict rather than a proxy because the same dict is
json.dumps'd into the audit record and hashed through
stable_hash. Cost measured at 1.05 us before against 4.31 us after for
a realistic 3-key/2-level arguments dict — once per tool call, against the
I/O the tool is about to do.
tool is deliberately NOT frozen: it is the resolved ToolPort, an
object the registry built, and rewriting a caller's object is the line
deep_freeze refuses to cross.
dataclass
¶
JSON-schema advertisement of a tool — what the provider needs to see to call it.
parameters is a dict SUBCLASS — a FrozenDict, frozen deeply at
construction. It is json.dumps'd straight into provider payloads
(openai_compat sends schema.parameters or {}, anthropic sends
it as input_schema) and read back through dataclasses.asdict,
neither of which survives a MappingProxyType — which is exactly why the
freeze is a subclass and not a view. See kernel._frozen.
The freeze matters most here because a schema is SHARED, not per-call: a
ToolRegistry hands the same ToolSchema object to every
ChatRequest for the life of the process, so one adapter doing
schema.parameters["properties"]["x"]["type"] = "number" to fix up one
provider's payload would have silently rewritten what every other provider
advertises from then on.
This is the WORST relative cost of the four — 0.74 us before against
12.3 us after for an 8-property schema, 16x — and the one that matters
least, because a schema is built once and then reused forever. Every
construction site is registration-time (@tool decoration, as_tool,
handoff, MCP discovery), and ToolRegistry.schemas() hands back the
SAME cached ToolSchema objects on every turn rather than rebuilding
them, so no per-turn path pays this. A 16x on a one-off is the right place
to spend it; a 16x per turn would not have been.
That mutable field is also what cost the frozen dataclass its generated
__hash__, and freezing does not give it back — a FrozenDict is
still a mapping, so still unhashable. See __hash__.
dataclass
¶
Usage(
input_tokens: int = 0,
output_tokens: int = 0,
cost_usd: float | None = 0.0,
cache_read_tokens: int = 0,
cache_write_tokens: int = 0,
cost_source: CostSource | None = None,
)
Per-call resource consumption. Token counts are exact; cost_usd is a
running APPROXIMATION — see the field note below.
Budget accumulates a Usage for the token totals and keeps its own
exact Decimal ledger for money. Budget.spent() is what a run
reconciles against; summing usage.cost_usd is not.
A cooperative cancellation signal shared across a run tree (via RunContext).
A caller/parent calls cancel(); patterns check the token at safe points (loop top, between steps
or tool calls) and raise_if_cancelled() → Cancelled. This is an abort — distinct from a
graceful TerminationCondition (which returns a Stop) and from a resource ceiling (MeterExceeded).
Shared down the tree, so cancelling a parent cancels its children.
Bases: RuntimeError
Raised by a cooperative cancellation check when the run (or its subtree) has been cancelled.
Stuck(failure: Failure)
Bases: AgentkitError
A loop bounded by recurrence hit a repeat: the same outcome signature twice.
An AgentkitError and not a new root, so the framework-wide
except AgentkitError: still catches it — this is the taxonomy in
kernel.errors, not a parallel one.
It carries the Failure rather than only a message because the escalating caller and the
quiet caller must be looking at the same finding. on_repeat chooses delivery; a handler
that catches Stuck reads exc.failure and gets byte-identical data to what
on_repeat="stop" would have returned — including partial_output, which is the last
answer the loop actually produced and the only artifact worth salvaging from a stall.
dataclass
¶
CircuitBreaker(
name: str,
fail_threshold: int = 5,
cooldown: float = 15.0,
clock: Callable[[], float] = monotonic,
state: BreakerState = "closed",
_fails: int = 0,
_opened_at: float = 0.0,
_probe_started_at: float = 0.0,
_lock: Lock = Lock(),
_last_error: BaseException | None = None,
)
Per-dependency breaker. CLOSED → (fail_threshold consecutive fails) → OPEN → (cooldown) → HALF_OPEN → one probe → CLOSED on success / OPEN on failure.
property
¶
The failure that opened the breaker, or None. Lets a refused call
report a cause instead of a bare CircuitOpen.
Admit this caller, or refuse. The half_open state is itself the
single-probe gate: the caller that flips open→half_open IS the probe;
every other caller sees half_open and is refused until that probe
resolves. The lock is what makes "the caller that flips it" exactly one
caller, across threads as well as within an event loop.
A success closes the breaker from CLOSED or HALF_OPEN — never from OPEN.
The class docstring describes the state machine as
CLOSED -> OPEN -> (cooldown) -> HALF_OPEN -> CLOSED. There is no
OPEN -> CLOSED edge, and there was one in the code: this method
set state = "closed" unconditionally.
That edge is reachable at any concurrency above one, which is the
normal case for the documented pattern of ONE breaker shared per
dependency (client.py builds CircuitBreaker("agentkit.llm")
once). Several calls are in flight, enough fail to trip the breaker,
and then a straggler that started BEFORE the trip finishes
successfully and reports it — reopening the gate and sending the herd
straight back at the failing provider. Measured: a 300-second cooldown
skipped entirely by one late success.
Ignoring the report while OPEN is the honest reading of it: that call was admitted under the old state, so it is evidence about the past, not about whether the dependency has recovered. Only the single probe admitted after the cooldown speaks to that.
Count a failure. exc is optional and remembered only so a later
refused call can name what went wrong.
Hand the HALF_OPEN probe slot back after an outcome that says NOTHING
about dependency health — a PERMANENT contract failure (401 / 403 /
content filter) or a BaseException such as cancellation.
Neither record_success nor record_failure fits. record_failure
would fold the outcome into _fails, breaking the deliberate rule that
a contract failure is not an upstream health signal; record_success
would grant health credit for a call that never proved the dependency
recovered. Doing NEITHER — which is what happened — wedged the breaker in
half_open forever, because the only exits were those two methods.
So this is the third, neutral edge: consume nothing, grant nothing. The state returns to OPEN and the cooldown RESTARTS.
Restarting it is a correction. The first version deliberately did not,
reasoning that a permanently-failing dependency should keep surfacing
its real 401 rather than hide behind a CircuitOpen. The effect was a
breaker that does not brake: with a 60s cooldown, twenty consecutive
calls spanning 2ms ALL reached the dead dependency, because an
already-elapsed cooldown makes every caller a fresh probe, not just the
next one.
The premise was wrong anyway — run_with_resilience raises
CircuitOpen(...) from last, so the real error is already reachable
as __cause__ on the calls that are refused. Nothing was being
masked; the rate limiting was simply absent.
Bases: RuntimeError
Raised when a per-dependency breaker is OPEN (classified TRANSIENT → caller may degrade).
Bases: Protocol
The metrics seam. Implementations may be sync (OTel SDK does
its own async dispatch internally) so the methods are plain
def — easier to call from middleware without await
overhead.
Operations are best-effort: implementations MUST NOT raise into the run on metric-write failure (log + drop instead).
Increment a monotonic counter by value (default 1).
Standard names follow OTel semantic conventions:
gen_ai.client.request.count, gen_ai.client.error.count.
record_histogram(
name: str,
value: int | float,
*,
tags: Mapping[str, str] | None = None,
) -> None
Record a single observation in a distribution. Used for
token counts, durations, costs. Standard names:
gen_ai.client.token.usage,
gen_ai.client.operation.duration.
Default observer: drops everything. Lets a run emit freely with nothing attached (zero-config).
dataclass
¶
Observation(
kind: ObservationKind,
agent: str = "",
render: str = "",
run_id: str = "",
payload: Any = None,
parent_id: str | None = None,
trace_context: TraceContext | None = None,
)
A structured, product-facing record. payload is machine-readable; render is a short human line.
Frozen: an Observation is fanned out to observers (audit sinks,
WebSocket forwarders, rollup buffers) that may retain the ref
alongside the run's live emit loop. Post-emit mutation of
render/payload would corrupt the record every subsequent
consumer sees — the same reason TraceContext is frozen.
NO seq / ts, deliberately — they used to be here and were
removed rather than populated. Measured before the removal:
Observation(kind="tool_result", payload={}).seq was 0 and
.ts was 0.0, and NOTHING in the package ever set them — not
RunContext.emit, not the signal channel, not RollupObserver,
not any adapter. __hash__ nonetheless folded both into the stream
key and its docstring described seq as "that emitter's monotonic
counter", so the type promised an audit ordering it never delivered.
Populating them was the other option and was rejected on evidence, not taste:
- The channel already orders itself, more strongly.
emitis awaited, so a sink observes emissions in emission order:CollectingObserver.itemsis append-ordered andQueueObserver.stream()yields in insertion order. Aseqwould be a second, weaker source of truth for something the structure guarantees — and it would be wrong in the one case a field could beat arrival order (cross-process replay), because a per-RunContextcounter restarts at 0 in everychild()while a counter on the sharedServiceswould be shared by any two runs an application wires to oneServices. - Durable ordering is already owned by the store. The Postgres
log adapter allocates
seq BIGSERIAL PRIMARY KEYper row and reads backORDER BY seq. An audit stream that outlives the process gets its sequence there, from the only component that can make it total. - One truthful seam is not enough. Observations are constructed
at four sites (
RunContext.emit,SignalChannel'ssignal.emitted,RollupObserver's roll-up, and the testing fake ctx). Stamping only the emit seam re-tells the same lie in three smaller places. tshad no owner either. Wall-clock display and operational timing both already live on the spantrace_contextdeep-links into; a second, unpopulated timestamp on the record only invites the monotonic-vs-wall-clock question with no consumer to answer it.
A consumer that genuinely needs a per-sink sequence or arrival stamp should keep it in the SINK, beside the record — the sink is what knows when it saw the observation, and one shared record cannot carry a different number for each of the three fan-out targets anyway.
Bases: Protocol
Producer side of the channel. Concrete observers add a consumer surface (e.g. stream()).
close() is part of the Protocol. Rollup/buffering observers
need a shutdown hook to flush their buffered tail; without it the
trailing summary is silently dropped at process exit.
NoopObserver implements a no-op — passthrough impls can ignore
it entirely and Protocol satisfaction is unchanged.
Bases: Protocol
Operational tracing seam. Distinct from ObserverPort (the
product-facing observation channel) — tracing records what the
framework's middleware/capabilities did, with spans + structured
attributes, for debugging and performance analysis. Implementations
bridge to OpenTelemetry or structured logging.
span(name, kind, **attrs) is a context manager (sync or async —
both forms are supported) that yields a span object with .set(key,
value) and .add_event(name, **fields). The default NoopTrace
in runtime/context.py does nothing; production wires an adapter.
current_span_id() returns the currently-open span's
(trace_id, span_id) for in-process consumers (e.g. ctx.emit) to
attach as Observation.trace_context. Returns None when no span
is open (cold emit / no tracer configured).
add_event_to_current_span(name, **fields) drops a span event on
whatever span is currently open, without the caller having to hold
a reference to it. Used by ctx.emit to mirror CRITICAL observations
into the trace timeline. No-op when no span is open.
Bases: Protocol
Decides whether a span should be recorded.
Called by the tracing middleware BEFORE opening the span; when
it returns False no span is opened, no attributes stamped,
no replay record written. The whole observability pipeline
short-circuits for that call.
Implementations should be FAST (microseconds) — they're on the hot path. A 1ms sampler erases its own value.
should_sample(
*,
operation: str,
correlation_id: str,
attrs: Mapping[str, str | int | float] | None = None,
) -> bool
Return True to record the span, False to skip.
operation is the span name ("chat", "execute_tool",
"invoke_agent", ...). correlation_id is the run id —
ratio-based samplers hash it so all spans in one run share
the decision. attrs is the request-time attribute set
(useful for rule-based sampling — "always sample errors,"
"always sample org X").
Sample a deterministic ratio of runs.
Hashes correlation_id modulo a fixed denominator; keeps the
span if the hash falls below ratio * denominator. Result:
every span within the SAME run shares the keep/drop decision,
so traces are coherent (no orphan child spans). Across runs,
you get the configured ratio in aggregate.
Ratio 0.0 = drop everything; 1.0 = keep everything (same as AlwaysOn).
compose_failures(
failures: Sequence[Failure | None],
*,
source: str = "composite",
) -> Failure | None
Aggregate child failures into one (or None if there are none; passthrough a single one).
Category rule: PERMANENT if any child is permanent; TRANSIENT only if
all are transient; else UNKNOWN. The aggregate is retriable for any
non-PERMANENT category, matching Failure.of and
run_with_resilience. Children are preserved.
chain(
middlewares: list[Middleware | BaseMiddleware],
terminal: Handler,
) -> Handler
Fold the chain right so middlewares[0] is outermost: m0(m1(… terminal)).
Accepts raw Middleware functions and/or BaseMiddleware instances (adapted via as_middleware()), so
the two styles mix freely in one chain. Each is an async generator over the single streaming contract;
the terminal is the seam stream (llm.stream(**req) / a one-item tool stream). An empty list returns
the terminal unchanged.
async
¶
Reduce a stream to the operation's result — a chat result is just its collected stream. A chat
assembles its Deltas into an LLMResult; any other op returns its single (last) yielded item.
async
¶
The single item of a one-item (tool) stream.
assemble_deltas(deltas: list[Delta]) -> LLMResult
Pure reducer: fold streamed Deltas into the final LLMResult (content concatenated; usage /
finish_reason / model / provider / tool_calls / parsed taken from the deltas that carry them).
Delta.partial is deliberately DROPPED here, and that is not an
oversight — do not "fix" it. partial is a per-delta,
in-progress value produced by a tolerant parse; parsed is the
single strict-validated result. Lifting both onto LLMResult
would give the result type two competing typed fields where the
tolerant one (with possibly-unset required fields) could shadow the
strict one. This reducer only ever runs on a COMPLETE delta list,
so by the time it is called there is no in-progress state left to
carry: parsed is the answer.
In-flight partials reach consumers through the streaming path
instead — the cognitions forward Delta.partial onto
StreamEvent.partial_output per message_delta. See
:class:StreamEvent.
async
¶
gather_best_effort(
coros: list[Awaitable[R]], *, sem: Semaphore
) -> list[R | Failure]
Best-effort fan-out: bound concurrency, isolate failures (each slot is a result OR a
:class:~agentkit.core.contracts.errors.Failure wrapping the raised exception). Use when one
sub-agent/tool failing must NOT cancel the others.
Wrapping the exception in a Failure (rather than returning the raw exception object)
lets a parent treat the failed slot as first-class data: it carries a classified
category, a source naming the slot (gather_best_effort[i]), and the originating
cause, so a caller can retry / route around / escalate uniformly. It also disambiguates
"the coroutine returned an Exception value" from "the coroutine raised" — the raw-
exception design conflated the two.
The nesting rule from :func:gather_bounded applies here identically, and
the isolation this function provides does NOT soften it: a deadlock is not a
failure any slot can hold, so a nested fan-out over a private semaphore
hangs exactly as it would there. Pass sem=ctx.semaphore() whenever the
coroutines may fan out again.
async
¶
Run coroutines concurrently, at most sem._value at once, preserving input order.
A failure cancels the rest (structured concurrency) and surfaces as an ExceptionGroup.
Pass sem=ctx.semaphore(), not a semaphore of your own, whenever the
coroutines may themselves fan out. A permit is held for the WHOLE duration
of each coroutine, so a nested call draws from a pool its own ancestors have
already drained — and the result is a permanent hang, not a slowdown.
Measured with one hand-rolled asyncio.Semaphore(2) shared across levels,
a two-deep two-wide tree never returns; at Semaphore(4), three-deep does
the same. The cliff moves with the cap, so a tree that works in testing can
still wedge in production once it gets one level deeper.
ctx.semaphore() returns a pool keyed on ctx.depth, which makes the
cycle structurally impossible: an ancestor at depth d only ever holds
permits from pool d, and its children draw from pool d+1. See
:meth:agentkit.core.runtime.Budget.semaphore for the full rationale and the
bound this trades for (max_concurrency per level, so worst-case
in-flight work is max_concurrency * (max_depth + 1)).
A private semaphore is fine for a FLAT fan-out — a batch of independent tool calls or provider requests that do not fan out again.
async
¶
Run [(agent, task), …] concurrently — each under ctx.child(), bounded by the tree semaphore.
Returns AgentResults in input order. In best_effort mode a failed slot holds a
:class:~agentkit.core.contracts.errors.Failure (with the raised exception on .cause), not the
raw exception, so callers can inspect it as first-class data. The shared budget accrues
across all of them, and the depth cap applies tree-wide. The CONCURRENCY cap does not:
ctx.semaphore() is keyed on depth, so max_concurrency bounds each level
separately (see :meth:agentkit.core.runtime.Budget.semaphore for why a single tree-wide
pool deadlocks).
Two opt-in coordination seams engage automatically when the caller wired
them onto ctx:
- ActorBudget slicing — when
ctx.actor_budgetis set, carve one slice per child (1/Nof each axis of the parent's remaining) viareserve_for_childBEFORE dispatch. Each child's ctx sees its OWNActorBudget(with the slice asmax_*), and after every child returns wesettle_childon the parent's book so the parent'sused_*reflects what was actually spent (capped at reservation). If any reservation would exceed the parent's cap, we release the already-reserved siblings and re-raiseBudgetExhaustedbefore any child runs — fail-fast semantics. - SignalChannel attach — when the coordinator's ctx has
signal_channelset AND the childAgenthas its ownchannelfield, we callchild.channel.attach_parent(ctx.signal_channel.merge_inbox)soDataSignals emitted by the child fan up to the coordinator's merge inbox without polling.
Both features are strict no-ops when the caller doesn't opt in. Missing
fields on structural stubs (NullCtx) are tolerated via getattr.
The single sync bridge for sync hosts to drive async agentkit.
If no event loop is running in this thread → asyncio.run(coro). If a loop IS already running
(e.g. nested inside an async caller), spin up a fresh worker thread with its own loop and run the
coroutine there to completion — avoids "loop already running" while keeping a blocking signature.
async
¶
attempt_until_stuck(
fn: Callable[[], Awaitable[R]],
*,
fingerprint: Callable[[R], Any],
on_repeat: OnRepeat = "escalate",
max_attempts: int = 4,
ctx: Ctx | None = None,
) -> R | Failure
Attempt · fingerprint the outcome · attempt again on a different fingerprint · stop on a repeat.
::
answer = await attempt_until_stuck(
lambda: run_one(),
fingerprint=lambda outcome: outcome.failure_signature,
on_repeat="escalate",
max_attempts=4, # a BACKSTOP, not the bound
)
fingerprint returning None IS the success signal. There is no second "did it
work?" predicate, and the omission is deliberate: the natural fingerprint of a successful
outcome is "no failure signature", which the motivating expression
(outcome.failure_signature) already spells None. A separate is_done= would let
the two disagree — an outcome reported done while still carrying a signature, or the reverse
— and there is no defensible behaviour for that state. One function, one answer: a signature
means not yet, None means done, and fn's return value is handed straight back.
Signatures are compared against EVERY signature seen, not the previous one. This is the
case worth being careful about, and the reason the adjacent compare is wrong is
A, B, A, B, …: a run oscillating between two dead ends never repeats consecutively, so a
previous == current check never fires and the loop silently degrades to being bounded by
max_attempts after all. Measured against the naive version on ["A","B","A","B","A","B"]
with max_attempts=6: 6 attempts and no repeat detected, versus 3 attempts here — the
alternation is a cycle, and re-treading ground you have already covered is not progress no
matter how the visits are interleaved.
Comparison is by stable_hash of the signature rather than by hash() or ==, so a
signature may be unhashable — a dict or list is the shape an LLM-authored or
schema-validation signature naturally has ({"missing": ["a", "b"]}). Requiring hashability
would push every caller into hand-serialising, and str() would reintroduce the memory
addresses stable_hash exists to keep out of fingerprints. It also means the key is the
same basis as cache and idempotency keys, so two processes agree on what "the same signature"
means. The hash only indexes, though: a matching key is then confirmed with == before
a repeat is declared, because stable_hash is deliberately lossy for values it cannot
describe structurally and a collision here is a PERMANENT verdict rather than a cache miss.
See _same_signature. The practical consequence for a caller: a signature type with no
value equality (a plain object without __eq__) can no longer be falsely called stuck,
but its genuine repeats will not be detected either — that run is bounded by max_attempts
and reported as exhaustion. Signatures should be values: strings, dicts, lists, tuples,
enums, frozen dataclasses.
Raising attempts are not fingerprinted, ever — that is the seam with
run_with_resilience. An exception is a fault; it is classified, backed off and
breaker-counted by that function. An attempt that returned is a result, and only results
have outcomes to fingerprint. Nothing is caught here, so an exception propagates untouched.
The two therefore compose by nesting rather than overlapping::
await attempt_until_stuck(
lambda: run_with_resilience(call_model, max_attempts=3),
fingerprint=score,
)
Collapsing that into one loop was tried and is wrong twice over: a transient timeout would burn a semantic attempt (and get a fresh "signature" each time, so the stall detector could never fire), and the semantic retries would multiply the transient ones — 4 × 3 round trips at a provider that is already rate-limiting.
A raising fingerprint propagates for the same reason, stated in reverse: swallowing it and
substituting a placeholder gives every attempt a distinct signature, which is a loop that can
never detect a repeat. A broken fingerprint must fail loudly, not quietly turn this back into
a counted retry.
What comes back. House style, following gather_best_effort: the failure is data.
- success →
fn's outcome, unwrapped; - repeat → a
FailurecategorisedPERMANENT/retriable=False. Not a guess, which is whatclassify's substring matching is doing when it says PERMANENT: the run has demonstrated that another identical attempt produces an identical result. Delivered byraise Stuck(failure)underescalate, returned understop; - exhaustion → a
FailurecategorisedUNKNOWN/retriable=True.
Exhaustion returns in both modes, and that asymmetry is the point of the whole helper:
on_repeat names the repeat. Hitting the backstop while every signature still differs means
the run was moving and ran out of a budget the caller chose — the fix is a larger backstop,
not an escalation. Escalating it would put "you are going in circles" and "you were making
progress" through the same handler, which is precisely the confusion a count causes.
max_attempts < 1 raises ValueError instead of returning an exhausted Failure. A
zero-attempt Failure is indistinguishable from a real one, so a caller would log and page
on an upstream failure for work that never ran. max_attempts=1 is legal and degenerate: a
single shot, where no recurrence is possible and any signature exhausts. An on_repeat
outside {"escalate", "stop"} raises ValueError from the same block, for a sharper
version of the reason: an unrecognised mode must not quietly resolve to the non-raising one.
ctx is optional and, when given, check_cancelled() runs at the top of every attempt —
the same safe point ReActCognition and LedgerPolicy use. Including the first
attempt: a run cancelled before this was reached must not pay for one more model call to find
out. Mid-attempt cancellation needs nothing here, because Cancelled raised inside fn
is an exception and exceptions are not caught.
Full-jitter exponential backoff (prevents retry storms across workers).
Substring-classify an exception as TRANSIENT / PERMANENT / UNKNOWN.
TRANSIENT wins on collision. Alternative orderings that checked PERMANENT
first, so ValidationError("request timed out") matched "validation"
in the PERMANENT list and never got retried — exactly the failure mode
the resilience layer exists to handle. Two reasons TRANSIENT must win:
- TRANSIENT signals (
timeout,5xx,rate limit) are inherently more specific than generic PERMANENT substrings ("invalid","validation") that frequently appear in transient infra errors. - The conservative choice on collision is to retry — wasting one retry on a permanent error costs the backoff delay; failing fast on a transient error costs the whole call.
Substring matching is brittle by nature; for stronger guarantees,
callers should pass a custom classify_fn to run_with_resilience.
async
¶
run_with_resilience(
fn: Callable[[], Any],
*,
breaker: CircuitBreaker | None = None,
max_attempts: int = 3,
classify_fn: Callable[
[BaseException], ErrorClass
] = classify,
sleep: Callable[[float], Any] | None = None,
rng: Any = random,
) -> Any
Run async fn with classification + jittered retry + optional circuit breaker (async-first;
the one resilience entry point). await fn(); backoff via await asyncio.sleep (injectable
async sleep for deterministic tests).
PERMANENT errors fail fast; TRANSIENT/UNKNOWN retry up to max_attempts; an OPEN breaker raises CircuitOpen (itself TRANSIENT — the caller decides degrade vs propagate).