The Ultimate MCP Cheatsheet

// Claude Desktop — stdio transport (local MCP server)
{
  "mcpServers": {
    "filesystem": {
      "command": "npx",
      "args": ["-y", "@anthropic/mcp-filesystem", "/Users/me/projects"]
    },
    "github": {
      "command": "npx",
      "args": ["-y", "@anthropic/mcp-github"],
      "env": { "GITHUB_TOKEN": "ghp_xxx" }
    }
  }
}

// Step 1: Client sends initialize
{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "initialize",
  "params": {
    "protocolVersion": "2025-03-26",
    "capabilities": {
      "sampling": {},
      "roots": { "listChanged": true }
    },
    "clientInfo": {
      "name": "my-claude-app",
      "version": "1.0.0"
    }
  }
}

// Step 2: Server responds
{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "protocolVersion": "2025-03-26",
    "capabilities": {
      "tools": { "listChanged": true },
      "resources": { "subscribe": true },
      "prompts": { "listChanged": true },
      "logging": {}
    },
    "serverInfo": {
      "name": "my-mcp-server",
      "version": "0.1.0"
    }
  }
}

// Step 3: Client sends initialized notification (no id = no response expected)
{
  "jsonrpc": "2.0",
  "method": "notifications/initialized"
}

// ── Request (expects response) ──
{ "jsonrpc": "2.0", "id": 1, "method": "tools/list" }

// ── Notification (no response expected) ──
{ "jsonrpc": "2.0", "method": "notifications/initialized" }

// ── Success Response ──
{
  "jsonrpc": "2.0", "id": 1,
  "result": {
    "tools": [
      { "name": "get_weather", "description": "Get weather", "inputSchema": { ... } }
    ]
  }
}

// ── Error Response ──
{
  "jsonrpc": "2.0", "id": 2,
  "error": { "code": -32601, "message": "Method not found" }
}

# ── Basic MCP Server (stdio transport) ──
from mcp.server import Server
from mcp.server.stdio import stdio_server
import mcp.types as types
import asyncio

# Create server instance
server = Server("weather-server")

@server.list_tools()
async def list_tools() -> list[types.Tool]:
    """Declare available tools."""
    return [
        types.Tool(
            name="get_weather",
            description="Get current weather for a city",
            inputSchema={
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "City name"},
                    "units": {"type": "string", "enum": ["celsius", "fahrenheit"],
                              "default": "celsius"}
                },
                "required": ["city"]
            }
        ),
        types.Tool(
            name="get_forecast",
            description="Get 5-day weather forecast",
            inputSchema={
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "City name"}
                },
                "required": ["city"]
            }
        )
    ]

@server.call_tool()
async def call_tool(
    name: str,
    arguments: dict
) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
    """Handle tool invocations."""
    if name == "get_weather":
        city = arguments["city"]
        units = arguments.get("units", "celsius")
        # Your actual API call here
        weather_data = fetch_weather(city, units)
        return [types.TextContent(type="text", text=str(weather_data))]
    elif name == "get_forecast":
        city = arguments["city"]
        forecast = fetch_forecast(city)
        return [types.TextContent(type="text", text=str(forecast))]
    raise ValueError(f"Unknown tool: {name}")

async def main():
    async with stdio_server() as (read_stream, write_stream):
        await server.run(read_stream, write_stream, server.create_initialization_options())

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

# ── MCP Server with Streamable HTTP Transport ──
from mcp.server import Server
from mcp.server.streamable_http import streamablehttp_server
import mcp.types as types

server = Server("my-http-server")

@server.list_tools()
async def list_tools() -> list[types.Tool]:
    return [
        types.Tool(
            name="search",
            description="Search the knowledge base",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "Search query"}
                },
                "required": ["query"]
            }
        )
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "search":
        results = search_knowledge_base(arguments["query"])
        return [types.TextContent(type="text", text=results)]
    raise ValueError(f"Unknown tool: {name}")

# Run with Streamable HTTP (production-ready)
import uvicorn
from starlette.applications import Starlette
from starlette.routing import Mount

app = Starlette(
    routes=[Mount("/mcp", app=streamablehttp_server(server))]
)
uvicorn.run(app, host="0.0.0.0", port=8000)

# ── MCP Client (stdio transport) ──
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
import asyncio

async def main():
    # Define server parameters
    server_params = StdioServerParameters(
        command="python",
        args=["weather_server.py"],
        env={"API_KEY": "your-api-key"}
    )

    # Connect and interact
    async with stdio_client(server_params) as (read_stream, write_stream):
        async with ClientSession(read_stream, write_stream) as session:
            # Step 1: Initialize
            await session.initialize()
            print("✓ Connected to MCP server")

            # Step 2: List available tools
            tools_result = await session.list_tools()
            for tool in tools_result.tools:
                print(f"  Tool: {tool.name} — {tool.description}")

            # Step 3: Call a tool
            result = await session.call_tool(
                "get_weather",
                arguments={"city": "San Francisco", "units": "fahrenheit"}
            )
            print(f"  Result: {result.content[0].text}")

asyncio.run(main())

# ── MCP Client (Streamable HTTP transport) ──
from mcp.client.streamable_http import streamablehttp_client
from mcp import ClientSession
import asyncio

async def main():
    # Connect via HTTP
    async with streamablehttp_client("http://localhost:8000/mcp") as (
        read_stream, write_stream, _
    ):
        async with ClientSession(read_stream, write_stream) as session:
            await session.initialize()

            # List tools
            tools = await session.list_tools()
            print(f"Available tools: {[t.name for t in tools.tools]}")

            # Call tool
            result = await session.call_tool(
                "search",
                arguments={"query": "machine learning"}
            )
            for content in result.content:
                print(content.text)

            # List resources
            resources = await session.list_resources()
            for res in resources.resources:
                print(f"Resource: {res.uri} — {res.name}")

asyncio.run(main())

# ── Advanced Tool with Annotations, Errors, and Progress ──
from mcp.server import Server
import mcp.types as types

server = Server("advanced-server")

@server.list_tools()
async def list_tools():
    return [
        types.Tool(
            name="delete_record",
            description="Delete a record from the database (irreversible)",
            inputSchema={
                "type": "object",
                "properties": {
                    "table": {"type": "string"},
                    "id": {"type": "integer"}
                },
                "required": ["table", "id"]
            },
            annotations=types.ToolAnnotations(
                destructiveHint=True,       # Warn: destructive operation
                idempotentHint=False,       # Not idempotent
                readOnlyHint=False,         # Modifies state
                openWorldHint=False         # No external access
            )
        ),
        types.Tool(
            name="search_web",
            description="Search the web for information",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {"type": "string"}
                },
                "required": ["query"]
            },
            annotations=types.ToolAnnotations(
                readOnlyHint=True,
                openWorldHint=True          # Accesses external network
            )
        )
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "delete_record":
        # Validate before destructive action
        table = arguments["table"]
        record_id = arguments["id"]

        # Return structured error if validation fails
        if not table.isidentifier():
            return [types.TextContent(
                type="text",
                text=json.dumps({"error": "Invalid table name", "code": "VALIDATION_ERROR"})
            )]

        # Perform deletion
        await db.execute(f"DELETE FROM {table} WHERE id = {record_id}")
        return [types.TextContent(
            type="text",
            text=json.dumps({"success": True, "deleted_id": record_id})
        )]
    raise ValueError(f"Unknown tool: {name}")

# ── Server with Resources & Subscriptions ──
from mcp.server import Server
import mcp.types as types

server = Server("resource-server")

@server.list_resources()
async def list_resources():
    return [
        types.Resource(
            uri="file:///app/logs/app.log",
            name="Application Logs",
            description="Recent application log entries",
            mimeType="text/plain"
        ),
        types.Resource(
            uri="file:///app/config/settings.json",
            name="Settings",
            description="Application configuration",
            mimeType="application/json"
        ),
    ]

@server.read_resource()
async def read_resource(uri: str):
    if uri == "file:///app/logs/app.log":
        content = read_log_file()
        return [types.TextResourceContents(
            uri=uri,
            mimeType="text/plain",
            text=content
        )]
    elif uri == "file:///app/config/settings.json":
        config = load_config()
        return [types.TextResourceContents(
            uri=uri,
            mimeType="application/json",
            text=json.dumps(config, indent=2)
        )]
    raise ValueError(f"Unknown resource: {uri}")

# ── Client: Reading Resources ──
# resources = await session.list_resources()
# content = await session.read_resource("file:///app/config/settings.json")

# ── Server with Prompt Templates ──
from mcp.server import Server
import mcp.types as types

server = Server("prompt-server")

@server.list_prompts()
async def list_prompts():
    return [
        types.Prompt(
            name="code_review",
            description="Generate a code review for the given code",
            arguments=[
                types.PromptArgument(
                    name="code",
                    description="The code to review",
                    required=True
                ),
                types.PromptArgument(
                    name="language",
                    description="Programming language",
                    required=False
                )
            ]
        ),
        types.Prompt(
            name="debug_error",
            description="Help debug an error message",
            arguments=[
                types.PromptArgument(
                    name="error_message",
                    description="The error message to debug",
                    required=True
                ),
                types.PromptArgument(
                    name="stack_trace",
                    description="Optional stack trace",
                    required=False
                )
            ]
        )
    ]

@server.get_prompt()
async def get_prompt(name: str, arguments: dict | None = None):
    args = arguments or {}
    if name == "code_review":
        code = args.get("code", "")
        lang = args.get("language", "unknown")
        return types.GetPromptResult(
            messages=[
                types.PromptMessage(
                    role="user",
                    content=types.TextContent(
                        type="text",
                        text=f"Please review the following {lang} code:\n\n"
                             f"--- CODE START ---\n{code}\n--- CODE END ---\n\n"
                             f"Focus on: bugs, performance, style, and best practices."
                    )
                )
            ]
        )
    raise ValueError(f"Unknown prompt: {name}")

# ── Client: Get Prompt ──
# result = await session.get_prompt("code_review", {"code": "def add(a,b): return a+b", "language": "python"})
# for msg in result.messages:
#     print(f"[{msg.role}]: {msg.content.text}")

# ── Server: Sampling & Logging ──
from mcp.server import Server
import mcp.types as types

server = Server("sampling-server")

# Server sends log messages to client
@server.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "analyze_data":
        # Send progress notifications
        await server.request_context.meta.send_progress(
            progress_token=arguments.get("_meta", {}).get("progressToken"),
            progress=1, total=3, message="Loading data..."
        )

        data = load_data()

        await server.request_context.meta.send_progress(
            progress_token=arguments.get("_meta", {}).get("progressToken"),
            progress=2, total=3, message="Analyzing..."
        )

        # Request LLM sampling for analysis
        if server.request_context.meta.sampling:
            result = await server.request_context.meta.send_sampling(
                messages=[
                    types.SamplingMessage(
                        role="user",
                        content=types.TextContent(
                            type="text",
                            text=f"Analyze this data and provide insights:\n{data}"
                        )
                    )
                ],
                max_tokens=500
            )
            analysis = result.content.text
        else:
            analysis = "Sampling not available, using basic analysis."

        await server.request_context.meta.send_progress(
            progress_token=arguments.get("_meta", {}).get("progressToken"),
            progress=3, total=3, message="Done!"
        )

        return [types.TextContent(type="text", text=analysis)]
    raise ValueError(f"Unknown tool: {name}")

# ── Production MCP Server with Error Handling ──
from mcp.server import Server
from mcp.server.streamable_http import streamablehttp_server
import mcp.types as types
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("mcp-server")

server = Server("production-server")

# Graceful error handling for all tools
@server.call_tool()
async def call_tool(name: str, arguments: dict):
    try:
        if name == "query_database":
            # Validate inputs
            query = arguments.get("query", "")
            if not query.strip():
                return [types.TextContent(
                    type="text",
                    text='{"error": "Query cannot be empty"}'
                )]

            # Execute with timeout
            result = await asyncio.wait_for(
                execute_query(query), timeout=30.0
            )
            return [types.TextContent(type="text", text=json.dumps(result))]

        return [types.TextContent(
            type="text",
            text=f'{{"error": "Unknown tool: {name}"}}'
        )]
    except asyncio.TimeoutError:
        logger.warning(f"Tool {name} timed out")
        return [types.TextContent(
            type="text",
            text='{"error": "Request timed out after 30s"}'
        )]
    except Exception as e:
        logger.error(f"Tool {name} failed: {e}")
        return [types.TextContent(
            type="text",
            text=f'{{"error": "Internal server error"}}'
        )]

# ── Dockerfile for MCP Server ──
FROM python:3.12-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY server.py .
EXPOSE 8000

CMD ["uvicorn", "server:app", "--host", "0.0.0.0", "--port", "8000"]

# ── Debug Utilities ──

# 1. Enable debug logging
import logging
logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s [%(levelname)s] %(name)s: %(message)s'
)

# 2. Quick test: Run server manually and test with raw JSON-RPC
# Start server: python server.py
# Send via stdin (for stdio transport):
# {"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-03-26","capabilities":{},"clientInfo":{"name":"test","version":"1.0"}}}
# {"jsonrpc":"2.0","method":"notifications/initialized"}
# {"jsonrpc":"2.0","id":2,"method":"tools/list"}

# 3. MCP Inspector (interactive debugging tool)
# $ npx @anthropic/mcp-inspector
# Opens web UI at http://localhost:6274
# - Inspect all messages in real-time
# - Test tool calls manually
# - Browse resources and prompts
# - View capabilities exchange