Duffel MCP Server

"""Flight search and booking tools.""" from datetime import datetime from ..server import mcp from ..models import ResponseFormat, FlightSearchInput, GetOfferInput, CreateOrderInput, GetOrderInput from ..api import make_api_request from ..formatters import format_json_response, format_markdown_flight_offer, format_currency, truncate_text @mcp.tool( name="duffel_search_flights", annotations={ "title": "Search Flights", "readOnlyHint": True, "destructiveHint": False, "idempotentHint": False, "openWorldHint": True } ) async def search_flights(params: FlightSearchInput) -> str: """ Search for available flights based on journey requirements. This tool creates an offer request and returns available flight options with pricing, schedules, and airline information. Each offer includes: - Total price and currency - Flight segments with timings - Airline and aircraft details - Cabin class and baggage allowance - Offer ID for booking Use this when users want to: - Find flights between destinations - Compare prices and schedules - Check availability for specific dates - Get flight options before booking Important notes: - Offers expire after 15-30 minutes (check expires_at) - Use passenger age instead of type for better accuracy - Round trips require 2 slices (outbound + return) - Direct flights: set max_connections=0 Returns flight offers in specified format (JSON or Markdown summary). """ request_data = { "data": { "slices": [ { "origin": slice_data.origin.upper(), "destination": slice_data.destination.upper(), "departure_date": slice_data.departure_date } for slice_data in params.slices ], "passengers": [] } } for passenger in params.passengers: if passenger.age is not None: request_data["data"]["passengers"].append({"age": passenger.age}) else: request_data["data"]["passengers"].append({"type": passenger.type.value}) if params.cabin_class: request_data["data"]["cabin_class"] = params.cabin_class.value if params.max_connections is not None: request_data["data"]["max_connections"] = params.max_connections try: response = await make_api_request( method="POST", endpoint="/air/offer_requests", data=request_data ) offers = response["data"].get("offers", []) if not offers: return "No flights found for the specified criteria. Try adjusting your search parameters (dates, cabin class, or connections)." if params.response_format == ResponseFormat.JSON: result = { "offer_request_id": response["data"]["id"], "total_offers": len(offers), "offers": offers[:10] # Limit to 10 offers for readability } return truncate_text(format_json_response(result)) else: # Markdown format lines = [ f"# Flight Search Results", f"**Offer Request ID**: `{response['data']['id']}`", f"**Total Offers Found**: {len(offers)}", "", f"Showing top {min(5, len(offers))} offers:", "" ] for offer in offers[:5]: lines.append(format_markdown_flight_offer(offer)) lines.append("---") lines.append("") if len(offers) > 5: lines.append(f"\n*{len(offers) - 5} more offers available. Use duffel_get_offer with specific offer ID for details.*") return truncate_text("\n".join(lines)) except Exception as e: return f"Error searching flights: {str(e)}\n\nTroubleshooting:\n- Verify airport/city codes are valid 3-letter IATA codes\n- Check dates are in future and in YYYY-MM-DD format\n- Ensure passenger count is reasonable (1-9)\n- Try broader search criteria if no results" @mcp.tool( name="duffel_get_offer", annotations={ "title": "Get Flight Offer Details", "readOnlyHint": True, "destructiveHint": False, "idempotentHint": True, "openWorldHint": True } ) async def get_offer(params: GetOfferInput) -> str: """ Retrieve detailed information and current pricing for a specific flight offer. This tool fetches the latest version of an offer, including: - Up-to-date pricing and availability - Complete flight schedule and routing - Passenger requirements and restrictions - Baggage allowance and cabin details - Cancellation and change policies Use this when: - User selects a flight from search results - Before booking to confirm current price - To check if an offer is still available - To get passenger IDs needed for booking Important: Always retrieve the offer immediately before booking to ensure pricing is current, as offers expire after 15-30 minutes. Returns offer details in specified format (JSON or Markdown). """ try: response = await make_api_request( method="GET", endpoint=f"/air/offers/{params.offer_id}" ) offer = response["data"] expires_at = datetime.fromisoformat(offer["expires_at"].replace("Z", "+00:00")) if expires_at < datetime.now().astimezone(): return ( f"⚠️ This offer has expired at {offer['expires_at']}.\n\n" "Offers typically expire 15-30 minutes after creation. " "Please perform a new flight search to get current offers." ) if params.response_format == ResponseFormat.JSON: return truncate_text(format_json_response(offer)) else: # Markdown format result = format_markdown_flight_offer(offer) passengers = offer.get("passengers", []) if passengers: result += "\n### Passengers\n" for i, pax in enumerate(passengers, 1): result += f"{i}. **ID**: `{pax['id']}` - Type: {pax.get('type', 'N/A')}\n" conditions = offer.get("conditions", {}) if conditions: result += "\n### Booking Conditions\n" if conditions.get("change_before_departure"): result += f"- Change allowed: {conditions['change_before_departure'].get('allowed', False)}\n" if conditions.get("refund_before_departure"): result += f"- Refund allowed: {conditions['refund_before_departure'].get('allowed', False)}\n" return truncate_text(result) except Exception as e: return f"Error retrieving offer: {str(e)}\n\nTroubleshooting:\n- Verify the offer ID is correct (starts with 'off_')\n- Check if the offer has expired\n- Offer might have been booked by someone else" @mcp.tool( name="duffel_create_order", annotations={ "title": "Create Flight Order (Book Flight)", "readOnlyHint": False, "destructiveHint": False, "idempotentHint": False, "openWorldHint": True } ) async def create_order(params: CreateOrderInput) -> str: """ Create a flight booking (order) for the specified offer and passengers. ⚠️ IMPORTANT: This creates a real booking and may involve payment. This action: - Creates a confirmed airline reservation - May charge the payment method (if not using test mode) - Issues a booking reference from the airline - Is typically non-refundable or has cancellation fees Before calling this tool: 1. Verify the offer is current using duffel_get_offer 2. Confirm all passenger details are accurate (names match IDs) 3. Check payment amount matches offer total 4. Ensure user understands booking terms Required data: - Offer ID from search results - Complete passenger details (names as on ID, date of birth, contact info) - Payment information matching offer total The tool returns: - Order ID for reference - Airline booking reference - Confirmation details - Ticket information Response codes: - 201: Order created successfully (immediate confirmation) - 200: Booking confirmed, details arriving soon (webhook notification will follow) - 202: Booking processing (check webhooks or email for confirmation) Use test mode tokens for testing to avoid actual charges. Returns order details in specified format (JSON or Markdown). """ order_data = { "data": { "selected_offers": [params.offer_id], "passengers": [ { "id": pax.id, "given_name": pax.given_name, "family_name": pax.family_name, "born_on": pax.born_on, "email": pax.email, "phone_number": pax.phone_number, "gender": pax.gender, **({"title": pax.title} if pax.title else {}) } for pax in params.passengers ], "payments": [ { "type": payment.type.value, "amount": payment.amount, "currency": payment.currency.upper() } for payment in params.payments ] } } try: response = await make_api_request( method="POST", endpoint="/air/orders", data=order_data ) status_code = response.get("meta", {}).get("status", 201) if status_code == 200: message = response.get("data", {}).get("message", "") return ( f"✅ Booking Confirmed!\n\n" f"{message}\n\n" f"The complete booking details will be available shortly. " f"You will receive:\n" f"- A webhook notification when the order is created\n" f"- An email to your support contact address\n\n" f"**Offer ID**: `{params.offer_id}`\n" f"Please check your webhooks or email for the order ID and booking reference." ) elif status_code == 202: return ( f"⏳ Booking Processing\n\n" f"Your booking request has been accepted and is being processed. " f"You will receive confirmation via:\n" f"- Webhook notification (order.created or order.creation_failed)\n" f"- Email to your support contact address\n\n" f"**Offer ID**: `{params.offer_id}`\n" f"This typically takes a few minutes. Please do not retry this request." ) else: order = response["data"] if params.response_format == ResponseFormat.JSON: return truncate_text(format_json_response(order)) else: # Markdown format lines = [ "# ✅ Booking Confirmed!", "", f"**Order ID**: `{order['id']}`", f"**Booking Reference**: {order.get('booking_reference', 'N/A')}", f"**Total**: {format_currency(order['total_amount'], order['total_currency'])}", "", "## Passengers" ] for i, pax in enumerate(order.get("passengers", []), 1): lines.append(f"{i}. {pax.get('given_name')} {pax.get('family_name')}") lines.append("") lines.append("## Flight Details") for i, slice_info in enumerate(order.get("slices", []), 1): origin = slice_info["origin"]["iata_code"] destination = slice_info["destination"]["iata_code"] lines.append(f"**Slice {i}**: {origin} → {destination}") lines.append(f"- Departure: {slice_info['departure_at']}") lines.append(f"- Arrival: {slice_info['arrival_at']}") lines.append("") lines.append("---") lines.append("*Save the Order ID and Booking Reference for future reference.*") return truncate_text("\n".join(lines)) except Exception as e: error_msg = str(e) troubleshooting = "\n\nTroubleshooting:\n" if "offer_expired" in error_msg: troubleshooting += "- The offer has expired. Perform a new flight search to get current offers.\n" elif "offer_no_longer_available" in error_msg: troubleshooting += "- The offer is no longer available. Search again or select a different offer.\n" elif "price_changed" in error_msg: troubleshooting += "- The price has changed. Retrieve the offer again to get updated pricing.\n" elif "validation" in error_msg.lower(): troubleshooting += "- Check passenger details are complete and accurate\n" troubleshooting += "- Verify names match government ID exactly\n" troubleshooting += "- Ensure date of birth is in YYYY-MM-DD format\n" troubleshooting += "- Confirm email and phone number are valid\n" elif "payment" in error_msg.lower(): troubleshooting += "- Verify payment amount matches offer total_amount exactly\n" troubleshooting += "- Confirm currency matches offer total_currency\n" troubleshooting += "- Check payment type is appropriate for your account\n" else: troubleshooting += "- Verify all passenger details are accurate\n" troubleshooting += "- Ensure payment amount and currency match the offer\n" troubleshooting += "- Check if using test mode token for testing\n" return f"Error creating order: {error_msg}{troubleshooting}" @mcp.tool( name="duffel_get_order", annotations={ "title": "Get Order Details", "readOnlyHint": True, "destructiveHint": False, "idempotentHint": True, "openWorldHint": True } ) async def get_order(params: GetOrderInput) -> str: """ Retrieve complete details for an existing flight order. This tool fetches: - Order status and booking reference - Flight itinerary and schedule - Passenger information - Payment and pricing details - Documents and tickets - Change and cancellation options Use this when: - User needs to review their booking - Checking order status - Before making changes or cancellations - Retrieving booking reference for airline website Returns order details in specified format (JSON or Markdown). """ try: response = await make_api_request( method="GET", endpoint=f"/air/orders/{params.order_id}" ) order = response["data"] if params.response_format == ResponseFormat.JSON: return truncate_text(format_json_response(order)) else: # Markdown format lines = [ "# Flight Order Details", "", f"**Order ID**: `{order['id']}`", f"**Booking Reference**: {order.get('booking_reference', 'N/A')}", f"**Status**: {order.get('status', 'N/A')}", f"**Total**: {format_currency(order['total_amount'], order['total_currency'])}", f"**Booked**: {order.get('created_at', 'N/A')}", "", "## Passengers" ] for i, pax in enumerate(order.get("passengers", []), 1): lines.append(f"{i}. {pax.get('given_name')} {pax.get('family_name')}") if pax.get('born_on'): lines.append(f" - DOB: {pax['born_on']}") lines.append("") lines.append("## Flight Itinerary") for i, slice_info in enumerate(order.get("slices", []), 1): origin = slice_info["origin"] destination = slice_info["destination"] lines.append(f"\n### Slice {i}: {origin['city_name']} → {destination['city_name']}") lines.append(f"**Departure**: {slice_info['departure_at']} ({origin['iata_code']})") lines.append(f"**Arrival**: {slice_info['arrival_at']} ({destination['iata_code']})") lines.append(f"**Duration**: {format_duration(slice_info.get('duration', ''))}") for j, segment in enumerate(slice_info.get("segments", []), 1): carrier = segment.get("marketing_carrier", {}) lines.append(f"\n**Flight {j}**: {carrier.get('name', 'N/A')} {segment.get('marketing_carrier_flight_number', '')}") lines.append(f"- Aircraft: {segment.get('aircraft', {}).get('name', 'N/A')}") conditions = order.get("conditions", {}) if conditions: lines.append("\n## Booking Conditions") change_before = conditions.get("change_before_departure", {}) if change_before: allowed = "✅ Yes" if change_before.get("allowed") else "❌ No" lines.append(f"- Changes before departure: {allowed}") if change_before.get("penalty_amount"): penalty = format_currency( change_before["penalty_amount"], change_before.get("penalty_currency", "") ) lines.append(f" - Change fee: {penalty}") refund_before = conditions.get("refund_before_departure", {}) if refund_before: allowed = "✅ Yes" if refund_before.get("allowed") else "❌ No" lines.append(f"- Refund before departure: {allowed}") return truncate_text("\n".join(lines)) except Exception as e: return f"Error retrieving order: {str(e)}\n\nTroubleshooting:\n- Verify the order ID is correct (starts with 'ord_')\n- Check if you have access to this order\n- Order might not exist or might have been cancelled"