A comprehensive testing and validation framework for monocle AI agent tracing. This package provides tools for validating agent behavior, tool invocations, inference responses, and overall AI system performance.
- Test Generator: Automatically generate test code from trace files
- Agentic Response: Verify that agent requests get the appropriate response.
- Agent Invocation: Verify that specific agents are invoked and delegate tasks correctly.
- Tool Validation: Ensure tools are called with expected inputs and produce expected outputs.
- Inference Testing: Test model inference responses against expected content.
- Cost/Performance/Quality: Verify token usage, duration limits, error states, and warnings.
- Evaluation: Integrate with Okahu or custom evaluation tools to validate LLM responses.
- Fluent API: Chain assertions using a readable, expressive builder pattern.
- Mock Tools: Simulate tool behavior without invoking external dependencies.
- Offline Testing: Assert against pre-recorded trace JSON files without running live agents.
The test tool runs your agent or workflow code with Monocle instrumentation enabled. It examines the traces generated by the genAI components used in your code (e.g. Google ADK, LangGraph, CrewAI, etc.) and verifies the test conditions you want to validate.
pip install monocle_test_toolsIf you are running tests locally without installing the package, register the pytest plugin in your conftest.py:
# conftest.py
pytest_plugins = ["monocle_test_tools.pytest_plugin"]When installed as a package, the plugin is registered automatically.
Decorate a test function with @MonocleValidator().monocle_testcase(test_cases) to run parameterized test cases automatically. Works with sync and async functions.
import pytest
from monocle_test_tools import TestCase, MonocleValidator
from adk_travel_agent import root_agent
agent_test_cases: list[dict] = [
# Test 1: Validate the end-to-end response with similarity matching
{
"test_input": ["Book a flight from San Francisco to Mumbai for 26th Nov 2025."],
"test_output": "A flight from San Francisco to Mumbai has been booked.",
"comparer": "similarity",
},
# Test 2: Validate specific span types (tool and agent invocations)
{
"test_input": ["Book a flight from San Francisco to Mumbai for 26th Nov 2025."],
"test_spans": [
{
"span_type": "agentic.turn",
"output": "A flight from San Francisco to Mumbai has been booked.",
"comparer": "similarity"
},
{
"span_type": "agentic.tool.invocation",
"entities": [
{"type": "tool", "name": "adk_book_flight"},
{"type": "agent", "name": "adk_flight_booking_agent"}
],
# Verify expected attributes on the matched span
"attributes": {"entity.1.type": "tool.adk"}
}
]
},
# Test 3: BERTScore evaluation on a span
{
"test_input": ["Book a flight from San Francisco to Mumbai for 26th Nov 2025."],
"test_spans": [
{
"span_type": "agentic.turn",
"eval": {
"eval": "bert_score",
"args": ["input", "output"],
"expected_result": {"Precision": 0.5, "Recall": 0.5, "F1": 0.5},
"comparer": "metric"
}
}
]
},
]
@MonocleValidator().monocle_testcase(agent_test_cases)
async def test_run_agents(my_test_case: TestCase):
await MonocleValidator().test_agent_async(root_agent, "google_adk", my_test_case)
if __name__ == "__main__":
pytest.main([__file__])Use the monocle_trace_asserter pytest fixture to write expressive, chainable assertions. Each assertion method filters the span context for subsequent calls.
import pytest
from my_agent import my_agent
@pytest.mark.asyncio
async def test_tool_called_with_input(monocle_trace_asserter):
await monocle_trace_asserter.run_agent_async(my_agent, "google_adk", "book a flight to Mumbai")
monocle_trace_asserter \
.called_tool("book_flight", agent_name="flight_booking_agent") \
.contains_input("Mumbai") \
.contains_output("booked")
@pytest.mark.asyncio
async def test_agent_under_limits(monocle_trace_asserter):
await monocle_trace_asserter.run_agent_async(my_agent, "langgraph", "summarize this document")
monocle_trace_asserter \
.called_agent("summarizer_agent") \
.under_token_limit(500) \
.under_duration(5.0, units="seconds", span_type="agent_invocation")Calling called_agent() followed by called_tool() narrows the span context to tools invoked within that specific agent:
# Verify tools called only within the flight booking agent, not other agents
monocle_trace_asserter \
.called_agent("adk_flight_booking_agent") \
.called_tool("adk_book_flight") \
.under_duration(0.1, units="minutes", span_type="tool_invocation")Every assertion method accepts an optional message= parameter to provide a clear error description when the assertion fails:
monocle_trace_asserter.called_tool(
"book_flight",
message="BOOKING FAILED: The flight booking tool was never invoked"
)
monocle_trace_asserter.under_token_limit(
500,
message="COST ALERT: Token usage exceeded the allowed budget"
)
monocle_trace_asserter.contains_input(
"Mumbai",
message="INPUT VALIDATION ERROR: Destination city was not passed to the tool"
)Assert specific or total invocation counts for agents and tools:
# Exact count: retry agent called exactly 3 times
monocle_trace_asserter.called_agent("retry_agent", count=3)
# Range: worker agent called 2-5 times
monocle_trace_asserter.called_agent("worker_agent", min_count=2, max_count=5)
# Min only: search tool called at least once
monocle_trace_asserter.called_tool("search_tool", min_count=1)
# Max only: expensive API called at most twice
monocle_trace_asserter.called_tool("expensive_api", max_count=2)
# Total agent invocations across all agents
monocle_trace_asserter.called_agents(count=10) # Exactly 10 agent calls total
monocle_trace_asserter.called_agents(min_count=5, max_count=15) # Between 5-15 calls
# Total tool invocations across all tools
monocle_trace_asserter.called_tools(max_count=20) # At most 20 tool calls total@pytest.mark.asyncio
async def test_adk_parallel_agents(monocle_trace_asserter):
await monocle_trace_asserter.run_agent_async(root_agent_parallel, "google_adk",
"Book a flight to Mumbai and a hotel for 4 nights.")
# Verify all agents in the parallel workflow were called
monocle_trace_asserter.called_agent("adk_parallel_booking_coordinator")
monocle_trace_asserter.called_agent("adk_flight_booking_agent")
monocle_trace_asserter.called_agent("adk_hotel_booking_agent")
# Verify both tools were invoked
monocle_trace_asserter.called_tool("adk_book_flight", "adk_flight_booking_agent")
monocle_trace_asserter.called_tool("adk_book_hotel", "adk_hotel_booking_agent")
# Verify output mentions both bookings
monocle_trace_asserter.contains_output("flight")
monocle_trace_asserter.contains_output("hotel")@MonocleValidator().monocle_testcase(agent_test_cases)
async def test_crewai_travel_agent(my_test_case: TestCase):
result = await execute_crewai_travel_request(my_test_case.test_input[0])
return resultFluent style:
@pytest.mark.asyncio
async def test_crewai_tool_invocation(monocle_trace_asserter):
crew = create_crewai_travel_crew("Book a hotel at Marriott in New York for 2 nights")
await monocle_trace_asserter.run_agent_async(crew, "crewai", travel_request)
monocle_trace_asserter \
.called_tool("crew_book_hotel", "CrewAI Hotel Booking Agent") \
.contains_output("success")LlamaIndex supports multiple runner methods — coordinator agents, ReActAgent (chat and query), and QueryEngine (sync and async):
@MonocleValidator().monocle_testcase(agent_test_cases)
async def test_llamaindex_agent(my_test_case: TestCase):
agent_workflow = await setup_agents()
await MonocleValidator().test_agent_async(agent_workflow, "llamaindex", my_test_case)Test cases can validate the full multi-agent coordinator hierarchy:
{
"test_spans": [
{"span_type": "agentic.invocation", "entities": [{"type": "agent", "name": "lmx_coordinator_05"}]},
{"span_type": "agentic.invocation", "entities": [{"type": "agent", "name": "lmx_flight_booking_agent_05"}]},
{"span_type": "agentic.tool.invocation", "entities": [
{"type": "tool", "name": "lmx_book_flight_tool_05"},
{"type": "agent", "name": "lmx_flight_booking_agent_05"}
]}
]
}Use session_id to associate multiple turns with the same session:
@MonocleValidator().monocle_testcase(agent_test_cases)
async def test_strands_agent(test_case: TestCase):
await MonocleValidator().test_agent_async(
root_agent, "strands", test_case, session_id="my_session"
)Run assertions against saved trace JSON files without invoking live agents. This is useful for unit tests and CI environments without API access.
from monocle_test_tools import TraceAssertion
def test_tool_invocation_from_trace(monocle_trace_asserter):
# Load spans from a saved trace file
monocle_trace_asserter.with_trace_source(source="file", trace_path="traces/trace1.json")
monocle_trace_asserter \
.called_tool("adk_book_hotel", "adk_hotel_booking_agent") \
.has_input("{'city': 'Mumbai', 'hotel_name': 'Marriot Intercontinental'}") \
.has_output("{'status': 'success', 'message': 'Successfully booked a stay...'}") \
.contains_input("Mumbai") \
.contains_output("Successfully booked") \
.does_not_contain_input("Delhi") \
.does_not_contain_output("failed")
monocle_trace_asserter.with_evaluation("okahu").check_eval("hallucination", "major_hallucination")
def test_agent_invocation_from_okahu(monocle_trace_asserter):
monocle_trace_asserter.with_trace_source(
source = "okahu", id="642dbd9d0dfcfdbdc8849f67f34c8a19", workflow_name=WORKFLOW_CC
).with_evaluation("okahu").check_eval("hallucination", "major_hallucination")
monocle_trace_asserter \
.called_agent("adk_hotel_booking_agent") \
.has_input("Book a flight from San Francisco...") \
.contains_output("I have booked a stay at Marriot Intercontinental") \
.does_not_have_output("cancel the booking")You can also use MonocleValidator directly for programmatic validation:
from monocle_test_tools import MonocleValidator, TestCase, TestSpan, Entity, DefaultComparer
from monocle_test_tools.span_loader import JSONSpanLoader
@pytest.fixture(scope="module")
def validator():
v = MonocleValidator()
spans = JSONSpanLoader.from_json("traces/trace1.json")
v.memory_exporter.export(spans)
yield v
v.memory_exporter.clear()
def test_tool_span(validator):
test_case = TestCase(
test_spans=[
TestSpan(
span_type="agentic.tool.invocation",
entities=[
Entity(type="tool", name="adk_book_hotel"),
Entity(type="agent", name="adk_hotel_booking_agent"),
],
input="{'city': 'Mumbai', 'hotel_name': 'Marriot Intercontinental'}",
output="{'status': 'success', ...}",
comparer=DefaultComparer(),
)
]
)
validator.validate(test_case)Automatically generate test code by analyzing trace files. The generator scans spans and creates Python test assertions for agents, tools, and outputs.
# Generate test code from a trace file
python -m monocle_test_tools.generate_test trace.json
# With custom test name
python -m monocle_test_tools.generate_test trace.json --test-name test_my_agent
# Save to file
python -m monocle_test_tools.generate_test trace.json > test_generated.pyimport pytest
from monocle_test_tools import TraceAssertion
from monocle_test_tools.span_loader import JSONSpanLoader
def test_generated(monocle_trace_asserter: TraceAssertion):
"""Auto-generated test from trace analysis."""
# Option 1: Load from JSON file
spans = JSONSpanLoader.from_json("path/to/trace.json")
# monocle_trace_asserter.validator.add_remote_spans(spans)
# Option 2: Load from Okahu
# from monocle_test_tools.span_loader import OkahuSpanLoader
# spans = OkahuSpanLoader.get_spans(workflow_name="your_workflow", trace_id="trace_id")
# monocle_trace_asserter.validator.add_remote_spans(spans)
# Option 3: Run agent directly
# from your_module import your_agent
# await monocle_trace_asserter.run_agent_async(your_agent, "framework_name", "user input")
asserter = monocle_trace_asserter
# Agent invocations with output checks
asserter.called_agent("travel_agent").contains_output("Successfully booked")
asserter.called_agent("hotel_agent").contains_output("Marriott reservation confirmed")
# Tool invocations
asserter.called_tool("book_flight", "travel_agent")
asserter.called_tool("book_hotel", "hotel_agent")from monocle_test_tools.test_generator import TestGenerator
# From local file
generator = TestGenerator.from_json_file("trace.json")
test_code = generator.generate_test_code(test_name="test_my_agent")
print(test_code)
# Write to file
generator.write_to_file("test_my_agent.py")
# From Okahu
generator = TestGenerator.from_okahu(trace_id="abc123", workflow_name="my_app")
print(generator.generate_test_code())The generator extracts:
- Agent invocations with output checks (first 80 chars as key phrase)
- Tool invocations with parent agent references
- Sorted by invocation order for readability
MockTool simulates tool calls without invoking real external services. Useful for testing agent logic in isolation.
Use {{placeholder}} syntax in the response to substitute values from tool input at runtime:
from monocle_test_tools import TestCase, MockTool, ToolType
test_case = TestCase(
test_input=["Book a flight from San Francisco to Mumbai for 26th Nov 2025."],
mock_tools=[
MockTool(
name="adk_book_flight",
type=ToolType.ADK,
response={
"status": "success",
"message": "Flight booked from {{from_airport}} to {{to_airport}}."
}
)
],
test_spans=[
{
"span_type": "agentic.tool.invocation",
"entities": [{"type": "tool", "name": "adk_book_flight"}],
"output": "{'status': 'success', 'message': 'Flight booked from San Francisco to Mumbai.'}"
}
]
)MockTool(
name="failing_tool",
type=ToolType.OPENAI,
raise_error=True,
error_message="Service unavailable"
)Supported ToolType values: tool.openai, tool.adk, tool.llama_index, tool.langgraph, tool.strands.
{
"test_name": "Optional name (default: 'monocle_test').",
"test_input": "Tuple of inputs passed to the workflow or agent function.",
"test_output": "Expected output compared against the actual result using 'comparer'.",
"comparer": "Comparison method. See Comparers section.",
"test_description": "Human-readable description of what the test verifies.",
"test_spans": "Array of TestSpan objects for specific interactions.",
"expect_errors": "If true, errors during the run are expected and do not fail the test.",
"expect_warnings": "If true, warnings during the run are expected.",
"mock_tools": "Array of MockTool objects to simulate tool behavior."
}Validation rules enforced by span type:
agentic.tool.invocation— first entity must be typetool; optional second must be typeagentagentic.delegation— requires at least twoagententities (delegator first, delegatee second)agentic.invocation— first entity must be typeagent
{
"span_type": "Type of interaction. See Span Types.",
"entities": "List of entities involved. Each has 'type' (tool|agent|inference) and 'name'.",
"input": "Expected input for this interaction.",
"output": "Expected output from this interaction.",
"test_type": "'positive' (default) or 'negative'. Negative tests assert the interaction does NOT occur.",
"eval": "Evaluation configuration. See Evaluation section.",
"expect_errors": "Whether errors are expected during this span.",
"expect_warnings": "Whether warnings are expected during this span.",
"comparer": "Comparison method. See Comparers."
}| Span Type | Constant | Description |
|---|---|---|
agentic.tool.invocation |
SpanType.TOOL_INVOCATION |
A tool was invoked by an agent |
agentic.invocation |
SpanType.AGENTIC_INVOCATION |
An agent was invoked |
agentic.turn |
SpanType.AGENTIC_REQUEST |
An end-to-end agentic request/response turn |
agentic.delegation |
SpanType.AGENTIC_DELEGATION |
An agent delegated to another agent |
inference |
SpanType.INFERENCE |
A model inference call |
| Entity Type | Constant | Description |
|---|---|---|
tool |
EntityType.TOOL |
A tool or function called by an agent |
agent |
EntityType.AGENT |
An agent or sub-agent |
inference |
EntityType.INFERENCE |
A model inference entity |
| Name | Class | Description |
|---|---|---|
default |
DefaultComparer |
Exact string equality (default) |
similarity |
SentenceComparer |
Semantic similarity via sentence-transformers (threshold: 0.8) |
bert_score |
BertScoreComparer |
BERTScore-based text similarity |
metric |
MetricComparer |
Numeric metric comparison (used with BERTScore eval results) |
token_match |
TokenMatchComparer |
Substring/token containment check |
Pass comparers as strings ("similarity") or class instances (SentenceComparer()). Extend BaseComparer to create a custom comparer.
Run directly on span input/output without external services:
{
"span_type": "agentic.turn",
"eval": {
"eval": "bert_score",
"args": ["input", "output"],
"expected_result": {"Precision": 0.5, "Recall": 0.5, "F1": 0.5},
"comparer": "metric"
}
}Submits traces to the Okahu evaluation service. Requires OKAHU_API_KEY.
Use with_evaluation("okahu") in the fluent API, then call check_eval() with an eval template name and expected result:
# Basic usage — single eval on the default fact (traces)
monocle_trace_asserter.with_evaluation("okahu").check_eval("sentiment", expected="positive")
# Using not_expected to assert a result is absent
monocle_trace_asserter.with_evaluation("okahu") \
.check_eval("toxicity", not_expected=["highly_toxic", "moderately_toxic", "mildly_toxic"])
# Chaining multiple evals
monocle_trace_asserter.with_evaluation("okahu") \
.check_eval("hallucination", expected="no_hallucination") \
.check_eval("bias", expected="unbiased") \
.check_eval("pii_leakage", not_expected="pii_leakage")
# Evaluating on a specific fact
monocle_trace_asserter.with_evaluation("okahu") \
.check_eval(fact_name="agentic_sessions", eval_name="role_adherence",
expected=["excellent_adherence", "good_adherence"],
not_expected=["poor_adherence", "no_adherence"])
# Evaluating filtered spans (after called_agent or called_tool)
monocle_trace_asserter.called_agent("adk_flight_booking_agent")
monocle_trace_asserter.with_evaluation("okahu") \
.check_eval("conversation_completeness", expected="complete")fact_name |
Description |
|---|---|
traces |
Whole traces (default) |
inferences |
Individual inference calls |
agentic_turns |
Agent request/response turns |
agentic_sessions |
Agent sessions |
agent_invocation |
Individual agent invocations |
tool_execution |
Individual tool executions |
commits |
Git commits |
conversations |
Conversations |
test_runs |
Test runs |
tests |
Tests |
The following eval template names have been validated against real traces in tests:
| Template | Supported facts | Example expected values |
|---|---|---|
sentiment |
traces, inferences, agentic_turns |
"positive", "negative", "neutral" |
bias |
traces, agentic_turns, agentic_sessions |
"unbiased" |
hallucination |
traces, agentic_sessions, agentic_turns |
"no_hallucination" |
toxicity |
traces, agentic_turns, agentic_sessions |
"non_toxic" |
pii_leakage |
traces, agentic_turns, agentic_sessions |
"pii_leakage" (use not_expected) |
frustration |
traces, conversations |
"ok" |
offtopic |
agentic_turns |
"on_topic" |
argument_correctness |
agentic_turns |
"correct" |
contextual_relevancy |
agentic_turns, agentic_sessions |
"highly_relevant" |
conversation_completeness |
agent_invocation |
"complete" |
summarization |
traces, agentic_sessions |
"excellent", "good" |
correctness |
agentic_sessions |
"correct" |
role_adherence |
agentic_sessions |
"excellent_adherence", "good_adherence" |
knowledge_retention |
agentic_sessions |
"excellent_retention", "good_retention" |
answer_relevancy |
agentic_sessions |
"yes" |
misuse |
agentic_sessions |
"clear_misuse" (use not_expected) |
Passing a template name or fact combination that does not exist in Okahu raises an
AssertionErrorwith the list of valid templates. Use@pytest.mark.xfailfor tests that are expected to fail.
| Agent type string | Framework |
|---|---|
google_adk |
Google Agent Development Kit (ADK) |
openai |
OpenAI Agents |
langgraph |
LangGraph |
crewai |
CrewAI |
llamaindex |
LlamaIndex |
strands |
Strands Agents |
msagent |
Microsoft Semantic Kernel / AutoGen |
The monocle_trace_asserter fixture provides a TraceAssertion instance. All assertion methods return self for chaining and accept an optional message= keyword argument for custom failure messages.
| Method | Description |
|---|---|
run_agent(agent, agent_type, *args) |
Run a sync agent |
await run_agent_async(agent, agent_type, *args, session_id=None) |
Run an async agent |
load_spans(spans) |
Load pre-recorded ReadableSpan objects for offline assertions |
| Method | Description |
|---|---|
called_tool(tool_name, agent_name=None, count=None, min_count=None, max_count=None) |
Assert a tool was called; narrows context to those spans. Optional: count for exact count, min_count/max_count for range |
does_not_call_tool(tool_name, agent_name=None) |
Assert a tool was NOT called |
called_agent(agent_name, count=None, min_count=None, max_count=None) |
Assert an agent was called; narrows context to those spans. Optional: count for exact count, min_count/max_count for range |
does_not_call_agent(agent_name) |
Assert an agent was NOT called |
called_agents(count=None, min_count=None, max_count=None) |
Assert total number of agent invocations across all agents. Optional: count for exact count, min_count/max_count for range |
called_tools(count=None, min_count=None, max_count=None) |
Assert total number of tool invocations across all tools. Optional: count for exact count, min_count/max_count for range |
| Method | Description |
|---|---|
has_input(expected) |
Exact input match |
has_any_input(*expected) |
Exact match against any of the given inputs |
does_not_have_input(unexpected) |
Assert input does not exactly match |
does_not_have_any_input(*unexpected) |
Assert none of the inputs exactly match |
contains_input(substring) |
Input contains the given substring |
contains_any_input(*substrings) |
Input contains any of the given substrings |
does_not_contain_input(substring) |
Input does not contain the substring |
does_not_contain_any_input(*substrings) |
Input does not contain any of the substrings |
| Method | Description |
|---|---|
has_output(expected) |
Exact output match |
has_any_output(*expected) |
Exact match against any of the given outputs |
does_not_have_output(unexpected) |
Assert output does not exactly match |
does_not_have_any_output(*unexpected) |
Assert none of the outputs exactly match |
contains_output(substring) |
Output contains the given substring |
contains_any_output(*substrings) |
Output contains any of the given substrings |
does_not_contain_output(substring) |
Output does not contain the substring |
does_not_contain_any_output(*substrings) |
Output does not contain any of the substrings |
| Method | Description |
|---|---|
under_token_limit(limit) |
Assert total tokens across current spans are under the limit |
under_duration(limit, units="seconds", span_type="workflow") |
Assert each span of the given type is under the duration limit. units: ms, seconds, minutes. span_type: workflow, agent_invocation, tool_invocation, agent_turn, inference |
| Method | Description |
|---|---|
with_evaluation(eval, eval_options=None) |
Configure the evaluator ("okahu", "bert_score", or a BaseEval instance) |
with_comparer(comparer) |
Override the comparer used by subsequent has_input, has_any_input, does_not_have_input, has_output, has_any_output, does_not_have_output assertions. Does not affect contains_* methods (those always use token/substring matching). Accepts a string key or class instance: "default" (exact match), "similarity" (semantic via sentence-transformers, threshold 0.8), "bert_score" (BERTScore similarity), "metric" (numeric metric comparison), "token_match" (substring containment), or any BaseComparer subclass instance. |
check_eval(eval_name, expected=None, not_expected=None, fact_name="traces") |
Run an evaluation and assert the result. expected and not_expected each accept a string or list of strings |
For use outside the fluent API or fixture:
validator = MonocleValidator()
# Run and validate a sync or async workflow function
validator.test_workflow(my_func, test_case)
await validator.test_workflow_async(my_func, test_case)
# Run and validate a named agent (sync or async)
validator.test_agent(agent, "langgraph", test_case)
await validator.test_agent_async(agent, "google_adk", test_case, session_id="abc")
# Token and duration checks (call after a run or after loading spans)
validator.check_completion_token_limits(max_output_tokens=200)
validator.check_total_token_limits(max_total_tokens=1000)
validator.check_duration_limits(max_duration=30.0, units="seconds", span_type="workflow")| Variable | Description | Default |
|---|---|---|
MONOCLE_EXPORTER |
Comma-separated exporter names (file, okahu, etc.) |
file |
MONOCLE_EXPORT_FAILED_TESTS_ONLY |
Set to true to export traces only for failed tests |
false |
MONOCLE_TEST_WORKFLOW_NAME |
Workflow name tag applied to all test traces | Git repo name |
MONOCLE_TRACE_OUTPUT_PATH |
Directory for file-based trace output | .monocle/test_traces |
OKAHU_API_KEY |
API key for Okahu evaluation and trace export | — |
OKAHU_EVALUATION_ENDPOINT |
Override the Okahu evaluation endpoint | https://eval.okahu.co/api |
LOCAL_RUN_ID |
Run identifier applied to all spans in a session | ISO datetime at session start |