agentkit.capabilities

Optional collaborators: RequestBuilder, Grounder, Compactor, Guardrail, Evaluator, Checkpointer, output-schema SchemaAdapter.

See the Capabilities concept for the mental model.

Optional collaborators over the seams.

Grounder module-attribute

Grounder = Callable[[Ctx, str], Awaitable[str]]

A grounder is anything async-callable as (ctx, task) -> str.

Why a callable and not a MemorySource plus k/where knobs: those knobs are retrieval mechanics, not prompt-assembly concerns. RequestBuilder should not know that the grounding came from a vector store at all — only that some text is available for this task. The caller bakes the retrieval policy (k, where filter, query derivation, even the choice of source — vector index, MCP server, static fixtures) in once at wiring time, then hands the bound callable to the RequestBuilder.

A VectorMemory (or any MemorySource) is adapted to a Grounder by writing a small async wrapper that queries it and formats the returned MemoryItems:

async def grounder(ctx, task):
    items = await mem.query(task, k=5, ctx=ctx)
    return "\n".join(f"[{i.source}] {i.content}" for i in items)
builder = RequestBuilder(prompt=..., grounder=grounder)

For role-specific tuning, the wrapper stays adjacent to the agent definition without leaking into RequestBuilder's signature.

Checkpointer dataclass

Checkpointer(
    port: CheckpointPort, clock: ClockPort | None = None
)

Capability over a CheckpointPort.

snapshot() bumps the version, stamps created_at, and saves. resume() returns the latest checkpoint (or None if there's nothing to resume from). at_version() is the time-travel hook — return a specific historical snapshot, or None if it doesn't exist. delete() clears all versions for a run_id (terminal cleanup).

snapshot async

snapshot(
    run_id: str,
    state: dict[str, Any],
    *,
    status: CheckpointStatus = RUNNING,
    metadata: dict[str, Any] | None = None,
    ctx: Any = None,
) -> Checkpoint

Persist a new versioned snapshot. Returns the saved Checkpoint so the caller can log/expose its version. Failures from port.save propagate — the caller decides whether a checkpoint-save failure should abort the run.

state and metadata are deep-copied at the seam. Checkpoint is a frozen shell, but its state / metadata dicts are stored by reference; InMemoryStore returns those refs on resume. Without the copy, a caller that keeps a live handle to the scratch dict (or a resumer that pops entries from cp.state) would mutate the durable record — parity with any serializing backend (Redis, Postgres) would break silently. Deep-copy models the wire semantics locally so in-memory tests stay honest about the durability contract.

ctx is optional: when supplied and it carries a TracePort via ctx.trace, a checkpointer.snapshot span is opened around the save so operators can see checkpoint activity in the trace timeline. Callers without a ctx (bare unit tests, offline replayers) leave it unset and the span helper degrades to a nullcontext.

resume async

resume(
    run_id: str, *, include_terminal: bool = False
) -> Checkpoint | None

Return the latest RESUMABLE checkpoint for run_id, or None.

By default, a terminal snapshot (DONE / FAILED) is treated as "no resumable state" and this returns None — a naive "resume if any checkpoint exists" wiring MUST NOT re-run a finished job. Callers that legitimately want to inspect the terminal snapshot (an audit UI, an operator tool, replay / time-travel debugging) opt in explicitly with include_terminal=True.

status on the loaded checkpoint may be a raw string (some producers, and the Checkpoint shape's own tests, pass strings directly) rather than a CheckpointStatus enum instance; because CheckpointStatus is a StrEnum, wrapping the value normalises both cases without an isinstance branch.

at_version async

at_version(run_id: str, version: int) -> Checkpoint | None

Return the checkpoint at exactly version, or None if missing. The time-travel / replay seam.

list_versions async

list_versions(run_id: str) -> list[int]

Return all known versions for run_id in ascending order.

delete async

delete(run_id: str) -> None

Clear all checkpoints for run_id. Idempotent — deleting an unknown run is a no-op.

Also evicts the per-run asyncio.Lock stashed in _run_locks — without this, a long-lived Checkpointer serving many short-lived runs (a shared engine pool, a coordinator spawning throwaway children) would accumulate one lock per distinct run_id for the process lifetime. Deletion is the producer's signal that a run is over, so it's the natural eviction point. The lock is dropped only AFTER port.delete succeeds — a failed backend delete leaves the lock in place so a retry stays serialised with any in-flight snapshot for the same run.

Compactor

Bases: Protocol

The single seam every compaction policy implements.

A Compactor takes a transcript and returns a (possibly smaller) transcript. Implementations own their own threshold check — a no-op return on a transcript that is already under budget is expected, and lets the caller call compact() unconditionally without a guard at the call site.

ImportanceFilteringCompactor dataclass

ImportanceFilteringCompactor(
    filterer: Any,
    model: str = "",
    max_tokens: int = 12000,
    keep_recent: int = 2,
    estimate: Callable[
        [list[Message]], int
    ] = _approx_tokens,
)

Uses an LLM to identify and keep only the most important turns.

SlidingWindowCompactor dataclass

SlidingWindowCompactor(keep_recent: int = 10)

Strictly keeps only the system prompt and the N most recent turns.

SummarizationCompactor dataclass

SummarizationCompactor(
    summarizer: Any,
    model: str = "",
    max_tokens: int = 12000,
    keep_recent: int = 4,
    estimate: Callable[
        [list[Message]], int
    ] = _approx_tokens,
)

Summarizes the older middle of the conversation using an LLM.

TruncationCompactor dataclass

TruncationCompactor(
    max_tokens: int = 12000,
    keep_recent: int = 4,
    estimate: Callable[
        [list[Message]], int
    ] = _approx_tokens,
)

Drops the oldest messages (excluding system prompt) to fit within token limits.

Evaluator

Evaluator(
    code_checks: dict[str, Callable[[Any], bool]]
    | None = None,
    *,
    judge_model: str = "",
    rubric: str = "",
)

judge async

judge(
    ctx: Any, *, goal: str, output: Any
) -> dict[str, Any]

Run the judge model through the invoker chain — tracing + retry + meter all fire. Direct self._judge_llm.complete() would bypass middleware: a flaky judge couldn't be retried and its cost wouldn't land on the run's budget. Routing through ctx.invoker.chat() keeps the judge on the same rails as every other model call.

Off-hot-path: caller is expected to fire-and-forget on a worker / background task. Returns {} on any failure (broken judge result, JSON parse failure, missing invoker) so monitoring records the empty result without crashing the run.

OutputCoercionError

OutputCoercionError(
    message: str,
    *,
    raw: str | dict[str, Any],
    errors: list[str] | None = None,
)

Bases: Exception

Raised when a model response can't be coerced into the declared schema.

Carries the validation diagnostics (errors) so the retry middleware can reflect them back to the model as a user message — "your previous response failed these checks: …, please try again" — which is the single most effective fix for the "almost-correct JSON" failure mode. The original raw payload (raw) is preserved for logging and for the transcript so a human reviewing a failed run can see exactly what the model emitted.

Wrapping every flavour's native exception (Pydantic ValidationError, stdlib json.JSONDecodeError, our own missing-field errors) under a single type means callers don't have to know which adapter is in play to write a retry policy.

raw instance-attribute

raw = raw

The exact payload (string or dict) the model produced — preserved verbatim for logging, transcripts, and the "show me the failure" UI.

errors instance-attribute

errors = errors or []

Per-field validation diagnostics as plain strings. The retry middleware joins these into a user message; the eval harness asserts on them.

SchemaAdapter

Bases: Protocol[T]

Uniform interface over any output-schema flavour the user declares on an Agent.

Implementations exist for Pydantic BaseModel, stdlib dataclass, attrs class, and raw JSON Schema dict. New flavours plug in by satisfying this protocol — the adapt() dispatcher is the single place that picks one for a given user input.

name instance-attribute

name: str

Used as the tool name on Anthropic structured-output mode and as the schema label inside the system-prompt fallback. Defaults to the class __name__ for Python types; pass-through for raw JSON Schema dicts (caller can name them).

python_type instance-attribute

python_type: type[T]

The concrete type instances are coerced INTO. dict for the raw-JsonSchema flavour. Used for AgentResult[T] generic stamping and isinstance checks at the boundary, so callers can write assert isinstance(result.output, Schema) without caring which adapter is in play.

json_schema

json_schema() -> dict[str, Any]

Render the JSON Schema dict the LLM (or its structured-output mode) will see. Must be self-contained — no $ref to external documents — because some providers reject external refs.

parse

parse(raw: str | dict[str, Any]) -> T

Coerce a raw model response into a typed instance. Accepts both: - str: the model emitted a JSON string (text-mode fallback path). - dict: a provider's structured-output mode already parsed it for us.

Raises :class:OutputCoercionError (NOT the underlying ValidationError / json.JSONDecodeError / KeyError) so the retry middleware has a single exception type to catch and a structured diagnostics list to reflect back to the model. Never returns None — partial / unfinished outputs go through :meth:partial_parse instead.

partial_parse

partial_parse(raw: str) -> T | None

Tolerant parse for streaming. Returns the best-effort partial object the so-far buffer permits — e.g. a Pydantic model with whatever fields have already arrived, missing required fields left unset. Returns None if the buffer has zero useful structure yet (empty, pure whitespace, a single open bracket the parser can't resolve to a useful prefix).

Implementations build the partial OBJECT via the type's bypass-init path (BaseModel.model_construct / object.__new__ + setattr) so missing required fields don't raise. The streaming output_coerce middleware calls this on each text delta and lifts the result onto Delta.partial when it changes — the consumer sees a steadily more-complete partial without waiting for the terminal delta. The strict :meth:parse still runs at end-of-stream and remains the source of truth for the final typed object.

serialize

serialize(value: T) -> dict[str, Any]

Round-trip the typed instance back to a JSON-serialisable dict. Used by the durable run-store (so a resumed run can read back the same output the original run produced) and by the transcript replay for evals.

validate

validate(value: Any) -> T

Validate an arbitrary Python value against the schema.

Distinct from :meth:parse (which expects raw JSON or dict from a model response). validate handles the tool-result case, where the value coming in is whatever Python object the tool's function produced — could be an already-instantiated model, a dict the tool built by hand, or a primitive when the schema wanted an object.

• Already-typed instance (e.g. MyModel(...) passed when output_schema=MyModel) — fast-path, returns as-is.
• Dict → run through coercion (typically via parse) to build the typed instance.
• Anything else (wrong type entirely) → raise :class:OutputCoercionError.

Used by the tool-result schema check on FunctionTool, where the tool's return value must match its declared output_schema.

BuiltRequest dataclass

BuiltRequest(
    messages: list[Message],
    prompt_version: str,
    approx_tokens: int,
)

The output of RequestBuilder.build(). Carries the messages ready to send AND the cross-cutting signals the caller's invoker and telemetry need:

• prompt_version lets the caller stamp the agent's output for attribution (so a regression maps back to the exact template).
• approx_tokens is a crude pre-call estimate the caller can use for budget pre-checks (the authoritative count comes back from the provider in Usage; this is just a guardrail).

RequestBuilder dataclass

RequestBuilder(
    prompt: Prompt,
    grounder: Grounder | None = None,
    compactor: Compactor | None = None,
    reground_every_turn: bool = False,
    budget_check: BudgetCheck | None = None,
)

Assembles the LLM input. One instance per agent role per run is typical — the seed prompt and grounding policy don't change inside a run, only the task and the growing transcript do.

Required

prompt: the versioned seed system prompt (an agentkit.prompts.Prompt).

Optional

grounder: an async callable (ctx, task) -> str injected on the first turn (or every turn if reground_every_turn). Empty string means "no grounding for this task" and produces no grounding message. The caller owns the policy (k, where, query derivation) — RequestBuilder just calls. compactor: a Compactor to fold the transcript when it grows. Applied AFTER the new turn is appended, to the WorkingContext.messages tail. The cache-stable prefix is never touched — compactors must not pretend to. Kept- recent tail always includes the message the LLM is about to answer. reground_every_turn: when True, the grounder is invoked on every turn, and the previous grounding message is dropped before the new one is appended. The default (False) preserves the legacy first-turn-only behaviour, which keeps the prefix cacheable. budget_check: optional (approx_tokens) -> None callable invoked at the end of build(). Raise to abort — the RequestBuilder surfaces no judgement of its own about what's "too big," that policy lives entirely with the caller.

build async

build(
    task: str,
    wc: WorkingContext,
    ctx: Ctx,
    *,
    output_adapter: SchemaAdapter[Any] | None = None,
) -> BuiltRequest

Append a turn to wc and return the messages to send.

First turn (empty prefix AND empty messages): the system prompt + (optional) grounding land in wc.prefix — the cache-stable head — and the user task is appended to wc.messages. Subsequent turns: just append the user task (and re-ground if reground_every_turn).

output_adapter: when supplied, the adapter's JSON Schema is rendered into wc.prefix.schema_block on the first turn — a cache-stable third pinned system message after the prompt and grounding. Never written into wc.messages (cache invariant).

Mutates wc in place so the caller's running blackboard stays the source of truth for the transcript. The returned messages list is a fresh copy of wc.assembled() — safe to hand to the invoker without worrying about post-call mutation racing the next turn.