📡 FastMCP Protocol Client#

Direct FastMCP Client (Protocol Plumbing) demystifies the Model Context Protocol (MCP) by exercising it with zero LLM involvement. Before putting an AI model in the driver's seat, this demo shows that MCP is a clean, deterministic client-server RPC protocol for discovering and invoking tools across process boundaries.

When you run the demo, 01_mcp_client.py spawns cricket_server.py as a child Python process connected via standard input/output (stdio_client) and walks through the three core MCP lifecycle calls in order:

1. Handshake (await session.initialize()): Client and server exchange protocol versions and capability metadata.
2. Tool Discovery (await session.list_tools()): The client asks the server what tools it offers and receives the auto-generated menu cards (get_live_score and get_player_stats).
3. Direct Invocation (await session.call_tool(...)): The client calls both tools directly with your inputs (e.g., "India vs West Indies" and "Shubman Gill") and returns the parsed JSON payloads.

1. The M + N Standardization Problem That MCP Solves

Without a standard protocol, connecting M AI applications (IDEs, chat UIs, custom agents) to N tool sources (databases, GitHub, Slack, sports APIs) requires M × N custom integrations. With MCP, each tool source is wrapped once as an MCP Server and each application implements an MCP Client once — reducing the ecosystem integration cost from M × N to M + N.

2. FastMCP Schema Generation on the Server

Look at cricket_server.py: there is no hand-written JSON Schema anywhere. By decorating functions with @mcp.tool(description=...) and annotating parameters with Field(description=...), FastMCP("Cricket-Score-Server") automatically synthesizes the MCP tool name, description, and JSON Schema inputSchema exposed by tools/list.

3. The Golden Rule of stdio Transport: Never Print Logs to stdout

When an MCP server runs over transport="stdio", the parent client and child server exchange newline-delimited JSON-RPC 2.0 messages over the child process's stdin and stdout pipes. If cricket_server.py called a normal print("debug...") to stdout, it would corrupt the JSON-RPC stream! That is why every diagnostic log in cricket_server.py explicitly writes to file=sys.stderr.

4. Unwrapping MCP Tool Results

An MCP call_tool() response returns a list of typed content blocks (TextContent, ImageContent, etc.). The helper _parse() reads res.content[0].text and parses the JSON string back into a Python dictionary for the API response.

src/python/01_mcp_client.py and src/python/cricket_server.py implement a complete, LLM-free Model Context Protocol (MCP) client-server interaction over standard input/output (stdio):

• 01_mcp_client.py — Spawns the MCP server as a subprocess over stdio, performs the protocol handshake, discovers registered tools and their JSON schemas, and invokes a tool directly with structured arguments.
• cricket_server.py — A standalone FastMCP server exposing two cricket data tools (get_live_score and get_player_stats) backed by an in-memory mock data store and communicating over stdio.

1. Direct MCP Client Module (01_mcp_client.py)

• Pydantic Payload Contracts (McpClientRequest, McpToolDescriptor, McpClientResponse):
  • McpClientRequest defines the default call parameters: tool_name: str = "get_live_score" and tool_args: dict = {"teams": "India vs West Indies"}.
  • McpToolDescriptor represents a single discovered tool with name: str, description: str, and input_schema: dict.
  • McpClientResponse wraps the list of discovered server_tools: list[McpToolDescriptor] and the raw string result: str returned by the invoked tool.
• Step ① — Subprocess Server Parameters (server_script & server): Resolves Path(__file__).with_name("cricket_server.py") and constructs StdioServerParameters(command=sys.executable, args=[server_script]) so the client spawns the server over stdio.
• async def run_mcp_client(tool_name: str, tool_args: dict) -> dict: Executes the 3-step MCP client protocol flow without an LLM:
  • Step ② — Open stdio Transport & Session: Enters async with stdio_client(server) as (read, write): to launch cricket_server.py and attach to its stdin/stdout streams, then opens async with ClientSession(read, write) as session: to frame JSON-RPC 2.0 messages.
  • Step ③ — STEP 1 Handshake (await session.initialize()): Sends the MCP initialize request to negotiate protocol version compatibility and server capabilities.
  • Step ④ — STEP 2 Tool Discovery (await session.list_tools()): Sends the MCP tools/list RPC request and extracts .tools, returning each registered tool's name, description, and auto-generated JSON Schema (inputSchema).
  • Step ⑤ — STEP 3 Direct Tool Call (await session.call_tool(tool_name, tool_args)): Sends the MCP tools/call RPC frame with tool_name and tool_args, receiving a response whose .content list holds typed content blocks (such as TextContent).
  • Step ⑥ — Format Return Dict: Serializes each discovered tool into {"name": t.name, "description": t.description or "", "input_schema": t.inputSchema} and extracts result.content[0].text. Exiting the async with blocks cleanly shuts down the server subprocess.

2. FastMCP Cricket Server (cricket_server.py)

• Server Instance (mcp = FastMCP("Cricket-Score-Server")): Registers tool functions, synthesizes JSON Schemas from Python type hints and Pydantic Field annotations, and handles JSON-RPC 2.0 dispatch over stdio.
• Mock Data Store (LIVE_MATCHES & PLAYER_STATS):
  • LIVE_MATCHES maps "india vs west indies" to a live scorecard dictionary containing series, match ("2nd ODI"), status ("LIVE"), team scores ("287/6 (50 overs)", "198/5 (38.2 overs)"), target (288), required ("90 runs from 70 balls"), and series_score.
  • PLAYER_STATS maps lowercase player names ("shubman gill", "shai hope") to series batting records (team, series_runs, innings, highest).
• @mcp.tool(...) def get_live_score(teams: str = Field(...)) -> dict:
  • Step ① — Diagnostic Logging to stderr: Writes trace logs with print(..., file=sys.stderr) because stdout is reserved strictly for MCP JSON-RPC wire frames.
  • Step ② — Normalization & Lookup: Normalizes teams (mapping "windies" to "west indies") and returns the matching scorecard from LIVE_MATCHES (or an {"error": ...} dict).
• @mcp.tool(...) def get_player_stats(player_name: str = Field(...)) -> dict:
  • Step ①: Logs the call to sys.stderr.
  • Step ②: Normalizes player_name.strip().lower() and returns the player's stats dictionary from PLAYER_STATS (or an {"error": ...} dict).
• Entry Point (mcp.run(transport="stdio")): Starts the JSON-RPC loop reading from sys.stdin and writing responses to sys.stdout.

3. End-to-End MCP Protocol Flow

1. run_mcp_client(tool_name, tool_args) uses stdio_client(server) to spawn cricket_server.py as a child process connected via stdin/stdout pipes.
2. ClientSession executes the three core MCP lifecycle calls in order: initialize() (handshake), list_tools() (schema discovery), and call_tool(tool_name, tool_args) (tool invocation).
3. cricket_server.py dispatches the call to get_live_score or get_player_stats, logs diagnostics to stderr, and returns the serialized JSON result over stdout alongside the discovered tool schemas.

Direct FastMCP Client Service.

"""Direct FastMCP Client Service.

Demonstrates the core Model Context Protocol (MCP) plumbing without an LLM:
1. Starts `cricket_server.py` as a subprocess and talks to it over stdio.
2. Performs the MCP protocol handshake (`session.initialize()`).
3. Lists all registered tools and their JSON schemas from the server (`tools/list`).
4. Invokes an MCP tool directly (`get_live_score` or `get_player_stats`) with structured arguments (`tools/call`).
"""

import sys
from pathlib import Path

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from pydantic import BaseModel


class McpClientRequest(BaseModel):
    """Input payload for Direct MCP Client."""

    tool_name: str = "get_live_score"
    tool_args: dict = {"teams": "India vs West Indies"}


class McpToolDescriptor(BaseModel):
    name: str
    description: str
    input_schema: dict


class McpClientResponse(BaseModel):
    """Output payload from Direct MCP Client."""

    server_tools: list[McpToolDescriptor]
    result: str


# ① Start cricket_server.py (found next to this file) over stdio using the active Python interpreter
server_script = str(Path(__file__).with_name("cricket_server.py"))
server = StdioServerParameters(command=sys.executable, args=[server_script])


async def run_mcp_client(tool_name: str, tool_args: dict) -> dict:
    """Execute the 3-step MCP client protocol flow (no LLM involved)."""
    # ② Open an MCP stdio transport pipe to the server subprocess
    async with stdio_client(server) as (read, write):
        async with ClientSession(read, write) as session:
            # ③ STEP 1: Perform the MCP protocol handshake
            await session.initialize()

            # ④ STEP 2: Discover all available tools and their JSON input schemas
            tools = (await session.list_tools()).tools

            # ⑤ STEP 3: Call one tool directly with structured arguments (no LLM involved)
            result = await session.call_tool(tool_name, tool_args)

    # ⑥ The server returns the tool's dict as JSON text (or an error message); pass that text back as-is
    return {
        "server_tools": [
            {"name": t.name, "description": t.description or "", "input_schema": t.inputSchema}
            for t in tools
        ],
        "result": result.content[0].text if result.content else "(empty response)",
    }

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_client_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 concepts to solidify how the Model Context Protocol (MCP) operates under the hood before introducing an LLM agent:

1. Model Context Protocol (MCP) Solving the M × N Problem: MCP is an open client-server standard that decouples AI hosts (M applications) from external data and tool providers (N servers), reducing ecosystem integration complexity from M × N custom connectors to M + N reusable implementations.
2. MCP Works Without an LLM: At its foundation, MCP is a deterministic JSON-RPC 2.0 wire protocol. Calling initialize(), list_tools(), and call_tool() directly from Python proves the transport and tool plumbing work independently of model inference.
3. Three-Phase MCP Session Lifecycle: Every MCP client interaction follows a strict sequence: (1) await session.initialize() to negotiate protocol versions and capabilities, (2) await session.list_tools() to discover tool names, descriptions, and JSON input schemas, and (3) await session.call_tool(name, args) to execute a tool with structured arguments.
4. Standard Input/Output (stdio) Subprocess Transport: Using StdioServerParameters(command=sys.executable, args=[server_script]) with stdio_client(server) spawns the MCP server as a local child process and communicates over its stdin and stdout pipes — requiring zero network ports or firewall configuration.
5. The Golden Rule of stdio Logging (sys.stderr Only): Because stdout carries newline-delimited JSON-RPC 2.0 frames from the server back to the client, any regular print() statement on the server would corrupt the protocol stream. Server diagnostic logs must always write to file=sys.stderr.
6. Automatic JSON Schema Synthesis with FastMCP: Decorating Python functions with @mcp.tool(description=...) and annotating parameters with Pydantic Field(description=...) lets FastMCP inspect type hints at registration time and auto-generate the JSON Schema inputSchema returned by tools/list.
7. Typed Content Blocks in Tool Responses: MCP tool executions return a response containing a list of typed content objects (result.content, such as TextContent or ImageContent), where the serialized JSON payload is accessed via result.content[0].text.
8. Process Isolation & Clean Teardown: Wrapping stdio_client(server) and ClientSession(read, write) in nested async with context managers guarantees that child server processes and OS pipes are closed deterministically as soon as the request completes.
9. Async-to-Sync Bridging in Flask: Because the official Python mcp SDK is built on anyio/asyncio coroutines while Flask route handlers are synchronous, mcp_client_route() runs asyncio.run(run_mcp_client(...)) to execute the async protocol session and return a standard JSON HTTP response.
10. Graceful Domain Error Handling: Rather than raising unhandled exceptions that crash the server subprocess, MCP tools normalize inputs (e.g., mapping "windies" to "west indies") and return structured {"error": ...} dictionaries when a requested match or player is not found.