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 HarvestTokenMiddleware(BaseHTTPMiddleware): """Middleware to extract Harvest credentials from query parameters.""" 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 Harvest credentials from query parameters access_token = request.query_params.get('access_token') account_id = request.query_params.get('account_id') if access_token and account_id: logger.info("✅ Extracted Harvest credentials from query params") auth.HarvestAuthProvider.set_credentials_for_request( access_token=access_token, account_id=account_id ) logger.info("✅ Context variables set for request") else: missing = [] if not access_token: missing.append('access_token') if not account_id: missing.append('account_id') logger.warning(f"Missing Harvest 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 the spec file from the package directory 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(HarvestTokenMiddleware) uvicorn.run( app, host="0.0.0.0", port=get_port_from_environment(), log_level="info", )