`agentkit.testing`¶

Test doubles for every Protocol — FakeLLM, FakeFetch, FakeSearch, FakeMemory, FakeTool, FakeCompactor, FakeGrounder, FakeClock — plus the make_test_ctx() builder that wires them into a RunContext.

Canonical test utilities for agentkit consumers.

What lives here: - make_test_ctx(...) — factory that builds a REAL RunContext for tests, wired with the FakeLLM-backed Invoker + noop Trace/Observer defaults. Suitable for any test that needs a real ctx.

What lives under fakes/: - FakeCtx — minimum Ctx that RECORDS spans (different from agentkit.runtime.NullCtx, which records nothing). Use when a test needs to assert on what spans were opened. - FakeLLM, FakeClock, FakeFetch, FakeSearch — port doubles. - FakeGrounder, FakeCompactor — capability doubles for RequestBuilder tests.

Naming convention: Fake* is the standard for test doubles. Null* / Noop* are production-grade null-object patterns; they live in their respective runtime/kernel modules, NOT here.

FakeClock ¶

FakeClock(start: float = 0.0)

Deterministic offline ClockPort. start is the initial unix timestamp; now() returns the current virtual time; sleep(s) advances it by s (no real wall-clock wait).

advance ¶

advance(seconds: float) -> None

Test-only escape hatch: move the clock forward without recording a sleep call (useful for simulating external time passing between agent steps).

FakeCompactor `dataclass` ¶

FakeCompactor(
    sentinel: str = "[COMPACTED]", called: bool = False
)

RequestBuilder test helper. Replaces every transcript with a single sentinel message so a test can detect that compaction ran.

FakeCtx ¶

FakeCtx(*, scope: Scope | None = None)

Minimal Ctx that RECORDS spans + observations.

Use when a test needs to assert on what spans were opened (e.g., RequestBuilder stamps agentkit.prompt.version — a test for that contract needs a tracer that captures attributes). Distinct from agentkit.runtime.NullCtx: NullCtx absorbs operations and records nothing; FakeCtx records spans for assertion. Both are valid — pick based on whether you need to assert on what was recorded.

tracer.spans is also exposed as ctx.spans for tests that fold the tracer onto the ctx itself. Either access form works.

spans `property` ¶

spans: list[tuple[str, str, dict[str, Any]]]

Convenience alias mirroring (name, kind, attrs) tuples on the ctx itself, so tests can assert on ctx.spans directly instead of reaching through ctx.trace.spans.

FakeFetch ¶

FakeFetch(fixtures: dict[str, FetchResponse])

Deterministic offline FetchPort. fixtures is {url: FetchResponse}; unknown URLs raise KeyError(url) so tests fail loud instead of silently falling back to a stub response.

FakeGrounder `dataclass` ¶

FakeGrounder(
    block: str = "", calls: list[tuple[Any, str]] = list()
)

RequestBuilder test helper. Records (ctx, task) calls; returns a canned block (empty string allowed to test the skip path).

FakeLLM ¶

FakeLLM(
    responses: dict[str, str]
    | Callable[..., str]
    | str = "{}",
    *,
    usage: Usage | None = None,
    fail_times: int = 0,
    fail_exc: BaseException | None = None,
    delay: float = 0.0,
    turns: list[Turn] | None = None,
)

Deterministic offline LLM. dict {substring_of_user: content} | callable | str for single answers; FakeLLM.script([Turn(...), …]) replays a multi-step tool loop across chat calls.

stream `async` ¶

stream(
    *,
    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,
) -> AsyncIterator[Delta]

The streaming primitive: emits content in small chunks then a terminal delta carrying usage/finish/tool_calls. _enter() raises BEFORE the first delta, so it's a pre-stream failure the retry middleware can re-invoke (honoring fail_times).

FakeMemory `dataclass` ¶

FakeMemory(
    name: str = "fake_memory",
    responses: dict[str, list[MemoryItem]] = dict(),
    default: list[MemoryItem] = list(),
    items: list[MemoryItem] | None = None,
)

A MemorySource-Protocol-conforming fake.

Two modes:

responses={"q": [MemoryItem(...), ...], ...} — per-query lookup. Misses fall through to default (defaults to []).
items=[MemoryItem(...), ...] — static fixture mode. query ignores the query string entirely and returns the first k items regardless. Useful when a test doesn't care about retrieval semantics, only that the agent received some items.

Every query is recorded as (query, k, where) on self.queries; every write batch is appended to self.writes. Lets tests assert mem.queries[0] == ("foo", 3, None) or mem.writes == [[MemoryItem("a", "src")]].

FakeSearch ¶

FakeSearch(
    hits: list[SearchHit] | dict[str, list[SearchHit]],
)

Deterministic offline SearchPort. Pass a list (returned for every query) or a dict {substring_of_query: [SearchHit, ...]} (first substring match wins; unknown queries → []).

FakeTool `dataclass` ¶

FakeTool(
    name: str = "fake_tool",
    description: str = "fake tool for unit tests",
    responder: Any = None,
    side_effecting: bool = False,
    requires_approval: bool = False,
    output_schema: dict[str, Any] | None = None,
    schema: ToolSchema = _default_schema(),
)

A Tool-Protocol-conforming fake.

Construct with a static responder value (returned verbatim from every run) OR with a callable responder(args) -> Any that may be sync or async. Any exception the callable raises propagates, letting a test exercise the error path.

Every run call is appended to self.calls as the (copied) args dict, so tests can do assert tool.calls == [{"x": 1}, {"x": 2}].

The default schema is the permissive {"type": "object"} — any JSON object accepted. Override per-test if you want the Protocol's advertised schema to reflect a real shape.

RecordingSpan ¶

RecordingSpan(name: str, kind: str, attrs: dict[str, Any])

A span that records its name / kind / attributes / events. Drop-in for the production span surface — exposes .set(key, value) and .add_event(name, **fields). Attributes start from the kwargs passed to span() and are mutated by set().

RecordingTracer ¶

RecordingTracer(
    *,
    exporter: Callable[[RecordingSpan], None] | None = None,
)

In-process TracePort that records every span. Optional exporter callback receives each span object on close (mirrors the OTel SpanExporter shape at the granularity tests care about).

Satisfies the TracePort Protocol structurally: - span(name, kind, **attrs) — context manager yielding a RecordingSpan - current_span_id() — returns the open span's TraceContext (synthetic ids) or None - add_event_to_current_span(name, **fields) — drops an event on the open span

Turn `dataclass` ¶

Turn(
    content: str = "",
    tool_calls: tuple[ToolCall, ...] = (),
    usage: Usage | None = None,
)

One scripted reply for FakeLLM.script — a final answer, or a turn that requests tools.

make_test_ctx ¶

make_test_ctx(
    *,
    llm: Any = None,
    invoker: Any = None,
    scope: Scope | None = None,
    budget: Budget | None = None,
    store: Any = None,
    vector: Any = None,
    checkpointer: Any = None,
    observer: Any = None,
    trace: Any = None,
    cancel: CancellationToken | None = None,
    chat_middleware: Sequence[Any] = (),
    tool_middleware: Sequence[Any] = (),
    meters: Sequence[Any] = (),
    correlation_id: str = "test-run",
    autonomy: AutonomyLiteral = "auto",
) -> RunContext

Build a RunContext for tests with the knobs the call site cares about and sensible no-op defaults for everything else.

Two ways to wire the LLM seam: pass invoker= (a fully-built Invoker) OR pass llm= and we'll wrap it with Invoker(llm=..., chat_middleware=chat_middleware). Tests that don't need an Invoker at all leave both unset — they get a RunContext with services.invoker = None, which is fine for capabilities tests (RequestBuilder, Compactor) that never invoke an LLM.

Scope() is the zero-tenant default. Override with scope=Scope(org_id, domain_id) for tenant-isolation tests.

observer / trace / checkpointer / store / vector default to None here, which we omit when constructing Services so its built-in NoopObserver / NoopTrace factories still kick in — passing None explicitly would clobber them and break code that calls ctx.trace.span(...) or ctx.observer.emit(...).

agentkit.testing¶