"""Trip management tools for MCP server"""
from typing import Optional
from mcp.server import Server
from mcp.types import Tool, TextContent
import mcp.types as types
def register_trip_tools(server: Server, trip_db):
"""Register trip-related MCP tools"""
@server.call_tool()
async def search_trips(name: str, arguments: dict) -> list[types.TextContent]:
"""
Search for trips by destination, date range, or status.
Args:
destination (str, optional): Filter by destination (partial match)
start_date (str, optional): Filter trips starting after this date (YYYY-MM-DD)
end_date (str, optional): Filter trips ending before this date (YYYY-MM-DD)
status (str, optional): Filter by status (completed, upcoming, cancelled)
limit (int, optional): Maximum number of results (default: 50)
"""
destination = arguments.get("destination")
start_date = arguments.get("start_date")
end_date = arguments.get("end_date")
status = arguments.get("status")
limit = arguments.get("limit", 50)
if status and status not in ["completed", "upcoming", "cancelled"]:
return [TextContent(type="text", text="Error: status must be one of: completed, upcoming, cancelled")]
try:
results = trip_db.search(
destination=destination,
start_date=start_date,
end_date=end_date,
status=status,
limit=limit
)
if not results:
return [TextContent(type="text", text="No trips found matching the criteria")]
# Format results
output = f"Found {len(results)} trip(s):\n\n"
for trip in results:
output += f"Trip ID: {trip['trip_id']}\n"
output += f"Customer ID: {trip['customer_id']}\n"
output += f"Destination: {trip['destination']}\n"
output += f"Dates: {trip['start_date']} to {trip['end_date']}\n"
output += f"Cost: ${trip['cost']:,.2f}\n"
output += f"Status: {trip['status']}\n"
output += f"Travelers: {trip['num_travelers']}\n"
output += f"Type: {trip['trip_type']}\n"
if trip['notes']:
output += f"Notes: {trip['notes']}\n"
output += "-" * 50 + "\n"
return [TextContent(type="text", text=output)]
except Exception as e:
return [TextContent(type="text", text=f"Error searching trips: {str(e)}")]
@server.call_tool()
async def get_trip_history(name: str, arguments: dict) -> list[types.TextContent]:
"""
Get trip history for a specific customer.
Args:
customer_id (int): The customer ID
limit (int, optional): Maximum number of trips to return (default: 50)
"""
try:
customer_id = int(arguments.get("customer_id"))
except (TypeError, ValueError):
return [TextContent(type="text", text="Error: customer_id must be a valid integer")]
limit = arguments.get("limit", 50)
try:
trips = trip_db.get_by_customer(customer_id, limit)
if not trips:
return [TextContent(type="text", text=f"No trips found for customer ID {customer_id}")]
# Calculate statistics
total_spent = sum(trip['cost'] for trip in trips if trip['status'] != 'cancelled')
completed_trips = sum(1 for trip in trips if trip['status'] == 'completed')
upcoming_trips = sum(1 for trip in trips if trip['status'] == 'upcoming')
cancelled_trips = sum(1 for trip in trips if trip['status'] == 'cancelled')
# Format output
output = f"TRIP HISTORY FOR CUSTOMER {customer_id}\n"
output += "=" * 50 + "\n\n"
output += f"Total Trips: {len(trips)}\n"
output += f" Completed: {completed_trips}\n"
output += f" Upcoming: {upcoming_trips}\n"
output += f" Cancelled: {cancelled_trips}\n"
output += f"Total Spent: ${total_spent:,.2f}\n\n"
output += "TRIP DETAILS\n"
output += "-" * 50 + "\n\n"
for trip in trips:
output += f"Trip ID: {trip['trip_id']}\n"
output += f"Destination: {trip['destination']}\n"
output += f"Dates: {trip['start_date']} to {trip['end_date']}\n"
output += f"Cost: ${trip['cost']:,.2f}\n"
output += f"Status: {trip['status']}\n"
output += f"Travelers: {trip['num_travelers']}\n"
output += f"Type: {trip['trip_type']}\n"
output += f"Booked: {trip['booking_date']}\n"
if trip['notes']:
output += f"Notes: {trip['notes']}\n"
output += "-" * 50 + "\n"
return [TextContent(type="text", text=output)]
except Exception as e:
return [TextContent(type="text", text=f"Error retrieving trip history: {str(e)}")]
