monocle-test-tools 0.8.10

Testing and validation framework for monocle AI agent tracing

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

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

---

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, and outputs.

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

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

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

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

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 Types

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

---

Comparers

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.

---

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

Okahu fact_name values

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

Okahu evaluation templates (examples)

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

Method	Description
get_eval_report()	The eval report stashed by the last check_eval (dict with a scenarios list)
get_eval_failures()	The subset of report scenarios whose status is not pass
write_eval_report(path)	Write the report to a JSON file

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 schema: run_id, scenario, trace_id, expected, actual, status (pass/fail/error), explanation, total_tokens, claim_verdicts, hallucination_types, entity_match_check, and (for the filtered flow) fact_id, workflow, job_id. 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:

Column	Required	Maps to	Notes
case_id	✅	(test id)	Unique per row; duplicates are rejected at load time
fact_id	✅	with_trace_source(id=...)	Identifier of the evaluated fact — equals the trace id when fact_name=traces, a finer-grained id otherwise
workflow_name	✅	with_trace_source(workflow_name=...)	Okahu workflow name
fact_name		check_eval(fact_name=...)	Fact grain (defaults to traces)
expected	one of these	check_eval(expected=...)	Multi-value: pipe-delimited (a|b) or a JSON array
not_expected	one of these	check_eval(not_expected=...)	Multi-value, same as expected
max_tokens		under_token_limit(...)	Token-budget guard rail
max_duration_ms		under_duration(..., units="ms")	Effort/latency guard rail
notes		(ignored)	Free-form documentation for curators

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)

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

---

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.

Method	Description
with_trace_source(source="local", **kwargs)	Choose where spans come from: "local" (in-memory, default), "file" (local .monocle/*.json), or "okahu" (cloud). File/Okahu kwargs: id, trace_path (file), fact_name (trace/session/scope), scope_name, workflow_name (required for Okahu). Okahu time-window (filtered) mode: pass start_time + end_time (both required) with workflow_name and no id to evaluate all discovered facts in a window (see Time-window evaluation)
with_comparer(comparer)	Override the comparer for subsequent has_* assertions (see the Comparers table). Accepts a string key or BaseComparer instance
with_evaluation(eval, eval_options=None)	Configure the evaluator ("okahu", "bert_score", or a BaseEval instance) before check_eval
with_mock_tool(mock_tool)	Register a MockTool to simulate tool behavior during a subsequent run_agent/run_agent_async (fluent equivalent of TestCase.mock_tools). Call once per mock tool

Run agents

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

Span selectors (narrow context for subsequent 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

Input assertions

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

Output assertions

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

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.

Method	Description
has_attribute(attribute_name, expected=None)	Assert at least one span carries the attribute attribute_name (optionally equal to expected); narrows context to matching spans (key/value accepted as legacy aliases)
does_not_have_attribute(attribute_name, expected=None)	Assert no span carries the attribute attribute_name (optionally equal to expected) (key/value accepted as legacy aliases)
has_event(event_name, attribute_name=None, expected=None)	Assert at least one span has an event named event_name (optionally carrying attribute attribute_name, optionally equal to expected); narrows context to matching spans
where(attribute=None, event=None, predicate=None)	Generic selector: narrow context to spans matching all given criteria — attribute ({name: expected} mapping), event ({"name":..., "attributes":{...}}), and/or predicate (Callable[[Span], bool]). has_attribute/has_event are wrappers over this.
does_not_match(attribute=None, event=None, predicate=None)	Assert no span matches all the given criteria (same arguments as where)

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.

Method	Description
has_scope(scope_name, expected_value=None)	Assert a span has the scope (optionally equal to expected_value)
has_any_scope(scope_name, *expected_values)	Assert a span has the scope with any of the given values
does_not_have_scope(scope_name, unexpected_value=None)	Assert no span has the scope (optionally with unexpected_value)
does_not_have_any_scope(scope_name, *unexpected_values)	Assert no span has the scope with any of the given values
contains_scope(scope_name, expected_substring)	Assert the scope value contains the substring
contains_any_scope(scope_name, *expected_substrings)	Assert the scope value contains any of the substrings
does_not_contain_scope(scope_name, unexpected_substring)	Assert the scope value does not contain the substring
does_not_contain_any_scope(scope_name, *unexpected_substrings)	Assert the scope value does not contain any of the substrings

Performance assertions

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

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

Method	Description
check_eval(eval_name=None, expected=None, not_expected=None, fact_name="traces", template_path=None, *, template=None, min_facts=1, fail_threshold=0, max_facts=None)	Run an evaluation and assert the result. Provide exactly one of eval_name (a standard Okahu template), template_path (a custom-template JSON file), or template (an inline custom-template dict). expected/not_expected each accept a string or list of strings. min_facts/fail_threshold/max_facts apply only in time-window (filtered) mode (see Time-window evaluation) and raise otherwise
get_eval_report() / get_eval_failures() / write_eval_report(path)	Read the report stashed by the last check_eval — the full report dict, the non-pass scenarios, or write it to JSON. Same shape in single-fact and filtered modes

---

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

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
OKAHU_MAX_FACTS	Runaway guard for time-window (filtered) evals: fail loudly if a filter discovers more than this many facts (see Time-window evaluation)	1000
MONOCLE_EVAL_MATRIX	Set (to any value, optionally a path) to enable the opt-in eval-result-matrix recorder (see Eval result matrix). Off when unset	unset (off)
LOCAL_RUN_ID	Run identifier applied to all spans in a session	ISO datetime at session start

---