Session 4: Production MCP Deployment - Enterprise-Grade Infrastructure¶

# src/production_mcp_server.py - Enterprise MCP Server

import os
import json
import structlog  # Enterprise structured logging
from datetime import datetime
from typing import Dict, Any, Optional
from mcp.server.fastmcp import FastMCP
import aioredis
import asyncio
from prometheus_client import Counter, Histogram, Gauge, start_http_server
from tenacity import retry, stop_after_attempt, wait_exponential
import time

# Configure enterprise-grade structured logging

structlog.configure(
    processors=[
        structlog.stdlib.filter_by_level,
        structlog.stdlib.add_logger_name,
        structlog.stdlib.add_log_level,
        structlog.stdlib.PositionalArgumentsFormatter(),
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.StackInfoRenderer(),
        structlog.processors.format_exc_info,
        structlog.processors.UnicodeDecoder(),
        structlog.processors.JSONRenderer()  # JSON for log aggregation
    ],
    context_class=dict,
    logger_factory=structlog.stdlib.LoggerFactory(),
    cache_logger_on_first_use=True,
)

logger = structlog.get_logger()

# Enterprise-grade Prometheus metrics

# Request metrics

request_count = Counter(
    'mcp_requests_total', 
    'Total MCP requests', 
    ['method', 'status', 'client_id', 'version']
)
request_duration = Histogram(
    'mcp_request_duration_seconds', 
    'MCP request duration in seconds',
    ['method', 'endpoint'],
    buckets=(0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0)
)

# System health metrics

active_connections = Gauge('mcp_active_connections', 'Active MCP connections')
cpu_usage = Gauge('mcp_cpu_usage_percent', 'CPU usage percentage')
memory_usage = Gauge('mcp_memory_usage_bytes', 'Memory usage in bytes')

# Business metrics

tool_usage = Counter('mcp_tool_usage_total', 'Tool usage counts', ['tool_name', 'success'])
cache_hits = Counter('mcp_cache_hits_total', 'Cache hit/miss counts', ['type'])
ratelimit_violations = Counter('mcp_ratelimit_violations_total', 'Rate limit violations')

# Start Prometheus metrics server

start_http_server(9090)  # Metrics available at :9090/metrics
logger.info("Prometheus metrics server started", port=9090)

@app.route('/health')
async def health_check():
    """Kubernetes liveness probe - is the service running?"""
    return {"status": "healthy", "timestamp": datetime.utcnow().isoformat()}

@app.route('/ready')
async def readiness_check():
    """Kubernetes readiness probe - is the service ready to handle traffic?"""
    checks = {
        "redis": await check_redis_connection(),
        "mcp_servers": await check_mcp_servers(),
        "disk_space": check_disk_space(),
    }

    all_healthy = all(checks.values())
    status_code = 200 if all_healthy else 503

    return {
        "status": "ready" if all_healthy else "not_ready",
        "checks": checks,
        "timestamp": datetime.utcnow().isoformat()
    }, status_code

class ProductionMCPServer:
    """
    Production-ready MCP server with comprehensive monitoring and caching.

    This server includes all the features needed for production deployment:
    - Redis caching for improved performance
    - Prometheus metrics for monitoring
    - Structured logging for debugging
    - Health checks for load balancers
    - Environment-based configuration
    """

    def __init__(self, name: str = "Production MCP Server"):
        self.mcp = FastMCP(name)
        self.redis_client: Optional[aioredis.Redis] = None
        self.cache_ttl = int(os.getenv('CACHE_TTL', '300'))  # 5 minutes default
        self.start_time = time.time()

        # Configuration from environment variables - essential for cloud deployment
        self.config = {
            'redis_url': os.getenv('REDIS_URL', 'redis://localhost:6379'),
            'max_request_size': int(os.getenv('MAX_REQUEST_SIZE', '1048576')),  # 1MB
            'rate_limit': int(os.getenv('RATE_LIMIT', '100')),  # requests per minute
            'environment': os.getenv('ENVIRONMENT', 'development')
        }

        self._setup_tools()
        self._setup_monitoring()

    async def initialize(self):
        """
        Initialize async resources like Redis connections.

        This separation allows the server to start synchronously but
        initialize expensive resources asynchronously.
        """
        try:
            self.redis_client = await aioredis.from_url(
                self.config['redis_url'],
                encoding="utf-8",
                decode_responses=True
            )
            logger.info("Redis connection established")
        except Exception as e:
            logger.warning(f"Redis connection failed: {e}. Running without cache.")

    def _setup_tools(self):
        """Set up MCP tools with decorators for monitoring."""

        @self.mcp.tool()
        @self._monitor_tool
        async def process_data(data: Dict[str, Any], operation: str = "transform") -> Dict:
            """
            Process data with specified operation, demonstrating caching patterns.

            This tool shows how to implement caching in production MCP servers.
            The cache key is based on the input data hash, ensuring consistency.
            """

            # Generate deterministic cache key based on input
            cache_key = f"process:{operation}:{hash(json.dumps(data, sort_keys=True))}"

            # Check cache first - this can dramatically improve response times
            cached = await self._get_cache(cache_key)
            if cached:
                logger.info(f"Cache hit for operation: {operation}")
                return cached

            # Perform actual processing
            logger.info(f"Processing data with operation: {operation}")
            result = {
                "operation": operation,
                "input_size": len(json.dumps(data)),
                "processed_at": datetime.now().isoformat(),
                "result": self._perform_operation(data, operation),
                "cache_status": "miss"
            }

            # Cache the result for future requests
            await self._set_cache(cache_key, result)
            return result

        @self.mcp.tool()
        @self._monitor_tool
        async def health_check() -> Dict:
            """
            Health check endpoint for load balancers and monitoring systems.

            This is crucial for production deployment - load balancers use this
            to determine if the server is ready to handle requests.
            """
            checks = {
                "status": "healthy",
                "timestamp": datetime.now().isoformat(),
                "uptime_seconds": time.time() - self.start_time,
                "environment": self.config['environment'],
                "version": "1.0.0"
            }

            # Check external dependency health
            if self.redis_client:
                try:
                    await self.redis_client.ping()
                    checks["redis"] = "connected"
                except Exception as e:
                    checks["redis"] = "disconnected"
                    checks["status"] = "degraded"
                    logger.warning(f"Redis health check failed: {e}")
            else:
                checks["redis"] = "not_configured"

            return checks

# src/cloud_run_adapter.py

from fastapi import FastAPI, Request, Response
from fastapi.responses import JSONResponse, StreamingResponse
import json
import asyncio
from typing import AsyncIterator
import os
import logging

from src.production_mcp_server import ProductionMCPServer

# Configure logging for Cloud Run (structured logging recommended)

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI(
    title="MCP Server on Cloud Run",
    description="Production MCP server deployed on Google Cloud Run",
    version="1.0.0"
)

# Global server instance

server = ProductionMCPServer()

@app.on_event("startup")
async def startup_event():
    """
    Initialize server on startup.

    Cloud Run containers start fresh for each deployment,
    so we initialize our server and its dependencies here.
    """
    logger.info("Initializing MCP server for Cloud Run...")
    await server.initialize()
    logger.info("MCP server ready to handle requests")

@app.on_event("shutdown")
async def shutdown_event():
    """
    Cleanup on shutdown.

    Proper cleanup ensures graceful shutdown when Cloud Run
    terminates the container.
    """
    logger.info("Shutting down MCP server...")
    if server.redis_client:
        await server.redis_client.close()

@app.post("/mcp")
async def handle_mcp_request(request: Request):
    """
    Handle MCP requests over HTTP.

    This endpoint converts HTTP requests to MCP JSON-RPC format
    and routes them to the appropriate MCP tools.
    """
    try:
        body = await request.json()
        logger.info(f"Received MCP request: {body.get('method', 'unknown')}")

        # Route to appropriate handler based on method
        method = body.get("method", "")
        params = body.get("params", {})

        if method == "tools/list":
            # List available tools
            tools = server.mcp.list_tools()
            return JSONResponse(content={
                "jsonrpc": "2.0",
                "result": tools,
                "id": body.get("id")
            })

        elif method.startswith("tools/call"):
            # Execute a specific tool
            tool_name = params.get("name")
            tool_params = params.get("arguments", {})

            # Find and execute the requested tool
            tool = server.mcp.get_tool(tool_name)
            if tool:
                result = await tool(**tool_params)
                return JSONResponse(content={
                    "jsonrpc": "2.0",
                    "result": result,
                    "id": body.get("id")
                })
            else:
                return JSONResponse(
                    content={
                        "jsonrpc": "2.0",
                        "error": {"code": -32601, "message": f"Tool '{tool_name}' not found"},
                        "id": body.get("id")
                    },
                    status_code=404
                )

        else:
            return JSONResponse(
                content={
                    "jsonrpc": "2.0",
                    "error": {"code": -32601, "message": f"Method '{method}' not found"},
                    "id": body.get("id")
                },
                status_code=404
            )

    except json.JSONDecodeError:
        return JSONResponse(
            content={
                "jsonrpc": "2.0",
                "error": {"code": -32700, "message": "Parse error"},
                "id": None
            },
            status_code=400
        )
    except Exception as e:
        logger.error(f"Unexpected error: {str(e)}")
        return JSONResponse(
            content={
                "jsonrpc": "2.0",
                "error": {"code": -32603, "message": "Internal error"},
                "id": body.get("id", None)
            },
            status_code=500
        )

@app.get("/health")
async def health_check():
    """Health check endpoint for Cloud Run."""
    try:
        health = await server.mcp.get_tool("health_check")()

        if health.get("status") == "healthy":
            return health
        else:
            return JSONResponse(content=health, status_code=503)

    except Exception as e:
        return JSONResponse(
            content={"status": "unhealthy", "error": str(e)},
            status_code=503
        )

@app.get("/metrics")
async def metrics():
    """Prometheus metrics endpoint."""
    from prometheus_client import generate_latest
    return Response(content=generate_latest(), media_type="text/plain")

# src/lambda_handler.py

import json
import os
import asyncio
from typing import Dict, Any
import logging
from mangum import Mangum

# Import our FastAPI app from the Cloud Run adapter

from src.cloud_run_adapter import app

# Configure logging for Lambda

logger = logging.getLogger()
logger.setLevel(logging.INFO)

# Create Mangum handler to convert ASGI app to Lambda handler

handler = Mangum(app, lifespan="off")

def lambda_handler_direct(event: Dict[str, Any], context: Any) -> Dict[str, Any]:
    """
    Direct Lambda handler for MCP requests without FastAPI overhead.

    This approach provides maximum control and minimum cold start time
    for simple MCP operations.

    Args:
        event: Lambda event containing the HTTP request
        context: Lambda context with runtime information

    Returns:
        HTTP response in Lambda proxy integration format
    """
    try:
        # Log request for debugging
        logger.info(f"Lambda invoked with event: {json.dumps(event, default=str)}")

        # Parse the HTTP request body
        body = json.loads(event.get('body', '{}'))
        method = body.get('method', '')

        # Handle different MCP methods
        if method == 'tools/list':
            # Return list of available tools
            tools = [
                {
                    "name": "process_data",
                    "description": "Process data with various operations",
                    "inputSchema": {
                        "type": "object",
                        "properties": {
                            "data": {"type": "object"},
                            "operation": {"type": "string", "enum": ["transform", "validate", "analyze"]}
                        }
                    }
                },
                {
                    "name": "health_check",
                    "description": "Check server health status",
                    "inputSchema": {"type": "object"}
                }
            ]

            return {
                'statusCode': 200,
                'headers': {
                    'Content-Type': 'application/json',
                    'Access-Control-Allow-Origin': '*'
                },
                'body': json.dumps({
                    'jsonrpc': '2.0',
                    'result': {"tools": tools},
                    'id': body.get('id')
                })
            }

        elif method.startswith('tools/call'):
            # Execute specific tool
            result = asyncio.run(execute_tool(body))
            return {
                'statusCode': 200,
                'headers': {
                    'Content-Type': 'application/json',
                    'Access-Control-Allow-Origin': '*'
                },
                'body': json.dumps(result)
            }

        else:
            return {
                'statusCode': 404,
                'headers': {'Content-Type': 'application/json'},
                'body': json.dumps({
                    'jsonrpc': '2.0',
                    'error': {'code': -32601, 'message': 'Method not found'},
                    'id': body.get('id')
                })
            }

    except json.JSONDecodeError:
        return {
            'statusCode': 400,
            'headers': {'Content-Type': 'application/json'},
            'body': json.dumps({
                'jsonrpc': '2.0',
                'error': {'code': -32700, 'message': 'Parse error'}
            })
        }
    except Exception as e:
        logger.error(f"Lambda handler error: {str(e)}")
        return {
            'statusCode': 500,
            'headers': {'Content-Type': 'application/json'},
            'body': json.dumps({
                'jsonrpc': '2.0',
                'error': {'code': -32603, 'message': 'Internal error'}
            })
        }

async def execute_tool(body: Dict[str, Any]) -> Dict[str, Any]:
    """
    Execute MCP tool based on request body.

    This function demonstrates how to handle MCP tool execution
    in the Lambda environment.
    """
    params = body.get('params', {})
    tool_name = params.get('name')
    tool_args = params.get('arguments', {})

    try:
        if tool_name == 'process_data':
            # Simulate data processing
            data = tool_args.get('data', {})
            operation = tool_args.get('operation', 'transform')

            result = {
                "operation": operation,
                "processed_at": "2024-01-01T00:00:00Z",
                "lambda_context": {
                    "aws_region": os.environ.get('AWS_REGION'),
                    "function_name": os.environ.get('AWS_LAMBDA_FUNCTION_NAME')
                },
                "result": data  # In real implementation, process the data
            }

        elif tool_name == 'health_check':
            result = {
                "status": "healthy",
                "platform": "aws_lambda",
                "region": os.environ.get('AWS_REGION'),
                "memory_limit": os.environ.get('AWS_LAMBDA_FUNCTION_MEMORY_SIZE')
            }

        else:
            raise ValueError(f"Unknown tool: {tool_name}")

        return {
            'jsonrpc': '2.0',
            'result': result,
            'id': body.get('id')
        }

    except Exception as e:
        return {
            'jsonrpc': '2.0',
            'error': {'code': -32603, 'message': str(e)},
            'id': body.get('id')
        }

# monitoring/monitor.py

from prometheus_client import start_http_server, Counter, Histogram, Gauge
import time
import asyncio
import aiohttp
from typing import List, Dict, Optional
import logging
import json
from dataclasses import dataclass
from datetime import datetime, timedelta

# Configure structured logging for production environments

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

@dataclass
class ServerHealthStatus:
    """Represents the health status of a single MCP server."""
    url: str
    status: str  # 'healthy', 'unhealthy', 'error'
    response_time: Optional[float]
    last_check: datetime
    error_message: Optional[str] = None
    details: Optional[Dict] = None

class MCPServerMonitor:
    """
    Comprehensive monitoring system for MCP servers.

    This monitor provides:
    - Health checking with configurable intervals
    - Prometheus metrics collection
    - Alerting when servers become unhealthy
    - Historical data tracking
    """

    def __init__(self, server_urls: List[str], check_interval: int = 30):
        self.server_urls = server_urls
        self.check_interval = check_interval
        self.server_status: Dict[str, ServerHealthStatus] = {}
        self.failure_counts: Dict[str, int] = {url: 0 for url in server_urls}

        # Prometheus metrics for comprehensive monitoring
        self.health_check_total = Counter(
            'mcp_health_checks_total',
            'Total health checks performed',
            ['server', 'status']
        )

        self.response_time = Histogram(
            'mcp_response_time_seconds',
            'Response time for MCP requests',
            ['server', 'method'],
            buckets=[0.1, 0.5, 1.0, 2.0, 5.0, 10.0, float('inf')]
        )

        self.server_availability = Gauge(
            'mcp_server_availability',
            'Server availability (1=up, 0=down)',
            ['server']
        )

        self.consecutive_failures = Gauge(
            'mcp_consecutive_failures',
            'Number of consecutive failures',
            ['server']
        )

    async def check_health(self, session: aiohttp.ClientSession, url: str) -> ServerHealthStatus:
        """
        Perform comprehensive health check on a single server.

        This includes:
        - Basic connectivity test
        - Health endpoint validation
        - Response time measurement
        - Error categorization
        """
        start_time = time.time()

        try:
            # Make health check request with timeout
            async with session.get(
                f"{url}/health", 
                timeout=aiohttp.ClientTimeout(total=10)
            ) as response:

                response_time = time.time() - start_time

                if response.status == 200:
                    # Parse health check response
                    try:
                        health_data = await response.json()

                        # Reset failure count on success
                        self.failure_counts[url] = 0

                        # Update metrics
                        self.health_check_total.labels(server=url, status='success').inc()
                        self.server_availability.labels(server=url).set(1)
                        self.response_time.labels(server=url, method='health').observe(response_time)
                        self.consecutive_failures.labels(server=url).set(0)

                        return ServerHealthStatus(
                            url=url,
                            status='healthy',
                            response_time=response_time,
                            last_check=datetime.now(),
                            details=health_data
                        )

                    except json.JSONDecodeError:
                        return ServerHealthStatus(
                            url=url,
                            status='unhealthy',
                            response_time=response_time,
                            last_check=datetime.now(),
                            error_message="Invalid JSON response"
                        )
                else:
                    # HTTP error status
                    self.failure_counts[url] += 1
                    self.health_check_total.labels(server=url, status='error').inc()
                    self.server_availability.labels(server=url).set(0)
                    self.consecutive_failures.labels(server=url).set(self.failure_counts[url])

                    return ServerHealthStatus(
                        url=url,
                        status='unhealthy',
                        response_time=response_time,
                        last_check=datetime.now(),
                        error_message=f"HTTP {response.status}"
                    )

        except asyncio.TimeoutError:
            self.failure_counts[url] += 1
            self.health_check_total.labels(server=url, status='timeout').inc()
            self.server_availability.labels(server=url).set(0)
            self.consecutive_failures.labels(server=url).set(self.failure_counts[url])

            return ServerHealthStatus(
                url=url,
                status='error',
                response_time=time.time() - start_time,
                last_check=datetime.now(),
                error_message="Request timeout"
            )

        except Exception as e:
            self.failure_counts[url] += 1
            self.health_check_total.labels(server=url, status='error').inc()
            self.server_availability.labels(server=url).set(0)
            self.consecutive_failures.labels(server=url).set(self.failure_counts[url])

            return ServerHealthStatus(
                url=url,
                status='error',
                response_time=time.time() - start_time,
                last_check=datetime.now(),
                error_message=str(e)
            )

    async def check_all_servers(self, session: aiohttp.ClientSession) -> List[ServerHealthStatus]:
        """Check health of all configured servers concurrently."""
        tasks = [
            self.check_health(session, url)
            for url in self.server_urls
        ]

        results = await asyncio.gather(*tasks, return_exceptions=True)

        # Handle any exceptions from the gather
        health_statuses = []
        for i, result in enumerate(results):
            if isinstance(result, Exception):
                # Create error status for failed check
                health_statuses.append(ServerHealthStatus(
                    url=self.server_urls[i],
                    status='error',
                    response_time=None,
                    last_check=datetime.now(),
                    error_message=str(result)
                ))
            else:
                health_statuses.append(result)

        return health_statuses