{
  "_comment": "Canonical fast-moving facts the mcp-ops skill encodes. scripts/check-mcp-facts.py asserts the skill prose still names these SDK packages + spec URL and still carries a dated currency note (--offline), and probes the npm/PyPI registries + the spec URL for drift (--live). Edit deliberately: a change here is a skill-content decision, not housekeeping.",
  "schema": "claude-mods.mcp-ops.facts/v1",
  "as_of": "2026-07-05",
  "spec": {
    "url": "https://modelcontextprotocol.io/specification/latest",
    "prose_token": "modelcontextprotocol.io/specification",
    "_comment": "The MCP specification URL cited in See Also (moved off the spec. subdomain, which stopped resolving by 2026-07). --offline asserts the token is named in the skill prose; --live expects the URL to answer HTTP 200 (after redirects)."
  },
  "packages": {
    "@modelcontextprotocol/sdk": {
      "registry": "npm",
      "role": "TypeScript SDK",
      "sampled_major": 1,
      "track_major": true,
      "prose_token": "@modelcontextprotocol/sdk",
      "_comment": "The official TS SDK. --live flags drift if the latest dist-tag leaves major 1 (a 2.x would be an API-break release warranting a skill review)."
    },
    "@modelcontextprotocol/inspector": {
      "registry": "npm",
      "role": "MCP Inspector (dev/debug UI)",
      "sampled_major": 0,
      "track_major": false,
      "prose_token": "@modelcontextprotocol/inspector",
      "_comment": "The interactive inspector invoked via `npx`. Only resolution is checked live (0.x versioning is chatty); a 404 means it was renamed/removed."
    },
    "mcp": {
      "registry": "pypi",
      "role": "Python SDK (FastMCP ships inside as mcp.server.fastmcp)",
      "sampled_major": 1,
      "track_major": true,
      "prose_token": "mcp[cli]",
      "_comment": "The official Python SDK on PyPI, installed via `uv add mcp[cli]`. --live flags drift if the latest release leaves major 1."
    },
    "fastmcp": {
      "registry": "pypi",
      "role": "FastMCP high-level framework (gofastmcp.com)",
      "sampled_major": 3,
      "track_major": true,
      "prose_token": "fastmcp",
      "_comment": "The standalone FastMCP package, documented at major 3 (constructor identity-only, async ctx state, fastmcp meta namespace). Named in prose as FastMCP and via gofastmcp.com; matched case-insensitively."
    }
  }
}

# Resources and Prompts

## Resource Overview

Resources provide **read-only data** to the LLM. They are identified by URIs and can be static or dynamic.

```
Resource URI format:  scheme://authority/path
Examples:
  file:///workspace/readme.md
  db://myapp/users
  config://settings
  api://github/repos/user/repo
```

## Resource URIs

### URI Scheme Design

| Scheme | Use Case | Example |
|--------|----------|---------|
| `file://` | Local filesystem | `file:///workspace/src/main.py` |
| `db://` | Database objects | `db://myapp/tables/users` |
| `config://` | Configuration | `config://app/settings` |
| `api://` | External API data | `api://github/repos` |
| `schema://` | Schema definitions | `schema://db/users` |
| `log://` | Log files/entries | `log://app/errors/today` |
| `docs://` | Documentation | `docs://api/endpoints` |

### URI Templates

URI templates allow parameterized resources:

```python
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("resource-server")

# Static URI - single resource
@mcp.resource("config://app/settings")
def get_settings() -> str:
    """Return application settings."""
    return json.dumps(load_settings(), indent=2)

# URI template - parameterized resource
@mcp.resource("db://app/tables/{table_name}")
def get_table_data(table_name: str) -> str:
    """Return data from a database table."""
    if table_name not in ALLOWED_TABLES:
        raise ValueError(f"Table {table_name} not accessible")
    rows = db.query(f"SELECT * FROM {table_name} LIMIT 100")
    return json.dumps(rows, default=str, indent=2)

# Template with multiple parameters
@mcp.resource("api://github/{owner}/{repo}/info")
def get_repo_info(owner: str, repo: str) -> str:
    """Return GitHub repository information."""
    resp = httpx.get(f"https://api.github.com/repos/{owner}/{repo}")
    return resp.text
```

### TypeScript Resources

```typescript
// Static resource
server.resource("settings", "config://app/settings", async (uri) => ({
  contents: [{
    uri: uri.href,
    mimeType: "application/json",
    text: JSON.stringify(loadSettings(), null, 2),
  }],
}));

// Resource template
server.resource(
  "table_data",
  new ResourceTemplate("db://app/tables/{table_name}", { list: undefined }),
  async (uri, { table_name }) => ({
    contents: [{
      uri: uri.href,
      mimeType: "application/json",
      text: JSON.stringify(await db.query(`SELECT * FROM ${table_name} LIMIT 100`)),
    }],
  })
);
```

## Static Resources

Resources backed by files, configs, or constant data:

```python
import os
import json

@mcp.resource("config://app/environment")
def get_environment() -> str:
    """Return current environment configuration."""
    return json.dumps({
        "node_env": os.environ.get("NODE_ENV", "development"),
        "debug": os.environ.get("DEBUG", "false"),
        "version": "1.0.0",
    }, indent=2)

@mcp.resource("docs://api/openapi")
def get_openapi_spec() -> str:
    """Return the OpenAPI specification."""
    with open("openapi.yaml") as f:
        return f.read()

@mcp.resource("schema://database/migrations")
def get_migration_status() -> str:
    """Return database migration status."""
    migrations = get_applied_migrations()
    pending = get_pending_migrations()
    return json.dumps({
        "applied": [m.name for m in migrations],
        "pending": [m.name for m in pending],
        "current_version": migrations[-1].version if migrations else "none",
    }, indent=2)
```

## Dynamic Resources

Resources that fetch data on-demand from external sources:

```python
import httpx

@mcp.resource("api://weather/{city}")
async def get_weather(city: str) -> str:
    """Return current weather for a city."""
    async with httpx.AsyncClient() as client:
        resp = await client.get(
            "https://api.weather.example.com/current",
            params={"city": city},
            timeout=10.0,
        )
        return resp.text

@mcp.resource("db://app/stats")
def get_app_stats() -> str:
    """Return application statistics."""
    stats = {
        "total_users": db.count("users"),
        "active_sessions": db.count("sessions", {"active": True}),
        "requests_today": db.count("requests", {"date": today()}),
        "error_rate": db.query_scalar(
            "SELECT COUNT(*) FILTER (WHERE status >= 500)::float / COUNT(*) FROM requests WHERE date = $1",
            today(),
        ),
    }
    return json.dumps(stats, indent=2)

@mcp.resource("log://app/errors/recent")
def get_recent_errors() -> str:
    """Return the most recent application errors."""
    errors = db.query(
        "SELECT timestamp, level, message, stack_trace FROM logs "
        "WHERE level = 'ERROR' ORDER BY timestamp DESC LIMIT 20"
    )
    return json.dumps(errors, default=str, indent=2)
```

## MIME Types

| MIME Type | Use For | Example |
|-----------|---------|---------|
| `text/plain` | Plain text, logs | Default if unspecified |
| `application/json` | Structured data | API responses, configs |
| `text/markdown` | Formatted docs | README, documentation |
| `text/html` | Web content | Rendered pages |
| `image/png` | PNG images | Charts, screenshots (base64) |
| `image/jpeg` | JPEG images | Photos (base64) |
| `application/pdf` | PDF documents | Reports (base64) |

### Setting MIME Types

```python
# Python - FastMCP infers from return type, or specify explicitly
@mcp.resource("docs://readme", mime_type="text/markdown")
def get_readme() -> str:
    """Return the project README."""
    with open("README.md") as f:
        return f.read()

# Binary content with base64
@mcp.resource("images://logo")
def get_logo() -> bytes:
    """Return the application logo."""
    with open("logo.png", "rb") as f:
        return f.read()  # FastMCP handles base64 encoding
```

```typescript
// TypeScript - specify mimeType in contents
server.resource("readme", "docs://readme", async (uri) => ({
  contents: [{
    uri: uri.href,
    mimeType: "text/markdown",
    text: await fs.readFile("README.md", "utf-8"),
  }],
}));

// Binary content
server.resource("logo", "images://logo", async (uri) => ({
  contents: [{
    uri: uri.href,
    mimeType: "image/png",
    blob: (await fs.readFile("logo.png")).toString("base64"),
  }],
}));
```

## Resource Subscriptions

Clients can subscribe to resource changes and receive notifications:

```python
from mcp.server.fastmcp import FastMCP, Context

mcp = FastMCP("subscription-demo")

# Track subscriptions
_config_version = 0

@mcp.resource("config://app/settings")
def get_settings() -> str:
    return json.dumps(load_settings(), indent=2)

@mcp.tool()
async def update_setting(key: str, value: str, ctx: Context) -> str:
    """Update a configuration setting."""
    global _config_version
    save_setting(key, value)
    _config_version += 1

    # Notify subscribed clients that the resource changed
    await ctx.request_context.session.send_resource_updated("config://app/settings")
    return f"Updated {key} = {value}"
```

## Resource Listing

Servers expose available resources via `resources/list`:

```python
# FastMCP handles listing automatically for registered resources.
# For dynamic resources, implement custom listing:

@mcp.resource("db://app/tables/{table_name}")
def get_table(table_name: str) -> str:
    """Read data from a database table."""
    return json.dumps(db.query(f"SELECT * FROM {table_name} LIMIT 50"), default=str)

# Override resource listing to show available tables
# (FastMCP's resource template handles this automatically when
#  the template is registered with a list callback)
```

### Pagination for Large Resource Lists

When you have many resources, consider chunking or metadata:

```python
@mcp.resource("db://app/tables/{table}/page/{page}")
def get_table_page(table: str, page: str) -> str:
    """Read a page of data from a database table.

    Args:
        table: Table name
        page: Page number (1-based)
    """
    page_num = int(page)
    offset = (page_num - 1) * 50
    rows = db.query(f"SELECT * FROM {table} LIMIT 50 OFFSET {offset}")
    total = db.count(table)
    return json.dumps({
        "rows": rows,
        "page": page_num,
        "total_pages": (total + 49) // 50,
        "total_rows": total,
    }, default=str, indent=2)
```

---

## Prompt Templates

Prompts are pre-written templates that suggest how the LLM should approach a task.

### Basic Prompts

```python
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("prompt-server")

@mcp.prompt()
def code_review(language: str, filepath: str) -> str:
    """Generate a code review prompt for the given file."""
    return (
        f"Please review the following {language} code from {filepath}. "
        f"Focus on:\n"
        f"1. Bug risks and edge cases\n"
        f"2. Performance issues\n"
        f"3. Code style and readability\n"
        f"4. Security concerns\n"
        f"5. Suggestions for improvement\n"
    )

@mcp.prompt()
def explain_error(error_message: str, context: str = "") -> str:
    """Generate a prompt to explain an error message."""
    prompt = f"Explain this error message and suggest how to fix it:\n\n```\n{error_message}\n```"
    if context:
        prompt += f"\n\nContext:\n{context}"
    return prompt
```

### TypeScript Prompts

```typescript
server.prompt(
  "code_review",
  "Generate a code review prompt",
  {
    language: z.string().describe("Programming language"),
    filepath: z.string().describe("Path to the file to review"),
  },
  async ({ language, filepath }) => ({
    messages: [{
      role: "user",
      content: {
        type: "text",
        text: `Review this ${language} code from ${filepath}. Check for bugs, performance, style, and security.`,
      },
    }],
  })
);
```

### Prompt Arguments

```python
from typing import Optional

@mcp.prompt()
def sql_query_help(
    task: str,
    database_type: str = "postgresql",
    tables: Optional[str] = None,
) -> str:
    """Help write a SQL query for the given task.

    Args:
        task: What the query should do
        database_type: Target database (postgresql, mysql, sqlite)
        tables: Comma-separated list of relevant tables
    """
    prompt = f"Write a {database_type} SQL query to: {task}\n"
    if tables:
        prompt += f"\nRelevant tables: {tables}"
        prompt += "\nPlease query the table schemas first if you need to understand the structure."
    prompt += "\n\nRequirements:\n"
    prompt += "- Use parameterized queries (no string interpolation)\n"
    prompt += "- Include appropriate indexes if suggesting schema changes\n"
    prompt += "- Add comments explaining complex joins or subqueries\n"
    return prompt
```

### Multi-Turn Prompts

Prompts can include multiple messages for conversation setup:

```python
@mcp.prompt()
def debug_session(error_type: str, language: str) -> list[dict]:
    """Start a debugging session for a specific error type."""
    return [
        {
            "role": "system",
            "content": f"You are a {language} debugging expert. Help the user systematically debug their {error_type} error.",
        },
        {
            "role": "user",
            "content": (
                f"I'm encountering a {error_type} in my {language} code. "
                "Please help me debug it step by step. "
                "Start by asking me for the error message and relevant code."
            ),
        },
    ]
```

```typescript
server.prompt(
  "debug_session",
  "Start a debugging session",
  {
    error_type: z.string().describe("Type of error (e.g., TypeError, ConnectionError)"),
    language: z.string().describe("Programming language"),
  },
  async ({ error_type, language }) => ({
    messages: [
      {
        role: "assistant",
        content: {
          type: "text",
          text: `I'll help you debug your ${error_type} in ${language}. Let's work through this systematically.\n\nFirst, can you share:\n1. The full error message and stack trace\n2. The relevant code section\n3. What you've already tried`,
        },
      },
    ],
  })
);
```

### Prompts that Reference Resources

Combine prompts with resources for context-aware interactions:

```python
@mcp.resource("schema://db/{table}")
def get_table_schema(table: str) -> str:
    """Return the schema for a database table."""
    schema = db.get_schema(table)
    return json.dumps(schema, indent=2)

@mcp.prompt()
def optimize_query(table: str, slow_query: str) -> list[dict]:
    """Help optimize a slow SQL query with schema context."""
    return [
        {
            "role": "user",
            "content": [
                {
                    "type": "resource",
                    "resource": {
                        "uri": f"schema://db/{table}",
                        "text": get_table_schema(table),
                        "mimeType": "application/json",
                    },
                },
                {
                    "type": "text",
                    "text": (
                        f"This query against the `{table}` table is slow:\n\n"
                        f"```sql\n{slow_query}\n```\n\n"
                        "Please suggest optimizations, including index recommendations."
                    ),
                },
            ],
        },
    ]
```

### Guided Workflows

Use prompts to define multi-step workflows:

```python
@mcp.prompt()
def migration_workflow(source_db: str, target_db: str) -> list[dict]:
    """Guide through a database migration workflow."""
    return [
        {
            "role": "user",
            "content": (
                f"I need to migrate data from {source_db} to {target_db}. "
                "Please guide me through these steps:\n\n"
                "1. Analyze the source schema\n"
                "2. Create the target schema (with any needed transformations)\n"
                "3. Write the migration script\n"
                "4. Create validation queries to verify the migration\n"
                "5. Suggest a rollback plan\n\n"
                "Let's start with step 1."
            ),
        },
    ]

@mcp.prompt()
def api_design(service_name: str, endpoints: str = "") -> str:
    """Help design a REST API."""
    prompt = f"Help me design a REST API for the {service_name} service.\n"
    if endpoints:
        prompt += f"\nPlanned endpoints:\n{endpoints}\n"
    prompt += (
        "\nFor each endpoint, specify:\n"
        "- HTTP method and path\n"
        "- Request/response schemas\n"
        "- Authentication requirements\n"
        "- Rate limiting considerations\n"
        "- Error responses\n"
    )
    return prompt
```

## Prompt Best Practices

| Practice | Why |
|----------|-----|
| Use descriptive argument names | LLM and client UIs show argument names |
| Provide defaults for optional args | Reduces friction for common cases |
| Include structured instructions | Numbered lists guide the LLM's approach |
| Reference resources when relevant | Gives the LLM concrete data to work with |
| Keep prompts focused | One task per prompt, not multi-purpose |
| Test with real LLM conversations | Prompts that read well may not work well |

## Combining Resources and Tools

A common pattern: resources provide context, tools perform actions:

```python
# Resource: provides data for the LLM to understand
@mcp.resource("db://app/tables/{table}/schema")
def get_schema(table: str) -> str:
    """Return the schema for a database table."""
    return json.dumps(db.get_schema(table), indent=2)

@mcp.resource("db://app/tables/{table}/stats")
def get_stats(table: str) -> str:
    """Return statistics for a database table."""
    return json.dumps({
        "row_count": db.count(table),
        "size_bytes": db.table_size(table),
        "last_modified": db.last_modified(table).isoformat(),
    }, indent=2)

# Tool: performs actions using the context from resources
@mcp.tool()
def optimize_table(table: str, strategy: str = "auto") -> str:
    """Optimize a database table.

    Args:
        table: Table name to optimize
        strategy: Optimization strategy: auto, vacuum, reindex, analyze
    """
    if strategy == "auto":
        stats = json.loads(get_stats(table))
        if stats["row_count"] > 1_000_000:
            strategy = "vacuum"
        else:
            strategy = "analyze"

    result = db.optimize(table, strategy)
    return f"Optimized {table} using {strategy}: {result}"

# Prompt: guides the LLM to use resources and tools together
@mcp.prompt()
def db_health_check() -> str:
    """Run a database health check."""
    return (
        "Please check the health of the database:\n"
        "1. Read the schema for each table\n"
        "2. Check the stats for each table\n"
        "3. Identify any tables that need optimization\n"
        "4. Run optimize_table on any that need it\n"
        "5. Summarize the results\n"
    )
```

# MCP Server Architecture

## Protocol Overview

MCP uses **JSON-RPC 2.0** as its wire protocol. Every message is a JSON object with:

- **Requests**: `{"jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": {...}}`
- **Responses**: `{"jsonrpc": "2.0", "id": 1, "result": {...}}`
- **Notifications**: `{"jsonrpc": "2.0", "method": "notifications/progress", "params": {...}}` (no `id`, no response expected)

The protocol is transport-agnostic. The same JSON-RPC messages flow over stdio pipes, SSE streams, or HTTP requests.

## Server Lifecycle

```
┌──────────────┐     ┌──────────────────────┐     ┌─────────┐
│ Uninitialized │────→│    Initializing       │────→│  Ready  │
└──────────────┘     │                      │     └────┬────┘
                     │ Client sends          │          │
                     │ initialize request    │     ┌────┴────┐
                     │ Server responds with  │     │ Serving │
                     │ capabilities          │     │ requests│
                     │ Client sends          │     └────┬────┘
                     │ initialized notif     │          │
                     └──────────────────────┘     ┌────┴────┐
                                                  │Shutdown │
                                                  └─────────┘
```

### Phase 1: Initialization

Client sends `initialize` with its capabilities and protocol version. Server responds with its own capabilities.

```json
// Client → Server
{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "initialize",
  "params": {
    "protocolVersion": "2025-03-26",
    "capabilities": {},
    "clientInfo": { "name": "claude-code", "version": "1.0.0" }
  }
}

// Server → Client
{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "protocolVersion": "2025-03-26",
    "capabilities": {
      "tools": { "listChanged": true },
      "resources": { "subscribe": true, "listChanged": true },
      "prompts": { "listChanged": true },
      "logging": {}
    },
    "serverInfo": { "name": "my-server", "version": "1.0.0" }
  }
}
```

### Phase 2: Initialized Notification

Client sends `notifications/initialized` to confirm. Server transitions to ready state.

### Phase 3: Serving

Server handles requests: `tools/list`, `tools/call`, `resources/list`, `resources/read`, `prompts/list`, `prompts/get`, `completion/complete`.

### Phase 4: Shutdown

Transport closes (process exit for stdio, connection close for HTTP/SSE). Server cleans up resources.

## Capability Negotiation

Servers declare what they support during initialization:

| Capability | Meaning | Sub-capabilities |
|-----------|---------|------------------|
| `tools` | Server offers tools | `listChanged` - notify on tool list changes |
| `resources` | Server offers resources | `subscribe` - clients can subscribe to changes; `listChanged` |
| `prompts` | Server offers prompts | `listChanged` - notify on prompt list changes |
| `logging` | Server can emit log messages | (none) |
| `experimental` | Experimental features | Varies by implementation |

## FastMCP Server Setup (Python)

FastMCP is the recommended high-level API for Python MCP servers. Two variants exist:
the frozen 1.x-era version bundled in the official SDK (`from mcp.server.fastmcp import
FastMCP`, used below) and the actively developed standalone `fastmcp` package at major 3
(`from fastmcp import FastMCP`). The decorator surface shown here works on both; on
standalone FastMCP 3, transport configuration moves out of the constructor to serve time,
and `ctx.set_state()`/`ctx.get_state()` are async.

### Basic Server

```python
from mcp.server.fastmcp import FastMCP

# Create server with metadata
mcp = FastMCP(
    "my-server",
    version="1.0.0",
    description="A server that does useful things",
)

@mcp.tool()
def greet(name: str) -> str:
    """Greet a user by name."""
    return f"Hello, {name}!"

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

### Server with Dependencies

```python
from mcp.server.fastmcp import FastMCP
import httpx

mcp = FastMCP("api-server")

# Dependencies are injected per-request via FastMCP's Context
@mcp.tool()
async def fetch_data(url: str) -> str:
    """Fetch data from a URL."""
    async with httpx.AsyncClient() as client:
        resp = await client.get(url, timeout=30.0)
        resp.raise_for_status()
        return resp.text
```

### Lifespan Handlers

Use lifespan to manage resources that live for the server's entire lifetime:

```python
from contextlib import asynccontextmanager
from mcp.server.fastmcp import FastMCP

@asynccontextmanager
async def lifespan(server: FastMCP):
    """Initialize and cleanup server resources."""
    # Startup: create connection pools, load config
    db = await create_db_pool()
    server.state["db"] = db
    try:
        yield
    finally:
        # Shutdown: cleanup
        await db.close()

mcp = FastMCP("db-server", lifespan=lifespan)

@mcp.tool()
async def query_db(sql: str, ctx: Context) -> str:
    """Run a read-only SQL query."""
    db = ctx.server.state["db"]
    results = await db.fetch(sql)
    return json.dumps(results, default=str)
```

### FastMCP Context Object

The `Context` parameter gives tools access to server internals:

```python
from mcp.server.fastmcp import FastMCP, Context

mcp = FastMCP("context-demo")

@mcp.tool()
async def long_operation(items: list[str], ctx: Context) -> str:
    """Process items with progress reporting."""
    results = []
    for i, item in enumerate(items):
        await ctx.report_progress(i, len(items))
        result = await process_item(item)
        results.append(result)
        await ctx.info(f"Processed {item}")  # Log to client
    return json.dumps(results)
```

Context provides:
- `ctx.report_progress(current, total)` - send progress notifications
- `ctx.info(message)`, `ctx.debug(message)`, `ctx.warning(message)`, `ctx.error(message)` - logging
- `ctx.read_resource(uri)` - read another resource from within a tool
- `ctx.server` - access the FastMCP server instance and its state
- `ctx.request_context` - access the low-level request context and session

### Running with Different Transports

```python
# stdio (default) - for Claude Desktop / Claude Code
mcp.run()
mcp.run(transport="stdio")

# SSE - for web clients
mcp.run(transport="sse", host="0.0.0.0", port=8000)

# Streamable HTTP - for production
mcp.run(transport="streamable-http", host="0.0.0.0", port=8000)
```

## TypeScript SDK Server Setup

### Basic Server with McpServer

```typescript
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";

const server = new McpServer({
  name: "my-server",
  version: "1.0.0",
});

// Tools, resources, and prompts registered via server methods
server.tool("greet", "Greet a user", { name: z.string() }, async ({ name }) => ({
  content: [{ type: "text", text: `Hello, ${name}!` }],
}));

const transport = new StdioServerTransport();
await server.connect(transport);
```

### Low-Level Server API

For maximum control, use the `Server` class directly:

```typescript
import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import {
  CallToolRequestSchema,
  ListToolsRequestSchema,
} from "@modelcontextprotocol/sdk/types.js";

const server = new Server(
  { name: "low-level-server", version: "1.0.0" },
  { capabilities: { tools: { listChanged: true } } }
);

server.setRequestHandler(ListToolsRequestSchema, async () => ({
  tools: [
    {
      name: "greet",
      description: "Greet a user",
      inputSchema: {
        type: "object" as const,
        properties: {
          name: { type: "string", description: "User name" },
        },
        required: ["name"],
      },
    },
  ],
}));

server.setRequestHandler(CallToolRequestSchema, async (request) => {
  if (request.params.name === "greet") {
    const { name } = request.params.arguments as { name: string };
    return {
      content: [{ type: "text", text: `Hello, ${name}!` }],
    };
  }
  throw new McpError(ErrorCode.MethodNotFound, `Unknown tool: ${request.params.name}`);
});
```

### McpError for Typed Errors

```typescript
import { McpError, ErrorCode } from "@modelcontextprotocol/sdk/types.js";

// Standard JSON-RPC error codes
throw new McpError(ErrorCode.InvalidParams, "Missing required field: query");
throw new McpError(ErrorCode.MethodNotFound, "Unknown tool: foo");
throw new McpError(ErrorCode.InternalError, "Database connection failed");

// Custom error codes (use negative numbers per JSON-RPC spec)
throw new McpError(-32001, "Rate limit exceeded");
```

## Multi-Tool Server Organization

### By Domain (Recommended)

```python
# tools/files.py
from mcp.server.fastmcp import FastMCP

def register_file_tools(mcp: FastMCP):
    @mcp.tool()
    def read_file(path: str) -> str:
        """Read contents of a file."""
        with open(path) as f:
            return f.read()

    @mcp.tool()
    def write_file(path: str, content: str) -> str:
        """Write content to a file."""
        with open(path, "w") as f:
            f.write(content)
        return f"Wrote {len(content)} bytes to {path}"

    @mcp.tool()
    def list_files(directory: str) -> str:
        """List files in a directory."""
        import os
        entries = os.listdir(directory)
        return "\n".join(entries)

# tools/database.py
def register_db_tools(mcp: FastMCP):
    @mcp.tool()
    def query(sql: str) -> str:
        """Execute a read-only SQL query."""
        ...

    @mcp.tool()
    def list_tables() -> str:
        """List all database tables."""
        ...

# server.py
from mcp.server.fastmcp import FastMCP
from tools.files import register_file_tools
from tools.database import register_db_tools

mcp = FastMCP("multi-tool-server")
register_file_tools(mcp)
register_db_tools(mcp)

if __name__ == "__main__":
    mcp.run()
```

### Tool Namespacing

Prefix tool names to avoid collisions when multiple servers are active:

```python
@mcp.tool(name="myapp_search")  # Not just "search"
def search(query: str) -> str:
    ...

@mcp.tool(name="myapp_create_item")  # Not just "create"
def create_item(title: str) -> str:
    ...
```

## Middleware Patterns

### Request Logging

```python
from mcp.server.fastmcp import FastMCP
import logging

logger = logging.getLogger("mcp-server")
mcp = FastMCP("logged-server")

# Use lifespan for server-level middleware
@asynccontextmanager
async def lifespan(server: FastMCP):
    logger.info("Server starting up")
    yield
    logger.info("Server shutting down")

# For per-tool logging, use a decorator
def logged_tool(func):
    async def wrapper(*args, **kwargs):
        logger.info(f"Tool called: {func.__name__} with {kwargs}")
        try:
            result = await func(*args, **kwargs) if asyncio.iscoroutinefunction(func) else func(*args, **kwargs)
            logger.info(f"Tool {func.__name__} succeeded")
            return result
        except Exception as e:
            logger.error(f"Tool {func.__name__} failed: {e}")
            raise
    wrapper.__name__ = func.__name__
    wrapper.__doc__ = func.__doc__
    wrapper.__annotations__ = func.__annotations__
    return wrapper
```

### Rate Limiting

```python
import time
from collections import defaultdict

class RateLimiter:
    def __init__(self, max_calls: int, window_seconds: int):
        self.max_calls = max_calls
        self.window = window_seconds
        self.calls: dict[str, list[float]] = defaultdict(list)

    def check(self, key: str) -> bool:
        now = time.time()
        self.calls[key] = [t for t in self.calls[key] if now - t < self.window]
        if len(self.calls[key]) >= self.max_calls:
            return False
        self.calls[key].append(now)
        return True

rate_limiter = RateLimiter(max_calls=60, window_seconds=60)

@mcp.tool()
async def rate_limited_api_call(endpoint: str, ctx: Context) -> str:
    """Call an API with rate limiting."""
    if not rate_limiter.check("api"):
        return "Rate limit exceeded. Please wait before making more requests."
    async with httpx.AsyncClient() as client:
        resp = await client.get(f"https://api.example.com/{endpoint}")
        return resp.text
```

### Caching Middleware

```python
import time
from functools import wraps

def cached(ttl_seconds: int = 300):
    """Cache tool results for the given TTL."""
    cache: dict[str, tuple[float, str]] = {}

    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            key = f"{func.__name__}:{args}:{kwargs}"
            if key in cache:
                cached_at, result = cache[key]
                if time.time() - cached_at < ttl_seconds:
                    return result
            result = await func(*args, **kwargs) if asyncio.iscoroutinefunction(func) else func(*args, **kwargs)
            cache[key] = (time.time(), result)
            return result
        return wrapper
    return decorator

@mcp.tool()
@cached(ttl_seconds=60)
async def get_weather(city: str) -> str:
    """Get current weather for a city (cached for 60s)."""
    async with httpx.AsyncClient() as client:
        resp = await client.get(f"https://weather.api/current?city={city}")
        return resp.text
```

## Error Handling

### Python Error Handling

```python
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("robust-server")

@mcp.tool()
async def safe_operation(input: str) -> str:
    """Perform an operation with proper error handling."""
    try:
        result = await do_something(input)
        return json.dumps({"status": "success", "data": result})
    except ValueError as e:
        # User-facing error: return as tool result with error flag
        # FastMCP handles this by returning content with isError=True
        raise ValueError(f"Invalid input: {e}")
    except httpx.HTTPError as e:
        raise RuntimeError(f"API request failed: {e}")
    except Exception as e:
        # Log internally, return safe message
        logger.exception("Unexpected error in safe_operation")
        raise RuntimeError("An unexpected error occurred. Check server logs.")
```

### TypeScript Error Handling

```typescript
server.tool("safe_operation", "Do something safely", { input: z.string() }, async ({ input }) => {
  try {
    const result = await doSomething(input);
    return { content: [{ type: "text", text: JSON.stringify(result) }] };
  } catch (error) {
    // Return error as tool result (visible to LLM)
    return {
      content: [{ type: "text", text: `Error: ${error.message}` }],
      isError: true,
    };
  }
});
```

### Error Categories

| Error Type | Handling | Example |
|-----------|----------|---------|
| Invalid input | Return clear message, `isError: true` | "Missing required field: query" |
| Auth failure | Return message suggesting config check | "API key invalid. Check MY_API_KEY env var" |
| External API error | Return status + retry suggestion | "GitHub API returned 503. Try again in 30s" |
| Internal error | Log details, return safe message | "Internal error. Check server logs" |
| Timeout | Return partial results if available | "Operation timed out. Partial results: ..." |

## Logging

### Python Logging

```python
from mcp.server.fastmcp import FastMCP, Context

mcp = FastMCP("logged-server")

@mcp.tool()
async def debug_tool(data: str, ctx: Context) -> str:
    """A tool with comprehensive logging."""
    await ctx.debug(f"Received data: {data[:100]}")
    await ctx.info("Processing started")

    try:
        result = process(data)
        await ctx.info(f"Processing complete: {len(result)} items")
        return json.dumps(result)
    except Exception as e:
        await ctx.error(f"Processing failed: {e}")
        raise
```

### Log Levels

| Level | Use For |
|-------|---------|
| `debug` | Detailed diagnostic info, request/response bodies |
| `info` | Normal operations, progress updates |
| `warning` | Recoverable issues, deprecation notices |
| `error` | Failures that affect the current operation |

## Graceful Shutdown

```python
import signal
from contextlib import asynccontextmanager
from mcp.server.fastmcp import FastMCP

@asynccontextmanager
async def lifespan(server: FastMCP):
    # Startup
    db_pool = await create_pool()
    http_client = httpx.AsyncClient()
    server.state["db"] = db_pool
    server.state["http"] = http_client

    try:
        yield
    finally:
        # Cleanup: close connections, flush buffers
        await http_client.aclose()
        await db_pool.close()
        logger.info("Server shutdown complete")

mcp = FastMCP("graceful-server", lifespan=lifespan)
```

## Connection and Session Management

### Session Isolation

Each client connection gets its own session. Do not share mutable state between sessions without synchronization:

```python
# BAD: Shared mutable state without locks
results_cache = {}  # All sessions share this dict unsafely

# GOOD: Per-session state via context
@mcp.tool()
async def track_calls(ctx: Context) -> str:
    """Track how many times this session called this tool."""
    session = ctx.request_context.session
    if not hasattr(session, "call_count"):
        session.call_count = 0
    session.call_count += 1
    return f"This session has made {session.call_count} calls"

# GOOD: Shared state with proper locking
import asyncio
_lock = asyncio.Lock()
_shared_cache: dict = {}

@mcp.tool()
async def cached_lookup(key: str) -> str:
    async with _lock:
        if key not in _shared_cache:
            _shared_cache[key] = await expensive_lookup(key)
        return _shared_cache[key]
```

### Notifying Clients of Changes

When your server's available tools, resources, or prompts change at runtime:

```python
@mcp.tool()
async def enable_advanced_tools(ctx: Context) -> str:
    """Dynamically register new tools and notify the client."""
    register_advanced_tools(ctx.server)
    # Notify client that tool list has changed
    await ctx.request_context.session.send_resource_list_changed()
    return "Advanced tools enabled"
```

## Project Structure

### Python (Recommended Layout)

```
my-mcp-server/
├── pyproject.toml
├── src/
│   └── my_server/
│       ├── __init__.py
│       ├── server.py          # FastMCP instance + main()
│       ├── tools/
│       │   ├── __init__.py
│       │   ├── files.py       # File operation tools
│       │   └── api.py         # API wrapper tools
│       ├── resources/
│       │   ├── __init__.py
│       │   └── config.py      # Configuration resources
│       └── prompts/
│           ├── __init__.py
│           └── workflows.py   # Prompt templates
└── tests/
    ├── test_tools.py
    └── test_resources.py
```

### TypeScript (Recommended Layout)

```
my-mcp-server/
├── package.json
├── tsconfig.json
├── src/
│   ├── index.ts               # Server setup + transport
│   ├── tools/
│   │   ├── files.ts
│   │   └── api.ts
│   ├── resources/
│   │   └── config.ts
│   └── prompts/
│       └── workflows.ts
└── tests/
    ├── tools.test.ts
    └── resources.test.ts
```

## Claude Desktop Configuration

```json
{
  "mcpServers": {
    "my-python-server": {
      "command": "uv",
      "args": ["run", "--directory", "/path/to/my-server", "python", "-m", "my_server"],
      "env": {
        "API_KEY": "your-key",
        "DATABASE_URL": "postgresql://localhost/mydb"
      }
    },
    "my-ts-server": {
      "command": "npx",
      "args": ["tsx", "/path/to/my-server/src/index.ts"],
      "env": {
        "API_KEY": "your-key"
      }
    }
  }
}
```

## Claude Code Configuration

In `.claude/settings.json` or project settings:

```json
{
  "mcpServers": {
    "my-server": {
      "command": "uv",
      "args": ["run", "--directory", "/path/to/server", "python", "server.py"],
      "env": {
        "API_KEY": "your-key"
      }
    }
  }
}
```

Or via CLI:

```bash
claude mcp add my-server -- uv run --directory /path/to/server python server.py
```

# Testing and Debugging

## MCP Inspector

The MCP Inspector is an interactive debugging tool for testing MCP servers without a full client setup.

### Installation and Usage

```bash
# Run directly with npx (no install needed)
npx @modelcontextprotocol/inspector

# Connect to a stdio server
npx @modelcontextprotocol/inspector -- uv run python server.py

# Connect to a stdio server with env vars
npx @modelcontextprotocol/inspector -e API_KEY=sk-key -- uv run python server.py

# Connect to an SSE server
npx @modelcontextprotocol/inspector --url http://localhost:8000/sse

# Connect to a Streamable HTTP server
npx @modelcontextprotocol/inspector --url http://localhost:8000/mcp
```

### What You Can Do with Inspector

| Feature | How |
|---------|-----|
| List tools | Click "Tools" tab to see all registered tools |
| Call tools | Fill in arguments and click "Call" |
| List resources | Click "Resources" tab |
| Read resources | Click any resource to see its contents |
| List prompts | Click "Prompts" tab |
| Get prompts | Fill in arguments and see rendered prompt |
| View messages | "Messages" tab shows raw JSON-RPC traffic |
| Test notifications | See server notifications in real-time |

### Inspecting Protocol Messages

The Inspector shows raw JSON-RPC messages. Use this to verify:

- Request format matches the MCP specification
- Response content is properly structured
- Error codes and messages are correct
- Notifications are sent at the right times

```
Example Inspector message log:

→ {"jsonrpc":"2.0","id":1,"method":"initialize","params":{...}}
← {"jsonrpc":"2.0","id":1,"result":{"capabilities":{"tools":{}},...}}
→ {"jsonrpc":"2.0","method":"notifications/initialized"}
→ {"jsonrpc":"2.0","id":2,"method":"tools/list"}
← {"jsonrpc":"2.0","id":2,"result":{"tools":[{"name":"search",...}]}}
→ {"jsonrpc":"2.0","id":3,"method":"tools/call","params":{"name":"search","arguments":{"query":"test"}}}
← {"jsonrpc":"2.0","id":3,"result":{"content":[{"type":"text","text":"..."}]}}
```

## Unit Testing Tools (Python)

### Testing with pytest

```python
# test_tools.py
import pytest
import json
from unittest.mock import AsyncMock, patch, MagicMock

# Test tool functions directly (they're just functions)
from my_server.server import search_docs, create_ticket

class TestSearchDocs:
    def test_basic_search(self):
        """Test search returns formatted results."""
        with patch("my_server.server.perform_search") as mock_search:
            mock_search.return_value = [
                MagicMock(title="Doc 1", snippet="First result"),
                MagicMock(title="Doc 2", snippet="Second result"),
            ]
            result = search_docs("test query")
            assert "Doc 1" in result
            assert "Doc 2" in result
            mock_search.assert_called_once_with("test query")

    def test_empty_search(self):
        """Test search with no results."""
        with patch("my_server.server.perform_search") as mock_search:
            mock_search.return_value = []
            result = search_docs("nonexistent")
            assert result == ""

    def test_search_error(self):
        """Test search handles errors gracefully."""
        with patch("my_server.server.perform_search") as mock_search:
            mock_search.side_effect = ConnectionError("API down")
            with pytest.raises(ConnectionError):
                search_docs("test")


class TestCreateTicket:
    def test_create_ticket(self):
        """Test ticket creation returns confirmation."""
        with patch("my_server.server.api") as mock_api:
            mock_api.create.return_value = MagicMock(id=42, url="https://example.com/42")
            result = create_ticket("Bug report", "Something broke", "high")
            assert "#42" in result
            assert "https://example.com/42" in result

    def test_create_ticket_validation(self):
        """Test ticket creation validates required fields."""
        # FastMCP validates types before calling the function
        # Test the function's own validation
        with pytest.raises(ValueError):
            create_ticket("", "body", "medium")  # Empty title
```

### Testing Async Tools

```python
import pytest
import asyncio

@pytest.mark.asyncio
async def test_async_tool():
    """Test an async tool handler."""
    with patch("my_server.server.httpx.AsyncClient") as mock_client:
        mock_response = AsyncMock()
        mock_response.text = '{"data": "test"}'
        mock_response.raise_for_status = MagicMock()
        mock_client.return_value.__aenter__ = AsyncMock(return_value=mock_response)
        mock_client.return_value.__aexit__ = AsyncMock(return_value=False)

        # Better: use a real mock client
        mock_instance = AsyncMock()
        mock_instance.get.return_value = mock_response

        with patch("my_server.server.httpx.AsyncClient") as MockClient:
            MockClient.return_value.__aenter__.return_value = mock_instance
            MockClient.return_value.__aexit__.return_value = False

            result = await fetch_data("https://example.com/api")
            assert "test" in result
```

### Testing with FastMCP Test Client

```python
import pytest
from mcp.server.fastmcp import FastMCP

# Create a test server
mcp = FastMCP("test-server")

@mcp.tool()
def add(a: int, b: int) -> str:
    """Add two numbers."""
    return str(a + b)

@mcp.resource("config://test")
def get_config() -> str:
    return '{"key": "value"}'

@pytest.mark.asyncio
async def test_tool_via_mcp():
    """Test tools through the MCP protocol layer."""
    async with mcp.test_client() as client:
        # List tools
        tools = await client.list_tools()
        assert any(t.name == "add" for t in tools)

        # Call a tool
        result = await client.call_tool("add", {"a": 2, "b": 3})
        assert result[0].text == "5"

@pytest.mark.asyncio
async def test_resource_via_mcp():
    """Test resources through the MCP protocol layer."""
    async with mcp.test_client() as client:
        resources = await client.list_resources()
        assert any(r.uri == "config://test" for r in resources)

        content = await client.read_resource("config://test")
        assert '"key"' in content[0].text
```

## Unit Testing Tools (TypeScript)

### Testing with vitest

```typescript
// tools.test.ts
import { describe, it, expect, vi, beforeEach } from "vitest";
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { InMemoryTransport } from "@modelcontextprotocol/sdk/inMemory.js";
import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { z } from "zod";

describe("search tool", () => {
  let server: McpServer;
  let client: Client;

  beforeEach(async () => {
    server = new McpServer({ name: "test", version: "1.0.0" });

    server.tool("search", "Search docs", { query: z.string() }, async ({ query }) => {
      // In real code, this calls an external service
      const results = await mockSearch(query);
      return {
        content: [{ type: "text", text: results.join("\n") }],
      };
    });

    // Connect via in-memory transport
    const [clientTransport, serverTransport] = InMemoryTransport.createLinkedPair();
    await server.connect(serverTransport);

    client = new Client({ name: "test-client", version: "1.0.0" });
    await client.connect(clientTransport);
  });

  it("should return search results", async () => {
    const result = await client.callTool({ name: "search", arguments: { query: "test" } });
    expect(result.content).toHaveLength(1);
    expect(result.content[0].type).toBe("text");
  });

  it("should list tools", async () => {
    const tools = await client.listTools();
    expect(tools.tools).toHaveLength(1);
    expect(tools.tools[0].name).toBe("search");
  });
});
```

### Testing Error Handling

```typescript
describe("error handling", () => {
  it("should return isError for invalid input", async () => {
    const result = await client.callTool({
      name: "query_db",
      arguments: { sql: "DELETE FROM users" },
    });
    expect(result.isError).toBe(true);
    expect(result.content[0].text).toContain("Only SELECT queries");
  });

  it("should handle tool not found", async () => {
    await expect(
      client.callTool({ name: "nonexistent", arguments: {} })
    ).rejects.toThrow();
  });
});
```

## Integration Testing

### Full Server Integration Test (Python)

```python
import pytest
import asyncio
from mcp import ClientSession
from mcp.client.stdio import stdio_client

@pytest.mark.asyncio
async def test_server_integration():
    """Test the full server by spawning it as a subprocess."""
    async with stdio_client(
        command="uv",
        args=["run", "python", "server.py"],
        env={"API_KEY": "test-key"},
    ) as (read, write):
        async with ClientSession(read, write) as session:
            # Initialize
            await session.initialize()

            # List tools
            tools = await session.list_tools()
            tool_names = [t.name for t in tools.tools]
            assert "search" in tool_names

            # Call a tool
            result = await session.call_tool("search", {"query": "test"})
            assert len(result.content) > 0
            assert result.content[0].type == "text"

            # List resources
            resources = await session.list_resources()
            assert len(resources.resources) > 0

            # Read a resource
            content = await session.read_resource("config://app")
            assert len(content.contents) > 0
```

### Integration Test with HTTP Transport

```python
import pytest
import httpx
import subprocess
import time

@pytest.fixture(scope="module")
def server_process():
    """Start the MCP server as a subprocess."""
    proc = subprocess.Popen(
        ["uv", "run", "python", "server.py", "--transport", "streamable-http", "--port", "8765"],
        env={**os.environ, "API_KEY": "test-key"},
    )
    time.sleep(2)  # Wait for server to start
    yield proc
    proc.terminate()
    proc.wait()

@pytest.mark.asyncio
async def test_http_server(server_process):
    """Test the server via HTTP transport."""
    async with httpx.AsyncClient(base_url="http://localhost:8765") as client:
        # Initialize
        resp = await client.post("/mcp", json={
            "jsonrpc": "2.0",
            "id": 1,
            "method": "initialize",
            "params": {
                "protocolVersion": "2025-03-26",
                "capabilities": {},
                "clientInfo": {"name": "test", "version": "1.0.0"},
            },
        })
        assert resp.status_code == 200
        result = resp.json()["result"]
        assert "capabilities" in result

        # List tools
        resp = await client.post("/mcp", json={
            "jsonrpc": "2.0",
            "id": 2,
            "method": "tools/list",
        })
        tools = resp.json()["result"]["tools"]
        assert len(tools) > 0
```

## Mock Clients

### Python Mock Client

```python
from unittest.mock import AsyncMock
from mcp.types import TextContent, CallToolResult

def create_mock_session():
    """Create a mock MCP session for testing."""
    session = AsyncMock()

    # Mock tool listing
    session.list_tools.return_value = MockToolList(tools=[
        MockTool(name="search", description="Search docs", inputSchema={
            "type": "object",
            "properties": {"query": {"type": "string"}},
            "required": ["query"],
        }),
    ])

    # Mock tool calls
    async def mock_call_tool(name, arguments):
        if name == "search":
            return CallToolResult(
                content=[TextContent(type="text", text="Mock result for: " + arguments["query"])]
            )
        raise ValueError(f"Unknown tool: {name}")

    session.call_tool = AsyncMock(side_effect=mock_call_tool)
    return session
```

## Protocol Debugging

### Logging JSON-RPC Messages

```python
import logging
import json

# Enable protocol-level logging
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger("mcp.protocol")

# Custom message logger
class MessageLogger:
    def log_request(self, method: str, params: dict, id: int):
        logger.debug(f"→ [{id}] {method}: {json.dumps(params, indent=2)}")

    def log_response(self, id: int, result: dict):
        logger.debug(f"← [{id}] Result: {json.dumps(result, indent=2, default=str)[:500]}")

    def log_notification(self, method: str, params: dict):
        logger.debug(f"→ (notif) {method}: {json.dumps(params, indent=2)}")

    def log_error(self, id: int, error: dict):
        logger.error(f"← [{id}] Error: {json.dumps(error, indent=2)}")
```

### Capturing Messages for Replay

```python
import json
from datetime import datetime

class MessageCapture:
    """Capture MCP messages for debugging and replay."""

    def __init__(self, output_file: str = "mcp_capture.jsonl"):
        self.output_file = output_file
        self.messages = []

    def capture(self, direction: str, message: dict):
        entry = {
            "timestamp": datetime.utcnow().isoformat(),
            "direction": direction,  # "sent" or "received"
            "message": message,
        }
        self.messages.append(entry)
        with open(self.output_file, "a") as f:
            f.write(json.dumps(entry) + "\n")

    def replay(self) -> list[dict]:
        """Load captured messages for analysis."""
        with open(self.output_file) as f:
            return [json.loads(line) for line in f]
```

### Common Protocol Issues

| Issue | Symptom | Diagnosis |
|-------|---------|-----------|
| Missing `jsonrpc` field | Server returns parse error | Check all messages include `"jsonrpc": "2.0"` |
| Wrong method name | Method not found error | Verify against spec: `tools/call` not `tool/call` |
| Missing `id` on request | No response received | All requests need unique `id`; notifications don't |
| `params` vs `arguments` | Tool gets empty args | `tools/call` uses `params.arguments` for tool args |
| Content format wrong | Client shows raw object | Must be `[{"type": "text", "text": "..."}]` |
| Protocol version mismatch | Initialize fails | Use `"2025-03-26"` (check spec for latest) |

## Common Issues and Solutions

### Server Not Starting

```bash
# Check 1: Can you run the server directly?
uv run python server.py
# If this fails, fix the Python/dependency issues first

# Check 2: Does the command path exist?
which uv    # or: which python, which npx
# Ensure the command is on PATH

# Check 3: Are dependencies installed?
cd /path/to/server && uv pip list | grep mcp

# Check 4: Check stderr for errors (stdio servers)
# Add to server.py:
import sys
print("Server starting...", file=sys.stderr)
```

### Tool Not Appearing in Client

```python
# Check 1: Does list_tools work?
# Use MCP Inspector to verify
npx @modelcontextprotocol/inspector -- uv run python server.py

# Check 2: Is the tool registered correctly?
# Verify with a simple test:
@mcp.tool()
def test_tool() -> str:
    """A test tool that always works."""
    return "Tool is working!"

# Check 3: Invalid inputSchema
# Ensure schema is valid JSON Schema
# Common mistake: using Python types instead of JSON Schema types
# BAD: {"type": "str"}
# GOOD: {"type": "string"}
```

### Auth Failures

```python
# Check 1: Are env vars set in the CLIENT config, not just your shell?
# Claude Desktop reads env from claude_desktop_config.json, NOT your shell profile

# Check 2: Verify env vars are accessible
@mcp.tool()
def debug_env() -> str:
    """Show environment variables (for debugging only)."""
    import os
    return json.dumps({
        k: v[:5] + "..." if len(v) > 5 else v
        for k, v in os.environ.items()
        if k.startswith("MY_")  # Only show your app's vars
    })

# Check 3: Token expiration
# Add logging to token refresh:
import sys
print(f"Token expires at: {expires_at}", file=sys.stderr)
```

### Timeout Errors

```python
# Check 1: Add timeouts to all external calls
async with httpx.AsyncClient(timeout=30.0) as client:
    resp = await client.get(url)

# Check 2: Break long operations into steps
@mcp.tool()
async def process_large_dataset(dataset_id: str, ctx: Context) -> str:
    """Process a large dataset in chunks with progress."""
    chunks = get_chunks(dataset_id)
    results = []
    for i, chunk in enumerate(chunks):
        await ctx.report_progress(i, len(chunks))
        results.append(await process_chunk(chunk))
    return json.dumps({"processed": len(results)})

# Check 3: Use streaming for large responses
# Return a summary instead of full data
@mcp.tool()
def query_large_table(table: str) -> str:
    """Query a table, returning summary + sample."""
    count = db.count(table)
    sample = db.query(f"SELECT * FROM {table} LIMIT 10")
    return json.dumps({
        "total_rows": count,
        "sample": sample,
        "message": f"Showing 10 of {count} rows. Use pagination for more.",
    }, default=str)
```

### JSON Parse Errors

```python
# Check 1: Don't print to stdout in stdio servers!
# stdout IS the protocol channel. Use stderr for logging.
import sys
print("Debug info", file=sys.stderr)  # Correct
# print("Debug info")  # WRONG - corrupts protocol stream

# Check 2: Ensure tool results are serializable
@mcp.tool()
def get_data() -> str:
    result = db.query("SELECT * FROM users")
    # BAD: datetime objects aren't JSON serializable by default
    # return json.dumps(result)

    # GOOD: handle non-serializable types
    return json.dumps(result, default=str)
```

## CI Testing

### GitHub Actions

```yaml
# .github/workflows/test-mcp.yml
name: Test MCP Server
on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Install uv
        uses: astral-sh/setup-uv@v4

      - name: Install dependencies
        run: uv sync

      - name: Run unit tests
        run: uv run pytest tests/ -v

      - name: Run integration tests
        run: uv run pytest tests/integration/ -v
        env:
          API_KEY: ${{ secrets.TEST_API_KEY }}

      - name: Test server starts
        run: |
          uv run python server.py &
          SERVER_PID=$!
          sleep 3
          # Verify server is running
          kill -0 $SERVER_PID 2>/dev/null && echo "Server started successfully"
          kill $SERVER_PID

      - name: Test with MCP Inspector
        run: |
          npx @modelcontextprotocol/inspector --test -- uv run python server.py
```

### Docker-Based Testing

```dockerfile
# Dockerfile.test
FROM python:3.12-slim
WORKDIR /app
COPY . .
RUN pip install uv && uv sync
RUN uv run pytest tests/ -v
```

```yaml
# docker-compose.test.yml
services:
  test:
    build:
      context: .
      dockerfile: Dockerfile.test
    environment:
      - API_KEY=test-key
      - DATABASE_URL=postgresql://postgres:test@db:5432/testdb
    depends_on:
      - db
  db:
    image: postgres:16
    environment:
      POSTGRES_PASSWORD: test
      POSTGRES_DB: testdb
```

```bash
docker compose -f docker-compose.test.yml up --build --abort-on-container-exit
```

## Performance Testing

### Concurrent Tool Calls

```python
import pytest
import asyncio
import time

@pytest.mark.asyncio
async def test_concurrent_tool_calls():
    """Test server handles concurrent requests correctly."""
    async with mcp.test_client() as client:
        start = time.time()

        # Fire 20 concurrent tool calls
        tasks = [
            client.call_tool("search", {"query": f"test-{i}"})
            for i in range(20)
        ]
        results = await asyncio.gather(*tasks)
        elapsed = time.time() - start

        assert len(results) == 20
        assert all(len(r) > 0 for r in results)
        print(f"20 concurrent calls completed in {elapsed:.2f}s")
```

### Large Payload Handling

```python
@pytest.mark.asyncio
async def test_large_response():
    """Test server handles large responses without issues."""
    async with mcp.test_client() as client:
        result = await client.call_tool("generate_report", {
            "size": "large",  # Generates a multi-KB response
        })
        assert len(result[0].text) > 10000
        # Verify it's valid JSON
        data = json.loads(result[0].text)
        assert "report" in data

@pytest.mark.asyncio
async def test_response_size_limit():
    """Verify server truncates oversized responses."""
    async with mcp.test_client() as client:
        result = await client.call_tool("get_all_data", {})
        text = result[0].text
        # Server should paginate or truncate
        assert len(text) < 1_000_000  # Under 1MB
```

### Memory Usage

```python
import tracemalloc

@pytest.mark.asyncio
async def test_memory_usage():
    """Test that tool calls don't leak memory."""
    tracemalloc.start()

    async with mcp.test_client() as client:
        snapshot1 = tracemalloc.take_snapshot()

        # Run 100 tool calls
        for i in range(100):
            await client.call_tool("search", {"query": f"test-{i}"})

        snapshot2 = tracemalloc.take_snapshot()
        stats = snapshot2.compare_to(snapshot1, "lineno")

        # Check no single allocation grew more than 10MB
        for stat in stats[:5]:
            assert stat.size_diff < 10 * 1024 * 1024, f"Memory leak detected: {stat}"

    tracemalloc.stop()
```

## Debugging Tools

### Claude Desktop Logs

```bash
# macOS
tail -f ~/Library/Logs/Claude/mcp-server-*.log

# Windows
Get-Content "$env:APPDATA\Claude\Logs\mcp-server-*.log" -Wait

# Look for:
# - Server startup errors
# - Tool call failures
# - Transport disconnections
```

### Custom Debug Server

Add a debug tool to your server for troubleshooting:

```python
import sys
import os
import platform

@mcp.tool()
def server_debug_info() -> str:
    """Return server diagnostic information (remove in production)."""
    return json.dumps({
        "python_version": sys.version,
        "platform": platform.platform(),
        "cwd": os.getcwd(),
        "env_vars": {k: "***" for k in os.environ if k.startswith("MY_")},
        "pid": os.getpid(),
        "argv": sys.argv,
    }, indent=2)
```

### Stderr Logging for stdio Servers

Since stdout is the protocol channel, use stderr for all debugging:

```python
import sys
import logging

# Configure logging to stderr
logging.basicConfig(
    stream=sys.stderr,
    level=logging.DEBUG,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
)
logger = logging.getLogger("my-server")

@mcp.tool()
def my_tool(query: str) -> str:
    logger.debug(f"my_tool called with query={query!r}")
    try:
        result = process(query)
        logger.info(f"my_tool succeeded: {len(result)} chars")
        return result
    except Exception:
        logger.exception("my_tool failed")
        raise
```

## Error Reproduction

### Capturing and Replaying Protocol Messages

```python
import json

class ProtocolRecorder:
    """Record MCP protocol messages for reproduction."""

    def __init__(self, output_path: str = "mcp_recording.jsonl"):
        self.output_path = output_path
        self._file = open(output_path, "w")

    def record(self, direction: str, message: dict):
        self._file.write(json.dumps({
            "direction": direction,
            "message": message,
            "timestamp": time.time(),
        }) + "\n")
        self._file.flush()

    def close(self):
        self._file.close()


class ProtocolReplayer:
    """Replay recorded messages against a server."""

    def __init__(self, recording_path: str):
        with open(recording_path) as f:
            self.entries = [json.loads(line) for line in f]

    async def replay(self, session):
        """Replay recorded messages and compare responses."""
        sent = [e for e in self.entries if e["direction"] == "sent"]
        received = [e for e in self.entries if e["direction"] == "received"]

        for i, entry in enumerate(sent):
            msg = entry["message"]
            if "id" in msg:
                # It's a request
                method = msg["method"]
                params = msg.get("params", {})
                if method == "tools/call":
                    result = await session.call_tool(
                        params["name"],
                        params.get("arguments", {}),
                    )
                    # Compare with recorded response
                    expected = received[i] if i < len(received) else None
                    if expected:
                        print(f"[{method}] Match: {result == expected['message']['result']}")
```

### Minimal Reproduction Script

When filing bug reports, create a minimal reproduction:

```python
#!/usr/bin/env python3
"""Minimal reproduction for MCP issue #XXX.

Run: uv run python repro.py
Test: npx @modelcontextprotocol/inspector -- uv run python repro.py
"""
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("repro-server")

@mcp.tool()
def trigger_bug(input: str) -> str:
    """This tool demonstrates the bug."""
    # Minimal code that triggers the issue
    result = problematic_operation(input)
    return result

if __name__ == "__main__":
    mcp.run()
```

# Tool Handlers

## Tool Schema Design

Every MCP tool has an `inputSchema` that follows JSON Schema. The schema tells the LLM what arguments the tool accepts.

### Basic Schema

```python
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("tools-demo")

# FastMCP generates the schema from type hints and docstring
@mcp.tool()
def search(query: str, max_results: int = 10) -> str:
    """Search for documents matching a query.

    Args:
        query: The search query string
        max_results: Maximum number of results to return (default: 10)
    """
    results = perform_search(query, limit=max_results)
    return "\n".join(f"- {r.title}: {r.snippet}" for r in results)
```

Generated schema:

```json
{
  "name": "search",
  "description": "Search for documents matching a query.",
  "inputSchema": {
    "type": "object",
    "properties": {
      "query": {
        "type": "string",
        "description": "The search query string"
      },
      "max_results": {
        "type": "integer",
        "description": "Maximum number of results to return (default: 10)",
        "default": 10
      }
    },
    "required": ["query"]
  }
}
```

### Complex Schemas with Pydantic

```python
from pydantic import BaseModel, Field
from enum import Enum
from typing import Optional

class Priority(str, Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"

class CreateTicket(BaseModel):
    title: str = Field(description="Short summary of the issue")
    body: str = Field(description="Detailed description")
    priority: Priority = Field(default=Priority.MEDIUM, description="Ticket priority level")
    labels: list[str] = Field(default_factory=list, description="Labels to apply")
    assignee: Optional[str] = Field(default=None, description="GitHub username to assign")

@mcp.tool()
def create_ticket(ticket: CreateTicket) -> str:
    """Create a new support ticket."""
    result = api.create_ticket(
        title=ticket.title,
        body=ticket.body,
        priority=ticket.priority.value,
        labels=ticket.labels,
        assignee=ticket.assignee,
    )
    return f"Created ticket #{result.id}: {result.url}"
```

### TypeScript Schema with Zod

```typescript
import { z } from "zod";

const PriorityEnum = z.enum(["low", "medium", "high", "critical"]);

server.tool(
  "create_ticket",
  "Create a new support ticket",
  {
    title: z.string().describe("Short summary of the issue"),
    body: z.string().describe("Detailed description"),
    priority: PriorityEnum.default("medium").describe("Ticket priority level"),
    labels: z.array(z.string()).default([]).describe("Labels to apply"),
    assignee: z.string().optional().describe("GitHub username to assign"),
  },
  async ({ title, body, priority, labels, assignee }) => {
    const result = await api.createTicket({ title, body, priority, labels, assignee });
    return {
      content: [{ type: "text", text: `Created ticket #${result.id}: ${result.url}` }],
    };
  }
);
```

### Schema Best Practices

| Practice | Why | Example |
|----------|-----|---------|
| Always add `description` to every field | LLM uses descriptions to decide what values to pass | `"description": "SQL query to execute"` |
| Use `enum` for fixed choices | Constrains LLM to valid values | `"enum": ["asc", "desc"]` |
| Set sensible `default` values | Reduces required arguments | `"default": 10` |
| Mark truly required fields only | Optional fields with defaults reduce friction | Only `query` required, not `max_results` |
| Use nested objects sparingly | Flat schemas are easier for LLMs | Prefer `title, body` over `ticket: {title, body}` |
| Keep descriptions under 100 chars | Long descriptions waste context | "Search query" not "The string to use for searching..." |

## Input Validation

### Python with Pydantic (FastMCP)

FastMCP automatically validates inputs against type hints:

```python
from pydantic import Field, field_validator

@mcp.tool()
def query_database(
    sql: str,
    limit: int = Field(default=100, ge=1, le=1000),
) -> str:
    """Execute a read-only SQL query.

    Args:
        sql: SQL SELECT query to execute
        limit: Maximum rows to return (1-1000)
    """
    if not sql.strip().upper().startswith("SELECT"):
        raise ValueError("Only SELECT queries are allowed")
    results = db.execute(f"{sql} LIMIT {limit}")
    return json.dumps(results, default=str)
```

### Custom Validators

```python
from pydantic import BaseModel, field_validator

class FileReadArgs(BaseModel):
    path: str
    encoding: str = "utf-8"

    @field_validator("path")
    @classmethod
    def validate_path(cls, v: str) -> str:
        import os
        # Prevent directory traversal
        resolved = os.path.realpath(v)
        allowed_root = os.path.realpath("/workspace")
        if not resolved.startswith(allowed_root):
            raise ValueError(f"Path must be within /workspace, got: {v}")
        return resolved

    @field_validator("encoding")
    @classmethod
    def validate_encoding(cls, v: str) -> str:
        allowed = {"utf-8", "ascii", "latin-1", "utf-16"}
        if v not in allowed:
            raise ValueError(f"Encoding must be one of: {allowed}")
        return v

@mcp.tool()
def read_file(args: FileReadArgs) -> str:
    """Read a file from the workspace."""
    with open(args.path, encoding=args.encoding) as f:
        return f.read()
```

### TypeScript with Zod

```typescript
server.tool(
  "query_database",
  "Execute a read-only SQL query",
  {
    sql: z.string()
      .refine((s) => s.trim().toUpperCase().startsWith("SELECT"), {
        message: "Only SELECT queries are allowed",
      }),
    limit: z.number().int().min(1).max(1000).default(100),
  },
  async ({ sql, limit }) => {
    const results = await db.query(`${sql} LIMIT ${limit}`);
    return {
      content: [{ type: "text", text: JSON.stringify(results) }],
    };
  }
);
```

## Return Types

### Text Content (Most Common)

```python
@mcp.tool()
def get_user(user_id: str) -> str:
    """Look up a user by ID."""
    user = db.get_user(user_id)
    return json.dumps({
        "id": user.id,
        "name": user.name,
        "email": user.email,
        "created_at": user.created_at.isoformat(),
    }, indent=2)
```

### Image Content

```python
import base64

@mcp.tool()
def generate_chart(data: str, chart_type: str = "bar") -> list:
    """Generate a chart image from data."""
    # Generate chart with matplotlib
    import matplotlib.pyplot as plt
    import io

    fig, ax = plt.subplots()
    parsed = json.loads(data)
    ax.bar(parsed["labels"], parsed["values"])
    ax.set_title(parsed.get("title", "Chart"))

    buf = io.BytesIO()
    fig.savefig(buf, format="png")
    buf.seek(0)
    plt.close(fig)

    image_b64 = base64.b64encode(buf.read()).decode("utf-8")

    # Return both image and text description
    return [
        {"type": "image", "data": image_b64, "mimeType": "image/png"},
        {"type": "text", "text": f"Generated {chart_type} chart with {len(parsed['labels'])} data points."},
    ]
```

### Embedded Resources

```python
@mcp.tool()
def analyze_file(path: str) -> list:
    """Analyze a file and return results with the file content as an embedded resource."""
    content = open(path).read()
    analysis = perform_analysis(content)

    return [
        {
            "type": "resource",
            "resource": {
                "uri": f"file://{path}",
                "mimeType": "text/plain",
                "text": content,
            },
        },
        {"type": "text", "text": f"Analysis:\n{analysis}"},
    ]
```

### Multiple Content Items

```python
@mcp.tool()
def compare_files(path_a: str, path_b: str) -> list:
    """Compare two files and show differences."""
    content_a = open(path_a).read()
    content_b = open(path_b).read()
    diff = compute_diff(content_a, content_b)

    return [
        {"type": "text", "text": f"## File A: {path_a}\n```\n{content_a}\n```"},
        {"type": "text", "text": f"## File B: {path_b}\n```\n{content_b}\n```"},
        {"type": "text", "text": f"## Differences\n```diff\n{diff}\n```"},
    ]
```

## Progress Notifications

For long-running operations, report progress so the client can display updates:

```python
@mcp.tool()
async def batch_process(items: list[str], ctx: Context) -> str:
    """Process a batch of items with progress reporting."""
    results = []
    total = len(items)

    for i, item in enumerate(items):
        await ctx.report_progress(i, total)
        await ctx.info(f"Processing item {i+1}/{total}: {item}")

        try:
            result = await process_single_item(item)
            results.append({"item": item, "status": "success", "result": result})
        except Exception as e:
            results.append({"item": item, "status": "error", "error": str(e)})

    await ctx.report_progress(total, total)

    succeeded = sum(1 for r in results if r["status"] == "success")
    return json.dumps({
        "summary": f"{succeeded}/{total} items processed successfully",
        "results": results,
    }, indent=2)
```

## Error Responses

### Returning Errors to the LLM

```python
@mcp.tool()
def delete_item(item_id: str) -> str:
    """Delete an item by ID."""
    try:
        item = db.get(item_id)
        if item is None:
            # Raise to signal error - FastMCP sets isError=True
            raise ValueError(f"Item {item_id} not found")
        if item.protected:
            raise PermissionError(f"Item {item_id} is protected and cannot be deleted")
        db.delete(item_id)
        return f"Deleted item {item_id}"
    except (ValueError, PermissionError):
        raise  # Re-raise known errors for the LLM
    except Exception as e:
        raise RuntimeError(f"Failed to delete item: {e}")
```

### TypeScript Error Responses

```typescript
server.tool("delete_item", "Delete an item", { item_id: z.string() }, async ({ item_id }) => {
  try {
    const item = await db.get(item_id);
    if (!item) {
      return {
        content: [{ type: "text", text: `Item ${item_id} not found` }],
        isError: true,
      };
    }
    await db.delete(item_id);
    return {
      content: [{ type: "text", text: `Deleted item ${item_id}` }],
    };
  } catch (error) {
    return {
      content: [{ type: "text", text: `Failed to delete: ${error.message}` }],
      isError: true,
    };
  }
});
```

### Error Best Practices

| Practice | Why |
|----------|-----|
| Always include actionable message | LLM can report to user or retry differently |
| Distinguish user errors from system errors | User errors: "Invalid SQL syntax"; system: "Database unavailable" |
| Never expose stack traces | Security risk; use structured error messages |
| Return `isError: true` for failures | Clients and LLMs can distinguish success from failure |
| Log internal details server-side | Use `ctx.error()` or logging for debugging |

## Tool Composition

### Tools Calling Other Tools

```python
@mcp.tool()
async def analyze_and_fix(filepath: str, ctx: Context) -> str:
    """Analyze code for issues and apply fixes."""
    # Read the file using a resource
    content = await ctx.read_resource(f"file://{filepath}")

    # Analyze
    issues = analyze_code(content)
    if not issues:
        return "No issues found"

    # Apply fixes
    fixed = apply_fixes(content, issues)

    # Write back (side effect)
    with open(filepath, "w") as f:
        f.write(fixed)

    return f"Fixed {len(issues)} issues in {filepath}:\n" + "\n".join(
        f"- {issue.description}" for issue in issues
    )
```

### Dependency Injection

```python
from dataclasses import dataclass

@dataclass
class AppDependencies:
    db: DatabasePool
    http: httpx.AsyncClient
    cache: dict

@asynccontextmanager
async def lifespan(server: FastMCP):
    deps = AppDependencies(
        db=await create_pool(),
        http=httpx.AsyncClient(),
        cache={},
    )
    server.state["deps"] = deps
    try:
        yield
    finally:
        await deps.http.aclose()
        await deps.db.close()

mcp = FastMCP("di-server", lifespan=lifespan)

@mcp.tool()
async def search_and_cache(query: str, ctx: Context) -> str:
    """Search with caching."""
    deps: AppDependencies = ctx.server.state["deps"]

    if query in deps.cache:
        return deps.cache[query]

    results = await deps.db.fetch("SELECT * FROM docs WHERE content LIKE $1", f"%{query}%")
    formatted = json.dumps(results, default=str)
    deps.cache[query] = formatted
    return formatted
```

## Batch Operations

```python
@mcp.tool()
async def bulk_update(
    updates: list[dict],
    continue_on_error: bool = True,
    ctx: Context = None,
) -> str:
    """Apply multiple updates, optionally continuing past errors.

    Args:
        updates: List of {id, field, value} objects
        continue_on_error: If true, skip failed items and continue
    """
    results = {"succeeded": [], "failed": []}

    for i, update in enumerate(updates):
        if ctx:
            await ctx.report_progress(i, len(updates))
        try:
            db.update(update["id"], {update["field"]: update["value"]})
            results["succeeded"].append(update["id"])
        except Exception as e:
            if continue_on_error:
                results["failed"].append({"id": update["id"], "error": str(e)})
            else:
                return json.dumps({
                    "error": f"Failed on item {update['id']}: {e}",
                    "completed": results["succeeded"],
                })

    return json.dumps({
        "summary": f"{len(results['succeeded'])} succeeded, {len(results['failed'])} failed",
        **results,
    }, indent=2)
```

## Idempotency

Design tools that are safe to retry:

```python
@mcp.tool()
def upsert_config(key: str, value: str) -> str:
    """Set a configuration value (idempotent - safe to retry).

    Args:
        key: Configuration key
        value: Configuration value
    """
    # Use upsert instead of insert to make retries safe
    db.execute(
        "INSERT INTO config (key, value) VALUES ($1, $2) "
        "ON CONFLICT (key) DO UPDATE SET value = $2",
        key, value,
    )
    return f"Config {key} = {value}"

@mcp.tool()
def create_item_idempotent(idempotency_key: str, title: str, body: str) -> str:
    """Create an item with an idempotency key (safe to retry).

    Args:
        idempotency_key: Unique key for this operation (e.g., UUID)
        title: Item title
        body: Item body
    """
    existing = db.get_by_idempotency_key(idempotency_key)
    if existing:
        return f"Item already exists: #{existing.id} (idempotent match)"
    item = db.create(title=title, body=body, idempotency_key=idempotency_key)
    return f"Created item #{item.id}"
```

## Side Effects and Confirmation

### Read-Only vs Mutating Tools

```python
# Read-only: no confirmation needed
@mcp.tool()
def list_users(role: str = "all") -> str:
    """List users, optionally filtered by role."""
    users = db.list_users(role=role if role != "all" else None)
    return json.dumps(users, default=str)

# Mutating: include clear description of what will change
@mcp.tool()
def delete_user(user_id: str, confirm: bool = False) -> str:
    """Permanently delete a user and all their data.

    WARNING: This action is irreversible. Set confirm=true to proceed.

    Args:
        user_id: The user ID to delete
        confirm: Must be true to actually perform the deletion
    """
    if not confirm:
        user = db.get_user(user_id)
        return (
            f"This will permanently delete user {user.name} ({user.email}) "
            f"and {user.data_count} associated records. "
            f"Call again with confirm=true to proceed."
        )
    db.delete_user(user_id)
    return f"User {user_id} has been permanently deleted."
```

### Dry Run Pattern

```python
@mcp.tool()
def refactor_imports(directory: str, dry_run: bool = True) -> str:
    """Reorganize import statements in Python files.

    Args:
        directory: Directory to process
        dry_run: If true, show what would change without modifying files
    """
    changes = analyze_imports(directory)

    if dry_run:
        summary = "\n".join(f"  {c.file}: {c.description}" for c in changes)
        return f"Dry run - {len(changes)} files would be modified:\n{summary}\n\nRun with dry_run=false to apply."

    for change in changes:
        apply_change(change)
    return f"Applied {len(changes)} import reorganizations."
```

## Schema Evolution

### Backwards-Compatible Changes

```python
# v1: Original tool
@mcp.tool()
def search_v1(query: str) -> str:
    """Search documents."""
    ...

# v2: Added optional fields (backwards compatible)
@mcp.tool()
def search(
    query: str,
    max_results: int = 10,        # New in v2
    include_archived: bool = False, # New in v2
) -> str:
    """Search documents with optional filters.

    Args:
        query: Search query
        max_results: Maximum results to return
        include_archived: Include archived documents in results
    """
    ...

# v3: Deprecated field (still accepted, but ignored)
@mcp.tool()
def search(
    query: str,
    max_results: int = 10,
    include_archived: bool = False,
    sort_by: str = "relevance",  # New in v3
    # Deprecated: use sort_by instead
    sort_order: str | None = None,  # Deprecated in v3
) -> str:
    """Search documents.

    Args:
        query: Search query
        max_results: Maximum results to return
        include_archived: Include archived documents
        sort_by: Sort results by: relevance, date, title
        sort_order: DEPRECATED - use sort_by instead
    """
    if sort_order is not None:
        # Handle deprecated parameter gracefully
        sort_by = sort_order
    ...
```

## Real-World Tool Examples

### File System Tools

```python
import os
import stat

@mcp.tool()
def read_file(path: str, encoding: str = "utf-8") -> str:
    """Read the contents of a file.

    Args:
        path: Absolute path to the file
        encoding: File encoding (default: utf-8)
    """
    path = os.path.realpath(path)
    if not os.path.isfile(path):
        raise FileNotFoundError(f"File not found: {path}")

    file_size = os.path.getsize(path)
    if file_size > 10 * 1024 * 1024:  # 10MB limit
        raise ValueError(f"File too large ({file_size} bytes). Maximum is 10MB.")

    with open(path, encoding=encoding) as f:
        return f.read()

@mcp.tool()
def write_file(path: str, content: str, create_dirs: bool = False) -> str:
    """Write content to a file.

    Args:
        path: Absolute path to write to
        content: Content to write
        create_dirs: Create parent directories if they don't exist
    """
    path = os.path.realpath(path)
    if create_dirs:
        os.makedirs(os.path.dirname(path), exist_ok=True)
    with open(path, "w") as f:
        f.write(content)
    return f"Wrote {len(content)} bytes to {path}"

@mcp.tool()
def list_directory(path: str, show_hidden: bool = False) -> str:
    """List files and directories at the given path.

    Args:
        path: Directory path to list
        show_hidden: Include hidden files (starting with .)
    """
    path = os.path.realpath(path)
    entries = []
    for entry in sorted(os.listdir(path)):
        if not show_hidden and entry.startswith("."):
            continue
        full_path = os.path.join(path, entry)
        info = os.stat(full_path)
        entry_type = "dir" if stat.S_ISDIR(info.st_mode) else "file"
        size = info.st_size if entry_type == "file" else ""
        entries.append(f"{'[D]' if entry_type == 'dir' else '[F]'} {entry} {size}")
    return "\n".join(entries) if entries else "(empty directory)"
```

### Database Query Tool

```python
import sqlite3
import json

@mcp.tool()
def query_sqlite(
    db_path: str,
    sql: str,
    params: list = None,
) -> str:
    """Execute a read-only SQL query against a SQLite database.

    Args:
        db_path: Path to the SQLite database file
        sql: SQL query (SELECT only)
        params: Optional query parameters for parameterized queries
    """
    sql_stripped = sql.strip().upper()
    if not sql_stripped.startswith("SELECT") and not sql_stripped.startswith("WITH"):
        raise ValueError("Only SELECT and WITH (CTE) queries are allowed")

    conn = sqlite3.connect(db_path)
    conn.row_factory = sqlite3.Row
    try:
        cursor = conn.execute(sql, params or [])
        rows = [dict(row) for row in cursor.fetchall()]
        columns = [desc[0] for desc in cursor.description] if cursor.description else []
        return json.dumps({
            "columns": columns,
            "rows": rows,
            "row_count": len(rows),
        }, indent=2, default=str)
    finally:
        conn.close()
```

### API Wrapper Tool

```python
import httpx
import os

@mcp.tool()
async def github_search(
    query: str,
    search_type: str = "repositories",
    per_page: int = 10,
) -> str:
    """Search GitHub for repositories, code, or issues.

    Args:
        query: GitHub search query
        search_type: Type of search: repositories, code, issues
        per_page: Results per page (1-100)
    """
    if search_type not in ("repositories", "code", "issues"):
        raise ValueError(f"Invalid search type: {search_type}")
    if not 1 <= per_page <= 100:
        raise ValueError("per_page must be between 1 and 100")

    token = os.environ.get("GITHUB_TOKEN")
    headers = {"Accept": "application/vnd.github.v3+json"}
    if token:
        headers["Authorization"] = f"Bearer {token}"

    async with httpx.AsyncClient() as client:
        resp = await client.get(
            f"https://api.github.com/search/{search_type}",
            params={"q": query, "per_page": per_page},
            headers=headers,
            timeout=30.0,
        )
        resp.raise_for_status()
        data = resp.json()

    items = data.get("items", [])
    if search_type == "repositories":
        results = [
            f"- [{item['full_name']}]({item['html_url']}) "
            f"({item['stargazers_count']} stars): {item.get('description', 'No description')}"
            for item in items
        ]
    elif search_type == "issues":
        results = [
            f"- [{item['title']}]({item['html_url']}) "
            f"({item['state']}): {item['repository_url'].split('/')[-1]}"
            for item in items
        ]
    else:
        results = [f"- {item['path']} in {item['repository']['full_name']}" for item in items]

    return f"Found {data['total_count']} results:\n" + "\n".join(results)
```

### Web Scraping Tool

```python
import httpx
from html.parser import HTMLParser

class TextExtractor(HTMLParser):
    def __init__(self):
        super().__init__()
        self.text_parts = []
        self._skip = False
        self._skip_tags = {"script", "style", "noscript"}

    def handle_starttag(self, tag, attrs):
        if tag in self._skip_tags:
            self._skip = True

    def handle_endtag(self, tag):
        if tag in self._skip_tags:
            self._skip = False

    def handle_data(self, data):
        if not self._skip:
            text = data.strip()
            if text:
                self.text_parts.append(text)

@mcp.tool()
async def fetch_webpage(url: str, extract_text: bool = True) -> str:
    """Fetch a webpage and optionally extract its text content.

    Args:
        url: URL to fetch
        extract_text: If true, extract text only; if false, return raw HTML
    """
    if not url.startswith(("http://", "https://")):
        raise ValueError("URL must start with http:// or https://")

    async with httpx.AsyncClient(follow_redirects=True) as client:
        resp = await client.get(url, timeout=30.0, headers={
            "User-Agent": "MCP-Server/1.0 (compatible; tool-fetch)"
        })
        resp.raise_for_status()

    if not extract_text:
        return resp.text[:50000]  # Limit raw HTML size

    extractor = TextExtractor()
    extractor.feed(resp.text)
    text = "\n".join(extractor.text_parts)

    # Truncate if too long
    if len(text) > 30000:
        text = text[:30000] + "\n\n[Truncated - content exceeds 30KB]"

    return text
```

# Transport and Authentication

## stdio Transport

### How It Works

The stdio transport communicates via **stdin** (client-to-server) and **stdout** (server-to-client). Each message is a JSON-RPC 2.0 object, one per line (newline-delimited).

```
Host Process                    MCP Server Process
┌──────────┐                    ┌──────────┐
│          │ ─── stdin ──────→  │          │
│  Client  │                    │  Server  │
│          │ ←── stdout ─────── │          │
└──────────┘                    └──────────┘
                                stderr → logs
```

**Key characteristics:**
- One client per server process (1:1 mapping)
- Host manages process lifecycle (spawn, restart, kill)
- No networking - everything is local
- stderr is used for logging (not protocol messages)
- Simplest transport, best for CLI tools and desktop integrations

### Python stdio Server

```python
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("my-server")

@mcp.tool()
def hello(name: str) -> str:
    """Say hello."""
    return f"Hello, {name}!"

if __name__ == "__main__":
    mcp.run()  # stdio is the default transport
```

### TypeScript stdio Server

```typescript
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";

const server = new McpServer({ name: "my-server", version: "1.0.0" });
// ... register tools ...

const transport = new StdioServerTransport();
await server.connect(transport);
```

### Claude Desktop Configuration

Location: `~/Library/Application Support/Claude/claude_desktop_config.json` (macOS) or `%APPDATA%/Claude/claude_desktop_config.json` (Windows)

```json
{
  "mcpServers": {
    "my-python-server": {
      "command": "uv",
      "args": ["run", "--directory", "/absolute/path/to/server", "python", "server.py"],
      "env": {
        "API_KEY": "sk-your-key",
        "LOG_LEVEL": "INFO"
      }
    },
    "my-node-server": {
      "command": "npx",
      "args": ["tsx", "/absolute/path/to/server/index.ts"],
      "env": {
        "API_KEY": "sk-your-key"
      }
    },
    "published-server": {
      "command": "uvx",
      "args": ["my-published-mcp-server"],
      "env": {}
    }
  }
}
```

### Claude Code Configuration

```json
// .claude/settings.json (project-level)
{
  "mcpServers": {
    "my-server": {
      "command": "uv",
      "args": ["run", "--directory", "/path/to/server", "python", "server.py"],
      "env": {
        "API_KEY": "sk-your-key"
      }
    }
  }
}
```

CLI shortcuts:

```bash
# Add a server
claude mcp add my-server -- uv run --directory /path/to/server python server.py

# Add with environment variables
claude mcp add my-server -e API_KEY=sk-key -- uv run python server.py

# List configured servers
claude mcp list

# Remove a server
claude mcp remove my-server
```

## SSE Transport

### How It Works

SSE (Server-Sent Events) uses HTTP for client-to-server requests and an EventSource stream for server-to-client messages.

```
Client                              Server
┌──────────┐                        ┌──────────┐
│          │ ── HTTP POST /sse ───→  │          │
│          │ ←── SSE event stream ── │          │
│          │                         │          │
│          │ ── HTTP POST /msg ───→  │          │
│          │ ←── (via SSE stream) ── │          │
└──────────┘                        └──────────┘
```

**Key characteristics:**
- HTTP-based, works through firewalls and proxies
- Server pushes events to client via EventSource
- Client sends requests via HTTP POST
- Multiple clients can connect simultaneously
- Good for development servers and internal tools

### Python SSE Server

```python
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("sse-server")

@mcp.tool()
def hello(name: str) -> str:
    return f"Hello, {name}!"

if __name__ == "__main__":
    mcp.run(transport="sse", host="0.0.0.0", port=8000)
```

### TypeScript SSE Server

```typescript
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { SSEServerTransport } from "@modelcontextprotocol/sdk/server/sse.js";
import express from "express";

const app = express();
const server = new McpServer({ name: "sse-server", version: "1.0.0" });

// ... register tools ...

app.get("/sse", async (req, res) => {
  const transport = new SSEServerTransport("/messages", res);
  await server.connect(transport);
});

app.post("/messages", async (req, res) => {
  // Handle incoming messages from client
  await transport.handlePostMessage(req, res);
});

app.listen(8000, () => console.log("SSE server running on port 8000"));
```

### SSE Reconnection

SSE connections can drop. The EventSource API handles automatic reconnection:

```typescript
// Client-side: EventSource handles reconnection automatically
const eventSource = new EventSource("http://localhost:8000/sse");
eventSource.onopen = () => console.log("Connected");
eventSource.onerror = (e) => console.log("Connection lost, reconnecting...");
```

Server-side, send keepalive comments to prevent connection timeout:

```python
# FastMCP handles this internally when using SSE transport
# For custom implementations, send periodic comments:
# : keepalive\n\n
```

## Streamable HTTP Transport

### How It Works

Streamable HTTP uses standard HTTP POST for requests and SSE for streaming responses. It supports both stateful (session-based) and stateless modes.

```
Client                              Server
┌──────────┐                        ┌──────────┐
│          │ ── POST /mcp ────────→ │          │
│          │    (JSON-RPC request)   │          │
│          │                         │          │
│          │ ←── SSE stream ──────── │          │
│          │    (JSON-RPC response)  │          │
│          │    (+ notifications)    │          │
└──────────┘                        └──────────┘

Session management via Mcp-Session-Id header
```

**Key characteristics:**
- Single HTTP endpoint for all communication
- Responses can be regular HTTP or SSE streams
- Session IDs for stateful operation, or fully stateless
- Load balancer and CDN friendly
- Full authentication support
- Recommended for production deployments

### Python Streamable HTTP Server

```python
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("http-server")

@mcp.tool()
def hello(name: str) -> str:
    return f"Hello, {name}!"

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

### TypeScript Streamable HTTP Server

```typescript
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamableHttp.js";
import express from "express";

const app = express();
app.use(express.json());

const server = new McpServer({ name: "http-server", version: "1.0.0" });
// ... register tools ...

app.post("/mcp", async (req, res) => {
  const transport = new StreamableHTTPServerTransport({
    sessionIdGenerator: () => crypto.randomUUID(),
  });
  await server.connect(transport);
  await transport.handleRequest(req, res);
});

app.listen(8000);
```

### Stateless Mode

For serverless or horizontally scaled deployments:

```typescript
const transport = new StreamableHTTPServerTransport({
  sessionIdGenerator: undefined,  // No session tracking
});
```

In stateless mode:
- No `Mcp-Session-Id` header
- Each request is independent
- Server reconstructs state from request context
- Works with any load balancer without sticky sessions

## Transport Selection Guide

| Scenario | Transport | Why |
|----------|-----------|-----|
| Claude Desktop integration | stdio | Native support, simplest setup |
| Claude Code tool | stdio | Direct process management |
| VS Code extension backend | stdio | Process managed by extension |
| Internal dashboard | SSE | Browser-friendly, real-time updates |
| Development/testing | SSE | Easy to inspect with browser |
| Production API | Streamable HTTP | Auth, scaling, load balancing |
| Serverless (Lambda, Workers) | Streamable HTTP (stateless) | No persistent connections needed |
| Multi-tenant SaaS | Streamable HTTP | Session isolation, auth per tenant |
| Mobile app backend | Streamable HTTP | Standard HTTP, auth support |

## Session Management

### Session IDs

For stateful HTTP transports, each client gets a unique session:

```typescript
const transport = new StreamableHTTPServerTransport({
  sessionIdGenerator: () => crypto.randomUUID(),
  // Optional: validate session IDs
  sessionValidator: async (sessionId) => {
    return await sessionStore.exists(sessionId);
  },
});
```

### Session State

```python
# Store per-session state
class SessionState:
    def __init__(self):
        self.user_id: str | None = None
        self.preferences: dict = {}
        self.history: list = []

# In FastMCP, access via context
@mcp.tool()
async def set_preference(key: str, value: str, ctx: Context) -> str:
    session = ctx.request_context.session
    if not hasattr(session, "state"):
        session.state = SessionState()
    session.state.preferences[key] = value
    return f"Set {key} = {value}"
```

### Session Cleanup

```python
from contextlib import asynccontextmanager
import asyncio

@asynccontextmanager
async def lifespan(server: FastMCP):
    # Start cleanup task
    cleanup_task = asyncio.create_task(cleanup_expired_sessions())
    try:
        yield
    finally:
        cleanup_task.cancel()

async def cleanup_expired_sessions():
    while True:
        await asyncio.sleep(300)  # Every 5 minutes
        expired = session_store.get_expired(max_age_seconds=3600)
        for session_id in expired:
            session_store.remove(session_id)
```

## Authentication

### API Keys in Environment Variables

The simplest auth pattern - pass credentials via environment configuration:

```python
import os

@mcp.tool()
async def call_api(endpoint: str) -> str:
    """Call an authenticated API."""
    api_key = os.environ.get("API_KEY")
    if not api_key:
        raise RuntimeError("API_KEY environment variable not set")

    async with httpx.AsyncClient() as client:
        resp = await client.get(
            f"https://api.example.com/{endpoint}",
            headers={"Authorization": f"Bearer {api_key}"},
            timeout=30.0,
        )
        resp.raise_for_status()
        return resp.text
```

Client config passes the key:

```json
{
  "mcpServers": {
    "api-server": {
      "command": "python",
      "args": ["server.py"],
      "env": {
        "API_KEY": "sk-your-api-key-here"
      }
    }
  }
}
```

### Bearer Token Authentication (HTTP transports)

For SSE and Streamable HTTP, authenticate incoming requests:

```python
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import JSONResponse

class AuthMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        # Skip auth for health checks
        if request.url.path == "/health":
            return await call_next(request)

        auth_header = request.headers.get("Authorization")
        if not auth_header or not auth_header.startswith("Bearer "):
            return JSONResponse({"error": "Missing or invalid Authorization header"}, status_code=401)

        token = auth_header[7:]
        if not await validate_token(token):
            return JSONResponse({"error": "Invalid token"}, status_code=403)

        # Attach user info to request state
        request.state.user = await get_user_from_token(token)
        return await call_next(request)
```

### OAuth2 PKCE Flow

For MCP servers that need user-level authentication:

```python
import secrets
import hashlib
import base64
from urllib.parse import urlencode

class OAuth2PKCEFlow:
    def __init__(self, client_id: str, auth_url: str, token_url: str, redirect_uri: str):
        self.client_id = client_id
        self.auth_url = auth_url
        self.token_url = token_url
        self.redirect_uri = redirect_uri

    def generate_auth_url(self) -> tuple[str, str]:
        """Generate authorization URL with PKCE challenge."""
        code_verifier = secrets.token_urlsafe(64)
        code_challenge = base64.urlsafe_b64encode(
            hashlib.sha256(code_verifier.encode()).digest()
        ).rstrip(b"=").decode()

        params = urlencode({
            "response_type": "code",
            "client_id": self.client_id,
            "redirect_uri": self.redirect_uri,
            "code_challenge": code_challenge,
            "code_challenge_method": "S256",
            "scope": "read write",
        })
        return f"{self.auth_url}?{params}", code_verifier

    async def exchange_code(self, code: str, code_verifier: str) -> dict:
        """Exchange authorization code for tokens."""
        async with httpx.AsyncClient() as client:
            resp = await client.post(self.token_url, data={
                "grant_type": "authorization_code",
                "client_id": self.client_id,
                "code": code,
                "redirect_uri": self.redirect_uri,
                "code_verifier": code_verifier,
            })
            resp.raise_for_status()
            return resp.json()

    async def refresh_token(self, refresh_token: str) -> dict:
        """Refresh an expired access token."""
        async with httpx.AsyncClient() as client:
            resp = await client.post(self.token_url, data={
                "grant_type": "refresh_token",
                "client_id": self.client_id,
                "refresh_token": refresh_token,
            })
            resp.raise_for_status()
            return resp.json()
```

### Token Management

```python
import time
import asyncio

class TokenManager:
    """Manage OAuth2 tokens with automatic refresh."""

    def __init__(self, oauth: OAuth2PKCEFlow):
        self.oauth = oauth
        self._tokens: dict = {}
        self._lock = asyncio.Lock()

    async def get_token(self) -> str:
        """Get a valid access token, refreshing if needed."""
        async with self._lock:
            if self._is_valid():
                return self._tokens["access_token"]

            if "refresh_token" in self._tokens:
                self._tokens = await self.oauth.refresh_token(self._tokens["refresh_token"])
                self._tokens["obtained_at"] = time.time()
                return self._tokens["access_token"]

            raise RuntimeError("No valid token available. User must re-authenticate.")

    def _is_valid(self) -> bool:
        if "access_token" not in self._tokens:
            return False
        expires_at = self._tokens.get("obtained_at", 0) + self._tokens.get("expires_in", 0)
        return time.time() < expires_at - 60  # 60s buffer

    def set_tokens(self, tokens: dict):
        """Store tokens after initial authorization."""
        self._tokens = {**tokens, "obtained_at": time.time()}
```

## Rate Limiting

### Per-Client Rate Limiting

```python
import time
from collections import defaultdict

class SlidingWindowRateLimiter:
    def __init__(self, max_requests: int, window_seconds: int):
        self.max_requests = max_requests
        self.window = window_seconds
        self.requests: dict[str, list[float]] = defaultdict(list)

    def allow(self, client_id: str) -> bool:
        now = time.time()
        # Remove expired entries
        self.requests[client_id] = [
            t for t in self.requests[client_id]
            if now - t < self.window
        ]
        if len(self.requests[client_id]) >= self.max_requests:
            return False
        self.requests[client_id].append(now)
        return True

    def remaining(self, client_id: str) -> int:
        now = time.time()
        active = [t for t in self.requests[client_id] if now - t < self.window]
        return max(0, self.max_requests - len(active))

# Usage in tools
rate_limiter = SlidingWindowRateLimiter(max_requests=100, window_seconds=60)

@mcp.tool()
async def rate_limited_tool(query: str, ctx: Context) -> str:
    """A rate-limited tool."""
    client_id = str(id(ctx.request_context.session))
    if not rate_limiter.allow(client_id):
        remaining_wait = rate_limiter.window
        raise RuntimeError(f"Rate limit exceeded. Try again in {remaining_wait}s.")
    return await process(query)
```

### Per-Tool Rate Limiting

```python
from functools import wraps

def rate_limit(max_calls: int, window: int):
    """Decorator to rate-limit individual tools."""
    limiter = SlidingWindowRateLimiter(max_calls, window)

    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            tool_name = func.__name__
            if not limiter.allow(tool_name):
                raise RuntimeError(f"Tool {tool_name} rate limit exceeded ({max_calls}/{window}s)")
            return await func(*args, **kwargs) if asyncio.iscoroutinefunction(func) else func(*args, **kwargs)
        return wrapper
    return decorator

@mcp.tool()
@rate_limit(max_calls=10, window=60)
async def expensive_api_call(query: str) -> str:
    """Call an expensive API (limited to 10 calls/minute)."""
    ...
```

## CORS Configuration

For web-based clients connecting to SSE or HTTP servers:

```python
from starlette.middleware.cors import CORSMiddleware

# If using Starlette/FastAPI alongside FastMCP
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000", "https://myapp.example.com"],
    allow_credentials=True,
    allow_methods=["GET", "POST", "OPTIONS"],
    allow_headers=["Authorization", "Content-Type", "Mcp-Session-Id"],
)
```

```typescript
// Express CORS
import cors from "cors";
app.use(cors({
  origin: ["http://localhost:3000", "https://myapp.example.com"],
  credentials: true,
  allowedHeaders: ["Authorization", "Content-Type", "Mcp-Session-Id"],
}));
```

## TLS/HTTPS for Production

Always use HTTPS for non-stdio transports in production:

```python
# Using uvicorn with TLS
import uvicorn

if __name__ == "__main__":
    uvicorn.run(
        app,
        host="0.0.0.0",
        port=443,
        ssl_keyfile="/path/to/key.pem",
        ssl_certfile="/path/to/cert.pem",
    )
```

Or terminate TLS at a reverse proxy (recommended):

```nginx
# nginx reverse proxy for MCP server
server {
    listen 443 ssl;
    server_name mcp.example.com;

    ssl_certificate /etc/letsencrypt/live/mcp.example.com/fullchain.pem;
    ssl_certificate_key /etc/letsencrypt/live/mcp.example.com/privkey.pem;

    location / {
        proxy_pass http://localhost:8000;
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "upgrade";
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_buffering off;  # Important for SSE
        proxy_cache off;      # Don't cache SSE streams
    }
}
```

## Proxy Patterns

### Reverse Proxy for Multiple MCP Servers

Route requests to different MCP servers based on path:

```nginx
# Serve multiple MCP servers from one domain
server {
    listen 443 ssl;
    server_name mcp.example.com;

    # Server A: database tools
    location /db/ {
        proxy_pass http://localhost:8001/;
        proxy_buffering off;
    }

    # Server B: file system tools
    location /files/ {
        proxy_pass http://localhost:8002/;
        proxy_buffering off;
    }

    # Server C: API integration tools
    location /api/ {
        proxy_pass http://localhost:8003/;
        proxy_buffering off;
    }
}
```

### Gateway Pattern

A single MCP server that routes to backend services:

```python
from mcp.server.fastmcp import FastMCP
import httpx

mcp = FastMCP("gateway")

# Route tool calls to backend MCP servers
BACKENDS = {
    "db_": "http://localhost:8001",
    "file_": "http://localhost:8002",
    "api_": "http://localhost:8003",
}

@mcp.tool()
async def gateway_call(tool_name: str, arguments: dict) -> str:
    """Route a tool call to the appropriate backend.

    Args:
        tool_name: Full tool name (e.g., db_query, file_read)
        arguments: Tool arguments as a JSON object
    """
    for prefix, backend_url in BACKENDS.items():
        if tool_name.startswith(prefix):
            async with httpx.AsyncClient() as client:
                resp = await client.post(
                    f"{backend_url}/mcp",
                    json={
                        "jsonrpc": "2.0",
                        "id": 1,
                        "method": "tools/call",
                        "params": {"name": tool_name, "arguments": arguments},
                    },
                )
                return resp.json()["result"]["content"][0]["text"]
    raise ValueError(f"No backend found for tool: {tool_name}")
```

#!/usr/bin/env python3
"""Staleness verifier for mcp-ops: the MCP SDK packages + spec URL the skill
names must stay real, named, and current.

mcp-ops tells the reader to install `@modelcontextprotocol/sdk` (npm),
`@modelcontextprotocol/inspector` (npm), `mcp[cli]` / `fastmcp` (PyPI), and
cites the spec at spec.modelcontextprotocol.io. Those are exactly the facts
that drift silently (SKILL-RESOURCE-PROTOCOL.md §7): an SDK is renamed, a
package is unpublished, the prose stops mentioning one the catalog still
lists, or the spec URL goes dead — and nobody notices for months. Two modes:

  --offline (default, safe for PR CI): structural consistency, no network.
    * assets/mcp-facts.json parses and carries the schema + an as_of date
    * every catalogued package's prose_token is still named somewhere in the
      skill prose (SKILL.md + references/*.md) — the catalog can't drift from
      the docs
    * the spec URL token is still cited in the prose
    * SKILL.md still carries a dated "verified as of <year>" currency note
  --live (scheduled freshness job, never a PR gate): does each package still
    resolve on its registry (npm latest / PyPI JSON), and has any tracked
    SDK's major moved off the sampled major? Does the spec URL answer 200?

Usage:   check-mcp-facts.py [--offline | --live] [--catalog FILE] [--skill DIR] [--json] [--timeout S]
Input:   argv flags only (no stdin).
Output:  stdout = findings (plain rows, or a --json envelope). Data only.
Stderr:  the verdict line, notices, errors.
Exit:    0 ok, 2 usage, 3 catalog/skill missing, 4 catalog unparseable,
         7 registries/spec unreachable (live, advisory — never a real failure),
         10 drift found (offline: unnamed/currency-note gone; live: package
            gone, SDK major moved, or spec URL dead)

Examples:
  check-mcp-facts.py --offline                 # PR CI: catalog ⇆ prose consistency
  check-mcp-facts.py --live                    # weekly: SDKs still resolve + spec 200
  check-mcp-facts.py --offline --json | jq '.data[]'
"""
from __future__ import annotations

import argparse
import json
import re
import sys
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path

EX_OK = 0
EX_USAGE = 2
EX_NOTFOUND = 3
EX_UNPARSEABLE = 4
EX_UNAVAILABLE = 7
EX_DRIFT = 10

SCHEMA = "claude-mods.mcp-ops.facts/v1"

HERE = Path(__file__).resolve().parent
DEFAULT_CATALOG = HERE.parent / "assets" / "mcp-facts.json"
DEFAULT_SKILL = HERE.parent

NPM_REGISTRY = "https://registry.npmjs.org"
PYPI_REGISTRY = "https://pypi.org/pypi"

CURRENCY_RE = re.compile(r"verified as of\s+(\d{4})", re.IGNORECASE)
AS_OF_RE = re.compile(r"^\d{4}-\d{2}-\d{2}$")


class Finding:
    __slots__ = ("check", "status", "detail")

    def __init__(self, check: str, status: str, detail: str) -> None:
        self.check = check
        self.status = status  # ok | drift | unavailable
        self.detail = detail

    def as_dict(self) -> dict:
        return {"check": self.check, "status": self.status, "detail": self.detail}


def load_catalog(path: Path) -> dict:
    if not path.is_file():
        print(f"error: facts catalog not found: {path}", file=sys.stderr)
        raise SystemExit(EX_NOTFOUND)
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
        if not isinstance(data, dict) or data.get("schema") != SCHEMA:
            raise ValueError(f"schema must be {SCHEMA!r}")
        if not AS_OF_RE.match(str(data.get("as_of", ""))):
            raise ValueError(f"as_of must be YYYY-MM-DD, got {data.get('as_of')!r}")
        if not isinstance(data.get("packages"), dict) or not data["packages"]:
            raise ValueError("'packages' must be a non-empty object")
        return data
    except (json.JSONDecodeError, KeyError, TypeError, ValueError) as exc:
        print(f"error: could not parse catalog {path}: {exc}", file=sys.stderr)
        raise SystemExit(EX_UNPARSEABLE)


def read_corpus(skill_dir: Path) -> tuple[str, str]:
    """Returns (skill_md_text, all_prose_text) across SKILL.md + references/*.md."""
    doc = skill_dir / "SKILL.md"
    if not doc.is_file():
        print(f"error: SKILL.md not found under {skill_dir}", file=sys.stderr)
        raise SystemExit(EX_NOTFOUND)
    skill_md = doc.read_text(encoding="utf-8", errors="replace")
    parts = [skill_md]
    ref_dir = skill_dir / "references"
    if ref_dir.is_dir():
        for ref in sorted(ref_dir.glob("*.md")):
            parts.append(ref.read_text(encoding="utf-8", errors="replace"))
    return skill_md, "\n".join(parts)


def check_offline(catalog: dict, skill_dir: Path) -> list[Finding]:
    skill_md, corpus = read_corpus(skill_dir)
    findings: list[Finding] = []
    lower_corpus = corpus.lower()

    # Currency note still dated.
    if CURRENCY_RE.search(skill_md):
        m = CURRENCY_RE.search(skill_md)
        findings.append(Finding("currency-note", "ok", f"currency note dated {m.group(1)}"))
    else:
        findings.append(Finding("currency-note", "drift",
                                "no dated 'verified as of <year>' currency note in SKILL.md"))

    # Spec URL token still cited.
    spec = catalog.get("spec", {})
    spec_token = str(spec.get("prose_token", ""))
    if spec_token and spec_token.lower() in lower_corpus:
        findings.append(Finding("spec-cited", "ok", f"{spec_token} named in prose"))
    else:
        findings.append(Finding("spec-cited", "drift",
                                f"spec token {spec_token!r} no longer named in skill prose"))

    # Every catalogued package still named in the prose.
    for name, meta in catalog.get("packages", {}).items():
        token = str(meta.get("prose_token", name))
        if token.lower() in lower_corpus:
            findings.append(Finding(f"pkg-named:{name}", "ok", "named in skill prose"))
        else:
            findings.append(Finding(f"pkg-named:{name}", "drift",
                                    f"prose_token {token!r} not found in skill prose"))
    return findings


def _fetch(url: str, timeout: float, accept: str = "application/json") -> tuple[str, object]:
    """Return (ok|notfound|unavailable, payload-or-status)."""
    req = urllib.request.Request(url, headers={"User-Agent": "claude-mods-mcp-ops-check/1",
                                               "Accept": accept})
    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            body = resp.read().decode("utf-8", errors="replace")
            return "ok", body
    except urllib.error.HTTPError as exc:
        if exc.code in (404, 410):
            return "notfound", exc.code
        return "unavailable", exc.code
    except (urllib.error.URLError, TimeoutError, OSError):
        return "unavailable", None


def _npm_latest(pkg: str, timeout: float) -> tuple[str, str]:
    """Return (status, version-or-status-detail). status in ok|notfound|unavailable."""
    url = f"{NPM_REGISTRY}/{urllib.parse.quote(pkg, safe='')}/latest"
    status, payload = _fetch(url, timeout)
    if status != "ok":
        return status, str(payload)
    try:
        ver = json.loads(payload).get("version", "")
        return "ok", ver
    except json.JSONDecodeError:
        return "unavailable", "bad-json"


def _pypi_latest(pkg: str, timeout: float) -> tuple[str, str]:
    url = f"{PYPI_REGISTRY}/{urllib.parse.quote(pkg, safe='')}/json"
    status, payload = _fetch(url, timeout)
    if status != "ok":
        return status, str(payload)
    try:
        ver = json.loads(payload).get("info", {}).get("version", "")
        return "ok", ver
    except json.JSONDecodeError:
        return "unavailable", "bad-json"


def _major(ver: str) -> str:
    """Leading integer component of a version string ('1.2.3' -> '1')."""
    m = re.match(r"\s*(\d+)", ver)
    return m.group(1) if m else ""


def check_live(catalog: dict, timeout: float) -> list[Finding]:
    findings: list[Finding] = []

    for name, meta in catalog.get("packages", {}).items():
        registry = meta.get("registry", "npm")
        if registry == "npm":
            status, info = _npm_latest(name, timeout)
        else:
            status, info = _pypi_latest(name, timeout)
        if status == "notfound":
            findings.append(Finding(f"npm:{name}", "drift",
                                    "package gone from registry — renamed/removed, review skill"))
            continue
        if status == "unavailable":
            findings.append(Finding(f"npm:{name}", "unavailable", "registry unreachable"))
            continue
        ver = str(info)
        if meta.get("track_major"):
            sampled = str(meta.get("sampled_major", ""))
            latest_major = _major(ver)
            if latest_major and latest_major != sampled:
                findings.append(Finding(f"npm:{name}", "drift",
                                        f"{name}@{ver} major {latest_major} != sampled {sampled} — review skill"))
            else:
                findings.append(Finding(f"npm:{name}", "ok", f"latest {ver} (major {latest_major})"))
        else:
            findings.append(Finding(f"npm:{name}", "ok", f"latest {ver}"))

    # Spec URL must answer 200 (follows redirects).
    spec_url = str(catalog.get("spec", {}).get("url", ""))
    if spec_url:
        status, info = _fetch(spec_url, timeout, accept="text/html,application/json")
        if status == "ok":
            findings.append(Finding("spec-url", "ok", f"{spec_url} answered 200"))
        elif status == "notfound":
            findings.append(Finding("spec-url", "drift", f"{spec_url} returned 4xx — spec URL dead"))
        else:
            findings.append(Finding("spec-url", "unavailable", f"{spec_url} unreachable"))
    return findings


def main(argv: list[str]) -> int:
    p = argparse.ArgumentParser(
        prog="check-mcp-facts.py",
        description="Verify mcp-ops' SDK packages + spec URL stay named (offline) and live (live).",
    )
    mode = p.add_mutually_exclusive_group()
    mode.add_argument("--offline", action="store_true", help="structural consistency, no network (default)")
    mode.add_argument("--live", action="store_true", help="probe npm/PyPI registries + the spec URL")
    p.add_argument("--catalog", default=str(DEFAULT_CATALOG), help="facts catalog JSON")
    p.add_argument("--skill", default=str(DEFAULT_SKILL), help="skill directory (SKILL.md + references/)")
    p.add_argument("--timeout", type=float, default=10.0, help="per-request timeout seconds (live)")
    p.add_argument("--json", action="store_true", help="emit a JSON envelope")
    p.add_argument("-q", "--quiet", action="store_true", help="suppress stderr progress/summary")
    try:
        args = p.parse_args(argv)
    except SystemExit as exc:
        return EX_USAGE if exc.code not in (0, None) else (exc.code or EX_OK)

    catalog = load_catalog(Path(args.catalog))
    live = args.live and not args.offline
    mode_name = "live" if live else "offline"
    findings = check_live(catalog, args.timeout) if live else check_offline(catalog, Path(args.skill))

    n_drift = sum(1 for f in findings if f.status == "drift")
    n_unavail = sum(1 for f in findings if f.status == "unavailable")

    if args.json:
        print(json.dumps({
            "data": [f.as_dict() for f in findings],
            "meta": {"mode": mode_name, "count": len(findings),
                     "drift": n_drift, "unavailable": n_unavail, "schema": SCHEMA},
        }, indent=2))
    else:
        for f in findings:
            print(f"{f.check}\t{f.status}\t{f.detail}")

    for f in findings:
        if f.status != "ok":
            print(f"  [{f.status.upper()}] {f.check}: {f.detail}", file=sys.stderr)
    if not args.quiet:
        print(f"-- {len(findings)} checks: {n_drift} drift, {n_unavail} unavailable", file=sys.stderr)

    if n_drift:
        return EX_DRIFT
    if n_unavail:
        return EX_UNAVAILABLE
    return EX_OK


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))

#!/usr/bin/env bash
# Offline self-test for the mcp-ops skill — structure, frontmatter, and the
# staleness-verifier contract (SKILL-RESOURCE-PROTOCOL §7, §10).
#
# Usage:   tests/run.sh
# Input:   none (self-contained; no network, no MCP server required)
# Output:  TAP-ish progress on stderr; final PASS/FAIL line.
# Exit:    0 all pass (or skipped on unsupported platform), 1 any failure.
#
# Examples:
#   tests/run.sh
#   bash skills/mcp-ops/tests/run.sh
set -uo pipefail

here="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
fail=0
pass=0
note() { printf '  %s %s\n' "$1" "$2" >&2; }
ok()   { pass=$((pass+1)); note "ok  " "$1"; }
bad()  { fail=$((fail+1)); note "FAIL" "$1"; }

# Resolve a *working* python (python3, else python). The bare `command -v` is not
# enough on Windows, where `python3` is a Microsoft Store stub that exits nonzero.
PY=""
for cand in python3 python; do
  if command -v "$cand" >/dev/null 2>&1 && "$cand" --version >/dev/null 2>&1; then
    PY="$cand"; break
  fi
done
if [ -z "$PY" ]; then
  echo "SKIP: no working python interpreter on this platform" >&2
  exit 0
fi

# 1. Required directories exist
for d in scripts references assets tests; do
  [ -d "$here/$d" ] && ok "dir $d/ exists" || bad "missing dir $d/"
done

# 2. SKILL.md frontmatter house rules
skill="$here/SKILL.md"
if [ -f "$skill" ]; then
  ok "SKILL.md present"
  grep -q '^name: mcp-ops$' "$skill" && ok "name matches directory" || bad "name != mcp-ops"
  grep -q '^license: MIT$' "$skill" && ok "license: MIT" || bad "missing license: MIT"
  grep -q '^  author: claude-mods$' "$skill" && ok "metadata.author" || bad "missing metadata.author"
else
  bad "SKILL.md missing"
fi

# 3. Every reference on disk is cited from SKILL.md (no dead weight)
for ref in "$here"/references/*.md; do
  base="references/$(basename "$ref")"
  grep -qF "$base" "$skill" && ok "cited: $base" || bad "uncited reference: $base"
done

# 4. Every SKILL.md-cited bundled resource exists on disk
for res in assets/mcp-facts.json scripts/check-mcp-facts.py; do
  [ -f "$here/$res" ] && ok "resource present: $res" || bad "missing resource: $res"
done

# 5. check-mcp-facts.py — staleness verifier contract (§7, §10), offline-safe
verifier="$here/scripts/check-mcp-facts.py"
catalog="$here/assets/mcp-facts.json"
ec() { local want="$1" lbl="$2"; shift 2; "$@" >/dev/null 2>&1; local got=$?
       [ "$got" = "$want" ] && ok "$lbl (exit $got)" || bad "$lbl (want $want got $got)"; }
if [ -f "$verifier" ]; then
  "$PY" -m py_compile "$verifier" && ok "verifier: py_compile clean" || bad "verifier: py_compile failed"
  grep -qE '^Examples:$' "$verifier" && ok "verifier: has Examples block" || bad "verifier: no Examples block (docstring)"
  "$PY" "$verifier" --help >/dev/null 2>&1 && ok "verifier: --help exits 0" || bad "verifier: --help nonzero"
  # Offline mode must pass on the skill's own content (internal consistency).
  ec 0 "verifier: --offline consistent"  "$PY" "$verifier" --offline
  # Bad flag → USAGE (exit 2); conflicting modes → USAGE.
  ec 2 "verifier: bad flag → exit 2"     "$PY" "$verifier" --bogus
  ec 2 "verifier: --offline --live → 2"  "$PY" "$verifier" --offline --live
  # stdout is data-only: --offline --json must emit parseable JSON.
  "$PY" "$verifier" --offline --json -q 2>/dev/null \
    | "$PY" -c 'import json,sys; d=json.load(sys.stdin); assert d["meta"]["schema"]=="claude-mods.mcp-ops.facts/v1"' \
    && ok "verifier: --json envelope parses (stdout clean)" || bad "verifier: --json envelope broken"
  # Error paths: missing catalog → 3, malformed catalog → 4, drift catalog → 10.
  TMP="$(mktemp -d)"; trap 'rm -rf "$TMP"' EXIT
  ec 3 "verifier: missing catalog -> 3"  "$PY" "$verifier" --offline --catalog "$TMP/nope.json"
  printf '{"packages":"x"}' > "$TMP/bad.json"
  ec 4 "verifier: malformed catalog -> 4" "$PY" "$verifier" --offline --catalog "$TMP/bad.json"
  # Minimal drift catalog (argv path so MSYS translates it): a package whose
  # prose_token is not in the real skill prose -> drift -> exit 10.
  printf '%s\n' '{"schema":"claude-mods.mcp-ops.facts/v1","as_of":"2026-07-05","spec":{"url":"https://spec.example","prose_token":"zzz.no.such.spec"},"packages":{"@no-such/sdk":{"registry":"npm","prose_token":"@no-such/sdk","sampled_major":1,"track_major":true}}}' > "$TMP/drift.json"
  ec 10 "verifier: unnamed package -> 10" "$PY" "$verifier" --offline --catalog "$TMP/drift.json"
  # cited from SKILL.md
  grep -qF "scripts/check-mcp-facts.py" "$skill" && ok "verifier: cited from SKILL.md" || bad "verifier: uncited"
else
  bad "check-mcp-facts.py missing"
fi

# 6. mcp-facts.json — parses, carries schema the verifier depends on
"$PY" -c "
import json, sys
d = json.load(open(sys.argv[1], encoding='utf-8'))
assert d['schema'] == 'claude-mods.mcp-ops.facts/v1'
assert d['packages'], 'no packages committed'
assert '@modelcontextprotocol/sdk' in d['packages']
assert d['spec']['url'].startswith('https://')
" "$catalog" && ok "mcp-facts.json schema + packages" || bad "mcp-facts.json invalid"

# 7. Currency note present near the top of the body
grep -qE 'verified as of [0-9]{4}' "$skill" && ok "currency note present" || bad "no dated currency note"

echo "mcp-ops self-test: $pass passed, $fail failed" >&2
[ "$fail" -eq 0 ]