agentkit.kernel¶
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.
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.
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').
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.
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).
Default observer: drops everything. Lets a run emit freely with nothing attached (zero-config).
dataclass
¶
Observation(
kind: ObservationKind,
seq: int = 0,
ts: float = 0.0,
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.
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.
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/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).
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
¶
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,
)
Per-dependency breaker. CLOSED → (fail_threshold consecutive fails) → OPEN → (cooldown) → HALF_OPEN → one probe → CLOSED on success / OPEN on failure.
Bases: RuntimeError
Raised when a per-dependency breaker is OPEN (classified TRANSIENT → caller may degrade).
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).
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.
dataclass
¶
A retrievable unit for semantic memory / RAG (metadata carries source + scope tags).
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, ...] = (),
)
One turn in a chat transcript.
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,
)
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 exposed as an immutable MappingProxyType: 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).
MappingProxyType is not natively copy.deepcopy-safe
(stdlib limitation — the view can't be pickled), which matters
because Checkpointer.snapshot deep-copies state that
contains ToolCalls. __deepcopy__ / __copy__ unwrap
the view into a plain dict at copy time; the fresh ToolCall
re-wraps it via __post_init__.
dataclass
¶
JSON-schema advertisement of a tool — what the provider needs to see to call it.
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.kernel.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.
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.
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.kernel.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; depth/concurrency caps apply tree-wide.
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.
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.
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).