7. MCP in Production

Overview

You've built MCP servers, MCP clients, integrated them into LangGraph agents, and explored the ecosystem of community servers. All of that worked on your machine — python server.py, stdio transport, a single client, zero authentication. Perfect for development. Completely unviable for production.

Production is a different game. Your MCP server is going to get requests from multiple simultaneous agents. Some will send malformed arguments. Others will make 500 calls per minute to a tool that queries an external API with rate limits of its own. A server will crash at 3AM and nobody will find out until a user reports that "the agent stopped working."

This capsule covers the real challenges of taking MCP to production: deploying servers in containers and the cloud, authentication and authorization, rate limiting, monitoring and observability, error handling that doesn't kill the agent when a server fails, and tool versioning that lets you evolve without breaking things. This isn't theory — it's what you need to implement before your first real user touches the system.

Deploying MCP Servers

From stdio to HTTP

In development, you used the stdio transport — the server runs as a subprocess of the agent. In production, you need HTTP transport (SSE or Streamable HTTP) so the server is an independent service multiple clients can reach.

DEVELOPMENT (stdio)                    PRODUCTION (HTTP)
─────────────────                       ──────────────────
┌──────────┐                            ┌──────────┐
│  Agent   │                            │ Agent 1  │──┐
│          │──stdio──│ Server │          └──────────┘  │
└──────────┘         │(subproc)│        ┌──────────┐  │  HTTPS    ┌──────────────┐
                                        │ Agent 2  │──┼──────────►│  MCP Server  │
                                        └──────────┘  │          │  (Cloud Run) │
                                        ┌──────────┐  │          └──────────────┘
                                        │ Agent 3  │──┘
                                        └──────────┘
One process, one client.                One service, N clients.

Containerizing an MCP Server

FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY server.py .
EXPOSE 8080
CMD ["python", "server.py"]

The server configured for SSE:

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("research-tools")

@mcp.tool()
async def search_papers(query: str, max_results: int = 5) -> str:
    """Search academic papers by topic."""
    return f"Results for '{query}'..."

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

Build and test:

docker build -t mcp-research-server .
docker run -p 8080:8080 mcp-research-server

Cloud Deployment

An example on Cloud Run:

gcloud run deploy mcp-research-server \
    --image gcr.io/my-project/mcp-research-server \
    --port 8080 \
    --min-instances 1 \
    --max-instances 10 \
    --allow-unauthenticated=false

--min-instances 1 avoids cold starts. --allow-unauthenticated=false requires auth.

Multiple servers as microservices

In production, you don't want a mega-server with 50 tools. You want small, specialized servers — each deployed, scaled, and monitored independently:

services:
  research-tools:
    build: ./servers/research
    ports: ["8081:8080"]
    environment:
      - API_KEY=${RESEARCH_API_KEY}

  file-manager:
    build: ./servers/files
    ports: ["8082:8080"]
    volumes:
      - ./workspace:/data

  database-access:
    build: ./servers/database
    ports: ["8083:8080"]
    environment:
      - DATABASE_URL=${DATABASE_URL}

If file-manager needs more capacity, you scale it without affecting research-tools. If database-access crashes, the others keep working.

Authentication and Authorization

Why auth isn't optional

An MCP server without authentication is a server anyone can call. If your server has an execute_query that runs SQL against your production database, you just exposed all your data to the world.

API Keys

The simplest approach for internal servers:

import os
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("secure-server")

VALID_KEYS = {
    "agent-research": os.environ.get("RESEARCH_AGENT_KEY"),
    "agent-support": os.environ.get("SUPPORT_AGENT_KEY"),
}

def require_auth(func):
    async def wrapper(*args, **kwargs):
        ctx = mcp.get_context()
        api_key = ctx.request_context.get("headers", {}).get("x-api-key")
        if api_key not in VALID_KEYS.values():
            return "Error: Unauthorized. Invalid or missing API key."
        return await func(*args, **kwargs)
    return wrapper

@mcp.tool()
@require_auth
async def search_papers(query: str) -> str:
    """Search academic papers."""
    return f"Authenticated results for '{query}'..."

The client sends the key in the headers:

async def connect_authenticated():
    headers = {"x-api-key": os.environ["MCP_API_KEY"]}
    async with sse_client(url="https://mcp-research.myapp.com/sse",
                          headers=headers) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            tools = await session.list_tools()
            print(f"Connected with {len(tools.tools)} tools")

Per-tool permissions

Not every client should be able to call every tool. A support agent doesn't need delete_user:

PERMISSIONS = {
    "agent-research": ["search_papers", "get_paper_details", "summarize_text"],
    "agent-support": ["search_papers", "get_paper_details"],
    "agent-admin": ["search_papers", "get_paper_details", "summarize_text",
                     "delete_paper", "update_index"],
}

def check_permission(client_id: str, tool_name: str) -> bool:
    return tool_name in PERMISSIONS.get(client_id, [])

Rate Limiting

Why limit requests

Your MCP server probably calls external APIs: OpenAI for embeddings, Google for searches, databases with limited connection pools. Without rate limiting, a hyperactive agent can burn your OpenAI quota in minutes or get you banned from an external API.

Rate limiting per client and per tool

import time
from collections import defaultdict

class RateLimiter:
    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 is_allowed(self, client_id: str) -> bool:
        now = time.time()
        self.requests[client_id] = [
            t for t in self.requests[client_id] if t > now - self.window
        ]
        if len(self.requests[client_id]) >= self.max_requests:
            return False
        self.requests[client_id].append(now)
        return True

Different tools have different costs. search_papers is cheap, but generate_embeddings calls OpenAI ($$$):

TOOL_LIMITS = {
    "search_papers":       RateLimiter(max_requests=100, window_seconds=60),
    "generate_embeddings": RateLimiter(max_requests=10,  window_seconds=60),
    "execute_query":       RateLimiter(max_requests=30,  window_seconds=60),
}

def check_rate_limit(client_id: str, tool_name: str) -> tuple[bool, str]:
    limiter = TOOL_LIMITS.get(tool_name)
    if not limiter:
        return True, ""
    if not limiter.is_allowed(client_id):
        return False, f"Rate limit exceeded for '{tool_name}'."
    return True, ""

Token budgets

Besides per-request rate limits, control total token spend with a TokenBudget that tracks consumption per client and resets daily. If an agent uses tools that call OpenAI, a limit of 100k tokens/day per client avoids surprises on the bill. The logic is simple: consume(client_id, tokens) -> bool — if the day's accumulated total exceeds the limit, reject.

Monitoring and Observability

Structured logging

Flat logs (print("Tool called")) are useless in production. You need structured logging:

import json
import logging
from datetime import datetime, timezone

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

def log_tool_call(tool_name: str, client_id: str, args: dict,
                  duration_ms: float, error: str | None = None):
    logger.info(json.dumps({
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "event": "tool_call",
        "tool": tool_name,
        "client": client_id,
        "duration_ms": round(duration_ms, 2),
        "result_length": 0 if error else -1,
        "error": error,
    }))

With structured logs you can filter by tool, by client, and detect latency anomalies.

Key metrics

Track per tool: total calls, error count, and latency (avg, p95, p99). A ServerMetrics dataclass with record(tool, duration_ms, error) and summary() gives you instant visibility. Store latencies in a deque(maxlen=10_000) to avoid memory leaks in long-running servers.

Alerting

Error Handling at Scale

Categorizing errors (server-side)

import httpx

def handle_tool_error(tool_name: str, error: Exception) -> dict:
    if isinstance(error, ValueError):
        return {"error": True, "category": "validation",
                "message": f"Invalid arguments: {error}", "retry": False}
    elif isinstance(error, httpx.TimeoutException):
        return {"error": True, "category": "timeout",
                "message": f"Timeout in '{tool_name}'", "retry": True}
    elif isinstance(error, httpx.HTTPStatusError):
        retryable = error.response.status_code >= 500
        return {"error": True, "category": "external_api",
                "message": f"The API responded {error.response.status_code}",
                "retry": retryable}
    else:
        return {"error": True, "category": "internal",
                "message": f"Internal error in '{tool_name}'", "retry": False}

The retry field tells the client whether retrying makes sense. A validation error doesn't get fixed by retrying. A timeout might.

Retry with exponential backoff (client-side)

import asyncio
import random

async def call_tool_with_retry(session, tool_name: str, arguments: dict,
                                max_retries: int = 3, base_delay: float = 1.0) -> str:
    for attempt in range(max_retries + 1):
        try:
            result = await asyncio.wait_for(
                session.call_tool(tool_name, arguments), timeout=30.0)

            if result.isError:
                error_data = json.loads(result.content[0].text)
                if not error_data.get("retry") or attempt == max_retries:
                    return f"Definitive error: {error_data['message']}"
            else:
                return result.content[0].text

        except (asyncio.TimeoutError, ConnectionError):
            if attempt == max_retries:
                return f"'{tool_name}' failed after {max_retries + 1} attempts"

        delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
        await asyncio.sleep(delay)

Graceful degradation

When a server isn't available, the agent shouldn't crash. The pattern: try calling the server → if it fails, mark it as down → use a local fallback if one exists → in the background, try to reconnect periodically. A ResilientMCPManager keeps a server_health: dict[str, bool] dict and, on detecting a failure, launches an async task that pings every 30 seconds until the server responds. Meanwhile, the agent keeps operating with the servers that do work.

Tool Versioning

The problem

Your server has search_papers(query: str). Three agents use it. Now you need to add date_range and change the result format. If you change it directly, you break all three.

Strategy 1: Backwards-compatible changes (preferred)

@mcp.tool()
async def search_papers(query: str, max_results: int = 5,
                        date_range: str | None = None,
                        output_format: str = "text") -> str:
    """Search academic papers by topic."""
    results = await _do_search(query, max_results, date_range)
    if output_format == "json":
        return json.dumps(results)
    return _format_as_text(results)

New parameters with defaults → existing clients keep working. Always the first option.

Strategy 2: Explicit tool versioning

For breaking changes, version the tool:

@mcp.tool()
async def search_papers(query: str) -> str:
    """⚠️ DEPRECATED — Use search_papers_v2. Removed 2025-09-01."""
    return _format_as_text(await _do_search(query))

@mcp.tool()
async def search_papers_v2(query: str, max_results: int = 5,
                            date_range: str | None = None) -> str:
    """[CURRENT] Search papers with advanced filters. Returns JSON."""
    return json.dumps(await _do_search(query, max_results, date_range))

Keep v1 while you migrate clients. When traffic to v1 reaches zero, you retire it.

The deprecation workflow

1. Add the new tool (v2) alongside the existing one (v1)
2. Mark v1 as deprecated in the description
3. Monitor v1's usage (per-tool metrics)
4. Notify the teams using v1
5. When v1's usage → 0, delete it

Comparison: Dev vs Production MCP

Every item in the "Production" column is a deliberate decision that prevents a future incident.

Connection to the Project

In this module's project (capsule 08), this capsule's techniques apply directly:

1. Deployment: The 3 MCP servers run in Docker containers with docker-compose
2. Authentication: Each server requires an API key sent in the connection headers
3. Error handling: The agent implements call_tool_with_retry with exponential backoff and graceful degradation
4. Monitoring: Each server logs tool calls in structured JSON — tool, duration, errors
5. Versioning: New optional parameters on existing tools, without breaking the agent's flow

When you finish the project, you'll have a mini MCP infrastructure with containers, auth, retry and logs — it would only need cloud deployment and alerts to be genuine production.

Troubleshooting

Problem 1: "Connection refused" when connecting to the HTTP server

Symptom: ConnectionRefusedError when connecting via SSE.

Cause: The server listens on 127.0.0.1 inside the container. From outside the container, that address isn't reachable.

Solution: Configure host="0.0.0.0":

mcp.run(transport="sse", host="0.0.0.0", port=8080)

Problem 2: Frequent timeouts in tools that call external APIs

Symptom: asyncio.TimeoutError in tools that query OpenAI, Google, etc.

Cause: The client's timeout (30s) is shorter than the API's response time under load.

Solution: Tiered timeouts — external API < tool < client:

API_TIMEOUT = 15       # httpx timeout for the external API
TOOL_TIMEOUT = 20      # the tool's internal timeout
CLIENT_TIMEOUT = 30    # the client's timeout

Problem 3: A memory leak in long-running servers

Symptom: The container's memory usage grows until the OOM killer kills it.

Cause: Accumulation in in-memory structures (metrics, rate limiter history) with no cleanup.

Solution: Periodic cleanup or a deque(maxlen=N) to cap the size:

from collections import deque
latencies: deque[float] = deque(maxlen=10_000)

Problem 4: Duplicate tools when connecting multiple servers

Symptom: Two servers expose search. The agent uses the wrong one.

Solution: Namespacing with server__tool (capsule 04):

namespaced = f"{server_name}__{tool.name}"

Exercises

Exercise 1: A rate limiter with a sliding window (Easy)

Implement a SlidingWindowLimiter with allow(client_id) -> bool and remaining(client_id) -> int. A maximum of N requests in W seconds. Test it: 5 requests in 10 seconds — the 6th gets rejected.

import time

class SlidingWindowLimiter:
    def __init__(self, max_requests: int, window_seconds: int):
        self.max_requests = max_requests
        self.window = window_seconds
        self.timestamps: dict[str, list[float]] = {}

    def allow(self, client_id: str) -> bool:
        now = time.time()
        self.timestamps.setdefault(client_id, [])
        self.timestamps[client_id] = [
            t for t in self.timestamps[client_id] if t > now - self.window
        ]
        if len(self.timestamps[client_id]) >= self.max_requests:
            return False
        self.timestamps[client_id].append(now)
        return True

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

limiter = SlidingWindowLimiter(max_requests=5, window_seconds=10)
for i in range(7):
    ok = limiter.allow("agent-1")
    print(f"Request {i+1}: {'✓' if ok else '✗'} (remaining: {limiter.remaining('agent-1')})")

Request 1: ✓ (remaining: 4)
Request 2: ✓ (remaining: 3)
Request 3: ✓ (remaining: 2)
Request 4: ✓ (remaining: 1)
Request 5: ✓ (remaining: 0)
Request 6: ✗ (remaining: 0)
Request 7: ✗ (remaining: 0)

Exercise 2: A structured logger for MCP tools (Easy)

Create a ToolLogger with log_call(tool, client, duration_ms, error) and get_recent(n). Use a deque(maxlen=10_000) to avoid memory leaks. Add error_rate() and avg_duration(tool).

from collections import deque
from datetime import datetime, timezone

class ToolLogger:
    def __init__(self, max_entries: int = 10_000):
        self.entries: deque[dict] = deque(maxlen=max_entries)

    def log_call(self, tool: str, client: str, duration_ms: float,
                 error: str | None = None):
        self.entries.append({
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "tool": tool, "client": client,
            "duration_ms": round(duration_ms, 2),
            "error": error, "success": error is None,
        })

    def get_recent(self, n: int = 10) -> list[dict]:
        return list(self.entries)[-n:]

    def error_rate(self) -> float:
        if not self.entries:
            return 0.0
        return sum(1 for e in self.entries if not e["success"]) / len(self.entries)

    def avg_duration(self, tool: str | None = None) -> float:
        relevant = [e for e in self.entries if tool is None or e["tool"] == tool]
        if not relevant:
            return 0.0
        return sum(e["duration_ms"] for e in relevant) / len(relevant)

log = ToolLogger()
log.log_call("search_papers", "agent-1", 142.5)
log.log_call("summarize_text", "agent-1", 523.1)
log.log_call("search_papers", "agent-2", 98.3)
log.log_call("generate_embeddings", "agent-1", 0, error="Timeout")

print(f"Error rate: {log.error_rate():.0%}")
print(f"Avg search: {log.avg_duration('search_papers'):.1f}ms")

Exercise 3: Retry with exponential backoff (Medium)

Implement an async retry_with_backoff(func, max_retries, base_delay). If it fails with a TimeoutError or ConnectionError, retry with a delay of base_delay * 2^attempt + jitter. Test it with a function that fails twice and succeeds on the 3rd.

import asyncio
import random

RETRYABLE = (TimeoutError, ConnectionError, OSError)

async def retry_with_backoff(func, max_retries=3, base_delay=1.0, max_delay=30.0):
    last_error = None
    for attempt in range(max_retries + 1):
        try:
            result = await func()
            if attempt > 0:
                print(f"  ✓ Success on attempt {attempt + 1}")
            return result
        except RETRYABLE as e:
            last_error = e
            if attempt == max_retries:
                break
            delay = min(base_delay * (2 ** attempt), max_delay)
            jitter = random.uniform(0, delay * 0.3)
            print(f"  ✗ Attempt {attempt+1} failed: {e}. Retry in {delay+jitter:.1f}s...")
            await asyncio.sleep(delay + jitter)
        except Exception as e:
            raise e
    raise last_error

call_count = 0
async def flaky_api():
    global call_count
    call_count += 1
    if call_count < 3:
        raise TimeoutError(f"The API didn't respond (attempt {call_count})")
    return {"status": "ok", "data": "papers found"}

async def main():
    global call_count
    call_count = 0
    result = await retry_with_backoff(flaky_api, max_retries=4, base_delay=0.1)
    print(f"  Result: {result}")

asyncio.run(main())

  ✗ Attempt 1 failed: The API didn't respond (attempt 1). Retry in 0.1s...
  ✗ Attempt 2 failed: The API didn't respond (attempt 2). Retry in 0.2s...
  ✓ Success on attempt 3
  Result: {'status': 'ok', 'data': 'papers found'}

Exercise 4: A health monitor for multiple servers (Medium)

Implement an MCPHealthMonitor with add_server(name, url), check_server(name, check_fn), and status_report(). A server gets marked "unhealthy" if the last 3 consecutive checks failed.

import asyncio
from dataclasses import dataclass, field
from collections import deque
from datetime import datetime, timezone

@dataclass
class HealthRecord:
    timestamp: str
    healthy: bool
    error: str | None = None

@dataclass
class ServerHealth:
    name: str
    history: deque = field(default_factory=lambda: deque(maxlen=50))

    @property
    def is_healthy(self) -> bool:
        recent = list(self.history)[-3:]
        if len(recent) < 3:
            return len(recent) == 0 or recent[-1].healthy
        return any(r.healthy for r in recent)

class MCPHealthMonitor:
    def __init__(self):
        self.servers: dict[str, ServerHealth] = {}

    def add_server(self, name: str):
        self.servers[name] = ServerHealth(name=name)

    async def check_server(self, name: str, check_fn) -> bool:
        server = self.servers[name]
        try:
            await asyncio.wait_for(check_fn(), timeout=5.0)
            server.history.append(HealthRecord(
                datetime.now(timezone.utc).isoformat(), True))
            return True
        except Exception as e:
            server.history.append(HealthRecord(
                datetime.now(timezone.utc).isoformat(), False, str(e)))
            return False

    def status_report(self) -> dict:
        return {n: {"healthy": s.is_healthy, "checks": len(s.history)}
                for n, s in self.servers.items()}

Exercise 5: Backwards-compatible tool evolution (Hard)

Create a versioned-tools server with analyze_text in 3 versions: v1 accepts only text, v2 adds an optional language, v3 adds an optional output_format. All 3 coexist as separate tools. Add a tools://migration-guide resource that documents the changes.

import json
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("versioned-tools")

async def _analyze(text: str, lang: str = "auto", fmt: str = "text") -> dict:
    words = len(text.split())
    return {"word_count": words, "char_count": len(text),
            "avg_word_len": round(len(text) / max(words, 1), 1),
            "language": lang, "format": fmt}

@mcp.tool()
async def analyze_text(text: str) -> str:
    """⚠️ DEPRECATED (v1) — Use analyze_text_v3. Removed 2025-10-01."""
    r = await _analyze(text)
    return f"Words: {r['word_count']}, Chars: {r['char_count']}"

@mcp.tool()
async def analyze_text_v2(text: str, language: str = "auto") -> str:
    """⚠️ DEPRECATED (v2) — Use analyze_text_v3. Removed 2025-12-01."""
    r = await _analyze(text, lang=language)
    return f"Words: {r['word_count']}, Avg: {r['avg_word_len']}, Lang: {r['language']}"

@mcp.tool()
async def analyze_text_v3(text: str, language: str = "auto",
                           output_format: str = "text") -> str:
    """[CURRENT] Analyze text with a configurable language and format."""
    r = await _analyze(text, lang=language, fmt=output_format)
    if output_format == "json":
        return json.dumps(r, indent=2)
    return f"Words: {r['word_count']}, Avg: {r['avg_word_len']}, Lang: {r['language']}"

@mcp.resource("tools://migration-guide")
def migration_guide() -> str:
    """A migration guide between tool versions."""
    return """v1→v2: Added language param (optional, default 'auto')
v2→v3: Added output_format param (optional, default 'text')
Deprecation: v1 removes 2025-10-01, v2 removes 2025-12-01"""

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

Summary

In this capsule you learned:

• Deployment: MCP servers in production run as HTTP services (SSE), containerized with Docker, deployed on Cloud Run, Railway, or any container platform. Stdout is no longer the transport — it's an independent service that scales horizontally
• Authentication and authorization: API keys to identify clients, per-tool permissions to control what each one can do. Without auth, your server is an open door
• Rate limiting: A sliding window per client and per tool, daily token budgets. It protects your external APIs, your infrastructure, and your budget
• Monitoring: Structured JSON logging, per-tool metrics (calls, errors, latency), health checks, alerts. Without monitoring, you're flying blind
• Error handling at scale: Categorizing errors (retryable vs definitive), exponential backoff with jitter, graceful degradation. The agent doesn't crash because a server failed
• Versioning: Backwards-compatible changes with optional parameters, tool versioning for breaking changes, a deprecation workflow with dates. Evolve without breaking

Next capsule: Project — Agent with MCP. You'll build Research Agent v4 connected to 3 MCP servers with containerized deployment, auth, retry, and monitoring.

Additional Resources

1. MCP Specification — Transports — The official specification for the stdio and HTTP transports
2. MCP Python SDK — SSE Server — SDK documentation for servers with the SSE transport
3. Docker Best Practices for Python — Docker's official guide to containerizing Python apps
4. Google Cloud Run Documentation — Serverless container deployment for MCP servers
5. Exponential Backoff and Jitter (AWS) — The reference article on retry strategies