Model Context Protocol (MCP)

Explore the requirements, architecture, and custom server implementations of the Model Context Protocol

📋 Table of Contents

• Chapter 1: Model Context Protocol (MCP) Introduction
• Chapter 2: Creating a Custom MCP Server

Chapter 1: Model Context Protocol (MCP) Introduction

As LLM applications scale, developers face a major integration problem: every database, API, and tool requires a custom, bespoke integration wrapper. The Model Context Protocol (MCP), created by Anthropic, is an open standard that acts like a universal “USB port” connecting AI agents to external data sources and tools.

1. What Problem Does MCP Solve?

Before MCP, integration was fragmented:

• The Custom Wrapper Mess: If you had 5 developer IDEs and 5 databases, you had to write 25 separate integration adapters.
• Lack of Standardization: There was no standard format defining how a database should advertise its tables, or how a web search API should register its parameters with an LLM.

MCP standardizes this interface. By defining a common protocol, any MCP-compatible Host can connect to any MCP Server instantly:

[ AI App / IDE (MCP Host) ] 
       │
       ▼ (Universal Protocol)
[ Stdio / SSE Transport ]
       │
       ▼
[ Database / GitHub / Search (MCP Server) ]

---

2. Architecture & Core Components

The Model Context Protocol divides responsibility among three distinct layers:

2.1 The Host

The orchestration application that communicates with the LLM and manages security permissions (e.g., Claude Desktop, Antigravity IDE, or your custom Python agent). The Host instantiates the clients.

2.2 The Client

A protocol component running inside the Host. It initiates connection sessions to MCP servers and translates tool schemas for the LLM.

2.3 The Server

A lightweight background process that exposes resources, prompts, and tools to the client.

• Resources: Read-only data sources (like local files, database records, or API responses).
• Prompts: Reusable prompt templates (like code review formats).
• Tools: Executable actions (like creating a GitHub issue, writing a file, or running a SQL query).

---

3. Connecting to an Existing MCP Server

MCP servers communicate with clients using transports. The most common transport is Stdio, where the host launches the server as a subprocess and communicates via standard input (stdin) and standard output (stdout).

Step 1: Install Python MCP SDK

To build MCP clients and servers, install the official SDK:

uv add mcp

Step 2: Configure Client Connection

To query an existing server (like the official GitHub MCP server), we initialize a stdio_client connection by launching the server process:

import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def run_mcp_client():
    # Configure parameters to spawn the GitHub MCP server using Node/npx
    # (Requires GITHUB_PERSONAL_ACCESS_TOKEN to be set in environment)
    server_params = StdioServerParameters(
        command="npx",
        args=["-y", "@modelcontextprotocol/server-github"],
        env={"GITHUB_PERSONAL_ACCESS_TOKEN": "your_github_token_here"}
    )
    
    # Establish stdio communication
    async with stdio_client(server_params) as (read_stream, write_stream):
        async with ClientSession(read_stream, write_stream) as session:
            # Initialize connection handshake
            await session.initialize()

Step 3: Retrieve and Execute Tools

Once connected, the client session queries the server to list its available tools and requests execution:

            # List tools registered by the GitHub server
            tools_response = await session.list_tools()
            print("Available Tools:")
            for tool in tools_response.tools:
                print(f"- {tool.name}: {tool.description}")
            
            # Execute a tool to fetch repository information
            result = await session.call_tool(
                "get_user_info",
                arguments={"username": "octocat"}
            )
            print("Output result:")
            print(result.content[0].text)

---

4. Combined Client Code Project

Below is a complete, runnable script illustrating how to connect to the GitHub MCP server to fetch repository details:

import asyncio
import os
from dotenv import load_dotenv
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

load_dotenv()

async def fetch_github_profile():
    # Make sure token is present
    token = os.getenv("GITHUB_PERSONAL_ACCESS_TOKEN")
    if not token:
        print("Error: GITHUB_PERSONAL_ACCESS_TOKEN environment variable not set.")
        return

    # Configure Stdio transport parameters for the GitHub MCP server
    server_params = StdioServerParameters(
        command="npx",
        args=["-y", "@modelcontextprotocol/server-github"],
        env={"GITHUB_PERSONAL_ACCESS_TOKEN": token}
    )

    print("Connecting to GitHub MCP Server...")
    async with stdio_client(server_params) as (read_stream, write_stream):
        async with ClientSession(read_stream, write_stream) as session:
            # Step 1: Handshake
            await session.initialize()
            print("Handshake Complete!\n")

            # Step 2: List available tools
            response = await session.list_tools()
            print("--- Available Server Tools ---")
            for t in response.tools:
                print(f"Tool: {t.name} -> {t.description[:80]}...")

            # Step 3: Call tool to fetch user profile details
            target_username = "octocat"
            print(f"\nCalling 'get_user_info' for username: '{target_username}'...")
            
            result = await session.call_tool(
                name="get_user_info",
                arguments={"username": target_username}
            )

            print("\n--- GitHub API Result ---")
            print(result.content[0].text)

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

---

5. Practice Exercises

Practice 1: Tool Parameters Audit

Explain why the client passes arguments as a dictionary (e.g. {"username": "octocat"}) during session.call_tool(), and how the server knows what parameters to expect.

Solution

• Schema Validation: During the handshake, session.list_tools() returns a list of tool objects, each containing an inputSchema (defined in JSON Schema format).
• API Contract: The server advertises exactly what keys and data types it expects (e.g., username must be a string). The client uses a standard dictionary to structure these arguments, allowing the server to validate them before executing the API request.

Back to Top

Chapter 2: Creating a Custom MCP Server

While prebuilt servers (like GitHub or PostgreSQL) are highly useful, production systems often require exposing internal company databases or custom computation engines. In this page, we build a custom FastMCP Math Server in Python and consume its tools in a separate agentic client application.

1. Defining the Custom Server using FastMCP

To build MCP servers easily, the SDK provides a high-level framework called FastMCP. FastMCP handles the JSON-RPC message serialization and transport setup automatically under the hood.

Step 1: Install FastMCP Dependency

Run in your terminal:

uv add "mcp[cli]"

Step 2: Write the Server Script (mcp_server.py)

We initialize a FastMCP instance and register tools using the @mcp.tool() decorator:

# mcp_server.py
from mcp.server.fastmcp import FastMCP

# Initialize FastMCP Server
mcp = FastMCP("MathServer")

@mcp.tool()
def add_numbers(a: float, b: float) -> float:
    """Adds two numbers together and returns the floating-point result."""
    return a + b

@mcp.tool()
def multiply_numbers(a: float, b: float) -> float:
    """Multiplies two numbers together and returns the floating-point result."""
    return a * b

if __name__ == "__main__":
    # Start stdio server loop
    mcp.run()

---

2. Consuming Custom Server Tools in an Agentic Client

Now, we build a client application that:

1. Spawns our mcp_server.py script as a subprocess stdio transport connection.
2. Queries the server’s available tools.
3. Binds those tools to a Chat Model so the LLM can invoke them.

Step 1: Initialize the Stdio Subprocess Param

We configure parameters to launch our server script:

from mcp import StdioServerParameters

server_params = StdioServerParameters(
    command="python3",
    args=["mcp_server.py"]
)

Step 2: Bind MCP Tools to ChatModel

We fetch the tool definitions from the server, format them for LangChain, and bind them:

# Retrieve tools from the active MCP session
tools_response = await session.list_tools()

# Format MCP tools for the LangChain bind_tools schema
langchain_tools = []
for tool in tools_response.tools:
    # Convert tool schemas dynamically
    langchain_tools.append({
        "name": tool.name,
        "description": tool.description,
        "parameters": tool.inputSchema
    })

# Bind tools to the LLM
llm_with_tools = llm.bind_tools(langchain_tools)

---

3. Client Implementation Script (mcp_client.py)

Here is the complete, self-contained client script that connects to our server subprocess, binds the exposed math tools to Gemini, and executes a calculation query:

# mcp_client.py
import asyncio
import os
from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

# Load API keys for LLM client
load_dotenv()

async def run_agentic_mcp():
    # Configure parameters to launch our local custom server file
    server_params = StdioServerParameters(
        command="python3",
        args=["mcp_server.py"]
    )

    print("Connecting to local custom Math MCP Server...")
    async with stdio_client(server_params) as (read_stream, write_stream):
        async with ClientSession(read_stream, write_stream) as session:
            # Step A: Handshake
            await session.initialize()
            print("Handshake Complete!\n")

            # Step B: List Tools from custom server
            response = await session.list_tools()
            print("--- Custom Server Tools Registered ---")
            for t in response.tools:
                print(f"Tool: {t.name} -> {t.description}")

            # Step C: Bind Server tools dynamically to LLM
            llm = init_chat_model("gemini-2.5-flash", model_provider="google_genai")
            
            # Format tool specifications
            tools_spec = []
            for t in response.tools:
                tools_spec.append({
                    "name": t.name,
                    "description": t.description,
                    "parameters": t.inputSchema
                })
            
            llm_with_tools = llm.bind_tools(tools_spec)

            # Step D: Invoke Agent with a math query
            query = "What is 453.25 + 98.75? Please use your calculator tools."
            print(f"\nUser Query: '{query}'")
            
            response_msg = llm_with_tools.invoke(query)
            
            # Step E: Handle Tool call
            if response_msg.tool_calls:
                call = response_msg.tool_calls[0]
                print(f"\nAgent requested tool execution: '{call['name']}' with arguments {call['args']}")
                
                # Execute tool execution request on custom MCP Server
                result = await session.call_tool(name=call['name'], arguments=call['args'])
                print(f"\nMCP Server Tool Return: {result.content[0].text}")
                
                # Feed back output to model to get final conversational response
                final_response = llm.invoke([
                    ("user", query),
                    response_msg,
                    ("user", f"Tool result was: {result.content[0].text}. Summarize the answer.")
                ])
                print(f"\nFinal Agent Response: {final_response.content}")
            else:
                print(f"\nFinal Agent Response: {response_msg.content}")

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

---

4. Practice Exercises

Practice 1: Adding a Division Tool

Modify the custom server script mcp_server.py to add a new division tool called divide_numbers(a: float, b: float) -> float. Define its docstring indicating that it divides a by b, and returns the result.

Solution

Add the following decorator and function to the server script mcp_server.py:

@mcp.tool()
def divide_numbers(a: float, b: float) -> float:
    """Divides a by b and returns the floating-point result. Raises error if b is zero."""
    if b == 0:
        raise ValueError("Cannot divide by zero.")
    return a / b