OpenAPI Lambda MCP Server

import logging import json from contextvars import ContextVar from typing import Dict, Optional from urllib.parse import parse_qs, urlencode, urlparse import httpx from awslabs.openapi_mcp_server.api.config import Config from awslabs.openapi_mcp_server.auth.auth_provider import AuthProvider from awslabs.openapi_mcp_server.auth.auth_factory import register_auth_provider logger = logging.getLogger(__name__) _current_access_token: ContextVar[Optional[str]] = ContextVar('harvest_access_token', default=None) _current_account_id: ContextVar[Optional[str]] = ContextVar('harvest_account_id', default=None) # Build parameter name mapping from OpenAPI spec at module load time _PARAMETER_NAME_MAPPING: Dict[str, str] = {} def _load_parameter_mapping(): """Load parameter name mappings from OpenAPI spec x-original-name extensions.""" try: with open('/var/task/openapi-spec.json') as f: spec = json.load(f) # Find all parameters with x-original-name for path, methods in spec.get('paths', {}).items(): for method, details in methods.items(): if method not in ['get', 'post', 'put', 'delete', 'patch', 'options', 'head']: continue for param in details.get('parameters', []): if 'x-original-name' in param: current_name = param.get('name') original_name = param['x-original-name'] _PARAMETER_NAME_MAPPING[current_name] = original_name if _PARAMETER_NAME_MAPPING: logger.info(f"🔧 Loaded parameter name mappings: {_PARAMETER_NAME_MAPPING}") except Exception as e: logger.warning(f"Could not load parameter mappings: {e}") # Load mappings at module import time _load_parameter_mapping() class HarvestAuth(httpx.Auth): """Custom HTTPX Auth class that dynamically adds Harvest auth headers per-request. Harvest API requires two headers: - Authorization: Bearer {access_token} - Harvest-Account-Id: {account_id} """ def __init__(self, provider: 'HarvestAuthProvider'): self.provider = provider def auth_flow(self, request: httpx.Request): """Add auth headers dynamically for each request.""" logger.info("🔐 HarvestAuth.auth_flow - Starting per-request auth") logger.info(f"🔐 Original request : {request.method} {request.url}") logger.info(f" headers : {request.headers}") logger.info(f" params : {request.url.params}") # Read credentials from context variables (set by middleware) access_token = _current_access_token.get() account_id = _current_account_id.get() logger.info(f"🔐 Context values - access_token: {bool(access_token)}, account_id: {bool(account_id)}") if not access_token or not account_id: missing = [] if not access_token: missing.append('access_token') if not account_id: missing.append('account_id') logger.warning(f"🔐 Missing Harvest credentials in request context: {', '.join(missing)}") yield request return # Fix parameter names using x-original-name mappings # This remaps parameters like from_ -> from for API compatibility if _PARAMETER_NAME_MAPPING: parsed = urlparse(str(request.url)) query_params = parse_qs(parsed.query, keep_blank_values=True) # Remap parameter names if they're in our mapping remapped_params = {} remapped_any = False for key, values in query_params.items(): if key in _PARAMETER_NAME_MAPPING: original_key = _PARAMETER_NAME_MAPPING[key] remapped_params[original_key] = values remapped_any = True logger.info(f"🔧 Remapping query param: {key} -> {original_key}") else: remapped_params[key] = values if remapped_any: # Reconstruct URL with remapped parameters new_query = urlencode(remapped_params, doseq=True) new_url = parsed._replace(query=new_query).geturl() from httpx import URL request.url = URL(new_url) logger.info(f"🔧 Remapped URL: {request.url}") logger.info("🔐 Adding Harvest auth headers to request") request.headers["Authorization"] = f"Bearer {access_token}" request.headers["Harvest-Account-Id"] = account_id yield request class HarvestAuthProvider(AuthProvider): def __init__(self, config: Config): self.config = config @property def provider_name(self) -> str: return "harvest" def is_configured(self) -> bool: return True def get_auth_headers(self) -> Dict[str, str]: """Get static authentication headers (not used - we use dynamic auth via get_httpx_auth). This is called during server initialization. Returns empty dict because Harvest auth is handled dynamically per-request via HarvestAuth.auth_flow(). """ logger.info("🔑 get_auth_headers() called (returning empty - using dynamic auth)") return {} def get_auth_params(self) -> Dict[str, str]: """Harvest uses header-based auth, not query params.""" return {} def get_auth_cookies(self) -> Dict[str, str]: """Harvest uses header-based auth, not cookies.""" return {} def get_httpx_auth(self) -> Optional[httpx.Auth]: """Get authentication object for HTTPX. Returns a custom Auth class that dynamically adds headers per-request. """ logger.info("🔐 Returning HarvestAuth instance for dynamic per-request auth") return HarvestAuth(self) @staticmethod def set_credentials_for_request(access_token: str, account_id: str) -> None: """Set credentials in context variables for the current request.""" _current_access_token.set(access_token) _current_account_id.set(account_id) # Auto-register the Harvest auth provider when this module is imported register_auth_provider("harvest", HarvestAuthProvider)