Task and state collaborators¶
Optional collaborators live with their owner; there is no capability aggregate package.
See the Capabilities concept for the mental model.
Requests and grounding¶
RequestBuilder capability — the single seam where the four engineering disciplines meet.
See base for the RequestBuilder / BuiltRequest / BudgetCheck surface and
grounder for the two grounding seams — the flattened Grounder and the
typed GroundingSource that keeps a retrieved item's provenance alive as far
as the prompt.
module-attribute
¶
A budget pre-check hook: (approx_tokens) -> None. Raise to abort
build() before the messages are returned. The RequestBuilder does not
prescribe an exception type — the caller chooses what BudgetExceeded,
PolicyDenied, or similar looks like for their system.
module-attribute
¶
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.
When the flattening in that wrapper is throwing away something the
application needs to act on, reach for GroundingSource instead — it
is the same seam with the items left intact.
module-attribute
¶
GroundingAdmit = Callable[[MemoryItem], bool]
Per-item admission predicate. False keeps the item out of the prompt
entirely — not merely out of the rendered text.
Deliberately one-item-at-a-time rather than a Sequence -> Sequence filter:
a whole-list hook is also a reordering and re-scoring hook, and ranking
belongs to the memory source (or its Reranker), which is the only layer
that knows what the scores mean. Keeping this shape means the order the
source returned is the order render sees.
module-attribute
¶
GroundingRender = Callable[[Sequence[MemoryItem]], str]
How admitted items become the text of the grounding block.
A pure, synchronous function on purpose. Rendering is string assembly over data already in hand; making it async would invite a second retrieval round trip into the middle of prompt assembly, where a failure has no honest recovery and the cost is invisible to the budget pre-check.
module-attribute
¶
GroundingSource = Callable[
[Ctx, str], Awaitable[Sequence[MemoryItem]]
]
The typed grounding seam: async-callable as (ctx, task) -> Sequence[MemoryItem].
Identical in spirit to Grounder — the caller still bakes the retrieval
policy (k, where, query derivation, which backend) in at wiring time and
RequestBuilder still treats the callable as opaque. The only difference is
what comes back, and that difference buys three things:
-
The items are inspectable before rendering.
RequestBuilder.admitis a per-item predicate that runs before the join, so an application can refuse an item outright — most usefully the model-authored kind::admit=lambda item: item.metadata.get("tier") != "inferred"
Filtering the rendered STRING instead cannot work: once joined, nothing identifies which span came from which item.
-
The rendering is a policy.
RequestBuilder.renderreceives the admitted items, so numbered citations, score-annotated lines or a JSON block are a one-liner rather than a rewrite of the retrieval wrapper. -
The items can be recorded beside the prefix. With
record_grounding=Truethe admitted items land onWorkingContext.scratchpad[GROUNDING_RECORD_KEY], so a finished run can say what it grounded ON, not only what it said. They land as JSON-safe{"content", "source", "score", "metadata"}dicts rather than liveMemoryItems, because the scratchpad is copied into the checkpoint blob and a record that cannot cross a resume is not an audit trail.
agentkit.state.memory.as_grounding_source adapts any MemorySource to this by
not flattening — strictly less code than as_grounder, which is the
usual sign the seam is at the right level.
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_versionlets the caller stamp the agent's output for attribution (so a regression maps back to the exact template).approx_tokensis a crude pre-call estimate the caller can use for budget pre-checks (the authoritative count comes back from the provider inUsage; this is just a guardrail).
dataclass
¶
RequestBuilder(
prompt: Prompt,
grounder: Grounder | None = None,
compactor: Compactor | None = None,
reground_every_turn: bool = False,
budget_check: BudgetCheck | None = None,
grounding: GroundingSource | None = None,
admit: GroundingAdmit | None = None,
render: GroundingRender | None = None,
record_grounding: bool = False,
)
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.agents.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.
"Turn" throughout means one call to build(), which is
one user turn of a conversation — NOT one step of a tool
loop. A ReActCognition.drive() calls build() exactly
once and then appends tool traffic to the tail directly,
so a 4-step ReAct run invokes the grounder once whatever
reground_every_turn says (measured: 4 LLM calls, 1
grounder call). The flag only bites when a caller reuses
one WorkingContext across successive agent.run(...)
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: whether grounding is retrieved once or
refreshed on every turn. This is a genuine trade, not a
compatibility shim — both settings are wrong for some
workload, so the knob stays and the caller picks.
False (the default) retrieves once, on the first turn, and
pins that block for the life of the context. Turn 5 is
answered with the evidence retrieved for turn 1's
question. Measured on a 5-turn conversation over a
5-fact handbook at k=1: 1 of 5 turns saw the fact that
actually answered it, and the prefix was bit-identical on
5 of 5 turns.
True re-invokes the grounder each turn and rebuilds
``PrefixContext.grounding`` with the result, so the block
tracks the current question — 4 of 5 turns saw their
answering fact in the same probe (the fifth was a
retrieval miss, not a staleness one). The price is that
rewriting the prefix invalidates the provider's KV cache
every turn: 1 of 5 prefixes matched turn 1. On a
4,000-token grounded prefix over a 20-turn conversation
with claude-sonnet-4-6, that is $0.2280 of fresh prefix
reads against $0.0228 of cache reads — 10x on the prefix
term, every turn, forever.
Rule of thumb: leave it False when the grounding is a
corpus the whole conversation shares (a handbook, a spec,
a codebase digest) and True when each turn asks about
something different and the retrieved evidence is the
answer. The default is False because the auto-wired
``Agent(memory=...)`` path constructs this builder with no
way to set the flag, so the default is what every
auto-wired agent gets — and silently paying 10x on the
prefix is a worse failure to hand someone by default than
grounding they can see is stale in the transcript.
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.
grounding: the TYPED alternative to grounder — an async
callable (ctx, task) -> Sequence[MemoryItem]. Same seam,
same place in the turn, same landing spot in the prefix; the
difference is that the items arrive intact, so admit can
refuse one and render can see its source and score.
Mutually exclusive with grounder (see __post_init__).
admit: optional per-item predicate applied to what grounding
returned, BEFORE render. A rejected item does not reach
the prompt at all. The rule worth having it for is
lambda i: i.metadata.get("tier") != "inferred" — a memory
a model wrote is not evidence, and this is the only point in
the pipeline where that is still checkable.
render: optional Sequence[MemoryItem] -> str deciding how the
admitted items become the grounding block. Defaults to
render_grounding ([source] content lines), which is
byte-identical to what as_grounder produced.
record_grounding: when True, the admitted items are stored on
wc.scratchpad[GROUNDING_RECORD_KEY] so a finished run can
report what it grounded ON, not only what it said. Off by
default — an audit trail nobody reads is still a write into
the caller's blackboard on every turn. Recorded as a tuple of
JSON-safe {"content", "source", "score", "metadata"} dicts,
not as live MemoryItems, because the scratchpad is durable
state — see _record. Rewritten (not appended to) on every
re-ground, so it always describes the prefix as it stands.
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.
render_grounding(items: Sequence[MemoryItem]) -> str
Render records as attributed evidence; callers decide admission first.
Evaluation¶
Eval capability — see base for the Evaluator class.
Evaluator(
code_checks: dict[str, Callable[[Any], bool]]
| None = None,
*,
judge_model: str = "",
rubric: str = "",
)
async
¶
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.
Context compaction¶
Compactors — policies for bounding a transcript before it hits the model.
See base for the Compactor Protocol and impls for the built-in concrete implementations.
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.
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.
dataclass
¶
Strictly keeps only the system prompt and the N most recent turns.
Checkpoints¶
Checkpointer capability — the durable-resume facade + the shape (de)serializers.
base.py houses the Checkpointer class (the ergonomic facade over CheckpointPort).
persistence.py houses the shape-conversion plumbing (Message/Usage/ToolCall ↔ dict, the
coordinator-state shape, and StoreBackedCheckpointStore — the CheckpointPort over a
plain StorePort that makes Services(store=...) a durable wiring) that the ReAct loop
and the coordinator policies use to round-trip mid-run state through a checkpoint.
Everything here is re-exported so callers say
from agentkit.state.checkpoints import … — they shouldn't have to know which
submodule a given name lives in.
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).
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 recursively freezes JSON-shaped state and
metadata payloads. Returned snapshot and adapter-read handles
cannot mutate nested dict/list values in place. The copy separates
the caller's original state graph before that freeze and models
the wire boundary for in-memory backends. Use an explicitly thawed
copy to edit state before taking another snapshot.
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.
A SECRET-TAINTED state is never persisted — see
:func:_is_tainted. This check lives here, at the single seam
every producer passes through, rather than in each cognition or
policy: there are seven snapshot call sites across the tool
loop and the coordinator policies, and a containment rule that
has to be re-implemented at each one is a rule that will be
missed at the eighth.
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.
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.
async
¶
Return all known versions for run_id in ascending order.
async
¶
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.
A minimal CheckpointPort over a generic StorePort.
This is what makes Services(store=...) a durable wiring. resolve_checkpointer
synthesizes one when a ctx.store is present and no Checkpointer was injected,
which is the ergonomic path every producer depends on: the tool loop, Workflow,
the coordinator policies and PlanPolicy's human gate all resolve through that one
function, and Workflow's own unpersisted-gate warning names Services(store=...)
to the user as one of the two ways to make a suspend resumable. Measured by deleting
the synthesis and running the suite: 23 tests fail across workflow gates, plan gates,
tool-loop approve/resume and coordinator slot isolation. It is the DEFAULT, not a
compatibility shim — an app with a StorePort already wired should not also have to
hand-construct a CheckpointPort to get resume.
Wire a real CheckpointPort adapter (InMemoryCheckpointStore, PostgresCheckpointStore)
when you want version HISTORY, which is the one thing this cannot give you.
The single-slot-per-run model is deliberate: only the latest version survives on the
store, so time-travel / replay is unavailable here and Checkpointer.list_versions()
returns at most one element. A generic KV cannot losslessly model a version history
without leaking a schema into StorePort, and resume — which reads latest and
nothing else — pays nothing for the limitation.
resolve_checkpointer(
ctx: Any, explicit: Checkpointer | None = None
) -> Checkpointer | None
The ONE durable-state resolution order, shared by every producer.
explicit (a producer's own checkpointer=) > ctx.checkpointer >
a StoreBackedCheckpointStore synthesized over ctx.store. None
when nothing is wired, which callers treat as "durable resume disabled".
That third step is an ERGONOMIC default, not a compatibility shim: an app
that already wired a StorePort gets durable resume without also
hand-constructing a CheckpointPort, and Workflow advertises
Services(store=...) to users as one of the two supported ways to make a
human-gate suspend resumable. It gives up version history (one slot per
run) and nothing else, which costs a resumer that only reads latest
exactly nothing.
This lives here rather than on any one producer because having two
resolution orders is how a seam silently stops working: Workflow
persisted only through ctx.store while the ReAct cognition preferred
ctx.checkpointer, so wiring the documented durable seam — a
Checkpointer — left workflow human-gates unpersisted, and the failure
surfaced later as "no suspended workflow
The KV key StoreBackedCheckpointStore reads and writes for run_id.
Public because "where is this run's checkpoint in my store?" is an operator
question — an admin tool clearing a stuck suspend, or a test asserting a
snapshot really reached the backend, needs the same derivation the port
uses, and hardcoding checkpoint:<id> in a second place is how the two
drift. tests/source/unit/state/adapters/test_durable_resume_backends.py asserts against
this key precisely so the store round trip stays pinned to it.
coord_state_from_dict(
d: dict[str, Any],
) -> tuple[
int, list[Message], dict[str, Any], list[Any], Usage
]
Rehydrate (turn, transcript, scratchpad, results, usage) from a Checkpoint.state.
The inverse of coord_state_to_dict. stop_reason and status aren't returned —
callers read them off the surrounding Checkpoint/state dict directly when they
matter (e.g. gating resume on status != "done").
coord_state_to_dict(
*,
turn: int,
transcript: list[Message],
scratchpad: dict[str, Any] | None,
results: list[Any],
usage: Usage,
stop_reason: str | None,
status: str,
) -> dict[str, Any]
Serialize a coordinator Agent's mid-run state into a flat dict suitable for
Checkpoint.state.
Shape: turn is the next turn index to execute on resume (so a snapshot taken
after completing turn N stores turn=N+1); transcript is the full message list
(works for blackboard mode too — on resume the caller's WorkingContext.messages is
repopulated from this); scratchpad is the blackboard scratchpad (or {} when no
blackboard); results is the per-turn AgentResult list serialized so far; usage
is the accumulated total; stop_reason is the termination/ceiling reason (typically
None mid-run, set on a "done" snapshot); status is "running" / "done" /
"failed". Re-uses msg_to_dict and usage_to_dict so leaf-agent and coordinator
checkpoints share a shape language.
Notes on what is intentionally NOT serialized
- The children dict — children are passed at construction; the same coordinator Agent must be rebuilt to resume.
- Termination-condition state — most built-ins are stateless or re-derivable
from the transcript (
MaxMessages/MaxTurnscounters == turn count).Timeout(wall-clock start),ExternalTermination(flag), and a statefulFunctionalTerminationpredicate are NOT re-derivable; on resume we.reset()and replay the rehydrated assistant deltas through the condition so cumulative counters catch up. Wall-clock conditions effectively restart their timer on resume — flagged for follow-up. - Policy state — selectors and assessors are treated as stateless (the built-ins read the last message / call a model). A policy with private state would need its own serialization seam.
dict_to_prefix(d: dict[str, Any]) -> PrefixContext
Inverse of prefix_to_dict.
Takes the block prefix_to_dict writes, and only that. An earlier
version also accepted None/absent and returned an empty
PrefixContext, so that a snapshot written before the context split
(system prompt inline as the first messages entry) still resumed.
That second shape has no writer: ReActCognition._save is the only
thing in the tree that produces a {"messages": ...} state and it has
always emitted prefix since the split, which landed before the first
release — 0.1.0 is the first published version and already contains
PrefixContext, so no release ever wrote the older shape.
A decoder branch that no writer can reach is not tolerance,
it is a silent default: a state that lost its prefix on the wire
would have rehydrated with an EMPTY system prompt and re-run the agent
with no instructions. Requiring the key turns that into a KeyError
naming the field, which is what the rest of rehydrate's required
set already does.
The per-field .get defaults below are a different thing and stay:
they let a caller hand-assembling a prefix block omit the parts it does
not use (these helpers are exported for exactly that), and they cost
nothing because each field's absence has one obvious meaning.
prefix_to_dict(p: PrefixContext) -> dict[str, Any]
Serialize a PrefixContext (the cache-stable head of a
WorkingContext). Round-trips the system prompt + grounding
messages + schema block — the three fields a v2 checkpoint must
rehydrate to reconstruct an identical prefix.
rehydrate(
saved: dict[str, Any],
) -> tuple[WorkingContext, Usage, int, bool]
Rebuild (WorkingContext, Usage, next_iteration, repaired) from
a checkpoint dict.
ONE shape, the one ReActCognition._save writes. Four keys are
REQUIRED — the prefix block (system prompt + grounding + schema),
the tail messages, the accrued usage and the next
iteration — and a missing one raises a KeyError naming it — an operator gets a diagnosable failure rather
than a run that quietly resumes with half its state. scratchpad /
limit / shared / repaired stay optional because each has a
single unambiguous default (empty, unbounded, unshared, unrepaired)
that is indistinguishable from having been written.
Serialize an AgentResult into a flat dict. The parsed field is passed through
raw — for a serializing port the caller is responsible for ensuring parsers produce
primitive/JSON-safe values (matches the workflow durable-resume contract).
Guardrails¶
Typed model output¶
Output-schema adapter layer — uniform bridge between "user declared this output shape on their Agent" and "framework can ship the matching JSON Schema to the LLM and coerce the response back into a typed instance."
Public surface:
* :class:`SchemaAdapter` — the Protocol every flavour satisfies.
* :class:`OutputCoercionError` — the single exception type the retry
middleware catches across all flavours.
* :func:`adapt` — dispatcher: takes any supported spec and
returns the right adapter.
The individual adapter classes (:class:PydanticAdapter,
:class:DataclassAdapter, :class:AttrsAdapter, :class:JsonSchemaAdapter)
are re-exported for type-hinting and direct construction; in normal use the
caller goes through adapt() and never names a concrete adapter.
Optional dependencies (pydantic, attrs, jsonschema, msgspec) are NEVER imported at module load — every probe is guarded so this subpackage stays importable in a minimal environment. The corresponding flavour simply becomes unavailable until its dep is installed.
Bases: Generic[T]
Adapter over an attrs-decorated class. Same fail-fast schema construction
as :class:DataclassAdapter.
instance-attribute
¶
The attrs class itself — used for isinstance at the boundary.
instance-attribute
¶
Defaults to the class __name__; caller can override.
Deep-copy so a provider-specific massage can't mutate the cached schema.
Same shape as the dataclass adapter's parse: JSON-decode if needed,
walk-and-construct, batch errors into one :class:OutputCoercionError.
Tolerant parse → build a partial attrs instance.
Mirrors the dataclass adapter. attrs's __init__ rejects
missing required positional args, so we bypass it: instantiate
via object.__new__ and setattr only the fields the
stream has produced. Seeds defaults for the rest so attribute
access doesn't raise on un-streamed fields.
Caveat: attrs's slotted classes (slots=True) reject
setattr for un-declared attributes. We catch and return
None in that case — the streaming partial is best-effort,
and the strict terminal parse remains the source of truth.
attrs.asdict walks recursively into nested attrs classes, lists,
and dicts.
Validate an arbitrary Python value against the attrs class.
Mirrors :meth:DataclassAdapter.validate — fast-path an instance,
coerce a dict through the type-tree walker, raise
:class:OutputCoercionError for anything else.
Bases: Generic[T]
Adapter over a stdlib @dataclass class. Constructor validates the
type-tree up-front — every field annotation is run through _type_to_schema
so an unsupported shape (set[X], Any, bare list) raises
TypeError at agent-build time rather than at first model response.
instance-attribute
¶
The dataclass type itself — used for isinstance at the boundary.
instance-attribute
¶
Defaults to the class __name__; caller can override.
Return a deep-ish copy so a downstream mutation (provider-specific massaging) can't corrupt the cached schema.
Parse JSON if needed, then walk the type tree and construct the
dataclass. All validation errors are batched into a single
:class:OutputCoercionError.
Tolerant parse → build a partial instance with only the fields the stream has supplied so far.
Bypasses __init__ (which would reject missing required fields)
via object.__new__ + setattr per present field. Fields
absent from the partial JSON stay unset on the instance — the
same convention Pydantic's model_construct uses. Returns
None on any failure; never raises.
Equality between partial instances goes through the dataclass's
synthesised __eq__, which compares every declared field.
Unset fields can cause __eq__ to raise AttributeError —
which the streaming middleware swallows when it compares
new != last. To keep that comparison usable, we ALSO seed
any declared default / default_factory so the instance is at
least readable without AttributeError for absent fields
that the dataclass itself defined a fallback for.
dataclasses.asdict walks recursively, converting nested dataclasses,
lists, and dicts into plain Python primitives. Round-trip guarantee holds
for the supported type subset.
Validate an arbitrary Python value against the dataclass.
Fast-paths an already-typed instance. Coerces a dict via the same
walker parse uses (so a tool that returns a hand-built dict
gets validated against the type-tree). Anything else (wrong type
entirely) raises :class:OutputCoercionError — the tool-shape
check has a clear single point to catch.
Adapter over a raw JSON Schema dict. python_type is dict because
that is the type instances are coerced into — there is no Python class
associated with the schema.
class-attribute
instance-attribute
¶
Always dict — the raw-schema flavour deliberately does not stamp a
Python class. Callers who want a typed result should use one of the other
adapters.
instance-attribute
¶
Defaults to the schema's title if present, else a bland "Output"
— raw dicts have no class name to fall back on, so the caller is
encouraged to pass name= when registering one.
Return an independent mutable wire schema from the frozen canonical contract.
JSON-decode and validate against the canonical host schema.
Tolerant parse → return the partial dict as-is.
No further coercion is done — the raw-schema flavour returns
dict instances on full parse, and the same convention
applies mid-stream. Schema validation is intentionally skipped
on the partial: a half-streamed object would routinely violate
required, and the strict terminal parse is the source of
truth for conformance.
Identity — the value already IS the dict. Returned as-is rather than deep-copied because the durable store treats serialized output as owned-by-store after the call returns.
Validate an arbitrary Python value against the raw JSON Schema.
The value must be an object, and every terminal parse/validation uses the canonical host validator. Provider-side guarantees do not replace host validation and missing dependencies never weaken the contract.
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.
instance-attribute
¶
The exact payload (string or dict) the model produced — preserved verbatim for logging, transcripts, and the "show me the failure" UI.
instance-attribute
¶
Per-field validation diagnostics as plain strings. The retry middleware joins these into a user message; the eval harness asserts on them.
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.
instance-attribute
¶
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).
instance-attribute
¶
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.
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.
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.
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.
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 tests.evals.
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 whenoutput_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.
Bases: Generic[T]
Adapter over a pydantic.BaseModel subclass.
Constructor takes the model class (not an instance). The adapter is stateless
beyond a reference to that class — safe to share across runs and across
threads/tasks (Pydantic's model_validate* are pure functions of input).
instance-attribute
¶
The BaseModel subclass itself — used for AgentResult[T] stamping and
isinstance(result.output, Model) checks at the boundary.
instance-attribute
¶
Defaults to the class __name__ (e.g. "WeatherReport"); the caller
can override when the model name would clash with another tool name or when
a more human-readable label is wanted in the system-prompt fallback.
Pydantic's own model_json_schema is the source of truth. We adjust for
provider "strict" expectations: OpenAI's structured-output strict mode
forbids default keys on required fields (it treats them as a hint that
the field is optional). Dropping the key when the field is in required
keeps the schema valid for both strict and non-strict consumers.
Dispatch on input type: a string goes through model_validate_json
(which parses + validates in one pass, surfacing JSON syntax errors with
line/column context), a dict goes through model_validate (the
provider already parsed the JSON for us in structured-output mode).
String inputs are first passed through :func:_strip_markdown_fences —
Claude and other instruction-tuned models routinely wrap JSON output in
json … fences even when the prompt asks for raw JSON.
Valid JSON can never start with a backtick, so unconditional stripping
is loss-less.
Every Pydantic ValidationError is wrapped in :class:OutputCoercionError
so the retry middleware has a single exception type to catch. The original
per-field errors are flattened to strings — preserving the dotted-path
location ("address.city: field required") which is exactly what we want
to reflect back to the model.
Tolerant parse → construct a partial model from whatever fields have already arrived.
Uses model_construct(**data) rather than model_validate so
missing required fields don't fail — the partial is intentionally
under-specified. On any failure (including incomplete JSON our
tolerant parser can't make sense of) returns None and the
caller pumps another delta. Never raises.
Side note on equality: Pydantic BaseModel instances compare
by field values, so the streaming middleware can cheaply
deduplicate identical partials with new == last.
model_dump returns plain Python types (dicts, lists, primitives) — no
Pydantic-specific objects leak into the durable store or transcript. That's
the round-trip guarantee: parse(json.dumps(serialize(v))) == v.
by_alias=True is what actually makes that guarantee true. Both of the
adapter's other surfaces speak ALIASES — model_json_schema() (validation
mode, what the model is shown) and model_validate* (what parse
accepts) — while a bare model_dump() emits FIELD NAMES. Measured on
user_name: str = Field(alias="userName"): serialize gave
{'user_name': 'bob'} and feeding that straight back to parse raised
OutputCoercionError: ['userName: Field required'], so every durable
rehydrate of an aliased model failed. Aliasing is a no-op for models without
aliases, and populate_by_name models (which accept both spellings)
round-trip either way.
by_alias=True is NOT enough on its own, because Pydantic has two
aliases and that flag picks the wrong one when they differ. It emits the
SERIALIZATION alias, while the schema and parse both speak the
VALIDATION alias. Measured across the three shapes:
Field(alias="userName") schema userName dump userName OK
Field(serialization_alias="userName") schema user_name dump userName FAILS
Field(validation_alias="uname",
serialization_alias="userName") schema uname dump userName FAILS
So the keys are remapped to the validation-facing name — exactly what
json_schema() advertises and parse accepts — rather than trusting
a dump mode to pick it. The previous docstring called the mismatch "a
modelling choice" whose fix was populate_by_name=True; that is true
of a single dump mode and not of the adapter, which knows both names and
can simply use the right one.
Keys the map does not know (a model with extra="allow") pass through
untouched — dropping them would lose data the model deliberately kept.
Validate an arbitrary Python value against the model.
Fast-paths an already-typed instance (isinstance(value, self._model))
— the common case when a tool function literally constructs and
returns a MyModel(...). Otherwise hands off to model_validate
(Pydantic accepts both dicts and dataclass-like / attrs-like objects),
wrapping any ValidationError in :class:OutputCoercionError so the
ToolShapeError fire site has a single exception type to catch.