Xây dựng Remote MCP Server trong môi trường Production: Streamable HTTP, OAuth 2.1 & Kiến trúc Gateway Doanh nghiệp

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

# Cấu hình logging có cấu trúc ra stderr
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)

# Danh mục công cụ sản xuất được cung cấp bởi server này
CATALOG: Dict[str, ToolDefinition] = {
    "query_sales_metrics": ToolDefinition(
        name="query_sales_metrics",
        description="Truy vấn số liệu doanh số tổng hợp và tỷ lệ hoàn tiền theo khu vực.",
        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="Lấy dữ liệu độ bão hòa CPU, mức dùng bộ nhớ và độ trễ p99 của cụm upstream.",
        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. Phòng vệ chống DNS rebinding
    if origin and origin not in ALLOWED_ORIGINS:
        logger.warning(f"Từ chối origin không được cấp quyền: {origin}")
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Forbidden: Header Origin không hợp lệ hoặc không đáng tin cậy"
        )

    # 2. Xác thực Bearer token
    if not authorization or not authorization.startswith("Bearer "):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Unauthorized: Thiếu hoặc sai định dạng Bearer token",
            headers={"WWW-Authenticate": "Bearer"}
        )
    
    token = authorization.split(" ", 1)[1].strip()
    if not hmac.compare_digest(token, EXPECTED_API_KEY):
        logger.warning("Từ chối nỗ lực xác thực token API sai")
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Forbidden: Thông tin xác thực không chính xác"
        )

@app.post("/mcp", response_model=None)
async def handle_mcp_request(
    request: Request,
    origin: Optional[str] = Header(None),
    authorization: Optional[str] = Header(None)
):
    # Xác minh các cổng bảo mật trước khi phân tích payload JSON-RPC
    verify_request_security(origin=origin, authorization=authorization)

    try:
        raw_body = await request.json()
        payload = JsonRpcRequest(**raw_body)
    except Exception as exc:
        logger.error(f"Lỗi phân tích cú pháp JSON-RPC: {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"Yêu cầu phương thức MCP: {payload.method} (id: {payload.id})")

    # 1. Bắt tay giao thức: Khởi tạo
    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. Kiểm tra kết nối ping
    elif payload.method == "ping":
        return {"jsonrpc": "2.0", "id": payload.id, "result": {}}

    # 3. Liệt kê danh sách công cụ
    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. Thực thi công cụ cụ thể
    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"Không tìm thấy công cụ '{tool_name}'"}
            }

        try:
            if tool_name == "query_sales_metrics":
                region = arguments.get("region", "apac")
                days = arguments.get("days", 30)
                output_text = f"Khu vực: {region.upper()} | Khung thời gian: {days} ngày gần nhất | Tổng GMV: $1,420,500 | Tỷ lệ hoàn tiền: 0.84%"
            elif tool_name == "inspect_system_health":
                cluster = arguments.get("cluster_id")
                output_text = f"Cụm {cluster}: Trạng thái HEALTHY | Độ trễ P99: 14.2ms | Bão hòa CPU: 42% | Error Budget: 99.98%"
            else:
                output_text = "Không có dữ liệu trả về"

            return {
                "jsonrpc": "2.0",
                "id": payload.id,
                "result": {
                    "content": [
                        {"type": "text", "text": output_text}
                    ],
                    "isError": False
                }
            }
        except Exception as err:
            logger.exception(f"Lỗi khi thực thi công cụ '{tool_name}': {err}")
            return {
                "jsonrpc": "2.0",
                "id": payload.id,
                "result": {
                    "content": [{"type": "text", "text": f"Lỗi thực thi nội bộ: {str(err)}"}],
                    "isError": True
                }
            }

    # Phương thức chưa được hỗ trợ
    return {
        "jsonrpc": "2.0",
        "id": payload.id,
        "error": {"code": -32601, "message": f"Phương thức '{payload.method}' chưa được cài đặt"}
    }

# /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;

        # Tắt bộ đệm phản hồi proxy để hỗ trợ streaming
        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"
      }
    }
  }
}