🏏 FastMCP Cricket Agent#

MCP-Enabled LLM Agent takes the exact same cricket_server.py child process from the previous demo and places an LLM (Groq's openai/gpt-oss-20b) in the decision loop. Instead of hardcoding tool schemas in the agent file, the agent discovers whatever tools the MCP server exposes at runtime, translates their schemas into OpenAI format, and lets the model decide which tool(s) to call to answer natural-language cricket questions.

Because live match scores ("India 287/6 vs. West Indies 198/5") and series batting stats ("Shubman Gill: 164 runs") do not exist in the model's pretraining weights, the agent must route your question to get_live_score or get_player_stats over MCP, read the server's JSON reply, and synthesize a grounded answer — while recording every {tool, args, result} step in a structured trace list.

1. Zero Hardcoded Tool Schemas (Runtime Discovery + Schema Bridge)

Notice that 02_mcp_agent.py never mentions get_live_score or get_player_stats anywhere in its code! At the start of each request, it calls await session.list_tools() and passes the returned MCP tool objects through a 10-line adapter function, to_openai_format(), which maps each MCP tool's name, description, and inputSchema into OpenAI's {"type": "function", "function": {..., "parameters": t.inputSchema}} format. If you add a third tool to cricket_server.py, the agent can use it immediately without touching 02_mcp_agent.py.

2. Security Isolation Between Agent Host and MCP Server

The LLM never connects directly to the MCP server, and the MCP server subprocess never receives your GROQ_API_KEY. The agent process acts as the secure broker: it talks to Groq over HTTPS using its API key, receives a JSON tool_calls prescription, executes await session.call_tool(...) over the local stdio pipe, and feeds the result back to Groq.

3. Handling Model-Specific Quirks (call_llm Retry Logic)

OpenAI's open-weight gpt-oss models were trained with a built-in browser.search tool in their pretraining environment and occasionally attempt to call it even when it is not in the offered tools list, causing Groq to return a 400 tool_use_failed error. The helper call_llm() catches that specific error and retries up to 3 times before falling back gracefully.

4. Structured Execution Traces for Downstream Evals

Every time the agent invokes an MCP tool, it appends {"tool": call.function.name, "args": args, "result": text} to trace. This trace powers both the UI visualization and the automated test assertions in the next demo (MCP Evals).

src/python/02_mcp_agent.py and src/python/cricket_server.py connect a Groq-hosted LLM to a FastMCP server via dynamic runtime tool discovery and an OpenAI-compatible ReAct tool-calling loop:

• 02_mcp_agent.py — Spawns cricket_server.py over stdio, discovers its tools at runtime with session.list_tools(), translates them into OpenAI function-calling schemas, and runs a multi-turn LLM tool-calling loop while recording an execution trace.
• cricket_server.py — The standalone FastMCP server exposing get_live_score and get_player_stats over stdio transport.

1. MCP-Enabled LLM Agent Module (02_mcp_agent.py)

• Pydantic Payload Contracts (McpAgentRequest & McpAgentResponse):
  • McpAgentRequest specifies the input question: str = "What is the score in India vs West Indies?".
  • McpAgentResponse defines the output payload: answer: str (the LLM's final natural-language answer) and trace: list[dict] (each MCP tool call's name, arguments, and raw JSON output).
• Step ① — Model & Subprocess Configuration (MODEL, GROQ_BASE_URL, server_script, server): Configures the Groq model (defaulting to "openai/gpt-oss-20b") and OpenAI-compatible base URL ("https://api.groq.com/openai/v1"), and builds StdioServerParameters(command=sys.executable, args=[server_script]) for cricket_server.py.
• def to_openai_format(mcp_tools): Bridges MCP tool descriptors to OpenAI's function-calling schema by mapping each tool's t.name, t.description, and t.inputSchema into {"type": "function", "function": {"name": t.name, "description": t.description or "", "parameters": t.inputSchema}}.
• def call_llm(llm, kwargs, retries=2): Wraps llm.chat.completions.create(**kwargs) with retry logic (up to 3 total attempts) to recover when an open-weight model attempts to invoke an undeclared built-in tool (triggering Groq's tool_use_failed error).
• async def run_mcp_agent(question: str) -> dict: Executes the end-to-end MCP agent loop:
  • Step ② — Client & Prompt Setup: Initializes the OpenAI client pointed at Groq, creates an empty trace = [] list, and seeds messages with a system prompt (instructing the assistant to use only provided functions) and the user's question.
  • Step ③ — MCP Handshake & Runtime Tool Discovery: Opens the subprocess pipe via stdio_client(server) and ClientSession(read, write), awaits session.initialize(), and converts (await session.list_tools()).tools with to_openai_format() so the agent discovers available tools dynamically without hardcoded schemas.
  • Step ④ — Multi-Step Tool-Calling Loop (up to 5 turns): Calls call_llm() with temperature=0 and the discovered tools. When not reply.tool_calls, the model has finished gathering facts and returns {"answer": reply.content or "", "trace": trace}.
  • Step ⑤ — MCP Tool Execution & Observation Feedback: Appends the assistant's tool_calls message to messages, then iterates over each requested call: parses args = json.loads(call.function.arguments), invokes the MCP server via await session.call_tool(call.function.name, args), records {"tool": call.function.name, "args": args, "result": text} in trace, and appends a {"role": "tool", "tool_call_id": call.id, "content": text} message for the next LLM turn.

2. FastMCP Cricket Server (cricket_server.py)

• mcp = FastMCP("Cricket-Score-Server"): Exposes LIVE_MATCHES and PLAYER_STATS through two decorated MCP tools (get_live_score and get_player_stats).
• get_live_score(teams) & get_player_stats(player_name): Log diagnostic traces to sys.stderr (Step ①) so stdout stays clean for JSON-RPC frames, normalize the input query (Step ②), and return the matching dictionary or an {"error": ...} object.

3. End-to-End Agent & MCP Server Flow

1. run_mcp_agent(question) spawns cricket_server.py over stdio, completes session.initialize(), fetches the MCP tool schemas via session.list_tools(), and converts them to OpenAI tool format with to_openai_format().
2. The agent sends the conversation messages and converted tool schemas to Groq via call_llm().
3. Whenever the LLM emits tool_calls, the agent forwards each tool call over the MCP stdio session via await session.call_tool(...), appends the step to trace, and feeds the JSON tool output back into messages until the LLM produces its final grounded answer.

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_agent_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 key concepts covering how an LLM agent discovers and invokes MCP tools dynamically at runtime:

1. Zero Hardcoded Tool Schemas: Unlike traditional tool-calling agents that hardcode a static TOOLS_SCHEMA array, an MCP agent discovers tools at runtime via await session.list_tools(). Adding a new `@mcp.tool` to cricket_server.py makes it immediately available to the agent without editing 02_mcp_agent.py.
2. MCP-to-OpenAI Schema Adapter (to_openai_format): MCP tools expose name, description, and inputSchema (JSON Schema). A lightweight 10-line adapter maps t.inputSchema directly into OpenAI's {"type": "function", "function": {..., "parameters": t.inputSchema}} contract.
3. ReAct Tool-Calling Loop over MCP: The agent alternates between reasoning (calling Groq's LLM with messages and tools) and acting (executing await session.call_tool(name, args) over stdio and appending a role="tool" observation message) for up to 5 turns until the model stops requesting tools.
4. Credential & Process Security Isolation: The MCP server subprocess (cricket_server.py) never receives the GROQ_API_KEY and has no network access to the LLM. The host process (02_mcp_agent.py) acts as the security broker between the remote LLM API and the local tool server.
5. Correlating Tool Calls via tool_call_id: Every tool execution result appended to messages under role="tool" must carry the exact tool_call_id=call.id from the assistant's tool_calls request so the LLM can match observations to its calls.
6. Resilience Against Pretrained Hallucinated Tools (call_llm): Open-weight models like openai/gpt-oss-20b sometimes try to invoke pretrained internal tools (e.g., browser.search), causing Groq to return 400 tool_use_failed. Catching that specific error and retrying up to 2 times recovers cleanly.
7. Deterministic Sampling (temperature=0): Setting temperature=0 in the tool-calling loop minimizes variance in tool selection and JSON argument extraction.
8. Structured Execution Traces (trace): Recording {"tool": call.function.name, "args": args, "result": text} on every MCP tool invocation provides full observability in the UI and enables deterministic trajectory assertions in downstream evaluation suites.
9. Bounded Iteration Guard (for _ in range(5)): Capping the agent loop at 5 iterations prevents infinite tool-calling cycles if a model repeatedly requests tools without converging on a final answer.