"""Travel Company MCP Server - Main entry point"""
import asyncio
import logging
import sys
from pathlib import Path
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
from .database import Database, CustomerDB, TripDB, RequestDB
# Determine project root (parent of src/)
PROJECT_ROOT = Path(__file__).parent.parent
LOGS_DIR = PROJECT_ROOT / "logs"
DATA_DIR = PROJECT_ROOT / "data"
# Ensure directories exist
LOGS_DIR.mkdir(exist_ok=True)
DATA_DIR.mkdir(exist_ok=True)
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler(LOGS_DIR / 'server.log'),
logging.StreamHandler(sys.stderr) # Use stderr for MCP
]
)
logger = logging.getLogger(__name__)
async def main():
"""Main server entry point"""
logger.info("Starting Travel Company MCP Server...")
# Initialize database
db_path = str(DATA_DIR / "travel_company.db")
db = Database(db_path)
# Create database access objects
customer_db = CustomerDB(db.conn)
trip_db = TripDB(db.conn)
request_db = RequestDB(db.conn)
# Create MCP server
server = Server("travel-company-mcp")
logger.info("Registering tools...")
# Register list_tools handler
@server.list_tools()
async def list_tools() -> list[Tool]:
return [
# Customer tools
Tool(
name="search_customers",
description="Search for customers by name, email, phone, or customer ID. Returns matching customer records with basic information.",
inputSchema={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "The search query (e.g., customer name, email, phone number, or ID)"
},
"search_by": {
"type": "string",
"description": "Field to search by",
"enum": ["name", "email", "phone", "customer_id"],
"default": "name"
}
},
"required": ["query"]
}
),
Tool(
name="get_customer_profile",
description="Get detailed customer profile including personal information and trip statistics (total trips, lifetime spending, last trip date).",
inputSchema={
"type": "object",
"properties": {
"customer_id": {
"type": "integer",
"description": "The customer ID to retrieve"
}
},
"required": ["customer_id"]
}
),
# Trip tools
Tool(
name="search_trips",
description="Search for trips by destination, date range, or status. Useful for finding trips to specific locations or within certain timeframes.",
inputSchema={
"type": "object",
"properties": {
"destination": {
"type": "string",
"description": "Filter by destination (partial match, e.g., 'Paris' or 'France')"
},
"start_date": {
"type": "string",
"description": "Filter trips starting after this date (format: YYYY-MM-DD)"
},
"end_date": {
"type": "string",
"description": "Filter trips ending before this date (format: YYYY-MM-DD)"
},
"status": {
"type": "string",
"description": "Filter by trip status",
"enum": ["completed", "upcoming", "cancelled"]
},
"limit": {
"type": "integer",
"description": "Maximum number of results to return",
"default": 50
}
}
}
),
Tool(
name="get_trip_history",
description="Get complete trip history for a specific customer, including statistics like total trips, spending, and trip details ordered by date.",
inputSchema={
"type": "object",
"properties": {
"customer_id": {
"type": "integer",
"description": "The customer ID to get trip history for"
},
"limit": {
"type": "integer",
"description": "Maximum number of trips to return",
"default": 50
}
},
"required": ["customer_id"]
}
),
# Request tools
Tool(
name="search_requests",
description="Search for customer information requests by email, destination, status, or date range. Useful for finding inquiries that need follow-up.",
inputSchema={
"type": "object",
"properties": {
"email": {
"type": "string",
"description": "Filter by customer email (partial match)"
},
"destination": {
"type": "string",
"description": "Filter by destination interest (partial match)"
},
"status": {
"type": "string",
"description": "Filter by request status",
"enum": ["pending", "contacted", "converted", "closed"]
},
"date_from": {
"type": "string",
"description": "Filter requests from this date (format: YYYY-MM-DD)"
},
"date_to": {
"type": "string",
"description": "Filter requests to this date (format: YYYY-MM-DD)"
},
"limit": {
"type": "integer",
"description": "Maximum number of results to return",
"default": 50
}
}
}
),
Tool(
name="get_pending_requests",
description="Get all pending customer requests from the last N days. Perfect for daily follow-up workflows and lead management.",
inputSchema={
"type": "object",
"properties": {
"days_back": {
"type": "integer",
"description": "Number of days to look back for pending requests",
"default": 30
}
}
}
)
]
# Register call_tool handler
@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
try:
# Customer tools
if name == "search_customers":
query = arguments.get("query", "")
search_by = arguments.get("search_by", "name")
if not query:
return [TextContent(type="text", text="Error: query parameter is required")]
if search_by not in ["name", "email", "phone", "customer_id"]:
return [TextContent(type="text", text="Error: search_by must be one of: name, email, phone, customer_id")]
results = customer_db.search(query, search_by)
if not results:
return [TextContent(type="text", text=f"No customers found matching '{query}' in {search_by}")]
output = f"Found {len(results)} customer(s):\n\n"
for customer in results:
output += f"Customer ID: {customer['customer_id']}\n"
output += f"Name: {customer['name']}\n"
output += f"Email: {customer['email']}\n"
output += f"Phone: {customer['phone']}\n"
output += f"Location: {customer['city']}, {customer['state']}\n"
output += f"Loyalty Tier: {customer['loyalty_tier']}\n"
output += f"Registered: {customer['registration_date']}\n"
output += "-" * 50 + "\n"
return [TextContent(type="text", text=output)]
elif name == "get_customer_profile":
try:
customer_id = int(arguments.get("customer_id"))
except (TypeError, ValueError):
return [TextContent(type="text", text="Error: customer_id must be a valid integer")]
profile = customer_db.get_profile(customer_id)
if not profile:
return [TextContent(type="text", text=f"Customer with ID {customer_id} not found")]
output = "CUSTOMER PROFILE\n"
output += "=" * 50 + "\n\n"
output += f"Customer ID: {profile['customer_id']}\n"
output += f"Name: {profile['name']}\n"
output += f"Email: {profile['email']}\n"
output += f"Phone: {profile['phone']}\n"
output += f"Address: {profile['address']}\n"
output += f"City: {profile['city']}, {profile['state']}, {profile['country']}\n"
output += f"Registration Date: {profile['registration_date']}\n"
output += f"Loyalty Tier: {profile['loyalty_tier']}\n\n"
output += "STATISTICS\n"
output += "-" * 50 + "\n"
stats = profile['statistics']
output += f"Total Trips: {stats['total_trips']}\n"
output += f"Lifetime Spending: ${stats['lifetime_spending']:,.2f}\n"
output += f"Last Trip: {stats['last_trip_date'] or 'N/A'}\n"
return [TextContent(type="text", text=output)]
# Trip tools
elif name == "search_trips":
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")]
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")]
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)]
elif name == "get_trip_history":
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)
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}")]
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')
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)]
# Request tools
elif name == "search_requests":
email = arguments.get("email")
destination = arguments.get("destination")
status = arguments.get("status")
date_from = arguments.get("date_from")
date_to = arguments.get("date_to")
limit = arguments.get("limit", 50)
if status and status not in ["pending", "contacted", "converted", "closed"]:
return [TextContent(type="text", text="Error: status must be one of: pending, contacted, converted, closed")]
results = request_db.search(
email=email,
destination=destination,
status=status,
date_from=date_from,
date_to=date_to,
limit=limit
)
if not results:
return [TextContent(type="text", text="No requests found matching the criteria")]
output = f"Found {len(results)} request(s):\n\n"
for req in results:
output += f"Request ID: {req['request_id']}\n"
output += f"Name: {req['name']}\n"
output += f"Email: {req['email']}\n"
if req['phone']:
output += f"Phone: {req['phone']}\n"
output += f"Destination Interest: {req['destination_interest']}\n"
output += f"Travel Dates: {req['travel_dates']}\n"
output += f"Travelers: {req['num_travelers']}\n"
output += f"Budget Range: {req['budget_range']}\n"
output += f"Message: {req['message']}\n"
output += f"Request Date: {req['request_date']}\n"
output += f"Status: {req['status']}\n"
output += "-" * 50 + "\n"
return [TextContent(type="text", text=output)]
elif name == "get_pending_requests":
days_back = arguments.get("days_back", 30)
try:
days_back = int(days_back)
if days_back < 1:
return [TextContent(type="text", text="Error: days_back must be a positive integer")]
except (TypeError, ValueError):
return [TextContent(type="text", text="Error: days_back must be a valid integer")]
requests = request_db.get_pending(days_back)
if not requests:
return [TextContent(type="text", text=f"No pending requests found in the last {days_back} days")]
output = f"PENDING REQUESTS (Last {days_back} days)\n"
output += "=" * 50 + "\n"
output += f"Total: {len(requests)}\n\n"
for req in requests:
output += f"Request ID: {req['request_id']}\n"
output += f"Name: {req['name']}\n"
output += f"Email: {req['email']}\n"
if req['phone']:
output += f"Phone: {req['phone']}\n"
output += f"Destination: {req['destination_interest']}\n"
output += f"Travel Dates: {req['travel_dates']}\n"
output += f"Travelers: {req['num_travelers']}\n"
output += f"Budget: {req['budget_range']}\n"
output += f"Message: {req['message']}\n"
output += f"Submitted: {req['request_date']}\n"
output += "-" * 50 + "\n"
return [TextContent(type="text", text=output)]
else:
return [TextContent(type="text", text=f"Unknown tool: {name}")]
except Exception as e:
logger.error(f"Error in tool {name}: {e}", exc_info=True)
return [TextContent(type="text", text=f"Error: {str(e)}")]
logger.info("Server ready with 6 tools registered")
# Run server with stdio transport
async with stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
server.create_initialization_options()
)
def run():
"""Entry point for running the server"""
try:
asyncio.run(main())
except KeyboardInterrupt:
logger.info("Server stopped by user")
except Exception as e:
logger.error(f"Server error: {e}", exc_info=True)
raise
if __name__ == "__main__":
run()
