14. Production MCP

flowchart TD
    subgraph DEV["Development"]
        PROTOTYPE["Local MCP Server\n(STDIO transport)"]
    end

    subgraph PROD["Production"]
        GATEWAY["API Gateway\n(Auth, Rate Limiting)"]
        LB["Load Balancer"]
        INST1["MCP Server Instance 1"]
        INST2["MCP Server Instance 2"]
        INST3["MCP Server Instance 3"]
        CACHE[("Redis Cache")]
        MON["Monitoring\n(Prometheus, Grafana)"]
        LOG[("Logs\n(Elasticsearch)")]
    end

    PROTOTYPE -->|"Deploy as HTTP"| GATEWAY
    GATEWAY --> LB
    LB --> INST1
    LB --> INST2
    LB --> INST3
    INST1 --> CACHE
    INST2 --> CACHE
    INST3 --> CACHE
    INST1 --> MON
    INST2 --> MON
    INST3 --> MON
    INST1 --> LOG
    INST2 --> LOG
    INST3 --> LOG

    style DEV fill:#3b82f6,color:#fff
    style PROD fill:#22c55e,color:#fff

flowchart TD
    REQ["Client Request"] --> AUTH{"Has valid\nauth token?"}
    AUTH -->|"No"| DENY["❌ 401 Unauthorized"]
    AUTH -->|"Yes"| CHECK{"Token valid\nnot expired?"}
    CHECK -->|"No"| DENY
    CHECK -->|"Yes"| PERM{"Has permission\nfor this tool?"}
    PERM -->|"No"| FORBID["❌ 403 Forbidden"]
    PERM -->|"Yes"| ALLOW["✅ Execute tool"]

    style DENY fill:#ef4444,color:#fff
    style FORBID fill:#ef4444,color:#fff
    style ALLOW fill:#22c55e,color:#fff

# MCP Server with authentication
from mcp.server import Server
from mcp.server.http import HTTPServerTransport
import jwt

# Auth middleware
async def authenticate_request(request):
    auth_header = request.headers.get("Authorization", "")
    token = auth_header.replace("Bearer ", "")

    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        return {
            "authenticated": True,
            "user_id": payload["sub"],
            "role": payload.get("role", "user"),
            "tenant": payload.get("tenant", "default")
        }
    except jwt.ExpiredSignatureError:
        raise PermissionError("Token expired")
    except jwt.InvalidTokenError:
        raise PermissionError("Invalid token")

# Authorization middleware
def authorize_tool(user, tool_name, arguments):
    """Check if user has permission to call a tool."""
    permissions = {
        "admin": ["*"],  # All tools
        "editor": ["search_docs", "get_file", "list_files"],
        "viewer": ["search_docs", "list_files"]
    }

    user_role = user.get("role", "viewer")
    allowed = permissions.get(user_role, [])

    if "*" not in allowed and tool_name not in allowed:
        raise PermissionError(f"Role '{user_role}' cannot call '{tool_name}'")

    return True

flowchart TD
    REQ["Client Request"] --> TLIMIT{"Token bucket\navailable?"}
    TLIMIT -->|"Yes"| CONS["Consume token\nProceed"]
    TLIMIT -->|"No"| RLIMIT{"Rate limit\nper client?"}
    RLIMIT -->|"Under limit"| CONS
    RLIMIT -->|"Exceeded"| WAIT["429 Too Many Requests\nRetry-After: 30s"]
    WAIT --> RETRY["Client retries\nafter wait"]
    RETRY --> TLIMIT

    style CONS fill:#22c55e,color:#fff
    style WAIT fill:#f59e0b,color:#fff

import time
from collections import defaultdict
import asyncio

class RateLimiter:
    """Token bucket rate limiter per client."""

    def __init__(self, rate: int, burst: int):
        self.rate = rate  # Requests per second
        self.burst = burst  # Maximum burst size
        self.tokens = defaultdict(lambda: burst)
        self.last_refill = defaultdict(time.time)

    async def check(self, client_id: str) -> bool:
        now = time.time()
        elapsed = now - self.last_refill[client_id]

        # Refill tokens
        self.tokens[client_id] = min(
            self.burst,
            self.tokens[client_id] + elapsed * self.rate
        )
        self.last_refill[client_id] = now

        if self.tokens[client_id] >= 1:
            self.tokens[client_id] -= 1
            return True
        return False

    async def get_retry_after(self, client_id: str) -> float:
        """Calculate how long client should wait."""
        deficit = 1 - self.tokens[client_id]
        return max(0, deficit / self.rate)

# Usage in production server
rate_limiter = RateLimiter(rate=10, burst=20)

@server.call_tool()
async def call_tool(name: str, arguments: dict, client_id: str = None):
    if not await rate_limiter.check(client_id):
        retry_after = await rate_limiter.get_retry_after(client_id)
        raise RateLimitError(
            f"Rate limit exceeded. Retry after {retry_after:.1f}s",
            retry_after=retry_after
        )
    # Execute tool...

sequenceDiagram
    participant Agent as AI Agent
    participant Server as MCP Server
    participant Metrics as Metrics Collector
    participant Monitor as Monitoring Dashboard
    participant Alert as Alert Manager

    Agent->>Server: tools/call
    Server->>Metrics: Increment tool_call_counter{name="search_docs"}
    Server->>Metrics: Record latency{name="search_docs", duration_ms=245}
    Server->>Agent: Result
    Metrics->>Monitor: Push metrics (Prometheus)

    Note over Monitor: Query: rate(tool_call_counter[5m])
    Note over Monitor: Alert if latency > 5s

    Monitor->>Alert: High latency detected
    Alert->>Alert: Send notification (PagerDuty/Slack)

import structlog

# Structured logging for MCP server
logger = structlog.get_logger()

@server.call_tool()
async def call_tool(name: str, arguments: dict, context: dict = None):
    request_id = context.get("request_id")
    client_id = context.get("client_id")

    logger.info("tool_call_started",
        request_id=request_id,
        client_id=client_id,
        tool_name=name,
        arguments_schema=list(arguments.keys())
    )

    try:
        result = await execute_tool(name, arguments)

        logger.info("tool_call_completed",
            request_id=request_id,
            tool_name=name,
            duration_ms=result.duration_ms,
            result_size=len(str(result.content))
        )

        return result
    except Exception as e:
        logger.error("tool_call_failed",
            request_id=request_id,
            tool_name=name,
            error=str(e),
            error_type=type(e).__name__
        )
        raise

flowchart TD
    subgraph USERS["Clients"]
        C1["Client 1"]
        C2["Client 2"]
        C3["Client 3"]
    end

    subgraph INFRA["Infrastructure"]
        DNS["DNS / Load Balancer"]
        GW["API Gateway\n(Rate Limit, Auth)"]
    end

    subgraph SERVERS["MCP Server Cluster"]
        S1["Instance 1"]
        S2["Instance 2"]
        S3["Instance 3"]
        S4["Instance N..."]
    end

    subgraph DATA["Data Layer"]
        CACHE[("Redis Cache")]
        DB[("PostgreSQL")]
    end

    C1 --> DNS
    C2 --> DNS
    C3 --> DNS
    DNS --> GW
    GW --> S1
    GW --> S2
    GW --> S3
    GW --> S4
    S1 --> CACHE
    S2 --> CACHE
    S3 --> DB
    S4 --> DB

    style USERS fill:#3b82f6,color:#fff
    style INFRA fill:#8b5cf6,color:#fff
    style SERVERS fill:#f59e0b,color:#fff
    style DATA fill:#22c55e,color:#fff

flowchart TD
    START["Security Review"] --> A["✅ Authentication configured?"]
    A --> B["✅ Authorization per tool?"]
    B --> C["✅ Input validation on all args?"]
    C --> D["✅ Rate limiting enabled?"]
    D --> E["✅ HTTPS/WSS enforced?"]
    E --> F["✅ Secrets in env vars not code?"]
    F --> G["✅ Output sanitization?"]
    G --> H["✅ Audit logging enabled?"]
    H --> I["✅ Dependency scanning?"]
    I --> J["✅ Penetration testing?"]
    J --> DONE["✅ Production Ready!"]

    style DONE fill:#22c55e,color:#fff