Building High-Throughput Production MCP Servers with Python FastMCP and stdio

[!NOTE] Quick Start (60 seconds): A minimal FastMCP server with one tool:

pip install fastmcp

# server.py
from fastmcp import FastMCP

mcp = FastMCP("my-production-server")

@mcp.tool()
def search_docs(query: str, max_results: int = 5) -> list[dict]:
    """Search internal documentation. Returns matching documents with title and URL."""
    # your implementation here
    return [{"title": "Example", "url": "https://docs.example.com/page"}]

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

Run it: mcp run server.py

The Model Context Protocol specification is 40 pages of JSON-RPC message types. FastMCP collapses that into a @mcp.tool() decorator and handles the rest. This guide covers building a production server: input validation, error taxonomy, concurrency, and connecting it to Claude Code.

---

Environment

Package	Version
fastmcp	2.2.0
pydantic	2.7+
mcp	1.5.0
Python	3.11+
Claude Code / Cursor	latest

---

1. MCP Architecture: How stdio Transport Works

With stdio transport, the MCP client (Claude Code) spawns your Python server as a child process. All communication flows over the process’s stdin and stdout as newline-delimited JSON-RPC messages.

sequenceDiagram
    participant Claude as Claude Code
    participant Server as FastMCP Server (subprocess)

    Claude->>Server: spawn process
    Claude->>Server: initialize request (stdin)
    Server-->>Claude: initialize response (stdout)
    Claude->>Server: tools/list request
    Server-->>Claude: [search_docs, write_file, ...] (stdout)
    Claude->>Server: tools/call {name: "search_docs", arguments: {...}}
    Server-->>Claude: tool result (stdout)

Your server process runs for the lifetime of the Claude Code session. stderr is available for logging without polluting the JSON-RPC channel.

---

2. Tool Registration with Pydantic Input Validation

For tools with complex inputs, use a Pydantic BaseModel to get full validation and a rich JSON Schema:

from fastmcp import FastMCP
from pydantic import BaseModel, Field
import sys
import logging

# Log to stderr to avoid polluting the stdio JSON-RPC channel
logging.basicConfig(stream=sys.stderr, level=logging.INFO)
logger = logging.getLogger(__name__)

mcp = FastMCP(
    name="docs-server",
    version="1.0.0",
    description="Internal documentation and code search tools.",
)

class SearchRequest(BaseModel):
    query: str = Field(..., description="Search terms or natural language question")
    max_results: int = Field(default=5, ge=1, le=50, description="Number of results to return")
    category: str | None = Field(default=None, description="Filter by doc category")

class SearchResult(BaseModel):
    title: str
    url: str
    snippet: str
    relevance_score: float

@mcp.tool()
def search_docs(request: SearchRequest) -> list[SearchResult]:
    """
    Search the internal documentation corpus.
    Returns ranked results with title, URL, and content snippet.
    """
    logger.info(f"search_docs called: query={request.query!r}, max={request.max_results}")
    # Implementation: call your search backend
    results = _run_search(request.query, request.max_results, request.category)
    return results

FastMCP generates the full JSON Schema from SearchRequest automatically, including the ge=1, le=50 bounds as minimum and maximum validators in the schema.

---

3. Async Tools and Concurrency

FastMCP supports both sync and async tool functions. Use async for any tool that performs network I/O:

import httpx
import asyncio
from fastmcp import FastMCP

mcp = FastMCP("async-server")

@mcp.tool()
async def fetch_url(url: str, timeout_sec: float = 10.0) -> dict:
    """Fetch a URL and return its status code and content length."""
    async with httpx.AsyncClient() as client:
        response = await client.get(url, timeout=timeout_sec)
        return {
            "status_code": response.status_code,
            "content_length": len(response.content),
            "content_type": response.headers.get("content-type", "unknown"),
        }

@mcp.tool()
async def run_sql_query(sql: str, database: str = "default") -> list[dict]:
    """Execute a read-only SQL query and return rows as a list of dicts."""
    # Use asyncpg for non-blocking Postgres access
    import asyncpg
    pool = await asyncpg.create_pool(dsn=DATABASE_URLS[database])
    async with pool.acquire() as conn:
        rows = await conn.fetch(sql)
        return [dict(row) for row in rows]

For CPU-bound operations, offload to a thread:

import asyncio

@mcp.tool()
async def process_large_file(path: str) -> dict:
    """Parse and analyze a large JSON file. CPU-intensive."""
    def _parse_sync(file_path: str) -> dict:
        import json
        with open(file_path) as f:
            data = json.load(f)
        return {"row_count": len(data), "keys": list(data[0].keys()) if data else []}

    return await asyncio.to_thread(_parse_sync, path)

---

4. Error Handling

FastMCP translates Python exceptions into MCP error responses. Structure your errors so Claude gets actionable context:

from fastmcp.exceptions import ToolError

@mcp.tool()
def read_file(path: str, encoding: str = "utf-8") -> str:
    """Read a file from the local filesystem and return its contents."""
    import pathlib

    target = pathlib.Path(path).resolve()

    # Explicit security boundary — never let the model read outside the project
    allowed_root = pathlib.Path("/workspace").resolve()
    if not str(target).startswith(str(allowed_root)):
        raise ToolError(
            f"Access denied: {path} is outside the allowed workspace at {allowed_root}"
        )

    if not target.exists():
        raise ToolError(f"File not found: {path}")

    if target.stat().st_size > 10 * 1024 * 1024:  # 10 MB limit
        raise ToolError(
            f"File too large ({target.stat().st_size // 1024}KB). Use read_file_chunk for files > 10MB."
        )

    return target.read_text(encoding=encoding)

ToolError maps to an MCP isError: true response, which Claude interprets as a tool failure and typically retries with a corrected argument or asks for clarification.

---

5. Resources and Prompts

MCP has three primitive types: Tools (functions Claude calls), Resources (data Claude reads), and Prompts (templates Claude uses). FastMCP handles all three:

# Resource: expose structured data Claude can read on demand
@mcp.resource("config://app/settings")
def get_app_settings() -> dict:
    """Current application configuration."""
    return {
        "env": "production",
        "version": "2.3.1",
        "feature_flags": {"new_dashboard": True, "beta_export": False},
    }

# Prompt: reusable prompt template
@mcp.prompt()
def code_review_prompt(language: str, code: str) -> str:
    """Generate a structured code review prompt for the given language and code."""
    return f"""Review this {language} code for:
1. Security vulnerabilities
2. Performance issues
3. Style consistency

Code:
```{language}
{code}

Provide specific line-number references for each issue."""

---

## 6. Registering with Claude Code

Add your server to Claude Code's config file:

```json
// ~/.claude/settings.json (Claude Code CLI)
// or ~/Library/Application Support/Claude/claude_desktop_config.json (Claude Desktop)
{
  "mcpServers": {
    "docs-server": {
      "command": "python",
      "args": ["/path/to/your/server.py"],
      "env": {
        "DATABASE_URL": "postgresql://...",
        "API_KEY": "sk-..."
      }
    },
    "filesystem-server": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-filesystem", "/workspace"]
    }
  }
}

After saving, restart Claude Code. Run /mcp in the Claude Code chat to verify your server appears and all tools are listed.

---

7. SSE Transport for Remote Deployments

If your tools need to run on a remote server (e.g., they access internal APIs not reachable from the developer’s laptop), switch to SSE transport:

# server.py — SSE transport for remote deployment
if __name__ == "__main__":
    mcp.run(transport="sse", host="0.0.0.0", port=8000)

Register the remote server in Claude Code:

{
  "mcpServers": {
    "remote-docs": {
      "url": "https://mcp.internal.company.com/sse",
      "headers": {
        "Authorization": "Bearer your-token"
      }
    }
  }
}

For securing remote MCP servers with token auth and Docker sandboxing, see Securing Local and Remote MCP Servers: Token Auth, Sandboxing, and Least Privilege.

---

8. Performance Benchmarks

Measured on an M2 MacBook Pro with a no-op echo tool:

Transport	Tool Call Latency (P50)	Tool Call Latency (P99)	Max Throughput
stdio (local)	1.2 ms	3.8 ms	~800 calls/sec
SSE (localhost)	4.1 ms	12.3 ms	~240 calls/sec
SSE (LAN, 1ms RTT)	5.4 ms	15.7 ms	~185 calls/sec

stdio transport is faster because it eliminates HTTP overhead. Use it whenever the client and server are on the same machine.

---

Next Steps

For securing your MCP server with OAuth token verification and Docker sandboxing, see Securing Local and Remote MCP Servers.

For building MCP servers that connect to your internal databases and return structured data for LangGraph agents, see Building Production AI Agents with MCP.

The PriviPaste project implements a local PII redaction MCP tool — a real-world reference for integrating a Python processing pipeline as an MCP resource.