"""Request 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_request_tools(server: Server, request_db):
"""Register request-related MCP tools"""
@server.call_tool()
async def search_requests(name: str, arguments: dict) -> list[types.TextContent]:
"""
Search for customer information requests.
Args:
email (str, optional): Filter by email (partial match)
destination (str, optional): Filter by destination interest
status (str, optional): Filter by status (pending, contacted, converted, closed)
date_from (str, optional): Filter requests from this date (YYYY-MM-DD)
date_to (str, optional): Filter requests to this date (YYYY-MM-DD)
limit (int, optional): Maximum number of results (default: 50)
"""
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")]
try:
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")]
# Format results
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)]
except Exception as e:
return [TextContent(type="text", text=f"Error searching requests: {str(e)}")]
@server.call_tool()
async def get_pending_requests(name: str, arguments: dict) -> list[types.TextContent]:
"""
Get pending customer requests from the last N days.
Args:
days_back (int, optional): Number of days to look back (default: 30)
"""
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")]
try:
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")]
# Format output
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)]
except Exception as e:
return [TextContent(type="text", text=f"Error retrieving pending requests: {str(e)}")]
