Test tools with fakes

A tool calls a CRM. In tests, give it a fake CRM instead, without patching imports.

flowchart LR
  P["production<br/>context(extras={crm: real client})"] --> T["customer_tier(email, ctx)"]
  Q["test<br/>context(extras={crm: FakeCrm})"] --> T
  T --> X["Executor<br/>retries, events"]

The tool

class Crm(Protocol):
    """What the tool needs from the CRM; the real client and the fake both fit."""

    async def find(self, email: str) -> dict[str, str] | None: ...


@registry.tool(retries=1)
async def customer_tier(email: str, ctx: ToolContext) -> str:
    """The customer's support tier."""
    crm: Crm = ctx.extras["crm"]  # injected by the caller, hidden from the model
    found = await crm.find(email)
    if found is None:
        raise ToolError(f"no customer {email}; ask for another address")
    return found["tier"]

ctx is hidden from the model's schema. ctx.extras reaches the tool unchanged (keys starting with aa. are reserved).

The fake

class FakeCrm:
    """Answers from a dict; fails the first `outages` calls, like a flaky network."""

    def __init__(self, rows: dict[str, dict[str, str]], outages: int = 0) -> None:
        self.rows, self.outages, self.calls = rows, outages, 0

    async def find(self, email: str) -> dict[str, str] | None:
        self.calls += 1
        if self.outages:
            self.outages -= 1
            raise ConnectionError("crm down")
        return self.rows.get(email)


async def no_wait(seconds: float) -> None:
    """Stands in for asyncio.sleep: retries do not wait."""

The tests

async def test_tier_survives_one_outage() -> None:
    crm, events = FakeCrm({"a@example.com": {"tier": "gold"}}, outages=1), EventLog()
    executor = Executor(registry, sink=events, sleep=no_wait)
    ctx = executor.context(extras={"crm": crm})
    result = await executor.run_call(
        ToolCall(name="customer_tier", arguments={"email": "a@example.com"}), ctx
    )
    assert (result.value, result.attempts, crm.calls) == ("gold", 2, 2)
    assert [e.data["error_kind"] for e in events.of_type("tool.retry")] == ["internal-error"]


async def test_unknown_customer_tells_the_model_what_to_do() -> None:
    executor = Executor(registry, sleep=no_wait)
    ctx = executor.context(extras={"crm": FakeCrm({})})
    result = await executor.run_call(
        ToolCall(name="customer_tier", arguments={"email": "b@example.com"}), ctx
    )
    assert result.error is not None and result.error.kind == "tool-error"
    assert result.error.message == "no customer b@example.com; ask for another address"

test_tier_survives_one_outage ok
test_unknown_customer_tells_the_model_what_to_do ok

Fake	Replaces
a client in ctx.extras	the network service
Executor(sleep=..., rng=..., clock=...)	backoff waits, jitter, time
EventLog()	your tracing backend
CallableModel(fn)	a model (Testing tools)

Pitfalls

Pitfall	Do instead
A module-global client	tests must patch it, and it breaks across event loops: inject it, or keep it in a LoopLocal
spec.invoke in tests of policy	it skips retries, approval and hooks: run through an Executor
Asserting on the text of crashes	assert on error.kind; crash text is never in the result

Files

File	Does
examples/cookbook_fakes.py	the tool, a fake CRM and two tests
tests/docs/test_cookbook.py	runs it offline and checks the output