Monocle Test Tools

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.

Features

• 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.

How does it work

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.

Installation

pip install monocle_test_tools

Project Setup

conftest.py

If 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.

Quick Start

Declarative API (@monocle_testcase)

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__])

Multi-turn testing (@monocle_multi_turn_testcase)

Single-turn tests run one input, collect one trace, and assert against it. Multi-turn testing runs an ordered list of turns against a single live agent session in one shot: the agent's own memory carries over between turns, spans from every turn are accumulated and grouped under one session id, and assertions can run per-turn or across the whole session.

This is what enables testing an "incomplete input -> agent asks for a clarification -> follow-up input" flow, plus session-wide evals such as hallucination and user sentiment across the whole conversation rather than a single turn.

Wrap the turns in a MultiTurnTestCase:

import pytest
from monocle_test_tools import MultiTurnTestCase, MonocleValidator
from adk_travel_agent import root_agent

multi_turn_cases: list[dict] = [
    {
        # A single session id groups every turn's spans (auto-generated if omitted).
        "session_id": "booking_session_1",
        "turns": [
            # Turn 1: incomplete input, the agent should ask for the destination.
            {
                "test_input": ["Book me a flight for 26th Nov 2025."],
                "test_spans": [
                    {
                        "span_type": "agentic.turn",
                        "output": "Which city would you like to fly to?",
                        "comparer": "similarity"
                    }
                ]
            },
            # Turn 2: the follow-up. The agent remembers the date from turn 1
            # because both turns share one live session.
            {
                "test_input": ["Fly to Mumbai."],
                "test_output": "A flight to Mumbai on November 26, 2025 has been booked.",
                "comparer": "similarity"
            }
        ],
        # Assertions across the accumulated spans of ALL turns.
        "session_spans": [
            {
                "span_type": "agentic.tool.invocation",
                "entities": [
                    {"type": "tool", "name": "adk_book_flight"},
                    {"type": "agent", "name": "adk_flight_booking_agent"}
                ]
            }
        ]
    }
]

@MonocleValidator().monocle_multi_turn_testcase(multi_turn_cases)
async def test_multi_turn(multi_turn_case: MultiTurnTestCase):
    await MonocleValidator().test_multi_turn_agent_async(root_agent, "google_adk", multi_turn_case)

if __name__ == "__main__":
    pytest.main([__file__])

Each turn is an ordinary TestCase, so per-turn test_output, test_spans, and evals work exactly as in single-turn tests, validated against that turn's spans. The session_spans block (and optional session_output) is validated against the spans accumulated across every turn.

Scoping an assertion to a specific turn. Every turn runs inside its own turn scope, so each span it produces carries a scope.turn_id attribute (alongside the shared scope.session_id). The turn id is the turn's own turn_id when set, otherwise its 1-based index ("1", "2", ...). Use it in a session_spans assertion's attributes to require that a span happened in a particular turn:

"turns": [
    {"test_input": ["Book me a flight for 26th Nov 2025."], "turn_id": "collect_info"},
    {"test_input": ["Fly to Mumbai."], "turn_id": "book"}
],
"session_spans": [
    {
        "span_type": "agentic.tool.invocation",
        "entities": [{"type": "tool", "name": "adk_book_flight"}],
        # the flight must have been booked in the "book" turn, not the first
        "attributes": {"scope.turn_id": "book"}
    }
]

This works with the fluent API too. If you run each turn as a separate run_agent_async call with the same session_id, the turns are numbered ("1", "2", ...). Then check a turn with has_scope:

await monocle_trace_asserter.run_agent_async(agent, "langgraph", "Book me a flight for 26th Nov 2025.", session_id="s")  # turn "1"
await monocle_trace_asserter.run_agent_async(agent, "langgraph", "Fly to Mumbai.", session_id="s")                      # turn "2"

monocle_trace_asserter.called_tool("adk_book_flight").has_scope("turn_id", "2")   # booked in turn 2

Chaining turn output into the next input. Use the {previous_output} placeholder in a later turn's test_input to feed the prior turn's result forward:

"turns": [
    {"test_input": ["Suggest a city to visit in November."]},
    {"test_input": ["Book me a flight to {previous_output}."]}
]

Session persistence across runners. The ADK runner keeps one in-memory session service alive for the whole multi-turn run, so agent memory persists between turns. Runners that delegate session continuity to their own framework keep memory across turns via the shared session_id too (LangGraph thread_id, Strands FileSessionManager, LlamaIndex chat store, MS Agent thread).

Fluent API (monocle_trace_asserter)

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")

Chaining span selectors to narrow context

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")

Custom assertion messages

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"
)

Count and aggregate assertions

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

Scope, attribute, and event assertions

Assert on monocle scopes, span attributes, and span events. has_scope, has_attribute, and has_event narrow the context to matching spans:

# Scope carried by the trace (value check, existence check, and substring check)
monocle_trace_asserter.has_scope("tenant_id", "customer-123")
monocle_trace_asserter.has_scope("tenant_id")                  # presence only
monocle_trace_asserter.contains_scope("tenant_id", "customer")

# Any-of and negative scope assertions
monocle_trace_asserter.has_any_scope("tenant_id", "customer-123", "customer-456")
monocle_trace_asserter.does_not_have_scope("tenant_id", "customer-999")

# Verify a span attribute on the matched tool invocation
monocle_trace_asserter \
    .called_tool("adk_book_flight") \
    .has_attribute("entity.1.type", "tool.adk")

Framework Examples

Google ADK

@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")

CrewAI

@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 result

Fluent 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

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"}
        ]}
    ]
}

Strands

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"
    )

AWS Bedrock AgentCore

Unlike the other runners, agentcore is a remote runner: it invokes an agent already deployed to AWS Bedrock AgentCore Runtime via boto3, so no agent framework needs to be installed locally. The agent argument is the deployed agent's Runtime ARN (the endpoint-qualified form is accepted and split automatically):

ARN = "arn:aws:bedrock-agentcore:us-east-1:<account>:runtime/<agent-id>"

response = monocle_trace_asserter.run_agent(
    ARN, "agentcore",
    "Book a flight from San Jose to Seattle for 22 Nov 2026",
    session_id=f"monocle_test_session_{uuid.uuid4().hex}",
)

• Region is taken from the ARN, so the agent is reached in the region it was deployed to regardless of your local AWS default region.
• Payload sent to the agent is {"prompt": <message>}, matching the BedrockAgentCoreApp entrypoint contract. Pass a dict to send a different shape verbatim.
• Response: the runner reads the response stream and JSON-decodes it. An agent that returns text (the common case) yields a plain str; an agent returning a JSON object yields a dict.
• session_id is sent as runtimeSessionId so the deployed agent keeps conversation context across turns. AWS requires at least 33 characters — Monocle's auto-generated session ids satisfy this, and shorter ids raise a ValueError rather than being silently rewritten. Omit it to let AgentCore generate one.
• Traces: spans are produced inside the deployed agent and exported by its own Monocle instrumentation, so they are not in the test process. Retrieve them by session — the agent stamps the runtimeSessionId onto its spans as scope.agentic.session, which Okahu indexes as the agent_sessions fact:

The runner retrieves them for you, so assertions apply to the remote spans directly:

# AGENTCORE_TRACE_WORKFLOW=<workflow the deployed agent exports under>
response = monocle_trace_asserter.run_agent(ARN, "agentcore", prompt, session_id=session_id)

monocle_trace_asserter.called_tool("book_flight_tool")   # asserts on the remote spans

This needs the workflow name the deployed agent reports under — its own, not the test's — via the AGENTCORE_TRACE_WORKFLOW environment variable or the runner's trace_workflow_name argument. Without it, retrieval is skipped and only the response is available to assert on. Allow a few seconds for the spans to reach Okahu after the call returns; the runner polls for up to MONOCLE_REMOTE_TRACE_TIMEOUT seconds (default 60).

To load a session explicitly instead — for example one from an earlier run, or when no workflow name is configured — use the trace source directly:

monocle_trace_asserter.with_trace_source(
    "okahu", id=session_id, fact_name="session", workflow_name="<agent's workflow name>",
).called_tool("book_flight_tool")

Allow a few seconds for the spans to reach Okahu after the call returns.

Offline Testing with Pre-Recorded Traces

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)

Test Generator

Automatically generate test code by analyzing trace files. The generator scans spans and creates Python test assertions for agents, tools, outputs, cost, and performance, plus evaluation assertions supplied as --eval parameters (built-in or custom, auto-detected).

Quick Start

# Generate test code from a trace file
python -m monocle_test_tools generate_test --trace-file trace.json

# With custom test name
python -m monocle_test_tools generate_test --trace-file trace.json --test-name test_my_agent

# Generate test code directly from an Okahu cloud trace
python -m monocle_test_tools generate_test --trace-id <trace_id> --workflow-name <workflow_name>

# Inject built-in eval assertions (--eval-source is required when using --eval)
python -m monocle_test_tools generate_test --trace-id <trace_id> --workflow-name <workflow_name> \
  --trace-source okahu --eval-source okahu \
  --eval hallucination=no_hallucination \
  --eval sentiment=positive

# Inject a custom eval template assertion
python -m monocle_test_tools generate_test --trace-id <trace_id> --workflow-name <workflow_name> \
  --trace-source okahu --eval-source okahu \
  --eval ./my_custom_eval.json=pass

# Specify a different fact_name for all injected evals (default: traces)
python -m monocle_test_tools generate_test --trace-id <trace_id> --workflow-name <workflow_name> \
  --eval-source okahu --eval hallucination=no_hallucination --eval-fact agentic_turns

# Only generate the loader for a specific trace source (file | okahu)
python -m monocle_test_tools generate_test --trace-file trace.json --trace-source file

# Save to file
python -m monocle_test_tools generate_test --trace-file trace.json > test_generated.py

# The old module path still works
python -m monocle_test_tools.generate_test trace.json

By default the generated test includes loader options for every supported trace source (file, Okahu, live agent run). Passing --trace-source file or --trace-source okahu emits only that loader (as active code).

The --eval type is auto-detected from the value (or forced with a builtin: / custom: prefix):

• Built-in: plain name (e.g. hallucination, sentiment) → check_eval("hallucination", ...)
• Custom: value ends with .json or is a path → check_eval(template_path="./my_eval.json", ...)

--eval-source is required when --eval is used. It sets the evaluator in the generated with_evaluation(...) call and drives how each --eval value is classified as built-in vs custom:

python -m monocle_test_tools generate_test --trace-file trace.json \
  --eval-source okahu --eval hallucination=no_hallucination
# ->  asserter.with_evaluation("okahu").check_eval("hallucination", expected="no_hallucination", ...)

Example Output

Generated tests load traces through the with_trace_source API and include cost (total tokens) and performance (turn duration) checks derived from the trace:

import pytest
from monocle_test_tools import TraceAssertion


def test_generated(monocle_trace_asserter: TraceAssertion):
    """Auto-generated test from trace analysis."""

    # Option 1: Load from a local trace file
    monocle_trace_asserter.with_trace_source("file", trace_path="path/to/trace.json")

    # Option 2: Load from Okahu
    # monocle_trace_asserter.with_trace_source("okahu", id="TRACE_ID", workflow_name="WORKFLOW_NAME")

    # 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")

    # Cost check: total tokens in the turn (derived from trace; adjust as needed)
    asserter.under_token_limit(1204)

    # Performance check: duration of the turn (derived from trace; adjust as needed)
    asserter.under_duration(5.2, units="seconds", span_type="agent_turn")

    # Eval assertions (from --eval parameters; require an eval service, e.g. Okahu)
    asserter.with_evaluation("okahu").check_eval("hallucination", expected="pass")

The Okahu cloud loader is pre-populated with the id (trace id) and workflow_name read from the trace itself, so it is ready to uncomment without editing placeholders.

With --trace-source file, only the file loader line is emitted:

    # Load traces from a local trace file
    monocle_trace_asserter.with_trace_source("file", trace_path="path/to/trace.json")

The generator extracts:

• Agent invocations with output checks (first 80 chars as key phrase)
• Tool invocations with parent agent references
• Total token usage for the turn, emitted as an under_token_limit() check
• Turn duration, emitted as an under_duration(..., span_type="agent_turn") check
• Trace id and workflow name, used to pre-populate the Okahu cloud loader
• Sorted by invocation order for readability

Mock Tools

MockTool simulates tool calls without invoking real external services. Useful for testing agent logic in isolation.

Response templates

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.'}"
        }
    ]
)

Simulating errors

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.

Fluent API

In the fluent API, register mock tools with with_mock_tool() before running the agent (call once per mock tool):

@pytest.mark.asyncio
async def test_with_mock_tool(monocle_trace_asserter):
    monocle_trace_asserter.with_mock_tool(
        MockTool(
            name="adk_book_flight",
            type=ToolType.ADK,
            response={"status": "success", "message": "Flight booked from {{from_airport}} to {{to_airport}}."}
        )
    )
    await monocle_trace_asserter.run_agent_async(root_agent, "google_adk",
                        "Book a flight from San Francisco to Mumbai for 26th Nov 2025.")

    monocle_trace_asserter.called_tool("adk_book_flight").contains_output("Flight booked")

Test Format Reference

TestCase

{
    "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."
}

TestSpan

Validation rules enforced by span type:

• agentic.tool.invocation — first entity must be type tool; optional second must be type agent
• agentic.delegation — requires at least two agent entities (delegator first, delegatee second)
• agentic.invocation — first entity must be type agent

{
    "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.",
    "attributes":       "Expected span attributes (key/value pairs) to verify on the matched span.",
    "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 Types

Entity Types

Comparers

Pass comparers as strings ("similarity") or class instances (SentenceComparer()). Extend BaseComparer to create a custom comparer.

Evaluators

BERTScore (local)

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"
    }
}

Okahu Evaluation (cloud)

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")

When fact_name resolves to multiple facts (e.g. agentic_turns across a multi-turn session), check_eval evaluates and asserts every fact — it fails if any single fact fails, and the error names each failing fact. To scope an eval to one turn instead, narrow the spans first (e.g. .where(attribute={"scope.turn_id": "2"})) so only that turn's fact is evaluated.

Okahu fact_name values

Okahu evaluation templates (examples)

The following eval template names have been validated against real traces in tests:

Passing a template name or fact combination that does not exist in Okahu raises an AssertionError with the list of valid templates. Use @pytest.mark.xfail for tests that are expected to fail.

Custom Evaluation Templates

Instead of a built-in eval_name, you can send your own template to the eval service. check_eval accepts exactly one of three mutually exclusive selectors: eval_name (a standard Okahu template), template_path (a path to a custom-template JSON file), or template (an inline custom-template dict).

# From a file
monocle_trace_asserter.called_agent("adk_flight_booking_agent")
monocle_trace_asserter.with_evaluation("okahu") \
    .check_eval(template_path="templates/my_custom_eval.json", expected="pass")

# Inline dict (no file needed) — pass the template as a keyword-only argument
monocle_trace_asserter.with_evaluation("okahu") \
    .check_eval(template={"name": "my_eval", "eval_prompt": "...", "structure_output": {...}},
                expected="pass")

For template_path, the file may be either the inner template ({"name": ..., "eval_prompt": ..., ...}) or the full API request body ({"template": {...}}) — the outer template key is unwrapped automatically. When eval_name is omitted, the template's name field is used as the eval name (falling back to "custom_eval"). Server-side validation errors (HTTP 400) surface as an AssertionError prefixed with Custom template validation failed:.

Time-window (filtered) evaluation

The examples above evaluate a specific trace/fact that you selected (by running an agent, or by with_trace_source("okahu", id=...)). You can instead evaluate every fact discovered in a time window for one or more workflows — without knowing the trace ids up front — by setting start_time/end_time on with_trace_source("okahu", ...). This runs the evaluation as a single asynchronous Okahu job that discovers the matching facts server-side, applies one blanket expected/not_expected to each, and reports per fact.

# Evaluate all traces for "my_app" in a time window against one blanket expectation
monocle_trace_asserter \
    .with_trace_source("okahu", workflow_name="my_app",
                       start_time="2026-07-21T00:00:00Z", end_time="2026-07-22T00:00:00Z") \
    .with_evaluation("okahu") \
    .check_eval("hallucination", expected="no_hallucination", min_facts=5)

• Mode is chosen by the source, not a separate method. Supplying a time window (vs. an id) makes the same check_eval run the filtered flow. Providing both id and a time window raises; filter mode requires both start_time and end_time and a workflow_name.
• fact_name (defaults to traces) selects the fact grain to discover.
• Filter-only check_eval parameters (raise a ValueError if used without a time window):
  • min_facts (default 1) — fail the run if fewer than this many facts were discovered (guards against a vacuous pass on an empty window).
  • fail_threshold (default 0) — allow up to this many non-matching facts before the assertion fails.
  • max_facts (default from OKAHU_MAX_FACTS, 1000) — runaway guard; fail loudly if the window discovers more than this many facts.
• Results are uniform. Both the filtered and the single-fact paths populate the same report, readable via the accessors below (filtered mode = N facts; single-fact mode = a 1-fact report), on pass and failure.

Eval result matrix

An opt-in pytest recorder captures a per-test eval-result matrix (scenario, expected/actual label, judge output, token cost, pass/fail/error) across a whole run — useful for measuring eval stability and token cost without hand-rolling a conftest.py recorder. It is off by default and changes no behavior unless enabled.

Enable it either way:

pytest --monocle-eval-matrix                       # writes test-eval-replay-matrix.json
pytest --monocle-eval-matrix=path/to/matrix.json   # custom path
MONOCLE_EVAL_MATRIX=1 pytest                        # via env var (default path)

A row is recorded for every test that calls check_eval. At session end the recorder writes {"generated_at": <UTC ISO8601>, "records": [ ... ]}, where each record has a fixed, template-agnostic schema: run_id, scenario, trace_id, expected, actual, status (pass/fail/error), explanation, total_tokens, (for the filtered flow) fact_id, workflow, job_id, and judge_output — the judge's structured output verbatim. No judge field is promoted to a top-level column, because every template defines its own structure_output: read claim_verdicts / hallucination_types / entity_match_check (hallucination), addressed_aspects / missing_aspects / completeness_score (conversation_completeness), bias_types (bias) and so on from judge_output. Path precedence: the --monocle-eval-matrix value, else MONOCLE_EVAL_MATRIX, else the default test-eval-replay-matrix.json.

CSV eval test cases

For curating many eval cases as a flat spreadsheet, monocle_test_tools exports a CSV adapter — load_cases_from_csv, CsvCase, and the @monocle_csv_cases decorator — that parametrizes a one-line test stub over the rows and drives each through the fluent check_eval path. Scope (v0): evaluation tests only (fixed to the okahu trace source).

Config in code, data in CSV. The test stub owns everything constant across the sheet (the evaluator, the eval template, the trace source); the CSV owns only what varies per row. One row = one test.

import os
from monocle_test_tools import monocle_csv_cases

TEMPLATE_PATH = os.path.join(os.path.dirname(__file__), "hallucination_test.json")

@monocle_csv_cases("cases.csv")
def test_cases(monocle_trace_asserter, case):
    case.run(monocle_trace_asserter.with_evaluation("okahu"), template_path=TEMPLATE_PATH)

Each row maps onto exactly three stable, eval-relevant fluent methods — check_eval (the label being tuned) plus two optional operational guard rails, under_token_limit and under_duration:

A row must declare an expected or not_expected label — a guard rail alone is not an eval test. load_cases_from_csv(path) can also be used standalone (returns a list of CsvCase) and reports load errors with the offending line N. A copy-ready example ships at examples/cases.example.csv + examples/csv_cases_example.py.

Supported Agent Frameworks (Runners)

Fluent API Reference (TraceAssertion)

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.

Configuration

Configure the asserter before running assertions. These methods return self for chaining but do not themselves assert.

Run agents

Span selectors (narrow context for subsequent assertions)

Input assertions

Output assertions

Attribute assertions

Filter the current spans down to those carrying a given span attribute, so subsequent chained assertions operate on the matching subset. When value is omitted, only the presence of the attribute is checked.

Scope assertions

Assert on monocle scope values attached to the current spans. has_*/contains_* narrow the context to matching spans. When the value/values argument is omitted on has_scope/does_not_have_scope, only presence/absence of the scope is checked.

Performance assertions

Evaluation

Configure the evaluator with with_evaluation (see the Configuration table) before calling check_eval. Note: with_comparer overrides the comparer for has_* assertions but does not affect contains_* methods (those always use token/substring matching).

Direct Validation Methods (MonocleValidator)

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")

Environment Variables