|
|
from mcp.server.fastmcp import FastMCP |
|
|
import sys, io, os, asyncio |
|
|
from tools.user_trip_summary import summarize_trip as db_summarize_trip |
|
|
from tools.fetch_trips_by_location import fetch_trips_by_location as fetch_trips |
|
|
from tools.web_search import get_travel_intelligence_briefing |
|
|
from tools.enhanced_personalized_planner import enhanced_create_personalized_plan |
|
|
from tools.travel_data_analyzer import analyze_travel_data_with_llm |
|
|
|
|
|
|
|
|
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace') |
|
|
sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8', errors='replace') |
|
|
|
|
|
mcp = FastMCP("persona-trip-mcp-server") |
|
|
|
|
|
@mcp.tool() |
|
|
async def web_search(destination: str, time_period: str, user_keywords: list[str] = None) -> str: |
|
|
""" |
|
|
Gathers real-time travel intelligence for a specific destination and time period. |
|
|
Use this tool BEFORE planning to check for advisories, strikes, events, and user-specific keywords. |
|
|
""" |
|
|
return await get_travel_intelligence_briefing(destination, time_period, user_keywords or []) |
|
|
|
|
|
@mcp.tool() |
|
|
async def travel_data_analyzer(user_question: str) -> str: |
|
|
""" |
|
|
This tool is designed to understand and respond to queries that require aggregation, comparison, or statistical insights from all users' past travel data. |
|
|
""" |
|
|
return await analyze_travel_data_with_llm(user_question) |
|
|
|
|
|
@mcp.tool() |
|
|
async def enhanced_personalized_travel_planner(user_name: str, new_destination: str, trip_duration_days: int, user_request: str, web_intelligence: str = None ) -> str: |
|
|
"""The main tool to generate a complete, personalized travel itinerary based on a user's travel history and latest web search result.""" |
|
|
return await enhanced_create_personalized_plan(user_name, new_destination, trip_duration_days, user_request, web_intelligence) |
|
|
|
|
|
@mcp.tool() |
|
|
async def summarize_trip(person_name: str, city: str) -> str: |
|
|
"""Summarizes the travel experience of a specific user in a specific city from the database.""" |
|
|
return await asyncio.to_thread(db_summarize_trip, person_name, city) |
|
|
|
|
|
@mcp.tool() |
|
|
async def fetch_trips_by_location(city: str) -> str: |
|
|
"""Retrieves all travel records related to a given city.""" |
|
|
return await asyncio.to_thread(fetch_trips, city) |
|
|
|
|
|
if __name__ == "__main__": |
|
|
mcp.run(transport='stdio') |
