🔌 Model Context Protocol (MCP) — Master Guide

"""Minimal MCP server with tool + resource + prompt."""
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("MyServer")

# Tool — model calls this when it needs to act
@mcp.tool()
def add_numbers(a: float, b: float) -> float:
    """Add two numbers together."""
    return a + b

# Resource — client pulls this as context
@mcp.resource("config://settings")
def get_settings() -> str:
    """Current application settings."""
    return '{"max_items": 100, "debug": false}'

# Prompt — user invokes this as a template
@mcp.prompt()
def analyze(topic: str, depth: str = "brief") -> str:
    """Generate an analysis prompt."""
    return f"Analyze '{topic}' with {depth} depth."

if __name__ == "__main__":
    mcp.run()  # stdio by default

# Run: python server.py
# Test: npx @modelcontextprotocol/inspector python server.py

"""Production MCP server over Streamable HTTP."""
import os
from mcp.server.fastmcp import FastMCP
from pydantic import BaseModel, validator

mcp = FastMCP(
    "ProductionServer",
    stateless_http=True,   # enables horizontal scaling
    json_response=True,    # load-balancer friendly
)

class SearchParams(BaseModel):
    query: str
    max_results: int = 10

    @validator('query')
    def not_empty(cls, v):
        if not v.strip():
            raise ValueError("Query cannot be empty")
        return v.strip()

    @validator('max_results')
    def reasonable_limit(cls, v):
        return min(v, 100)  # hard cap

@mcp.tool(annotations={"readOnlyHint": True})
def search(params: SearchParams) -> list[dict]:
    """Search the knowledge base. Read-only."""
    # Your search logic here
    return [{"id": 1, "content": f"Result for: {params.query}"}]

if __name__ == "__main__":
    mcp.run(
        transport="streamable-http",
        host="0.0.0.0",
        port=int(os.getenv("PORT", 8000)),
    )

# Deploy: docker run -p 8000:8000 -e API_KEY=... your-image
# Client connects to: http://your-host:8000/mcp

"""Connect to an MCP server and call its tools."""
import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def main():
    # Connect to a local stdio server
    server_params = StdioServerParameters(
        command="python3",
        args=["server.py"],
        env={"API_KEY": "your-key"},
    )

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # Step 1: Initialize (capability handshake)
            await session.initialize()

            # Step 2: Discover what tools are available
            tools = await session.list_tools()
            print("Available tools:")
            for tool in tools.tools:
                print(f"  - {tool.name}: {tool.description}")

            # Step 3: Call a specific tool
            result = await session.call_tool(
                "add_numbers",
                arguments={"a": 10, "b": 32}
            )
            print(f"Result: {result.content[0].text}")

            # Step 4: Read a resource
            resource = await session.read_resource("config://settings")
            print(f"Config: {resource.contents[0].text}")

asyncio.run(main())

"""LangGraph agent that uses MCP tools via langchain-mcp-adapters."""
# pip install langchain-mcp-adapters langgraph langchain-anthropic

import asyncio
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.prebuilt import create_react_agent
from langchain_anthropic import ChatAnthropic

async def run_agent_with_mcp(query: str):
    # Connect to one or more MCP servers
    async with MultiServerMCPClient({
        "filesystem": {
            "command": "uvx",
            "args": ["mcp-server-filesystem", "/tmp"],
            "transport": "stdio",
        },
        "web-search": {
            "url": "http://localhost:8001/mcp",
            "transport": "streamable_http",
        },
    }) as client:
        # Get all tools from all connected servers
        tools = client.get_tools()
        print(f"Loaded {len(tools)} tools from MCP servers")
        for t in tools:
            print(f"  [{t.name}] {t.description[:60]}...")

        # Create a LangGraph agent with these tools
        model = ChatAnthropic(model="claude-sonnet-4-6")
        agent = create_react_agent(model, tools)

        # Run it
        result = await agent.ainvoke({
            "messages": [{"role": "user", "content": query}]
        })
        return result["messages"][-1].content

# Usage:
# answer = asyncio.run(run_agent_with_mcp("Search for recent LangGraph docs"))
# print(answer)

"""Production tool with validation, rate limiting, audit log."""
import time
import logging
from collections import defaultdict
from functools import wraps
from mcp.server.fastmcp import FastMCP
from pydantic import BaseModel, validator

mcp = FastMCP("SecureServer")
logger = logging.getLogger("mcp_audit")

# Simple in-memory rate limiter
_call_counts = defaultdict(list)
RATE_LIMIT = 10  # calls per minute

def rate_limit(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        key = func.__name__
        now = time.time()
        # Clean calls older than 60 seconds
        _call_counts[key] = [t for t in _call_counts[key] if now - t < 60]
        if len(_call_counts[key]) >= RATE_LIMIT:
            raise RuntimeError(f"Rate limit exceeded for {key}: {RATE_LIMIT}/min")
        _call_counts[key].append(now)
        return func(*args, **kwargs)
    return wrapper

class WriteParams(BaseModel):
    content: str
    filename: str

    @validator('filename')
    def safe_filename(cls, v):
        # Prevent path traversal
        if '..' in v or '/' in v or '\\' in v:
            raise ValueError("Invalid filename — no path separators")
        if not v.endswith(('.txt', '.md', '.json')):
            raise ValueError("Only .txt/.md/.json allowed")
        return v

    @validator('content')
    def reasonable_length(cls, v):
        if len(v) > 10_000:
            raise ValueError("Content too large (max 10KB)")
        return v

@mcp.tool(annotations={"destructiveHint": False, "idempotentHint": False})
@rate_limit
def write_note(params: WriteParams) -> dict:
    """Write a note file. Validated, rate-limited, audit-logged."""
    logger.info(f"AUDIT: write_note called | file={params.filename} | size={len(params.content)}")
    path = f"/tmp/notes/{params.filename}"
    with open(path, 'w') as f:
        f.write(params.content)
    return {"status": "written", "path": path, "bytes": len(params.content)}

if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    mcp.run()

"""Test your MCP server's tools directly — no client needed."""
# pip install pytest pytest-asyncio mcp
# pytest test_server.py -v

import pytest
from mcp.server.fastmcp import FastMCP

# ── Import your server ──────────────────────────────────────
from server import mcp  # your FastMCP instance

# ── Unit test individual tool functions ─────────────────────
def test_add_numbers():
    """Test the tool function directly (fastest)."""
    from server import add_numbers
    assert add_numbers(2, 3) == 5
    assert add_numbers(-1, 1) == 0
    assert add_numbers(0.1, 0.2) == pytest.approx(0.3)

def test_add_numbers_validation():
    """Test input validation."""
    from server import add_numbers
    with pytest.raises((TypeError, ValueError)):
        add_numbers("not", "numbers")

# ── Integration test via MCP session ───────────────────────
@pytest.mark.asyncio
async def test_tool_via_mcp_session():
    """Test through the full MCP protocol stack."""
    from mcp import ClientSession
    from mcp.server.fastmcp.testing import create_test_client

    async with create_test_client(mcp) as client:
        # List tools
        tools = await client.list_tools()
        tool_names = [t.name for t in tools.tools]
        assert "add_numbers" in tool_names

        # Call the tool
        result = await client.call_tool(
            "add_numbers", {"a": 10, "b": 20}
        )
        assert not result.isError
        assert "30" in result.content[0].text

@pytest.mark.asyncio
async def test_resource():
    from mcp.server.fastmcp.testing import create_test_client
    async with create_test_client(mcp) as client:
        result = await client.read_resource("config://settings")
        assert result.contents[0].text is not None

# Dockerfile for your MCP server
FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
EXPOSE 8000
CMD ["python", "server.py"]

# requirements.txt
# mcp>=1.0.0
# fastmcp>=2.0.0  # standalone v2 (more features)
# pydantic>=2.0.0

# ── Run it ──────────────────────────────────────────────────
# docker build -t my-mcp-server .
# docker run -p 8000:8000 -e API_KEY=... my-mcp-server

# ── Claude Desktop config (~/.config/claude/claude_desktop_config.json) ──
# For a LOCAL server (stdio):
{
  "mcpServers": {
    "my-local-server": {
      "command": "python3",
      "args": ["/absolute/path/to/server.py"],
      "env": {"API_KEY": "your-key-here"}
    }
  }
}

# For a REMOTE server (HTTP) — Claude Desktop ≥ 0.8:
{
  "mcpServers": {
    "my-remote-server": {
      "type": "streamable-http",
      "url": "https://your-server.com/mcp",
      "headers": {"Authorization": "Bearer your-token"}
    }
  }
}

"""Track MCP tool calls in your LangGraph telemetry pipeline.
Adds per-tool span tracking to the tracer from the LangGraph kit."""

import time
from contextlib import asynccontextmanager
from langchain_mcp_adapters.client import MultiServerMCPClient
from telemetry.tracer import get_tracer  # from our LangGraph kit

class TracedMCPClient:
    """Wraps MultiServerMCPClient to emit telemetry spans per tool call."""

    def __init__(self, server_configs: dict):
        self._configs = server_configs
        self._client = None

    async def __aenter__(self):
        self._client = MultiServerMCPClient(self._configs)
        await self._client.__aenter__()
        return self

    async def __aexit__(self, *args):
        await self._client.__aexit__(*args)

    def get_traced_tools(self):
        """Return tools wrapped with telemetry spans."""
        tracer = get_tracer()
        raw_tools = self._client.get_tools()
        traced = []

        for tool in raw_tools:
            original_run = tool._run

            def make_traced(t, orig):
                def traced_run(*args, **kwargs):
                    with tracer.span(f"mcp_tool:{t.name}") as span:
                        t0 = time.perf_counter()
                        try:
                            result = orig(*args, **kwargs)
                            latency = (time.perf_counter() - t0) * 1000
                            span.record(0, 0, latency, status="ok")
                            return result
                        except Exception as e:
                            span.record(0, 0, 0, status="error", error=str(e))
                            raise
                return traced_run

            tool._run = make_traced(tool, original_run)
            traced.append(tool)

        return traced

# Usage with LangGraph agent:
# async with TracedMCPClient({"github": {...}}) as client:
#     tools = client.get_traced_tools()
#     agent = create_react_agent(model, tools)
#     result = await agent.ainvoke({"messages": [...]})