OpenAPI Lambda MCP Server

import logging import os import time from contextlib import contextmanager import uvicorn from starlette.requests import Request from starlette.responses import JSONResponse from starlette.middleware.base import BaseHTTPMiddleware from awslabs.openapi_mcp_server.api.config import Config from awslabs.openapi_mcp_server.server import create_mcp_server # Import custom auth provider to register it import auth import config as api_config logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class HealthCheckMiddleware(BaseHTTPMiddleware): """Middleware to handle health check requests.""" async def dispatch(self, request: Request, call_next): if request.method == "GET" and request.url.path == "/": return JSONResponse({"status": "healthy"}) response = await call_next(request) return response class ZohoTokenMiddleware(BaseHTTPMiddleware): async def dispatch(self, request: Request, call_next): logger.info(f"Incoming request: {request.method} {request.url.path}") logger.info(f"Request headers: {dict(request.headers)}") logger.info(f"Query parameters: {dict(request.query_params)}") # Extract OAuth credentials from query parameters client_id = request.query_params.get('client_id') client_secret = request.query_params.get('client_secret') refresh_token = request.query_params.get('refresh_token') if client_id and client_secret and refresh_token: logger.info(f"✅ Extracted OAuth credentials from query params") auth.ZohoAuthProvider.set_credentials_for_request( client_id=client_id, client_secret=client_secret, refresh_token=refresh_token ) logger.info(f"✅ Context variables set for request") else: missing = [] if not client_id: missing.append('client_id') if not client_secret: missing.append('client_secret') if not refresh_token: missing.append('refresh_token') logger.warning(f"Missing OAuth credentials in query parameters: {', '.join(missing)}") response = await call_next(request) return response @contextmanager def stopwatch(description: str): logger.info(f"start {description}") start = time.time() try: yield finally: elapsed = time.time() - start logger.info(f"end {description}: {elapsed:.3f}s") def get_port_from_environment() -> int: return int(os.getenv("PORT", "8080")) if __name__ == "__main__": with stopwatch("config creation"): # Use merged spec (with all external references resolved) # The API's justfile downloads, merges, and validates the spec # The Dockerfile copies it to /var/task/openapi-spec.json config = Config( api_name=api_config.API_NAME, api_base_url=api_config.API_BASE_URL, api_spec_path="/var/task/openapi-spec.json", transport="http", auth_type=api_config.AUTH_TYPE, version="0.1.0", ) with stopwatch("MCP server creation"): mcp = create_mcp_server(config) with stopwatch("HTTP app creation"): app = mcp.http_app( path="/", transport="http", stateless_http=True, ) # Add middleware (order matters: health check first, then auth) app.add_middleware(HealthCheckMiddleware) app.add_middleware(ZohoTokenMiddleware) uvicorn.run( app, host="0.0.0.0", port=get_port_from_environment(), log_level="info", )