Testing¶
Auto-generated from source docstrings (mkdocstrings). The complete public contract is agentmaker.__all__ plus each subpackage __all__.
agentmaker.testing: official test utilities for testing agents built on this framework, with no cost and no network.
When third parties write agents, the deterministic doubles here swap the LLM / embedding / checkpoint / hook
for local implementations so hermetic unit tests can run (mirroring Pydantic AI's TestModel). Not part of the
top-level agentmaker.__all__; import on demand via from agentmaker.testing import ScriptedLLM.
from agentmaker import Agent
from agentmaker.testing import ScriptedLLM
agent = Agent("test", ScriptedLLM(["hello"]))
assert agent.run("hi").final_output == "hello"
# Tool call: the script first emits a "call calculator" response, then the final answer.
llm = ScriptedLLM([ScriptedLLM.tool_call("calculator", {"expression": "1+1"}), "equals 2"])
ScriptedLLM
¶
Test LLM that emits preset responses in call order (duck-typed; does NOT inherit LLMClient, avoiding key validation / network).
Each script element is a str (a plain-text reply) or a ready-made LLMResponse (with tool_calls / usage /
etc.). chat consumes them in order; calling again after the script is exhausted raises AssertionError
(noting how many entries are missing). stream slices the next response's content. Use
ScriptedLLM.tool_call(...) to conveniently build a "request to call a tool" response. The duck-typed
contract mirrors LLMClient: provider / model / supports_function_calling / context_window / chat / stream.
__init__(script=None, *, model='test', provider='test', supports_function_calling=True, context_window=None)
¶
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
script
|
List of response entries, each a str (text reply) or an LLMResponse (for precise control over tool_calls / usage / etc.). |
None
|
|
supports_function_calling
|
bool
|
Model capability flag (a tool-enabled Agent validates against it at construction time); pass False to test the no-function-calling path. |
True
|
context_window
|
Optional[int]
|
Context window (None means unknown, so no window-budget reduction is triggered); pass a concrete value to test reduction / window budget. |
None
|
tool_call(name, arguments=None, *, call_id='call_1', content='')
staticmethod
¶
Build an LLMResponse representing "the model requests calling tool name(arguments)" (so you need not hand-craft the OpenAI tool_calls structure).
chat(messages, *, tools=None, **kwargs)
async
¶
Return the next response from the script (ignoring messages / tools, since the script determines test behavior).
stream(messages, *, tools=None, on_stats=None, **kwargs)
async
¶
Yield the next response's content in slices (roughly 8 characters each); at the end, invoke on_stats and write last_stream_stats.
Mirrors the real adapter contract: empty content yields NO empty chunk (an empty range simply skips the loop); on completion, regardless of whether there was content, assemble a StreamStats from the response's model / finish_reason / usage and hand it back, since harness.astream_llm relies on on_stats to collect usage for accounting. With tools passed, the full LLMResponse is yielded as the terminal item after the text slices (same contract as the real adapters), making the streaming tool loop hermetically testable.
FakeEmbedder
¶
MemoryCheckpointStore
¶
Bases: CheckpointStore
In-process in-memory checkpoint store (one ExecutionState JSON per scope): test HITL suspend / resume / crash recovery, without persisting to disk.
Implementing synchronous save / load / clear is enough: the base class's asave / aload / aclear wrap them with to_thread by default, so the async path comes for free.