⚙️ Session 4 Advanced: Cloud Deployment Strategies - Multi-Platform Production¶

# src/cloud_run_adapter.py - Bridging MCP and HTTP
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

# Cloud Run optimized logging
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():
    """
    Cloud Run Startup: Preparing for Production Traffic

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

@app.on_event("shutdown")
async def shutdown_event():
    """
    Graceful Shutdown: Cleaning Up Resources

    Proper cleanup ensures graceful shutdown when Cloud Run
    terminates containers during scaling or deployment.
    """
    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):
    """
    The Protocol Bridge: HTTP ↔ MCP JSON-RPC

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

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

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

        elif method.startswith("tools/call"):
            # Tool execution
            tool_name = params.get("name")
            tool_params = params.get("arguments", {})

            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
                )

    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():
    """Cloud Run Health Check: Service Readiness Validation"""
    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 for Monitoring Integration"""
    from prometheus_client import generate_latest
    return Response(content=generate_latest(), media_type="text/plain")

# src/lambda_handler.py - MCP Server in Serverless Function Form
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

# Lambda-optimized logging configuration
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: Maximum Performance, Minimum Overhead

    This approach provides:
    - Minimal cold start time
    - Direct control over execution flow
    - Maximum performance for simple operations
    - Full access to Lambda runtime context
    """
    try:
        # Request logging for debugging and monitoring
        logger.info(f"Lambda invoked",
                   request_id=context.aws_request_id,
                   remaining_time=context.get_remaining_time_in_millis())

        # Parse HTTP request body to extract MCP JSON-RPC data
        body = json.loads(event.get('body', '{}'))
        method = body.get('method', '')

        # Handle MCP protocol methods - Tools discovery
        if method == 'tools/list':
            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 and Lambda runtime 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 requested tool through async handler
            result = asyncio.run(execute_tool(body, context))
            return {
                'statusCode': 200,
                'headers': {
                    'Content-Type': 'application/json',
                    'Access-Control-Allow-Origin': '*'
                },
                'body': json.dumps(result)
            }

async def execute_tool(body: Dict[str, Any], context: Any) -> Dict[str, Any]:
    """Lambda Tool Execution: Stateless, Fast, and Monitored"""
    params = body.get('params', {})
    tool_name = params.get('name')
    tool_args = params.get('arguments', {})

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

            result = {
                "operation": operation,
                "processed_at": datetime.now().isoformat(),
                "lambda_context": {
                    "aws_region": os.environ.get('AWS_REGION'),
                    "function_name": os.environ.get('AWS_LAMBDA_FUNCTION_NAME'),
                    "memory_limit": os.environ.get('AWS_LAMBDA_FUNCTION_MEMORY_SIZE'),
                    "request_id": context.aws_request_id,
                    "remaining_time_ms": context.get_remaining_time_in_millis()
                },
                "result": data  # Process data here
            }

        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'),
                "cold_start": not hasattr(context, '_warm'),
                "execution_env": os.environ.get('AWS_EXECUTION_ENV')
            }

            # Mark as warm for subsequent invocations
            context._warm = True

        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')
        }