🧪 FastMCP Evals Suite#

MCP Agent Evaluation Suite treats the MCP Cricket Agent from the previous demo as a software system under test. Instead of eyeballing one or two prompts manually after editing a system prompt, model, or tool docstring, run_mcp_evals_suite() runs a repeatable 5-case golden evaluation suite (TEST_CASES) against the real run_mcp_agent() pipeline and computes an overall pass rate.

Every test case applies two independent deterministic assertions:

1. Tool Routing Assertion (tool_routing_passed): Did the agent call the expected MCP tool during its execution trace (e.g., get_live_score for match questions, get_player_stats for player questions, and None for "What is 2 + 2?")?
2. Ground-Truth Answer Assertion (answer_verified): Does the agent's final natural-language answer contain every required factual token in must_contain (e.g., "287", "90", "164", "78", "4")?

1. Why Exact String Equality (assert answer == expected) Fails for LLMs

In traditional unit testing, assert add(2, 2) == 4 checks for an exact value. With an LLM, "India scored 287/6 in 50 overs" and "India posted 287 for 6" are both 100% correct yet fail a naive == check. Effective AI evals test behavioral invariants (which tool was called) and factual containment (whether the ground-truth numbers appear in the response).

2. Dual-Layer Grading: Trajectory Evaluation + Outcome Evaluation

Grading only the final text isn't enough, and grading only the tool call isn't enough:

• Trajectory Check (tool_ok): Inspects tools_used = [step["tool"] for step in result["trace"]]. This catches an agent that called the wrong tool or wasted latency calling tools when none were needed.
• Outcome Check (answer_ok): Verifies all(word.lower() in answer.lower() for word in case["must_contain"]). This catches an agent that called the right tool but misread or omitted the returned number in its final reply.

3. The Importance of Negative Control Cases (Over-Triggering Check)

Look at Test Case 5: {"q": "What is 2 + 2?", "tool": None, "must_contain": ["4"]}. When an agent is given cricket tools, a poorly tuned prompt might try to call get_live_score on every single question. Setting expected_tool=None and asserting tools_used == [] verifies that the agent abstains from calling tools on general-knowledge questions.

4. Regression Testing Across Prompt and Model Changes

By aggregating passed_cases, total_cases, and accuracy_percentage into a single JSON scorecard, engineers can immediately see if swapping models or editing tool descriptions in cricket_server.py causes a routing or grounding regression.

src/python/03_mcp_evals.py, src/python/02_mcp_agent.py, and src/python/cricket_server.py implement an automated, unit-test-style evaluation harness for the MCP Cricket Agent:

• 03_mcp_evals.py — Defines the 5-case golden evaluation dataset (TEST_CASES) and runs each prompt through run_mcp_agent(), grading both tool-routing accuracy (trajectory evaluation) and factual answer containment (outcome evaluation).
• 02_mcp_agent.py — The system under test: connects the LLM to cricket_server.py over MCP and returns {"answer": ..., "trace": ...} for each evaluation prompt.
• cricket_server.py — The FastMCP server providing deterministic ground-truth cricket scores and player stats over stdio.

1. MCP Evaluation Suite Module (03_mcp_evals.py)

• Pydantic Scorecard Contracts (TestCaseEvalResult & McpEvalsResponse):
  • TestCaseEvalResult captures per-question grading: prompt: str, expected_tool: Optional[str], tools_used: list[str], tool_routing_passed: bool, answer_verified: bool, and overall_passed: bool.
  • McpEvalsResponse aggregates suite-level KPIs: total_cases: int, passed_cases: int, accuracy_percentage: int, and results: list[TestCaseEvalResult].
• Golden Evaluation Dataset (TEST_CASES, Step ②): Defines 5 structured test cases covering three distinct agent behaviors:
  • Live Match Routing (Cases 1 & 2): Questions about the India vs West Indies live score and runs needed to win → expect tool="get_live_score" and facts ["287"] / ["90"].
  • Player Stats Routing (Cases 3 & 4): Questions about Shubman Gill's series runs and Shai Hope's highest score → expect tool="get_player_stats" and facts ["164"] / ["78"].
  • Negative Control / Abstention (Case 5): "What is 2 + 2?" → expects tool=None (verifying the agent abstains from calling MCP tools on unrelated arithmetic) and fact ["4"].
• async def run_mcp_evals_suite() -> dict: Executes and grades all 5 cases sequentially:
  • Step ③ — Agent Invocation & Trace Extraction: Loops over each case in TEST_CASES, awaits mcp_agent.run_mcp_agent(case["q"]) (Step ①), extracts the list of invoked tool names tools_used = [step["tool"] for step in result["trace"]], and reads answer = result["answer"] or "".
  • Step ④ — Assertion 1: Tool Routing Check (tool_ok): Evaluates (case["tool"] in tools_used) if case["tool"] else (tools_used == []). When an MCP tool is expected, verifies it was called; when case["tool"] is None, verifies no tools were called.
  • Step ⑤ — Assertion 2: Ground-Truth Fact Check (answer_ok): Evaluates all(word.lower() in answer.lower() for word in case["must_contain"]) to confirm every required ground-truth token appears in the final answer.
  • Step ⑥ — Per-Case Verdict Aggregation: Computes ok = tool_ok and answer_ok, increments passed += ok, and appends the per-case result dictionary to results.
  • Step ⑦ — Suite Score Calculation: Returns total_cases, passed_cases, integer percentage accuracy_percentage = 100 * passed // len(TEST_CASES), and the per-case results list.

2. System Under Test: Agent & Server (02_mcp_agent.py & cricket_server.py)

• run_mcp_agent(question) in 02_mcp_agent.py: Spawns cricket_server.py over stdio, converts discovered MCP tools via to_openai_format(), runs the Groq tool-calling loop with call_llm(), and returns both answer and trace so 03_mcp_evals.py can inspect the exact tool trajectory.
• get_live_score & get_player_stats in cricket_server.py: Provide deterministic mock data (287/6, 90 runs from 70 balls, Gill's 164 runs, Hope's 78 highest score) so ground-truth assertions are 100% reproducible.

3. End-to-End Evaluation Flow

1. For each of the 5 TEST_CASES in 03_mcp_evals.py, run_mcp_evals_suite() calls await mcp_agent.run_mcp_agent(case["q"]).
2. run_mcp_agent() starts cricket_server.py over stdio, discovers its tools, queries the LLM, executes any requested MCP tools on cricket_server.py, and returns {"answer": ..., "trace": ...}.
3. run_mcp_evals_suite() grades each response on two axes — tool_routing_passed (trajectory check against trace) and answer_verified (outcome check against must_contain) — and computes accuracy_percentage across all 5 cases.

MCP Agent Evaluation Suite Service.

"""MCP Agent Evaluation Suite Service.

Runs unit-test style evaluation assertions against the MCP-enabled cricket agent (02_mcp_agent.py):
1. Tool routing assertion: verifies the LLM chooses the expected MCP tool (`get_live_score`, `get_player_stats`, or none).
2. Ground-truth verification: checks whether required factual tokens appear in the agent's final answer.
3. Suite scoring: aggregates pass/fail counts and overall accuracy percentage across all test cases.
"""

import importlib.util
import sys
from pathlib import Path
from typing import Optional

from pydantic import BaseModel

# ① Reuse the MCP agent loop from 02_mcp_agent.py in this folder
if "services.mcp_agent" in sys.modules:
    mcp_agent = sys.modules["services.mcp_agent"]
else:
    _spec = importlib.util.spec_from_file_location("mcp_agent", Path(__file__).with_name("02_mcp_agent.py"))
    mcp_agent = importlib.util.module_from_spec(_spec)
    _spec.loader.exec_module(mcp_agent)


class TestCaseEvalResult(BaseModel):
    prompt: str
    expected_tool: Optional[str]
    tools_used: list[str]
    tool_routing_passed: bool
    answer_verified: bool
    overall_passed: bool


class McpEvalsResponse(BaseModel):
    """Output payload from MCP Evals Suite."""

    total_cases: int
    passed_cases: int
    accuracy_percentage: int
    results: list[TestCaseEvalResult]


# ② Evaluation test suite: question, expected tool (None = no tool) and facts the answer must contain
TEST_CASES = [
    {"q": "What's the live score of India vs West Indies?", "tool": "get_live_score", "must_contain": ["287"]},
    {"q": "How many runs does West Indies need to win?", "tool": "get_live_score", "must_contain": ["90"]},
    {"q": "How many runs has Shubman Gill scored this series?", "tool": "get_player_stats", "must_contain": ["164"]},
    {"q": "Shai Hope's highest score in the series?", "tool": "get_player_stats", "must_contain": ["78"]},
    {"q": "What is 2 + 2?", "tool": None, "must_contain": ["4"]},
]


async def run_mcp_evals_suite() -> dict:
    """Execute the complete MCP evaluation suite."""
    results = []
    passed = 0

    # ③ Run every test case in turn: run the agent on one test case and apply both assertions
    for case in TEST_CASES:
        result = await mcp_agent.run_mcp_agent(case["q"])
        tools_used = [step["tool"] for step in result["trace"]]
        answer = result["answer"] or ""

        # ④ Assertion 1 (Tool Routing): the agent picked the expected tool (or no tool at all)
        tool_ok = (case["tool"] in tools_used) if case["tool"] else (tools_used == [])
        # ⑤ Assertion 2 (Ground Truth): the final answer contains every required fact
        answer_ok = all(word.lower() in answer.lower() for word in case["must_contain"])

        # ⑥ Combine both assertions into the per-case result
        ok = tool_ok and answer_ok
        passed += ok
        results.append({
            "prompt": case["q"],
            "expected_tool": case["tool"],
            "tools_used": tools_used,
            "tool_routing_passed": tool_ok,
            "answer_verified": answer_ok,
            "overall_passed": ok,
        })

    # ⑦ Compute the final suite score
    return {
        "total_cases": len(TEST_CASES),
        "passed_cases": passed,
        "accuracy_percentage": 100 * passed // len(TEST_CASES),
        "results": results,
    }

MCP-Enabled LLM Agent Service.

"""MCP-Enabled LLM Agent Service.

Connects an LLM to the FastMCP cricket server:
1. Starts `cricket_server.py` over stdio and discovers its tools with `tools/list`.
2. Converts the MCP tool descriptors into the OpenAI function-calling format.
3. Runs the tool-calling loop: the LLM asks for a tool, the client runs it with `tools/call`,
   and the tool output is fed back until the LLM gives a final answer.
"""

import json
import os
import re
import sys
from pathlib import Path

from dotenv import find_dotenv, load_dotenv
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from openai import OpenAI
from pydantic import BaseModel

load_dotenv(find_dotenv())


class McpAgentRequest(BaseModel):
    """Input payload for MCP Agent."""

    question: str = "What is the score in India vs West Indies?"


class McpAgentResponse(BaseModel):
    """Output payload from MCP Agent."""

    answer: str
    trace: list[dict]


# ① Model settings (Groq's OpenAI-compatible endpoint) and the MCP server to spawn over stdio
MODEL = os.getenv("GROQ_MODEL", "openai/gpt-oss-20b")
GROQ_BASE_URL = os.getenv("GROQ_BASE_URL", "https://api.groq.com/openai/v1")
server_script = str(Path(__file__).with_name("cricket_server.py"))
server = StdioServerParameters(command=sys.executable, args=[server_script])


def to_openai_format(mcp_tools):
    """MCP tool list -> format expected by OpenAI-compatible API."""
    return [
        {
            "type": "function",
            "function": {
                "name": t.name,
                "description": t.description or "",
                "parameters": t.inputSchema,
            },
        }
        for t in mcp_tools
    ]


def call_llm(llm, kwargs, retries=2):
    """Call the model, retrying when it tries to call a tool we never gave it."""
    last = None
    for _ in range(retries + 1):
        try:
            return llm.chat.completions.create(**kwargs).choices[0].message, None
        except Exception as e:
            last = str(e)
            if "tool_use_failed" not in last:
                return None, f"API ERROR: {last}"
    m = re.search(r'"name":\s*"([^"]+)"', last or "")
    tool = m.group(1) if m else "a tool"
    return None, f"MODEL TRIED TO CALL A TOOL WE NEVER GAVE IT ({tool}). Try running again."


async def run_mcp_agent(question: str) -> dict:
    """Run the MCP agent loop for a single user question; returns the answer and a tool trace."""
    # ② Initialize the Groq OpenAI-compatible client
    llm = OpenAI(api_key=os.getenv("GROQ_API_KEY"), base_url=GROQ_BASE_URL)
    trace = []

    messages = [
        {
            "role": "system",
            "content": (
                "You are a cricket assistant. Use ONLY the functions provided to get live data. "
                "Do not use any browser or code tools. Be brief."
            ),
        },
        {"role": "user", "content": question},
    ]

    # ③ Connect to the FastMCP cricket server over stdio and discover its tools
    async with stdio_client(server) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            tools = to_openai_format((await session.list_tools()).tools)

            # ④ Run the multi-step tool-calling loop (up to 5 turns)
            for _ in range(5):
                reply, error = call_llm(llm, {"model": MODEL, "messages": messages, "tools": tools, "temperature": 0})
                if error:
                    return {"answer": f"({error})", "trace": trace}

                # No tool requested -> this is the final answer
                if not reply.tool_calls:
                    return {"answer": reply.content or "", "trace": trace}

                # ⑤ Run each requested tool on the MCP server and feed the output back to the LLM
                messages.append(
                    {
                        "role": "assistant",
                        "content": reply.content or "",
                        "tool_calls": [
                            {
                                "id": c.id,
                                "type": "function",
                                "function": {"name": c.function.name, "arguments": c.function.arguments},
                            }
                            for c in reply.tool_calls
                        ],
                    }
                )
                for call in reply.tool_calls:
                    args = json.loads(call.function.arguments)
                    result = await session.call_tool(call.function.name, args)
                    text = result.content[0].text if result.content else "(empty)"
                    trace.append({"tool": call.function.name, "args": args, "result": text})
                    messages.append({"role": "tool", "tool_call_id": call.id, "content": text})

    return {"answer": "(stopped: too many steps)", "trace": trace}

Cricket FastMCP Server (`cricket_server.py`).

"""Cricket FastMCP Server (`cricket_server.py`).

Provides Model Context Protocol (MCP) tool endpoints for live cricket matches and player statistics:
1. `get_live_score(teams)`: Fetches the live scorecard, target, and match situation.
2. `get_player_stats(player_name)`: Fetches series batting statistics and highest scores.
3. Communicates over standard input/output (`stdio`) transport so MCP clients can spawn it as a subprocess.

Note: The match and player data below is mock/demo data for classroom exercises (no external API key needed).

Run locally (from this folder):
    python cricket_server.py
"""

import sys
from pydantic import Field
from mcp.server.fastmcp import FastMCP

# ---------------------------------------------------------------------------
# FastMCP Server Initialization
# ---------------------------------------------------------------------------
mcp = FastMCP("Cricket-Score-Server")

# ---------------------------------------------------------------------------
# Mock Data Store (Simulated Third-Party Sports API Data)
# ---------------------------------------------------------------------------
LIVE_MATCHES = {
    "india vs west indies": {
        "series": "India vs West Indies ODI Series, Sep-Oct 2026 (DEMO DATA)",
        "match": "2nd ODI",
        "status": "LIVE",
        "india": "287/6 (50 overs)",
        "west_indies": "198/5 (38.2 overs)",
        "target": 288,
        "required": "90 runs from 70 balls",
        "series_score": "India lead 1-0",
    }
}

PLAYER_STATS = {
    "shubman gill": {"team": "India", "series_runs": 164, "innings": 3, "highest": 150},
    "shai hope": {"team": "West Indies", "series_runs": 131, "innings": 2, "highest": 78},
}


# ---------------------------------------------------------------------------
# FastMCP Tool Definitions
# ---------------------------------------------------------------------------
@mcp.tool(description="Get the LIVE score of a cricket match. Use this for any question about the current score, who is winning, or runs required.")
def get_live_score(teams: str = Field(description="The two teams, e.g. 'India vs West Indies'")) -> dict:
    """Get the LIVE score of a cricket match."""
    # ① Log diagnostic trace to stderr (stdout is reserved strictly for MCP JSON-RPC frames)
    print(f"[server] get_live_score called with teams={teams!r}", file=sys.stderr)  # logs go to stderr, NEVER stdout
    # ② Normalize team names and look up the live match scorecard
    key = teams.strip().lower().replace("windies", "west indies")
    if key not in LIVE_MATCHES:
        return {"error": f"No live match found for '{teams}'. Try 'India vs West Indies'."}
    return LIVE_MATCHES[key]


@mcp.tool(description="Get a player's batting stats for the current series. Use this for questions about a specific player's runs.")
def get_player_stats(player_name: str = Field(description="Full player name, e.g. 'Shubman Gill'")) -> dict:
    """Get a player's batting stats for the current series."""
    # ① Log diagnostic trace to stderr
    print(f"[server] get_player_stats called with player_name={player_name!r}", file=sys.stderr)
    # ② Normalize player name and return series batting statistics
    key = player_name.strip().lower()
    if key not in PLAYER_STATS:
        return {"error": f"No stats found for '{player_name}'."}
    return PLAYER_STATS[key]


# ---------------------------------------------------------------------------
# Entry Point (Stdio Transport)
# ---------------------------------------------------------------------------
if __name__ == "__main__":
    mcp.run(transport="stdio")

Flask Blueprint server and route handler (mcp_evals_route) in src/python/app.py.

"""Flask server for 08-ai-mcp-evals."""

import asyncio
import importlib.util
import os
import sys
from pathlib import Path

from dotenv import find_dotenv, load_dotenv
from flask import Blueprint, Flask, jsonify, request
from flask_cors import CORS
from rate_limiter import check_rate_limit

load_dotenv(find_dotenv())

PATH_PREFIX = os.environ.get("PATH_PREFIX", "")
STATIC_DIR = Path(__file__).resolve().parents[1]
app = Flask(__name__, static_folder=str(STATIC_DIR))
CORS(app)

bp = Blueprint("main", __name__)

_CUR_DIR = Path(__file__).resolve().parent


def _load_service(filename: str, alias: str):
    # ① Resolve the numbered service file path and create its import spec
    file_path = _CUR_DIR / filename
    spec = importlib.util.spec_from_file_location(f"services.{alias}", file_path)
    if spec is None or spec.loader is None:
        raise ImportError(f"Cannot load module from {file_path}")
    # ② Register and execute the module so its functions can be called by routes
    mod = importlib.util.module_from_spec(spec)
    sys.modules[f"services.{alias}"] = mod
    spec.loader.exec_module(mod)
    return mod


mod_mcp_client = _load_service("01_mcp_client.py", "mcp_client")
mod_mcp_agent = _load_service("02_mcp_agent.py", "mcp_agent")
mod_mcp_evals = _load_service("03_mcp_evals.py", "mcp_evals")


@bp.before_request
def enforce_rate_limit():
    # ① Only rate-limit mutating POST requests
    if request.method == "POST":
        # ② Check whether this client has exceeded 30 requests per hour
        blocked, msg, retry_after = check_rate_limit(
            request, max_requests=30, window_seconds=3600
        )
        if blocked:
            # ③ Return HTTP 429 with Retry-After header when rate-limited
            resp = jsonify({"error": msg})
            resp.status_code = 429
            resp.headers["Retry-After"] = str(retry_after)
            return resp


@bp.route("/")
def index():
    # ① Read index.html and inject the runtime PATH_PREFIX into data-api-base
    with open(os.path.join(app.static_folder, "index.html"), encoding="utf-8") as f:
        html = f.read()
    html = html.replace('data-api-base=""', f'data-api-base="{PATH_PREFIX}"')
    # ② Return the patched HTML page
    return app.response_class(html, mimetype="text/html")


@bp.route("/css/<path:filename>")
def css(filename):
    return app.send_static_file(os.path.join("css", filename))


@bp.route("/js/<path:filename>")
def js(filename):
    return app.send_static_file(os.path.join("js", filename))


@bp.route("/explanation/<path:filename>")
@bp.route("/info/<path:filename>")
def explanation_pages(filename):
    return app.send_static_file(os.path.join("explanation", filename))


@bp.route("/theory/<path:filename>")
def theory_pages(filename):
    """Serve the theory documentation pages from src/theory."""
    return app.send_static_file(os.path.join("theory", filename))


@bp.route("/api/mcp-client", methods=["POST"])
def mcp_client_route():
    # ① Parse tool_name and tool_args from the JSON request body
    data = request.get_json(force=True, silent=True) or {}
    tool_name = data.get("tool_name") or "get_live_score"
    tool_args = data.get("tool_args") or {"teams": "India vs West Indies"}
    try:
        # ② Run the direct FastMCP client over stdio in 01_mcp_client.py
        res = asyncio.run(mod_mcp_client.run_mcp_client(tool_name, tool_args))
        return jsonify(res)
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


@bp.route("/api/mcp-agent", methods=["POST"])
def mcp_agent_route():
    # ① Parse user question from the JSON request body and reject empty strings
    data = request.get_json(force=True, silent=True) or {}
    question = (data.get("question") or data.get("message") or "").strip()
    if not question:
        return jsonify({"error": "Question cannot be empty"}), 400
    try:
        # ② Run the MCP-enabled Groq cricket agent in 02_mcp_agent.py
        res = asyncio.run(mod_mcp_agent.run_mcp_agent(question))
        res["result"] = res.get("answer", "")
        return jsonify(res)
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


@bp.route("/api/mcp-evals", methods=["POST"])
def mcp_evals_route():
    try:
        # ① Run the 5-case MCP evaluation suite in 03_mcp_evals.py and return scores
        res = asyncio.run(mod_mcp_evals.run_mcp_evals_suite())
        res["result"] = f"Evaluation Suite Accuracy: {res['accuracy_percentage']}% ({res['passed_cases']}/{res['total_cases']} passed)"
        return jsonify(res)
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


app.register_blueprint(bp, url_prefix=PATH_PREFIX)

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000)

Review these core evaluation principles for testing tool-calling AI agents reliably in CI and production:

1. Why Exact String Matching Fails for LLMs: Because an LLM can phrase the same factual answer in many valid ways (e.g., "India scored 287/6" vs. "India's score is 287 for 6"), naive assert answer == expected checks are brittle. Robust AI evals test behavioral invariants and factual token containment instead.
2. Dual-Layer Evaluation (Trajectory + Outcome): A complete agent eval grades both how the agent arrived at the answer (Trajectory / Tool Routing: did it invoke the right MCP tool?) and what it produced (Outcome / Grounding: does the final answer contain the ground-truth numbers returned by the tool?).
3. Why Trajectory Grading Alone Is Insufficient: An agent might call the right tool (get_live_score) and receive the right JSON payload, yet misread or omit the required number when synthesizing its final natural-language answer.
4. Why Outcome Grading Alone Is Insufficient: An agent might guess a plausible number from pretraining memory or call three unnecessary tools before answering, wasting latency and tokens. Checking tools_used catches routing regressions immediately.
5. Negative Control Cases (Testing Abstention): Including out-of-domain or general-knowledge prompts like "What is 2 + 2?" with expected_tool=None and asserting tools_used == [] verifies that the agent does not over-trigger tools when none are needed.
6. Deterministic Mock Servers for Reproducible Evals: Connecting the agent to a deterministic FastMCP server (cricket_server.py) with known mock data ensures that evaluation failures reflect genuine prompt, model, or schema bugs rather than fluctuating third-party API data.
7. Case-Insensitive Factual Containment (must_contain): Checking all(word.lower() in answer.lower() for word in case["must_contain"]) verifies that essential factual tokens ("287", "90", "164", "78", "4") appear in the response regardless of surrounding prose.
8. Trace-Driven Observability: Because run_mcp_agent() returns a structured trace list alongside answer, the eval harness can inspect [step["tool"] for step in result["trace"]] without parsing raw log output.
9. Regression Detection Across Prompt & Model Changes: Aggregating per-case booleans into a single accuracy_percentage scorecard lets engineers immediately measure whether editing a tool description in cricket_server.py or swapping models in 02_mcp_agent.py improves or degrades agent reliability.