Duffel MCP Server

"""API client for making requests to Duffel API.""" import os import json from typing import Any import httpx from ..config import API_BASE_URL, API_VERSION, API_TIMEOUT def get_api_token() -> str: """Get Duffel API token from environment variable.""" token = os.getenv("DUFFEL_ACCESS_TOKEN") if not token: raise ValueError( "DUFFEL_ACCESS_TOKEN environment variable not set. " "Please set it with your Duffel API token from https://app.duffel.com" ) return token def get_headers() -> dict[str, str]: """Get standard API headers with authentication.""" return { "Authorization": f"Bearer {get_api_token()}", "Content-Type": "application/json", "Accept": "application/json", "Duffel-Version": API_VERSION } async def make_api_request( method: str, endpoint: str, data: dict[str, Any] | None = None, params: dict[str, Any] | None = None ) -> dict[str, Any]: """ Make an API request to Duffel with proper error handling. Args: method: HTTP method (GET, POST, etc.) endpoint: API endpoint path (e.g., "/air/offer_requests") data: Request body data params: URL query parameters Returns: API response data Raises: ValueError: If request fails with details from Duffel error response """ url = f"{API_BASE_URL}{endpoint}" async with httpx.AsyncClient(timeout=API_TIMEOUT) as client: try: response = await client.request( method=method, url=url, headers=get_headers(), json=data, params=params ) response.raise_for_status() return response.json() except httpx.HTTPStatusError as e: # Parse Duffel error response for better error messages try: error_data = e.response.json() errors = error_data.get("errors", []) if errors: error_messages = [] for error in errors: msg = f"{error.get('title', 'Error')}: {error.get('message', 'Unknown error')}" if error.get('code'): msg += f" (code: {error['code']})" error_messages.append(msg) raise ValueError("\n".join(error_messages)) except (json.JSONDecodeError, KeyError): pass # Fallback to generic error raise ValueError(f"API request failed: {e.response.status_code} {e.response.text}")