Duffel MCP Server

"""Accommodation (stays) search and booking tools.""" from ..server import mcp from ..models import ( ResponseFormat, SearchAccommodationInput, GetAccommodationInput, GetAccommodationRatesInput, CreateStaysQuoteInput, CreateStaysBookingInput, GetStaysBookingInput, ) from ..api import make_api_request from ..formatters import ( format_json_response, format_markdown_accommodation, format_markdown_accommodation_rate, format_currency, truncate_text, ) @mcp.tool( name="duffel_search_accommodation", annotations={ "title": "Search Accommodation", "readOnlyHint": True, "destructiveHint": False, "idempotentHint": False, "openWorldHint": True } ) async def search_accommodation(params: SearchAccommodationInput) -> str: """ Search for available accommodations based on location, dates, and guest requirements. This tool creates a search for accommodation and returns available properties with: - Property details (name, location, ratings) - Search result IDs for fetching rates - Amenities and features - Guest ratings from multiple sources Use this when users want to: - Find hotels/accommodations in a location - Compare properties and amenities - Check availability for specific dates - Get accommodation options before booking Important notes: - Provide either location (lat/long/radius) OR accommodation_ids - Check-in and check-out dates must be in the future - Search results contain search_result_id needed for next step Workflow: 1. Search accommodation → get search_result_id 2. Get rates using search_result_id 3. Create quote from rate_id 4. Book using quote_id Returns accommodation search results in specified format (JSON or Markdown). """ if not params.location and not params.accommodation_ids: return "Error: Must provide either 'location' (latitude, longitude, radius) or 'accommodation_ids'" if params.location and params.accommodation_ids: return "Error: Provide either 'location' OR 'accommodation_ids', not both" request_data = { "data": { "check_in_date": params.check_in_date, "check_out_date": params.check_out_date, "guests": [ { "type": "adult", "count": params.guests.adult_count } ], "rooms": params.rooms } } if params.guests.child_count and params.guests.child_count > 0: request_data["data"]["guests"].append({ "type": "child", "count": params.guests.child_count }) if params.location: request_data["data"]["location"] = { "geographic_coordinates": { "latitude": params.location.latitude, "longitude": params.location.longitude }, "radius": params.location.radius } elif params.accommodation_ids: request_data["data"]["accommodation_ids"] = params.accommodation_ids try: response = await make_api_request( method="POST", endpoint="/stays/search", data=request_data ) search_results = response.get("data", []) if not search_results: return "No accommodations found for the specified criteria. Try adjusting your search parameters (location, dates, or number of guests)." if params.response_format == ResponseFormat.JSON: result = { "total_results": len(search_results), "search_results": search_results[:10] # Limit to 10 for readability } return truncate_text(format_json_response(result)) else: # Markdown format lines = [ f"# Accommodation Search Results", f"**Total Results Found**: {len(search_results)}", f"**Check-in**: {params.check_in_date}", f"**Check-out**: {params.check_out_date}", f"**Guests**: {params.guests.adult_count} adult(s)" + (f", {params.guests.child_count} child(ren)" if params.guests.child_count else ""), f"**Rooms**: {params.rooms}", "", f"Showing top {min(5, len(search_results))} results:", "" ] for result in search_results[:5]: accommodation = result.get("accommodation", {}) lines.append(format_markdown_accommodation(accommodation)) lines.append(f"**Search Result ID**: `{result.get('id', 'N/A')}`") lines.append("*Use this ID with duffel_get_accommodation_rates to see available rooms and prices*") lines.append("---") lines.append("") if len(search_results) > 5: lines.append(f"\n*{len(search_results) - 5} more accommodations available.*") return truncate_text("\n".join(lines)) except Exception as e: return f"Error searching accommodations: {str(e)}\n\nTroubleshooting:\n- Verify dates are in future and in YYYY-MM-DD format\n- Check latitude/longitude are valid coordinates\n- Ensure radius is between 1-100 km\n- Try broader search criteria if no results" @mcp.tool( name="duffel_get_accommodation", annotations={ "title": "Get Accommodation Details", "readOnlyHint": True, "destructiveHint": False, "idempotentHint": True, "openWorldHint": True } ) async def get_accommodation(params: GetAccommodationInput) -> str: """ Retrieve detailed information for a specific accommodation property. This tool fetches: - Property name and location details - Amenities and facilities - Check-in/check-out information - Ratings from multiple sources - Brand information - Photos and description Use this when: - User wants more details about a specific property - Need to verify property information - Want to see all amenities and features Returns accommodation details in specified format (JSON or Markdown). """ try: response = await make_api_request( method="GET", endpoint=f"/stays/accommodation/{params.accommodation_id}" ) accommodation = response["data"] if params.response_format == ResponseFormat.JSON: return truncate_text(format_json_response(accommodation)) else: # Markdown format result = format_markdown_accommodation(accommodation) check_in_info = accommodation.get("check_in_information", {}) if check_in_info: result += "\n### Check-in Information\n" result += f"- Check-in time: {check_in_info.get('check_in_start_time', 'N/A')}\n" result += f"- Check-out time: {check_in_info.get('check_out_time', 'N/A')}\n" return truncate_text(result) except Exception as e: return f"Error retrieving accommodation: {str(e)}\n\nTroubleshooting:\n- Verify the accommodation ID is correct (starts with 'acc_')\n- Check if the accommodation exists" @mcp.tool( name="duffel_get_accommodation_rates", annotations={ "title": "Get Accommodation Rates", "readOnlyHint": True, "destructiveHint": False, "idempotentHint": False, "openWorldHint": True } ) async def get_accommodation_rates(params: GetAccommodationRatesInput) -> str: """ Retrieve available rooms and rates for a specific search result. This tool fetches all available room options with: - Room types and bed configurations - Pricing for each rate option - Cancellation policies - Breakfast and meal inclusions - Rate IDs needed for creating quotes Use this when: - User selected an accommodation from search results - Need to see room options and prices - Want to compare different rate plans - Ready to proceed with booking Important: Each rate has a unique rate_id that must be used to create a quote. Returns accommodation rates in specified format (JSON or Markdown). """ try: response = await make_api_request( method="POST", endpoint="/stays/search_results/rates", data={ "data": { "search_result_id": params.search_result_id } } ) rates = response.get("data", []) if not rates: return "No rates available for this accommodation. It may be fully booked for your selected dates." if params.response_format == ResponseFormat.JSON: result = { "total_rates": len(rates), "rates": rates[:10] # Limit to 10 for readability } return truncate_text(format_json_response(result)) else: # Markdown format lines = [ f"# Available Rooms and Rates", f"**Total Options**: {len(rates)}", "", f"Showing top {min(5, len(rates))} rate options:", "" ] for rate in rates[:5]: lines.append(format_markdown_accommodation_rate(rate)) lines.append("*Use this Rate ID with duffel_create_stays_quote to get a booking quote*") lines.append("---") lines.append("") if len(rates) > 5: lines.append(f"\n*{len(rates) - 5} more rate options available.*") return truncate_text("\n".join(lines)) except Exception as e: return f"Error retrieving rates: {str(e)}\n\nTroubleshooting:\n- Verify the search_result_id is correct (starts with 'srs_')\n- The search might have expired - perform a new search\n- Try different dates if no rates available" @mcp.tool( name="duffel_create_stays_quote", annotations={ "title": "Create Stays Quote", "readOnlyHint": False, "destructiveHint": False, "idempotentHint": False, "openWorldHint": True } ) async def create_stays_quote(params: CreateStaysQuoteInput) -> str: """ Create a ready-to-book quote for a specific accommodation rate. This tool: - Confirms rate availability - Locks in the current price - Validates the booking before payment - Returns a quote_id for creating the booking Use this when: - User has selected a specific room/rate - Before collecting payment information - To confirm final pricing before booking Important: - Always create a quote before booking - Quotes may expire after some time - Price and availability are confirmed at this stage Returns quote details in specified format (JSON or Markdown). """ try: response = await make_api_request( method="POST", endpoint="/stays/quotes", data={ "data": { "rate_id": params.rate_id } } ) quote = response["data"] if params.response_format == ResponseFormat.JSON: return truncate_text(format_json_response(quote)) else: # Markdown format lines = [ "# Accommodation Quote", "", f"**Quote ID**: `{quote.get('id', 'N/A')}`", f"**Total Amount**: {format_currency(quote.get('total_amount', '0'), quote.get('total_currency', 'USD'))}", "", "## Accommodation Details" ] accommodation = quote.get("accommodation", {}) if accommodation: lines.append(f"**Property**: {accommodation.get('name', 'N/A')}") location = accommodation.get("location", {}) if location: lines.append(f"**Location**: {location.get('city_name', 'N/A')}, {location.get('country_name', 'N/A')}") room = quote.get("room", {}) if room: lines.append("") lines.append("## Room Details") lines.append(f"**Room**: {room.get('name', 'N/A')}") beds = room.get("beds", []) if beds: bed_desc = ", ".join([f"{bed.get('count', 1)} {bed.get('type', 'bed')}" for bed in beds]) lines.append(f"**Beds**: {bed_desc}") lines.append("") lines.append("## Stay Details") lines.append(f"**Check-in**: {quote.get('check_in_date', 'N/A')}") lines.append(f"**Check-out**: {quote.get('check_out_date', 'N/A')}") lines.append(f"**Nights**: {quote.get('nights', 'N/A')}") lines.append("") lines.append("---") lines.append("*Use this Quote ID with duffel_create_stays_booking to complete the booking*") return truncate_text("\n".join(lines)) except Exception as e: return f"Error creating quote: {str(e)}\n\nTroubleshooting:\n- Verify the rate_id is correct (starts with 'rat_')\n- The rate may no longer be available\n- Try selecting a different rate option" @mcp.tool( name="duffel_create_stays_booking", annotations={ "title": "Create Stays Booking", "readOnlyHint": False, "destructiveHint": False, "idempotentHint": False, "openWorldHint": True } ) async def create_stays_booking(params: CreateStaysBookingInput) -> str: """ Create a confirmed accommodation booking from a quote. ⚠️ IMPORTANT: This creates a real booking and charges payment. This action: - Creates a confirmed hotel/accommodation reservation - Charges the Duffel Balance (or specified payment method) - Issues a booking confirmation from the property - May be non-refundable or have cancellation fees Before calling this tool: 1. Verify the quote is current using duffel_create_stays_quote 2. Confirm all guest details are accurate 3. Ensure user understands booking terms and cancellation policy Required data: - Quote ID from quote creation - Guest details (names and contact information) - Optional special requests The tool returns: - Booking ID for reference - Property confirmation number - Check-in instructions - Booking details Use test mode tokens for testing to avoid actual charges. Returns booking details in specified format (JSON or Markdown). """ booking_data = { "data": { "quote_id": params.quote_id, "guests": [ { "given_name": guest.given_name, "family_name": guest.family_name, "email": guest.email, **({"phone_number": guest.phone_number} if guest.phone_number else {}) } for guest in params.guests ] } } if params.special_requests: booking_data["data"]["special_requests"] = params.special_requests try: response = await make_api_request( method="POST", endpoint="/stays/bookings", data=booking_data ) booking = response["data"] if params.response_format == ResponseFormat.JSON: return truncate_text(format_json_response(booking)) else: # Markdown format lines = [ "# ✅ Accommodation Booking Confirmed!", "", f"**Booking ID**: `{booking.get('id', 'N/A')}`", f"**Status**: {booking.get('booking_status', 'N/A')}", f"**Total**: {format_currency(booking.get('total_amount', '0'), booking.get('total_currency', 'USD'))}", "" ] accommodation = booking.get("accommodation", {}) if accommodation: lines.append("## Property Details") lines.append(f"**Name**: {accommodation.get('name', 'N/A')}") location = accommodation.get("location", {}) if location: lines.append(f"**Address**: {location.get('address', {}).get('line_one', 'N/A')}") lines.append(f"**City**: {location.get('city_name', 'N/A')}") lines.append("") lines.append("## Stay Details") lines.append(f"**Check-in**: {booking.get('check_in_date', 'N/A')}") lines.append(f"**Check-out**: {booking.get('check_out_date', 'N/A')}") lines.append(f"**Nights**: {booking.get('nights', 'N/A')}") guests = booking.get("guests", []) if guests: lines.append("") lines.append("## Guests") for i, guest in enumerate(guests, 1): lines.append(f"{i}. {guest.get('given_name')} {guest.get('family_name')}") payment_instructions = booking.get("payment_instructions", {}) if payment_instructions: lines.append("") lines.append("## Payment") lines.append(f"**Status**: {payment_instructions.get('status', 'N/A')}") lines.append("") lines.append("---") lines.append("*Save the Booking ID for future reference and modifications.*") lines.append("*Check your email for detailed confirmation and check-in instructions.*") return truncate_text("\n".join(lines)) except Exception as e: error_msg = str(e) troubleshooting = "\n\nTroubleshooting:\n" if "quote" in error_msg.lower(): troubleshooting += "- The quote may have expired. Create a new quote.\n" troubleshooting += "- Verify the quote_id is correct (starts with 'quo_')\n" elif "validation" in error_msg.lower(): troubleshooting += "- Check guest details are complete and accurate\n" troubleshooting += "- Verify email addresses are valid\n" troubleshooting += "- Ensure at least one guest is provided\n" else: troubleshooting += "- Verify all guest details are accurate\n" troubleshooting += "- Check if using test mode token for testing\n" troubleshooting += "- The accommodation may no longer be available\n" return f"Error creating booking: {error_msg}{troubleshooting}" @mcp.tool( name="duffel_get_stays_booking", annotations={ "title": "Get Stays Booking Details", "readOnlyHint": True, "destructiveHint": False, "idempotentHint": True, "openWorldHint": True } ) async def get_stays_booking(params: GetStaysBookingInput) -> str: """ Retrieve complete details for an existing accommodation booking. This tool fetches: - Booking status and confirmation details - Property information and address - Check-in/check-out dates - Guest information - Payment status - Cancellation policy Use this when: - User needs to review their booking - Checking booking status - Getting property contact information - Before making modifications or cancellations Returns booking details in specified format (JSON or Markdown). """ try: response = await make_api_request( method="GET", endpoint=f"/stays/bookings/{params.booking_id}" ) booking = response["data"] if params.response_format == ResponseFormat.JSON: return truncate_text(format_json_response(booking)) else: # Markdown format lines = [ "# Accommodation Booking Details", "", f"**Booking ID**: `{booking.get('id', 'N/A')}`", f"**Status**: {booking.get('booking_status', 'N/A')}", f"**Total**: {format_currency(booking.get('total_amount', '0'), booking.get('total_currency', 'USD'))}", f"**Booked**: {booking.get('created_at', 'N/A')}", "" ] accommodation = booking.get("accommodation", {}) if accommodation: lines.append("## Property Information") lines.append(f"**Name**: {accommodation.get('name', 'N/A')}") location = accommodation.get("location", {}) if location: address = location.get("address", {}) lines.append(f"**Address**: {address.get('line_one', 'N/A')}") if address.get('line_two'): lines.append(f" {address['line_two']}") lines.append(f"**City**: {location.get('city_name', 'N/A')}") lines.append(f"**Country**: {location.get('country_name', 'N/A')}") lines.append(f"**Postal Code**: {address.get('postal_code', 'N/A')}") lines.append("") lines.append("## Stay Details") lines.append(f"**Check-in**: {booking.get('check_in_date', 'N/A')}") lines.append(f"**Check-out**: {booking.get('check_out_date', 'N/A')}") lines.append(f"**Nights**: {booking.get('nights', 'N/A')}") room = booking.get("room", {}) if room: lines.append(f"**Room**: {room.get('name', 'N/A')}") guests = booking.get("guests", []) if guests: lines.append("") lines.append("## Guests") for i, guest in enumerate(guests, 1): lines.append(f"{i}. {guest.get('given_name')} {guest.get('family_name')}") if guest.get('email'): lines.append(f" Email: {guest['email']}") payment_instructions = booking.get("payment_instructions", {}) if payment_instructions: lines.append("") lines.append("## Payment") lines.append(f"**Status**: {payment_instructions.get('status', 'N/A')}") cancellation = booking.get("cancellation_timeline", {}) if cancellation: lines.append("") lines.append("## Cancellation Policy") lines.append(f"{cancellation.get('description', 'See booking terms')}") return truncate_text("\n".join(lines)) except Exception as e: return f"Error retrieving booking: {str(e)}\n\nTroubleshooting:\n- Verify the booking ID is correct (starts with 'bok_')\n- Check if you have access to this booking\n- Booking might not exist"