Duffel MCP Server

"""Airport search tools.""" from ..server import mcp from ..models import ResponseFormat, SearchAirportsInput, ListAirportsInput from ..api import make_api_request from ..formatters import format_json_response, truncate_text @mcp.tool( name="duffel_search_airports", annotations={ "title": "Search Airports", "readOnlyHint": True, "destructiveHint": False, "idempotentHint": True, "openWorldHint": True } ) async def search_airports(params: SearchAirportsInput) -> str: """ Search for airports by name, city, or IATA code. This tool helps users find correct airport codes for flight searches by: - Searching airport names (e.g., "Heathrow", "Charles de Gaulle") - Searching city names (e.g., "London", "Paris") - Validating IATA codes (e.g., "LHR", "CDG") Results include: - Airport name and IATA code - City and country information - GPS coordinates - Time zone Use this when: - User provides city/airport names instead of codes - Verifying airport codes before search - Finding all airports in a city - User unsure of exact airport code Returns matching airports in specified format (JSON or Markdown). """ try: response = await make_api_request( method="GET", endpoint="/air/airports", params={"limit": 200} ) airports = response["data"] query_lower = params.query.lower() matches = [] for airport in airports: if ( query_lower in airport.get("name", "").lower() or query_lower in airport.get("city_name", "").lower() or query_lower == airport.get("iata_code", "").lower() or query_lower in airport.get("iata_city_code", "").lower() ): matches.append(airport) if len(matches) >= params.limit: break if not matches: return ( f"No airports found matching '{params.query}'.\n\n" f"Try:\n" f"- A different spelling\n" f"- The city name instead of airport name\n" f"- A broader search term" ) if params.response_format == ResponseFormat.JSON: return truncate_text(format_json_response(matches)) else: # Markdown format lines = [ f"# Airport Search Results for '{params.query}'", f"", f"Found {len(matches)} matching airports:", "" ] for airport in matches: city = airport.get("city", {}) lines.append(f"## {airport.get('name', 'N/A')}") lines.append(f"**IATA Code**: `{airport.get('iata_code', 'N/A')}`") lines.append(f"**City**: {airport.get('city_name', 'N/A')}") if city: lines.append(f"**City Code**: `{city.get('iata_code', 'N/A')}`") lines.append(f"**Country**: {airport.get('iata_country_code', 'N/A')}") lines.append(f"**Time Zone**: {airport.get('time_zone', 'N/A')}") lines.append("") return truncate_text("\n".join(lines)) except Exception as e: return f"Error searching airports: {str(e)}\n\nTry a different search term or check your internet connection." @mcp.tool( name="duffel_list_airports", annotations={ "title": "List Airports", "readOnlyHint": True, "destructiveHint": False, "idempotentHint": True, "openWorldHint": True } ) async def list_airports(params: ListAirportsInput) -> str: """ List airports with optional country filter. This tool retrieves a paginated list of airports. Results can be filtered by country using ISO 3166-1 alpha-2 country codes (e.g., 'US', 'GB', 'FR'). Use this when: - Exploring available airports - Getting airports in a specific country - Building airport selection lists Note: For finding a specific airport, use duffel_search_airports instead. Returns airports in specified format (JSON or Markdown). """ try: query_params = {"limit": params.limit} if params.country_code: query_params["iata_country_code"] = params.country_code.upper() response = await make_api_request( method="GET", endpoint="/air/airports", params=query_params ) airports = response["data"] if not airports: return "No airports found with the specified criteria." if params.response_format == ResponseFormat.JSON: return truncate_text(format_json_response(airports)) else: # Markdown format title = f"# Airports" if params.country_code: title += f" in {params.country_code.upper()}" lines = [title, "", f"Showing {len(airports)} airports:", ""] for airport in airports: lines.append( f"- **{airport.get('iata_code', 'N/A')}**: " f"{airport.get('name', 'N/A')} - " f"{airport.get('city_name', 'N/A')}" ) return truncate_text("\n".join(lines)) except Exception as e: return f"Error listing airports: {str(e)}"