ADK eval harness starter

The gap between “the agent seems better” and “the agent is better” is a dataset and a loop. This is the minimal harness shape for running an ADK agent against a case set and scoring the results — small enough to read in one sitting, structured so judges and metrics can grow without rework.

Case format

Cases are data, not code. Keep them in version control next to the agent they test.

{
  "id": "refund-over-limit-001",
  "input": "I need a refund of $4,200 for order 88231",
  "expectations": {
    "must_call_tools": ["lookup_order", "check_refund_policy"],
    "must_not_call_tools": ["issue_refund"],
    "outcome": "escalates to a human approver because the amount exceeds the auto-approval limit"
  }
}

The harness

import asyncio
import json
from dataclasses import dataclass, field
from pathlib import Path

from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService

from my_agents import root_agent  # your ADK agent


@dataclass
class CaseResult:
    case_id: str
    final_response: str
    tool_calls: list[str] = field(default_factory=list)
    passed_checks: dict[str, bool] = field(default_factory=dict)


async def run_case(runner: Runner, case: dict) -> CaseResult:
    session = await runner.session_service.create_session(
        app_name="eval", user_id="eval-user"
    )
    result = CaseResult(case_id=case["id"], final_response="")

    async for event in runner.run_async(
        user_id="eval-user", session_id=session.id, new_message=case["input"]
    ):
        for call in event.get_function_calls() or []:
            result.tool_calls.append(call.name)
        if event.is_final_response() and event.content:
            result.final_response = event.content.parts[0].text

    exp = case["expectations"]
    result.passed_checks = {
        "required_tools": set(exp.get("must_call_tools", [])) <= set(result.tool_calls),
        "forbidden_tools": not set(exp.get("must_not_call_tools", []))
        & set(result.tool_calls),
        # Outcome checks need an LLM judge — see below.
    }
    return result


async def main() -> None:
    cases = [json.loads(p.read_text()) for p in Path("cases").glob("*.json")]
    runner = Runner(
        agent=root_agent, app_name="eval", session_service=InMemorySessionService()
    )
    results = [await run_case(runner, case) for case in cases]

    for r in results:
        status = "PASS" if all(r.passed_checks.values()) else "FAIL"
        print(f"{status}  {r.case_id}  {r.passed_checks}")


if __name__ == "__main__":
    asyncio.run(main())

Extending it

• Deterministic checks first. Tool-call assertions catch most regressions and cost nothing. Only reach for an LLM judge for the outcome expectation — score the final response against the expected outcome with a rubric prompt, and calibrate the judge against a handful of human-labeled cases before trusting it.
• Persist traces. Write each CaseResult (plus raw events) to JSONL per run; diffing two runs answers “what changed” faster than any dashboard.
• Gate on it. Wire the harness into CI so prompt and tool changes fail the build when pass-rate drops — that is the moment evals start paying for themselves.

The companion deep-dive, Evaluating multi-agent systems: a practical loop with Google’s ADK, covers dataset construction, judge calibration, and the failure modes this harness surfaces.