Remote MCP Servers in Production: Streamable HTTP, OAuth 2.1 & Enterprise Gateway Architecture

import hmac
import logging
from typing import Any, Dict, List, Optional
from fastapi import FastAPI, Header, HTTPException, Request, Response, status
from pydantic import BaseModel, Field

# Configure structured stderr logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s [%(levelname)s] %(name)s: %(message)s'
)
logger = logging.getLogger("remote-mcp-server")

app = FastAPI(
    title="Production Remote MCP Service",
    version="1.0.0",
    docs_url=None,
    redoc_url=None
)

ALLOWED_ORIGINS = {
    "https://console.anthropic.com",
    "https://cursor.com",
    "https://internal-agent-orchestrator.company.net"
}
EXPECTED_API_KEY = "sk-mcp-prod-9f8a3c2e1b4d5e6f7a8b9c0d"

class ToolDefinition(BaseModel):
    name: str
    description: str
    inputSchema: Dict[str, Any]

class JsonRpcRequest(BaseModel):
    jsonrpc: str = Field(default="2.0")
    id: Optional[Any] = None
    method: str
    params: Optional[Dict[str, Any]] = Field(default_factory=dict)

# Available production tools advertised by this server
CATALOG: Dict[str, ToolDefinition] = {
    "query_sales_metrics": ToolDefinition(
        name="query_sales_metrics",
        description="Query aggregated daily sales volume and refund metrics by region.",
        inputSchema={
            "type": "object",
            "properties": {
                "region": {"type": "string", "enum": ["apac", "emea", "amer"]},
                "days": {"type": "integer", "minimum": 1, "maximum": 90, "default": 30}
            },
            "required": ["region"]
        }
    ),
    "inspect_system_health": ToolDefinition(
        name="inspect_system_health",
        description="Fetch CPU saturation, memory utilization, and p99 latency for an upstream cluster.",
        inputSchema={
            "type": "object",
            "properties": {
                "cluster_id": {"type": "string", "pattern": "^[a-z0-9-]+$"}
            },
            "required": ["cluster_id"]
        }
    )
}

def verify_request_security(origin: Optional[str], authorization: Optional[str]) -> None:
    # 1. Defend against DNS rebinding
    if origin and origin not in ALLOWED_ORIGINS:
        logger.warning(f"Rejected unauthorized origin: {origin}")
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Forbidden: Invalid or untrusted Origin header"
        )

    # 2. Verify Bearer token
    if not authorization or not authorization.startswith("Bearer "):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Unauthorized: Missing or malformed Bearer token",
            headers={"WWW-Authenticate": "Bearer"}
        )
    
    token = authorization.split(" ", 1)[1].strip()
    if not hmac.compare_digest(token, EXPECTED_API_KEY):
        logger.warning("Rejected invalid API token attempt")
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Forbidden: Invalid authentication credentials"
        )

@app.post("/mcp", response_model=None)
async def handle_mcp_request(
    request: Request,
    origin: Optional[str] = Header(None),
    authorization: Optional[str] = Header(None)
):
    # Verify security gates prior to parsing JSON-RPC payload
    verify_request_security(origin=origin, authorization=authorization)

    try:
        raw_body = await request.json()
        payload = JsonRpcRequest(**raw_body)
    except Exception as exc:
        logger.error(f"Failed to parse JSON-RPC payload: {exc}")
        return Response(
            content='{"jsonrpc": "2.0", "error": {"code": -32700, "message": "Parse error"}, "id": null}',
            media_type="application/json",
            status_code=400
        )

    logger.info(f"Incoming MCP method: {payload.method} (id: {payload.id})")

    # 1. Protocol Handshake: Initialize
    if payload.method == "initialize":
        result = {
            "protocolVersion": "2026-11-01",
            "capabilities": {
                "tools": {"listChanged": False},
                "resources": {"subscribe": False, "listChanged": False}
            },
            "serverInfo": {
                "name": "enterprise-remote-mcp-gateway",
                "version": "1.0.0"
            }
        }
        return {"jsonrpc": "2.0", "id": payload.id, "result": result}

    # 2. Client ping / keepalive
    elif payload.method == "ping":
        return {"jsonrpc": "2.0", "id": payload.id, "result": {}}

    # 3. List available tools
    elif payload.method == "tools/list":
        tools_list = [tool.model_dump() for tool in CATALOG.values()]
        return {
            "jsonrpc": "2.0",
            "id": payload.id,
            "result": {"tools": tools_list}
        }

    # 4. Invoke a specific tool
    elif payload.method == "tools/call":
        params = payload.params or {}
        tool_name = params.get("name")
        arguments = params.get("arguments", {})

        if tool_name not in CATALOG:
            return {
                "jsonrpc": "2.0",
                "id": payload.id,
                "error": {"code": -32601, "message": f"Tool '{tool_name}' not found in catalog"}
            }

        # Dispatch execution safely
        try:
            if tool_name == "query_sales_metrics":
                region = arguments.get("region", "apac")
                days = arguments.get("days", 30)
                output_text = f"Region: {region.upper()} | Window: Last {days} days | Total Gross GMV: $1,420,500 | Refund Rate: 0.84%"
            elif tool_name == "inspect_system_health":
                cluster = arguments.get("cluster_id")
                output_text = f"Cluster {cluster}: Status HEALTHY | P99 Latency: 14.2ms | CPU Saturation: 42% | Error Budget: 99.98%"
            else:
                output_text = "No output produced"

            return {
                "jsonrpc": "2.0",
                "id": payload.id,
                "result": {
                    "content": [
                        {"type": "text", "text": output_text}
                    ],
                    "isError": False
                }
            }
        except Exception as err:
            logger.exception(f"Tool execution failure for '{tool_name}': {err}")
            return {
                "jsonrpc": "2.0",
                "id": payload.id,
                "result": {
                    "content": [{"type": "text", "text": f"Internal execution error: {str(err)}"}],
                    "isError": True
                }
            }

    # Unhandled method
    return {
        "jsonrpc": "2.0",
        "id": payload.id,
        "error": {"code": -32601, "message": f"Method '{payload.method}' not implemented"}
    }

# /etc/nginx/conf.d/mcp_gateway.conf
upstream mcp_cluster {
    server 127.0.0.1:8000 max_fails=3 fail_timeout=10s;
    keepalive 64;
}

limit_req_zone $binary_remote_addr zone=mcp_ip_limit:10m rate=15r/s;

server {
    listen 443 ssl http2;
    server_name mcp.company.internal;

    ssl_certificate /etc/ssl/certs/mcp_bundle.crt;
    ssl_certificate_key /etc/ssl/private/mcp.key;
    ssl_protocols TLSv1.3 TLSv1.2;

    location /mcp {
        limit_req zone=mcp_ip_limit burst=30 nodelay;

        # Disable proxy response buffering for streaming support
        proxy_buffering off;
        proxy_cache off;
        proxy_read_timeout 120s;
        proxy_send_timeout 120s;

        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;

        proxy_pass http://mcp_cluster;
        proxy_http_version 1.1;
        proxy_set_header Connection "";
    }
}

{
  "mcpServers": {
    "enterprise-analytics": {
      "url": "https://mcp.company.internal/mcp",
      "headers": {
        "Authorization": "Bearer sk-mcp-prod-9f8a3c2e1b4d5e6f7a8b9c0d",
        "Origin": "https://console.anthropic.com"
      }
    }
  }
}