""" Intelligent URL Builder Tools for Hotel MCP Server. This module provides MCP tools for building intelligent hotel reservation URLs that integrate with real hotel data and generate URLs for the accommodation booking system with proper query parameters. Supports building URLs for: - Hotel room reservations with real room data - Activity bookings with availability - Facility access with scheduling - Multi-language support with real translations """ import logging import os import re from datetime import date, datetime from typing import Any, Dict, List, Optional from urllib.parse import ( parse_qs, quote, quote_plus, unquote, unquote_plus, urlencode, urljoin, urlparse, urlunparse, ) logger = logging.getLogger(__name__) def _get_base_url_from_env() -> Optional[str]: """Get the hotel base URL from environment variables.""" return os.getenv("HOTEL_BASE_URL") def _safe_get_port(parsed_url): """Safely get port from parsed URL, handling invalid port ranges.""" try: return parsed_url.port except ValueError: # Port out of range, return None return None def _get_utm_params() -> Dict[str, str]: """Get UTM parameters from environment variables for tracking conversions.""" utm_params = {} # UTM Source - identifies the source (e.g., 'mcp-agent') utm_source = os.getenv("UTM_SOURCE", "hotel-mcp") if utm_source: utm_params["utm_source"] = utm_source # UTM Medium - identifies the medium (e.g., 'ai-agent') utm_medium = os.getenv("UTM_MEDIUM", "ai-agent") if utm_medium: utm_params["utm_medium"] = utm_medium # UTM Campaign - identifies the campaign (e.g., 'hotel-bookings-2024') utm_campaign = os.getenv("UTM_CAMPAIGN") if utm_campaign: utm_params["utm_campaign"] = utm_campaign # UTM Term - identifies paid search keywords utm_term = os.getenv("UTM_TERM") if utm_term: utm_params["utm_term"] = utm_term # UTM Content - identifies specific content/variant utm_content = os.getenv("UTM_CONTENT") if utm_content: utm_params["utm_content"] = utm_content return utm_params def _validate_email(email: str) -> bool: """Validate email format.""" if not email: return False email_pattern = r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$" return bool(re.match(email_pattern, email)) def _validate_date(date_str: str) -> bool: """Validate date format (YYYY-MM-DD) and that it's a valid date.""" if not date_str: return False try: datetime.strptime(date_str, "%Y-%m-%d") return True except ValueError: return False def _validate_phone(phone: str) -> bool: """Validate phone number format (basic validation).""" if not phone: return False # Remove common separators and spaces clean_phone = re.sub(r"[\s\-\(\)\+]", "", phone) # Check if it contains only digits and is reasonable length return clean_phone.isdigit() and 7 <= len(clean_phone) <= 15 def _sanitize_name(name: str) -> str: """Sanitize guest name by removing potentially harmful characters.""" if not name: return "" # Remove HTML tags and special characters, keep letters, spaces, hyphens, apostrophes sanitized = re.sub(r'[<>"\']', "", name) sanitized = re.sub(r"[^\w\s\-\']", "", sanitized) # Limit length and strip whitespace return sanitized.strip()[:100] def _validate_guest_counts(adults: int, children: int) -> Dict[str, Any]: """Validate guest counts and return validation result.""" errors = [] if adults < 1: errors.append("Adults count must be at least 1") elif adults > 20: errors.append("Adults count cannot exceed 20") if children < 0: errors.append("Children count cannot be negative") elif children > 10: errors.append("Children count cannot exceed 10") total_guests = adults + children if total_guests > 25: errors.append("Total guests cannot exceed 25") return { "valid": len(errors) == 0, "errors": errors, "adults": max(1, min(20, adults)), "children": max(0, min(10, children)), } def _validate_dates( check_in: Optional[str], check_out: Optional[str] ) -> Dict[str, Any]: """Validate check-in and check-out dates.""" errors = [] if check_in and not _validate_date(check_in): errors.append("Check-in date must be in YYYY-MM-DD format") if check_out and not _validate_date(check_out): errors.append("Check-out date must be in YYYY-MM-DD format") # If both dates are valid, check logical order if ( check_in and check_out and _validate_date(check_in) and _validate_date(check_out) ): try: checkin_date = datetime.strptime(check_in, "%Y-%m-%d").date() checkout_date = datetime.strptime(check_out, "%Y-%m-%d").date() today = date.today() if checkin_date < today: errors.append("Check-in date cannot be in the past") if checkout_date <= checkin_date: errors.append("Check-out date must be after check-in date") # Check for reasonable stay duration (max 30 days) stay_duration = (checkout_date - checkin_date).days if stay_duration > 30: errors.append("Stay duration cannot exceed 30 days") except ValueError: errors.append("Invalid date values") return {"valid": len(errors) == 0, "errors": errors} def _validate_and_sanitize_guest_data( guest_name: Optional[str], guest_email: Optional[str], guest_phone: Optional[str] ) -> Dict[str, Any]: """Validate and sanitize guest data.""" errors = [] sanitized_data = {} # Validate and sanitize name if guest_name: sanitized_name = _sanitize_name(guest_name) if not sanitized_name: errors.append("Guest name contains invalid characters") else: sanitized_data["guest_name"] = sanitized_name # Validate email if guest_email: if not _validate_email(guest_email): errors.append("Invalid email format") else: sanitized_data["guest_email"] = guest_email.lower().strip() # Validate phone if guest_phone: if not _validate_phone(guest_phone): errors.append("Invalid phone number format") else: sanitized_data["guest_phone"] = guest_phone.strip() return { "valid": len(errors) == 0, "errors": errors, "sanitized_data": sanitized_data, } def build_hotel_reservation_url( domain: Optional[str] = None, room_id: Optional[str] = None, room_data: Optional[Dict[str, Any]] = None, check_in: Optional[str] = None, check_out: Optional[str] = None, adults: int = 2, children: int = 0, guest_name: Optional[str] = None, guest_email: Optional[str] = None, guest_phone: Optional[str] = None, language: Optional[str] = None, ) -> Dict[str, Any]: """ Build intelligent hotel reservation URL using real room data. ⚠️ CRITICAL: This function REQUIRES roomType for the reservation form to work! The roomType parameter is ESSENTIAL for the hotel website to display the reservation form. Without it, visitors will see an empty page. This function integrates with the hotel MCP data to generate URLs for the /accommodation/reservations page with proper query parameters. Args: domain: The hotel website domain (e.g., "https://hotel.com"). If not provided, uses HOTEL_BASE_URL environment variable. Domain is required - either as parameter or environment variable. room_id: Room ID to get real data from MCP (REQUIRED for roomType) room_data: Pre-fetched room data with slug field (REQUIRED for roomType) check_in: Check-in date (YYYY-MM-DD format) check_out: Check-out date (YYYY-MM-DD format) adults: Number of adult guests (default: 2) children: Number of children (default: 0) guest_name: Guest name for pre-filling form guest_email: Guest email for pre-filling form guest_phone: Guest phone for pre-filling form language: Language for room data (es, en, fr, ca) Returns: Dictionary with the built URL and metadata Raises: ValueError: If neither room_id nor room_data with slug is provided (roomType is required for reservation form to work) """ try: # Resolve domain: explicit parameter takes precedence over environment variable final_domain = domain or _get_base_url_from_env() if not final_domain: raise ValueError( "Domain is required. Provide domain parameter or set HOTEL_BASE_URL environment variable." ) # Ensure domain has protocol if not final_domain.startswith(("http://", "https://")): final_domain = f"https://{final_domain}" # Validate guest counts guest_validation = _validate_guest_counts(adults, children) if not guest_validation["valid"]: raise ValueError( f"Invalid guest counts: {'; '.join(guest_validation['errors'])}" ) # Use validated guest counts validated_adults = guest_validation["adults"] validated_children = guest_validation["children"] # Validate dates date_validation = _validate_dates(check_in, check_out) if not date_validation["valid"]: raise ValueError(f"Invalid dates: {'; '.join(date_validation['errors'])}") # Validate and sanitize guest data guest_data_validation = _validate_and_sanitize_guest_data( guest_name, guest_email, guest_phone ) if not guest_data_validation["valid"]: raise ValueError( f"Invalid guest data: {'; '.join(guest_data_validation['errors'])}" ) sanitized_guest_data = guest_data_validation["sanitized_data"] # Build query parameters query_params = {} # Add validated guest information if "guest_name" in sanitized_guest_data: query_params["guestName"] = sanitized_guest_data["guest_name"] if "guest_email" in sanitized_guest_data: query_params["guestEmail"] = sanitized_guest_data["guest_email"] if "guest_phone" in sanitized_guest_data: query_params["guestPhone"] = sanitized_guest_data["guest_phone"] # Add validated booking dates if check_in: query_params["checkIn"] = check_in if check_out: query_params["checkOut"] = check_out # Add validated guest counts if validated_adults > 0: query_params["adults"] = str(validated_adults) if validated_children > 0: query_params["children"] = str(validated_children) # Add UTM tracking parameters for conversion tracking utm_params = _get_utm_params() query_params.update(utm_params) # Handle room data - either from room_data parameter or fetch from MCP room_info = None if room_data: room_info = room_data elif room_id: # Import here to avoid circular imports try: from . import rooms room_response = rooms.get_room_details(room_id, language=language) if room_response.get("status") != "error": room_info = room_response.get("item") except Exception as e: logger.warning(f"Could not fetch room data for {room_id}: {e}") # Add room type using the slug from database (not extracted from name) room_type_found = False if room_info: # Use the slug field directly from the database room_slug = room_info.get("slug") if room_slug: query_params["roomType"] = room_slug room_type_found = True else: # Fallback: if no slug, log warning and skip roomType logger.warning( f"No slug found for room {room_info.get('id', 'unknown')}" ) logger.debug(f"Room info available: {list(room_info.keys())}") # CRITICAL VALIDATION: roomType is REQUIRED for reservation form to work if not room_type_found: raise ValueError( "❌ CRITICAL: roomType is REQUIRED for reservation form to work!\n" "Without roomType, the reservation form will NOT be displayed.\n" "Solutions:\n" "1. Provide room_id parameter to fetch room data automatically\n" "2. Provide room_data parameter with slug field\n" "3. Use build_room_reservation_url_from_search() to find a room first\n" "4. Use build_multiple_reservation_urls() to get URLs with room types\n" "\nThe roomType parameter (e.g., 'ocean-deluxe', 'garden-deluxe') is essential " "for the hotel website to show the correct reservation form." ) # Build the complete URL reservation_url = build_url( base_url=final_domain, path="/accommodation/reservations", query_params=query_params, encode_params=True, ) # Add metadata about the room if available metadata = { "room_info": room_info, "parameters_used": list(query_params.keys()), "booking_summary": { "check_in": check_in, "check_out": check_out, "guests": {"adults": validated_adults, "children": validated_children}, "room_type": query_params.get("roomType", "Not specified"), }, "validation_applied": { "guest_data_sanitized": bool(sanitized_guest_data), "dates_validated": bool(check_in or check_out), "guest_counts_validated": True, "utm_tracking_added": bool(utm_params), }, "utm_tracking": utm_params if utm_params else None, } if reservation_url["status"] == "success": reservation_url["metadata"] = metadata reservation_url["message"] = "Hotel reservation URL built successfully" return reservation_url except Exception as e: logger.error(f"Error building hotel reservation URL: {e}") return { "url": None, "error": str(e), "message": "Failed to build hotel reservation URL", "status": "error", } def build_room_reservation_url_from_search( search_query: str, domain: Optional[str] = None, check_in: Optional[str] = None, check_out: Optional[str] = None, adults: int = 2, children: int = 0, language: Optional[str] = None, site_id: Optional[str] = None, ) -> Dict[str, Any]: """ Build reservation URL by searching for rooms and using the best match. This function searches the hotel's room inventory and generates a reservation URL for the best matching room. Args: search_query: Search query to find matching rooms domain: The hotel website domain. If not provided, uses HOTEL_BASE_URL environment variable. Domain is required - either as parameter or environment variable. check_in: Check-in date (YYYY-MM-DD format) check_out: Check-out date (YYYY-MM-DD format) adults: Number of adult guests children: Number of children language: Language for search and room data site_id: Hotel site ID for filtering Returns: Dictionary with reservation URL and search results """ try: # Import here to avoid circular imports from . import rooms # Search for matching rooms search_results = rooms.search_rooms( query=search_query, site_id=site_id, language=language, limit=5 ) if search_results.get("status") == "error" or not search_results.get("items"): return { "url": None, "error": "No rooms found matching search criteria", "message": f"No rooms found for query: {search_query}", "status": "error", "search_query": search_query, } # Use the first (best) match best_room = search_results["items"][0] # Build reservation URL using the found room (validation happens in build_hotel_reservation_url) reservation_url = build_hotel_reservation_url( domain=domain, room_data=best_room, check_in=check_in, check_out=check_out, adults=adults, children=children, language=language, ) # Add search metadata if reservation_url["status"] == "success": reservation_url["search_metadata"] = { "search_query": search_query, "total_matches": search_results.get("total_count", 0), "selected_room": { "id": best_room.get("id"), "name": best_room.get("name"), "type": best_room.get("type"), }, "other_options": len(search_results["items"]) - 1, } reservation_url[ "message" ] = f"Reservation URL built for best match: {best_room.get('name')}" return reservation_url except Exception as e: logger.error(f"Error building reservation URL from search: {e}") return { "url": None, "error": str(e), "message": "Failed to build reservation URL from search", "status": "error", "search_query": search_query, } def build_multiple_reservation_urls( domain: Optional[str] = None, room_ids: Optional[List[str]] = None, check_in: Optional[str] = None, check_out: Optional[str] = None, adults: int = 2, children: int = 0, language: Optional[str] = None, site_id: Optional[str] = None, limit: int = 10, ) -> Dict[str, Any]: """ Build multiple reservation URLs for different rooms. This function generates reservation URLs for multiple rooms, either from a provided list of room IDs or by fetching available rooms. Args: domain: The hotel website domain. If not provided, uses HOTEL_BASE_URL environment variable. Domain is required - either as parameter or environment variable. room_ids: List of specific room IDs (optional) check_in: Check-in date (YYYY-MM-DD format) check_out: Check-out date (YYYY-MM-DD format) adults: Number of adult guests children: Number of children language: Language for room data site_id: Hotel site ID for filtering limit: Maximum number of URLs to generate Returns: Dictionary with multiple reservation URLs """ try: # Import here to avoid circular imports from . import rooms reservation_urls = [] if room_ids: # Build URLs for specific room IDs for room_id in room_ids[:limit]: try: # Validation happens in build_hotel_reservation_url room_url = build_hotel_reservation_url( domain=domain, room_id=room_id, check_in=check_in, check_out=check_out, adults=adults, children=children, language=language, ) if room_url["status"] == "success": reservation_urls.append(room_url) except Exception as e: logger.warning(f"Failed to build URL for room {room_id}: {e}") else: # Get available rooms and build URLs rooms_list = rooms.list_rooms( site_id=site_id, language=language, limit=limit ) if rooms_list.get("status") != "error" and rooms_list.get("rooms"): for room in rooms_list["rooms"]: try: # Validation happens in build_hotel_reservation_url room_url = build_hotel_reservation_url( domain=domain, room_data=room, check_in=check_in, check_out=check_out, adults=adults, children=children, language=language, ) if room_url["status"] == "success": reservation_urls.append(room_url) except Exception as e: logger.warning( f"Failed to build URL for room {room.get('id')}: {e}" ) return { "reservation_urls": reservation_urls, "total_urls": len(reservation_urls), "booking_parameters": { "check_in": check_in, "check_out": check_out, "adults": adults, "children": children, "language": language, }, "domain": domain, "message": f"Generated {len(reservation_urls)} reservation URLs", "status": "success", } except Exception as e: logger.error(f"Error building multiple reservation URLs: {e}") return { "reservation_urls": [], "error": str(e), "message": "Failed to build multiple reservation URLs", "status": "error", } def build_url( base_url: str, path: Optional[str] = None, query_params: Optional[Dict[str, Any]] = None, fragment: Optional[str] = None, encode_params: bool = True, ) -> Dict[str, Any]: """ Build a complete URL from components. Args: base_url: The base URL (required) path: Additional path to append to base URL query_params: Dictionary of query parameters fragment: URL fragment (anchor) encode_params: Whether to URL encode query parameters Returns: Dictionary containing the built URL and components """ try: # Validate base URL if not base_url: raise ValueError("Base URL is required") # Parse base URL to validate it parsed_base = urlparse(base_url) if not parsed_base.scheme or not parsed_base.netloc: raise ValueError("Invalid base URL format") # Build the URL if path: # Ensure path starts with / if base_url doesn't end with / if not base_url.endswith("/") and not path.startswith("/"): path = "/" + path full_url = urljoin(base_url, path) else: full_url = base_url # Parse the URL to work with components parsed = urlparse(full_url) # Handle query parameters query_string = "" if query_params: if encode_params: # URL encode the parameters encoded_params = {} for key, value in query_params.items(): if isinstance(value, (list, tuple)): # Handle multiple values for same parameter encoded_params[key] = [str(v) for v in value] else: encoded_params[key] = str(value) query_string = urlencode(encoded_params, doseq=True) else: # Build query string without encoding query_parts = [] for key, value in query_params.items(): if isinstance(value, (list, tuple)): for v in value: query_parts.append(f"{key}={v}") else: query_parts.append(f"{key}={value}") query_string = "&".join(query_parts) # Combine existing query with new query params if parsed.query and query_string: combined_query = f"{parsed.query}&{query_string}" elif query_string: combined_query = query_string else: combined_query = parsed.query # Handle fragment final_fragment = fragment or parsed.fragment # Reconstruct the URL final_url = urlunparse( ( parsed.scheme, parsed.netloc, parsed.path, parsed.params, combined_query, final_fragment, ) ) return { "url": final_url, "components": { "scheme": parsed.scheme, "netloc": parsed.netloc, "hostname": parsed.hostname, "port": parsed.port, "path": parsed.path, "query": combined_query, "fragment": final_fragment, }, "query_params": parse_qs(combined_query) if combined_query else {}, "message": "URL built successfully", "status": "success", } except Exception as e: logger.error(f"Error building URL: {e}") return { "url": None, "error": str(e), "message": "Failed to build URL", "status": "error", } def parse_url(url: str) -> Dict[str, Any]: """ Parse a URL into its components. Args: url: The URL to parse Returns: Dictionary containing parsed URL components """ try: if not url: raise ValueError("URL is required") parsed = urlparse(url) # Parse query parameters query_params = parse_qs(parsed.query) if parsed.query else {} # Convert single-item lists to strings for cleaner output simplified_params = {} for key, values in query_params.items(): if len(values) == 1: simplified_params[key] = values[0] else: simplified_params[key] = values return { "original_url": url, "components": { "scheme": parsed.scheme, "netloc": parsed.netloc, "hostname": parsed.hostname, "port": parsed.port, "path": parsed.path, "params": parsed.params, "query": parsed.query, "fragment": parsed.fragment, }, "query_params": simplified_params, "is_valid": bool(parsed.scheme and parsed.netloc), "message": "URL parsed successfully", "status": "success", } except Exception as e: logger.error(f"Error parsing URL: {e}") return { "original_url": url, "error": str(e), "message": "Failed to parse URL", "status": "error", } def encode_url_component( text: str, component_type: str = "query", safe_chars: Optional[str] = None ) -> Dict[str, Any]: """ URL encode a text component. Args: text: Text to encode component_type: Type of component (query, path, fragment) safe_chars: Characters to not encode (optional) Returns: Dictionary containing encoded text and details """ try: if not text: raise ValueError("Text to encode is required") if component_type == "query": # Use quote_plus for query parameters (spaces become +) encoded = quote_plus(text, safe=safe_chars or "") elif component_type == "path": # Use quote for path components (spaces become %20) encoded = quote(text, safe=safe_chars or "/") elif component_type == "fragment": # Use quote for fragments encoded = quote(text, safe=safe_chars or "") else: # Default to quote encoded = quote(text, safe=safe_chars or "") return { "original": text, "encoded": encoded, "component_type": component_type, "safe_chars": safe_chars, "message": f"Text encoded for {component_type} component", "status": "success", } except Exception as e: logger.error(f"Error encoding URL component: {e}") return { "original": text, "error": str(e), "message": "Failed to encode URL component", "status": "error", } def decode_url_component( encoded_text: str, component_type: str = "query" ) -> Dict[str, Any]: """ URL decode a text component. Args: encoded_text: Encoded text to decode component_type: Type of component (query, path, fragment) Returns: Dictionary containing decoded text and details """ try: if not encoded_text: raise ValueError("Encoded text is required") if component_type == "query": # Use unquote_plus for query parameters (+ becomes space) decoded = unquote_plus(encoded_text) else: # Use unquote for other components decoded = unquote(encoded_text) return { "encoded": encoded_text, "decoded": decoded, "component_type": component_type, "message": f"Text decoded from {component_type} component", "status": "success", } except Exception as e: logger.error(f"Error decoding URL component: {e}") return { "encoded": encoded_text, "error": str(e), "message": "Failed to decode URL component", "status": "error", } def build_url_from_template( template: str, path_params: Optional[Dict[str, str]] = None, query_params: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: """ Build URL from a template with path parameter substitution. Args: template: URL template with {param} placeholders path_params: Dictionary of path parameters to substitute query_params: Dictionary of query parameters to add Returns: Dictionary containing the built URL and substitution details """ try: if not template: raise ValueError("URL template is required") # Find all path parameters in template path_param_pattern = r"\{([^}]+)\}" found_params = re.findall(path_param_pattern, template) # Start with the template url = template substitutions = {} # Substitute path parameters if path_params: for param_name, param_value in path_params.items(): placeholder = f"{{{param_name}}}" if placeholder in url: # URL encode the parameter value for path encoded_value = quote(str(param_value), safe="") url = url.replace(placeholder, encoded_value) substitutions[param_name] = { "original": param_value, "encoded": encoded_value, "substituted": True, } # Check for unsubstituted parameters remaining_params = re.findall(path_param_pattern, url) # Add query parameters if provided if query_params: query_string = urlencode(query_params, doseq=True) separator = "&" if "?" in url else "?" url = f"{url}{separator}{query_string}" return { "template": template, "url": url, "found_path_params": found_params, "substitutions": substitutions, "remaining_params": remaining_params, "query_params": query_params or {}, "fully_substituted": len(remaining_params) == 0, "message": "URL built from template successfully", "status": "success", } except Exception as e: logger.error(f"Error building URL from template: {e}") return { "template": template, "error": str(e), "message": "Failed to build URL from template", "status": "error", } def validate_url( url: str, allowed_schemes: Optional[List[str]] = None ) -> Dict[str, Any]: """ Validate a URL and check its components. Args: url: URL to validate allowed_schemes: List of allowed schemes (default: ['http', 'https']) Returns: Dictionary containing validation results """ try: if not url: raise ValueError("URL is required") if allowed_schemes is None: allowed_schemes = ["http", "https"] try: parsed = urlparse(url) except ValueError as e: # Handle invalid URL format (like invalid port) return { "url": url, "is_valid": False, "validation": { "has_scheme": False, "has_netloc": False, "scheme_allowed": False, "has_valid_hostname": False, "port_valid": False, }, "components": {}, "allowed_schemes": allowed_schemes, "error": str(e), "message": "URL validation completed with errors", "status": "success", } validation_results = { "has_scheme": bool(parsed.scheme), "has_netloc": bool(parsed.netloc), "scheme_allowed": parsed.scheme.lower() in [s.lower() for s in allowed_schemes], "has_valid_hostname": bool(parsed.hostname), "port_valid": True, # Will be updated below } # Validate port if present try: port = parsed.port if port is not None: validation_results["port_valid"] = 1 <= port <= 65535 except ValueError: # Invalid port format (port out of range) validation_results["port_valid"] = False # Overall validity is_valid = all( [ validation_results["has_scheme"], validation_results["has_netloc"], validation_results["scheme_allowed"], validation_results["port_valid"], ] ) return { "url": url, "is_valid": is_valid, "validation": validation_results, "components": { "scheme": parsed.scheme, "netloc": parsed.netloc, "hostname": parsed.hostname, "port": _safe_get_port(parsed), "path": parsed.path, "query": parsed.query, "fragment": parsed.fragment, }, "allowed_schemes": allowed_schemes, "message": "URL validation completed", "status": "success", } except Exception as e: logger.error(f"Error validating URL: {e}") return { "url": url, "is_valid": False, "error": str(e), "message": "Failed to validate URL", "status": "error", }