Building an Agent Evaluation Harness with MCP (Model Context Protocol): Complete Guide

Introduction to Agent Evaluation Harnesses

As AI agents grow more capable and autonomous, evaluating their behavior becomes one of the most important engineering challenges in the LLM stack. An agent evaluation harness is a structured framework that runs an agent against a suite of tasks, captures its outputs, decisions, tool calls, and intermediate reasoning, and scores them against expected outcomes. Unlike simple prompt benchmarks, an agent harness must account for multi-step trajectories, tool usage, error recovery, and non-deterministic behavior.

The Model Context Protocol (MCP) — an open standard introduced by Anthropic for connecting LLMs to external tools, data sources, and capabilities — turns out to be an ideal backbone for building such a harness. MCP standardizes how tools and resources are exposed, which means an evaluation harness built on MCP can plug into any compliant agent and any compliant tool server without bespoke integration code.

In this guide, we'll build a complete, working agent evaluation harness from scratch using MCP. We'll cover what MCP is, why it matters for evaluation, how to construct the harness, and best practices for getting reliable, reproducible results.

What Is the Model Context Protocol?

MCP is a JSON-RPC 2.0 based protocol that defines a standard way for AI applications to communicate with external servers that provide tools, resources, and prompts. An MCP server exposes capabilities; an MCP client (typically an agent runtime) consumes them. The protocol supports both stdio and HTTP+SSE transports, making it suitable for local development and production deployments alike.

Three core primitives matter for evaluation:

• Tools — callable functions the agent can invoke (e.g., search_web, run_python).
• Resources — read-only data the agent can access (e.g., a file, a database row, a knowledge base entry).
• Prompts — reusable prompt templates the agent can load.

For an evaluation harness, the key insight is that MCP lets us observe and instrument the agent's interactions with the outside world in a uniform way. Every tool call flows through the same protocol surface, so we can intercept, log, replay, and score them.

Why MCP Is a Natural Fit for Evaluation

Traditional eval harnesses are tightly coupled to specific agent frameworks (LangChain, CrewAI, AutoGen, etc.). This coupling makes it hard to compare agents across frameworks and forces evaluators to re-implement instrumentation for each new tool integration. MCP solves this by decoupling the agent from its tools.

Here's why that matters:

• Framework agnosticism — Any agent that speaks MCP can be evaluated by the same harness.
• Tool-level observability — Every tool call is a structured JSON-RPC message you can capture.
• Reproducibility — Captured tool calls can be replayed against deterministic mock servers.
• Composability — Swap real tool servers for mock or sandboxed ones without touching the agent.
• Cost control — Mock servers eliminate API spend during repeated eval runs.

Architecture of the Harness

Our harness will have five components:

• Task Suite — A set of evaluation tasks, each with an input prompt, expected behavior, and scoring rubric.
• Agent Runner — Launches the agent under test and connects it to MCP tool servers.
• Instrumented MCP Proxy — Sits between the agent and tool servers, recording every request and response.
• Scorers — Functions that evaluate the agent's trajectory and final output against the task's rubric.
• Reporter — Aggregates scores across tasks and emits a structured report.

The instrumented proxy is the heart of the design. Instead of connecting the agent directly to a real tool server, we connect it to our proxy, which forwards traffic to the real server (or a mock) while logging everything. This gives us a complete record of the agent's tool-using behavior without modifying the agent itself.

Setting Up the Project

We'll use Python with the official MCP SDK. Create a new project and install dependencies:

mkdir mcp-eval-harness && cd mcp-eval-harness
python -m venv .venv
source .venv/bin/activate
pip install mcp anthropic pydantic rich

Create the following directory structure:

mcp-eval-harness/
├── harness/
│   ├── __init__.py
│   ├── proxy.py
│   ├── runner.py
│   ├── scorers.py
│   ├── tasks.py
│   └── reporter.py
├── servers/
│   └── mock_tools.py
├── tasks/
│   └── suite.json
└── run_eval.py

Defining the Task Suite

Each task describes a scenario the agent must handle. We define tasks as JSON so they can be versioned, shared, and extended. A task contains the user prompt, a list of available tools, optional mock tool responses, and one or more scoring criteria.

// tasks/suite.json
{
  "tasks": [
    {
      "id": "math-001",
      "prompt": "What is 17 multiplied by 23? Use the calculator tool.",
      "tools": ["calculator"],
      "mock_responses": {
        "calculator": {"result": 391}
      },
      "scorers": [
        {
          "type": "final_answer_exact",
          "expected": "391"
        },
        {
          "type": "tool_called",
          "tool": "calculator",
          "min_calls": 1
        }
      ]
    },
    {
      "id": "research-001",
      "prompt": "Find the capital of Australia and then calculate the length of its name.",
      "tools": ["search", "calculator"],
      "mock_responses": {
        "search": {"answer": "Canberra"},
        "calculator": {"result": 8}
      },
      "scorers": [
        {
          "type": "tool_called",
          "tool": "search",
          "min_calls": 1
        },
        {
          "type": "tool_called",
          "tool": "calculator",
          "min_calls": 1
        },
        {
          "type": "tool_order",
          "sequence": ["search", "calculator"]
        },
        {
          "type": "final_answer_contains",
          "expected": "8"
        }
      ]
    }
  ]
}

Now let's define the Python data model for tasks:

# harness/tasks.py
from pydantic import BaseModel
from typing import Any

class ScorerSpec(BaseModel):
    type: str
    expected: str | None = None
    tool: str | None = None
    min_calls: int | None = None
    sequence: list[str] | None = None

class Task(BaseModel):
    id: str
    prompt: str
    tools: list[str]
    mock_responses: dict[str, Any] = {}
    scorers: list[ScorerSpec]

class TaskSuite(BaseModel):
    tasks: list[Task]

def load_suite(path: str) -> TaskSuite:
    import json
    with open(path) as f:
        return TaskSuite(**json.load(f))

Building the Mock Tool Server

For reproducible evaluation, we want deterministic tool responses. We'll build a mock MCP server that returns canned responses based on the task's mock_responses field. In production you might point at real servers, but mocks are essential for CI.

# servers/mock_tools.py
import sys
import json
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent

server = Server("mock-tools")

# This is populated at startup from stdin as a JSON line.
MOCK_STATE = {"responses": {}}

@server.list_tools()
async def list_tools() -> list[Tool]:
    tools = []
    for name, resp in MOCK_STATE["responses"].items():
        tools.append(Tool(
            name=name,
            description=f"Mock tool: {name}",
            inputSchema={"type": "object", "properties": {}}
        ))
    return tools

@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    if name in MOCK_STATE["responses"]:
        result = MOCK_STATE["responses"][name]
        return [TextContent(type="text", text=json.dumps(result))]
    return [TextContent(type="text", text=json.dumps({"error": "unknown tool"}))]

async def main():
    # Read mock configuration from the first line of stdin before MCP starts.
    line = sys.stdin.readline()
    if line.strip():
        MOCK_STATE["responses"] = json.loads(line)
    async with stdio_server() as (read, write):
        await server.run(read, write, server.create_initialization_options())

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

The Instrumented MCP Proxy

The proxy is the most important piece. It acts as an MCP server to the agent and as an MCP client to the upstream tool server. Every message that passes through is recorded into a trajectory log. This design means we never need to modify the agent or the tool server to capture data.

# harness/proxy.py
import json
import asyncio
from dataclasses import dataclass, field
from typing import Any

@dataclass
class Trajectory:
    task_id: str
    tool_calls: list[dict] = field(default_factory=list)
    final_output: str = ""

    def add_call(self, tool: str, arguments: dict, response: Any):
        self.tool_calls.append({
            "tool": tool,
            "arguments": arguments,
            "response": response
        })

class InstrumentedProxy:
    """
    A minimal in-process proxy that records tool calls.
    In a full implementation this would bridge two MCP transports;
    here we simulate it by wrapping a callable tool registry.
    """
    def __init__(self, tool_registry: dict, trajectory: Trajectory):
        self.registry = tool_registry
        self.trajectory = trajectory

    async def call(self, tool: str, arguments: dict) -> Any:
        if tool not in self.registry:
            result = {"error": f"unknown tool: {tool}"}
        else:
            handler = self.registry[tool]
            result = await handler(arguments) if asyncio.iscoroutinefunction(handler) else handler(arguments)
        self.trajectory.add_call(tool, arguments, result)
        return result

In a production harness, you'd implement the proxy as a real MCP server using stdio_server on the agent-facing side and an MCP client connection to the upstream server. The recording logic remains identical — intercept call_tool, log, forward.

Building the Agent Runner

The runner executes a single task: it sets up the tool registry, creates a trajectory, runs the agent loop, and returns the trajectory for scoring. We'll use the Anthropic API directly to keep the example framework-agnostic, but you could swap in any MCP-compatible agent.

# harness/runner.py
import json
import anthropic
from harness.proxy import InstrumentedProxy, Trajectory
from harness.tasks import Task

client = anthropic.Anthropic()

def build_tool_registry(task: Task) -> dict:
    """Build a registry of mock tool handlers from the task spec."""
    registry = {}
    for tool_name, response in task.mock_responses.items():
        # Closure captures the correct response per tool.
        def make_handler(resp):
            def handler(args):
                return resp
            return handler
        registry[tool_name] = make_handler(response)
    return registry

def run_task(task: Task, model: str = "claude-sonnet-4-20250514") -> Trajectory:
    trajectory = Trajectory(task_id=task.id)
    registry = build_tool_registry(task)
    proxy = InstrumentedProxy(registry, trajectory)

    # Build Anthropic tool definitions from the registry.
    tools = [
        {
            "name": name,
            "description": f"Tool: {name}",
            "input_schema": {"type": "object", "properties": {}}
        }
        for name in registry
    ]

    messages = [{"role": "user", "content": task.prompt}]

    while True:
        response = client.messages.create(
            model=model,
            max_tokens=1024,
            tools=tools,
            messages=messages
        )

        if response.stop_reason == "tool_use":
            messages.append({"role": "assistant", "content": response.content})
            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    result = proxy.call(block.name, block.input)
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": json.dumps(result)
                    })
            messages.append({"role": "user", "content": tool_results})
        else:
            # Extract final text output.
            for block in response.content:
                if block.type == "text":
                    trajectory.final_output += block.text
            break

    return trajectory

Implementing Scorers

Scorers turn a trajectory into a numeric score. We implement several common scorer types referenced in our task spec. Each scorer returns a float between 0 and 1 and a human-readable reason.

# harness/scorers.py
from harness.proxy import Trajectory
from harness.tasks import ScorerSpec

def score_final_answer_exact(traj: Trajectory, spec: ScorerSpec) -> tuple[float, str]:
    passed = spec.expected.strip().lower() in traj.final_output.lower()
    return (1.0 if passed else 0.0,
            f"Expected '{spec.expected}' in output")

def score_final_answer_contains(traj: Trajectory, spec: ScorerSpec) -> tuple[float, str]:
    passed = spec.expected.lower() in traj.final_output.lower()
    return (1.0 if passed else 0.0,
            f"Expected output to contain '{spec.expected}'")

def score_tool_called(traj: Trajectory, spec: ScorerSpec) -> tuple[float, str]:
    count = sum(1 for c in traj.tool_calls if c["tool"] == spec.tool)
    passed = count >= (spec.min_calls or 1)
    return (1.0 if passed else 0.0,
            f"Expected >= {spec.min_calls} calls to '{spec.tool}', got {count}")

def score_tool_order(traj: Trajectory, spec: ScorerSpec) -> tuple[float, str]:
    called = [c["tool"] for c in traj.tool_calls]
    # Check if the expected sequence appears as a subsequence.
    i = 0
    for tool in called:
        if i < len(spec.sequence) and tool == spec.sequence[i]:
            i += 1
    passed = i == len(spec.sequence)
    return (1.0 if passed else 0.0,
            f"Expected order {spec.sequence}, observed {called}")

SCORERS = {
    "final_answer_exact": score_final_answer_exact,
    "final_answer_contains": score_final_answer_contains,
    "tool_called": score_tool_called,
    "tool_order": score_tool_order,
}

def score_trajectory(traj: Trajectory, specs: list[ScorerSpec]) -> dict:
    results = []
    for spec in specs:
        fn = SCORERS.get(spec.type)
        if fn is None:
            results.append({"type": spec.type, "score": 0.0, "reason": "unknown scorer"})
            continue
        score, reason = fn(traj, spec)
        results.append({"type": spec.type, "score": score, "reason": reason})
    overall = sum(r["score"] for r in results) / len(results) if results else 0.0
    return {"task_id": traj.task_id, "scores": results, "overall": overall}

The Reporter

The reporter aggregates per-task results into a summary and prints a readable table.

# harness/reporter.py
from rich.console import Console
from rich.table import Table

def print_report(results: list[dict]):
    console = Console()
    table = Table(title="Agent Evaluation Results")
    table.add_column("Task ID", style="cyan")
    table.add_column("Overall", style="bold")
    table.add_column("Details", style="dim")

    for r in results:
        details = "; ".join(
            f"{s['type']}={s['score']:.1f}" for s in r["scores"]
        )
        color = "green" if r["overall"] == 1.0 else ("yellow" if r["overall"] >= 0.5 else "red")
        table.add_row(r["task_id"], f"[{color}]{r['overall']:.2f}[/{color}]", details)

    console.print(table)

    avg = sum(r["overall"] for r in results) / len(results) if results else 0
    console.print(f"\n[bold]Average score across {len(results)} tasks: {avg:.2f}[/bold]")

Putting It All Together

Finally, the entry point loads the suite, runs each task, scores it, and prints the report.

# run_eval.py
import asyncio
from harness.tasks import load_suite
from harness.runner import run_task
from harness.scorers import score_trajectory
from harness.reporter import print_report

def main():
    suite = load_suite("tasks/suite.json")
    results = []
    for task in suite.tasks:
        print(f"Running task {task.id}...")
        traj = run_task(task)
        result = score_trajectory(traj, task.scorers)
        result["trajectory"] = {
            "tool_calls": traj.tool_calls,
            "final_output": traj.final_output[:200]
        }
        results.append(result)

    print_report(results)

    # Optionally save full results to disk.
    import json
    with open("eval_results.json", "w") as f:
        json.dump(results, f, indent=2)

if __name__ == "__main__":
    main()

Run the harness:

python run_eval.py

You should see output like:

Running task math-001...
Running task research-001...

                  Agent Evaluation Results
┏━━━━━━━━━━━━━━┳━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┓
┃ Task ID      ┃ Overall ┃ Details                          ┃
┡━━━━━━━━━━━━━━╇━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┩
│ math-001     │ 1.00    │ final_answer_exact=1.0; tool... │
│ research-001 │ 0.75    │ tool_called=1.0; tool_called=... │
└──────────────┴─────────┴──────────────────────────────────┘

Average score across 2 tasks: 0.88

Best Practices

Building a harness is only half the battle. Getting trustworthy evaluation results requires discipline. Here are the practices that matter most:

• Use mock tools for CI, real tools for staging. Deterministic mocks let you catch regressions in agent logic. Run against real APIs periodically to catch drift in tool behavior.
• Run multiple seeds per task. Agents are stochastic. A single run is a sample, not a measurement. Run each task 3–5 times and report the mean and variance.
• Score trajectories, not just final answers. An agent that reaches the right answer through a dangerous or inefficient path is not equivalent to one that does so cleanly. Use tool-order and call-count scorers to capture process quality.
• Version your task suites. Treat your eval suite like test code: commit it, review changes, and track score history over time. A score drop after a suite change is meaningless without versioning.
• Isolate agent state between tasks. Never let one task's context bleed into the next. Each task gets a fresh conversation, fresh tool registry, and fresh trajectory.
• Capture full trajectories for debugging. Store the complete tool call log alongside scores. When a task fails, the trajectory is your debugging evidence.
• Use LLM-as-judge scorers for open-ended tasks. Exact-match scorers don't work for tasks like "write a summary." Add a scorer type that sends the output to a judge model with a rubric prompt.
• Watch for prompt leakage. If your mock tool returns the expected answer verbatim, a lazy agent might just echo it. Include distractor data in mocks and verify the agent actually reasons.

Extending the Harness

The harness above is a foundation. Here are directions worth exploring:

• Real MCP proxy transport — Implement the proxy as a true stdio-to-stdio MCP bridge so you can evaluate any MCP-speaking agent, not just the Anthropic direct integration shown here.
• Parallel execution — Run tasks concurrently with asyncio.gather to speed up large suites. Just be mindful of API rate limits.
• Regression dashboards — Store results in a database (SQLite is fine) and plot score trends over commits.
• Adversarial tasks — Include tasks where the correct behavior is to refuse a tool call or ask for clarification, to test safety and calibration.
• Cost tracking — Capture token counts from each API response and report cost per task alongside scores.

Conclusion

Evaluating AI agents is fundamentally harder than evaluating static models because agents act, decide, and recover across multiple steps. By building your evaluation harness on top of MCP, you get a framework-agnostic, observable, and reproducible system that can grow with your agent stack. The protocol's clean separation between agent and tool server means your instrumentation lives in one place — the proxy — rather than scattered across every integration. Start with the mock-based harness shown here, add LLM-as-judge scorers for open-ended tasks, and treat your eval suite with the same rigor you apply to production test suites. The agents you ship are only as trustworthy as the evaluation that backs them.