feat: add flixbus mcp tools rename map_tools files

src/app/main.py

@@ -4,7 +4,7 @@ import gradio as gr
 from src.common.logger import get_logger
 from src.common.config import settings
 from src.tools import register_tools
-from src.tools.map_tool import render_map_with_places
+from src.tools.map_tools import render_map_with_places
 
 logger = get_logger(__name__)
 

src/common/flixbus_client.py

@@ -210,5 +210,16 @@ class FlixBusClient:
         response.raise_for_status()
         return SearchResponse(**response.json())
 
+    def get_shop_url_for_search_trips(
+        self,
+        from_city_id: str,
+        to_city_id: str,
+        departure_date: str,  # Format: DD.MM.YYYY
+        adult: int = 1,
+        currency: str = "EUR",
+        locale: str = "en",
+    ) -> str:
+        return f"https://shop.global.flixbus.com/search?departureCity={from_city_id}&arrivalCity={to_city_id}&rideDate={departure_date}&adult={adult}&_locale={locale}&currency={currency}"
+
     async def close(self):
         await self.client.aclose()

src/tools/__init__.py

@@ -1,5 +1,14 @@
 def register_tools(gr):
-    from src.tools.map_tool import gm_walking_dir_tool, render_map_with_places_tool
+    from src.tools.map_tools import gm_walking_dir_tool, render_map_with_places_tool
+    from src.tools.flixbus_tools import (
+        get_city_info_tool,
+        search_cities_in_radius_tool,
+        get_tripe_options_list_tool,
+    )
 
     gr.api(gm_walking_dir_tool)
     gr.api(render_map_with_places_tool)
+
+    gr.api(get_city_info_tool)
+    gr.api(search_cities_in_radius_tool)
+    gr.api(get_tripe_options_list_tool)

src/tools/flixbus_tools.py

@@ -0,0 +1,229 @@
+import gradio as gr
+import json
+from typing import Optional
+from src.common.logger import get_logger
+from src.common.flixbus_client import FlixBusClient
+from src.common.utils import format_docstring
+from src.tools.schemas.flixbus_tools_schema import (
+    GetCityInfoResilt,
+    GetTripeOptionsListResult,
+    SearchCitiesInRadiusResult,
+    TripeOption,
+)
+
+
+logger = get_logger(__name__)
+
+##### get_city_info #####
+
+
+async def get_city_info(city_name: str) -> Optional[GetCityInfoResilt]:
+    """
+    Resolves a city name to its Flixbus UUID.
+    """
+    client = FlixBusClient()
+    try:
+        suggestions = await client.get_city_suggestions(q=city_name)
+        if not suggestions:
+            return None
+
+        city = suggestions[0]
+        return GetCityInfoResilt(
+            id=city.id,
+            name=city.name,
+            country=city.country,
+            latitude=city.location.lat,
+            longitude=city.location.lon,
+        )
+
+    except Exception as e:
+        logger.error(f"Error resolving city ID: {e}")
+        return None
+    finally:
+        await client.close()
+
+
+@format_docstring(template_args={"return_schema": GetCityInfoResilt._model_json_schema})
+@gr.mcp.tool(
+    name="get_city_info_tool",
+)
+async def get_city_info_tool(
+    city_name: str,
+) -> str:
+    """
+    Resolves a city name to its Flixbus UUID.
+
+    Args:
+        city_name: The name of the city to resolve.
+
+    Returns:
+        {return_schema}
+    """
+    city_info = await get_city_info(city_name)
+
+    if city_info is None:
+        return json.dumps(
+            {"error": f"City '{city_name}' not found in Flixbus database."}
+        )
+
+    return city_info.model_dump_json()
+
+
+##### search_cities_in_radius #####
+
+
+async def search_cities_in_radius(
+    lat: float, lon: float, radius_km: int = 300, limit: int = 20
+) -> SearchCitiesInRadiusResult:
+    """
+    Finds Flixbus cities within a bounding box roughly corresponding to the radius.
+    Note: This is an approximation using a bounding box.
+    """
+    client = FlixBusClient()
+    try:
+        # 1 degree of latitude is approx 111km
+        lat_delta = radius_km / 111
+        # 1 degree of longitude is approx 111km * cos(lat) - simplifying to 111km for rough box
+        lon_delta = radius_km / 111
+
+        top_left_lat = lat + lat_delta
+        top_left_lon = lon - lon_delta
+        bottom_right_lat = lat - lat_delta
+        bottom_right_lon = lon + lon_delta
+
+        response = await client.get_cms_cities(
+            top_left_lat=top_left_lat,
+            top_left_lon=top_left_lon,
+            bottom_right_lat=bottom_right_lat,
+            bottom_right_lon=bottom_right_lon,
+            limit=limit,
+        )
+
+        cities: list[GetCityInfoResilt] = []
+        for city in response.result:
+            cities.append(
+                GetCityInfoResilt(
+                    name=city.name,
+                    id=city.uuid,  # Using UUID as ID
+                    latitude=city.location.lat,
+                    longitude=city.location.lon,
+                    country=city.country,
+                )
+            )
+
+        return SearchCitiesInRadiusResult(cities=cities)
+    finally:
+        await client.close()
+
+
+@format_docstring(
+    template_args={"return_schema": SearchCitiesInRadiusResult._model_json_schema}
+)
+@gr.mcp.tool(
+    name="search_cities_in_radius_tool",
+)
+async def search_cities_in_radius_tool(
+    lat: float, lon: float, radius_km: int = 300, limit: int = 20
+) -> str:
+    """
+    Finds Flixbus cities within a bounding box roughly corresponding to the radius.
+    Note: This is an approximation using a bounding box.
+
+    Args:
+        lat: Latitude of the center point.
+        lon: Longitude of the center point.
+        radius_km: Search radius in kilometers, default is 300km.
+        limit: Maximum number of cities to return, default is 20.
+
+    Returns:
+        {return_schema}
+    """
+    result = await search_cities_in_radius(lat, lon, radius_km, limit)
+    return result.model_dump_json()
+
+
+##### get_trip_options_list #####
+
+
+async def get_tripe_options_list(
+    from_city_id: str, to_city_id: str, date: str
+) -> GetTripeOptionsListResult:
+    """
+    Searches for the trips between two cities.
+    """
+    client = FlixBusClient()
+
+    shop_url = client.get_shop_url_for_search_trips(
+        from_city_id=from_city_id, to_city_id=to_city_id, departure_date=date
+    )
+
+    empty_result = GetTripeOptionsListResult(
+        options=[], best_option=None, shop_url=shop_url
+    )
+
+    try:
+        response = await client.search_trips(
+            from_city_id=from_city_id, to_city_id=to_city_id, departure_date=date
+        )
+
+        if not response.trips:
+            return empty_result
+
+        options: list[TripeOption] = []
+        best_option: Optional[TripeOption] = None
+
+        tripe_options = response.trips[0].results
+
+        for _, trip_option in tripe_options.items():
+            if trip_option.status != "available":
+                continue
+
+            option = TripeOption(
+                price=trip_option.price.total,
+                departure_station_id=trip_option.departure.station_id,
+                arrival_station_id=trip_option.arrival.station_id,
+                departure_time=trip_option.departure.date.strftime(
+                    "%Y-%m-%dT%H:%M:%S%z"
+                ),
+                arrival_time=trip_option.arrival.date.strftime("%Y-%m-%dT%H:%M:%S%z"),
+                duration_hours=trip_option.duration.hours,
+                available_seats=trip_option.available.seats,
+            )
+
+            options.append(option)
+
+            if best_option is None or option.price < best_option.price:
+                best_option = option
+
+        return GetTripeOptionsListResult(
+            options=options, best_option=best_option, shop_url=shop_url
+        )
+
+    except Exception as e:
+        logger.error(f"Error finding trip: {e}")
+        return empty_result
+    finally:
+        await client.close()
+
+
+@format_docstring(
+    template_args={"return_schema": GetTripeOptionsListResult._model_json_schema}
+)
+@gr.mcp.tool(
+    name="get_tripe_options_list_tool",
+)
+async def get_tripe_options_list_tool(
+    from_city_id: str, to_city_id: str, date: str
+) -> str:
+    """
+    Searches for the trips between two cities.
+
+    Args:
+        from_city_id: Flixbus UUID of the departure city.
+        to_city_id: Flixbus UUID of the destination city.
+        date: Departure date in format DD.MM.YYYY.
+    Returns:
+        {return_schema}
+    """
+    result = await get_tripe_options_list(from_city_id, to_city_id, date)
+    return result.model_dump_json()

src/tools/{map_tool.py → map_tools.py}

@@ -1,10 +1,9 @@
-from unittest import result
 from urllib.parse import urlencode, quote_plus
 import folium
 import gradio as gr
 from src.common.logger import get_logger
 from src.common.utils import format_docstring
-from src.tools.schemas.map_tool_schema import (
+from src.tools.schemas.map_tools_schema import (
     MapRenderingResult,
     MapWalkingDirectionResult,
 )

src/tools/schemas/flixbus_tools_schema.py

@@ -0,0 +1,123 @@
+from typing import Optional
+from pydantic import BaseModel, Field
+
+
+###################
+
+
+class GetCityInfoResilt(BaseModel):
+    """
+    Contains information about a city
+    """
+
+    id: str = Field(..., description="Flixbus UUID of the city")
+    name: str = Field(..., description="Name of the city")
+    country: str = Field(..., description="Country of the city")
+    latitude: float = Field(..., description="Latitude of the city")
+    longitude: float = Field(..., description="Longitude of the city")
+
+
+GetCityInfoResilt._model_example = GetCityInfoResilt(
+    id="string", name="string", country="string", latitude=0.0, longitude=0.0
+)
+
+GetCityInfoResilt._model_json_schema = (
+    GetCityInfoResilt._model_example.model_dump_json()
+)
+
+###################
+
+
+class SearchCitiesInRadiusResult(BaseModel):
+    """
+    Contains a list of cities within a specified radius
+    """
+
+    cities: list[GetCityInfoResilt] = Field(
+        ..., description="List of cities within the specified radius"
+    )
+
+
+SearchCitiesInRadiusResult._model_example = SearchCitiesInRadiusResult(
+    cities=[GetCityInfoResilt._model_example]
+)
+
+SearchCitiesInRadiusResult._model_json_schema = (
+    SearchCitiesInRadiusResult._model_example.model_dump_json()
+)
+
+
+###################
+
+
+class TripeOption(BaseModel):
+    """
+    Contains information about a trip option
+    """
+
+    departure_station_id: Optional[str] = Field(
+        None, description="ID of the departure station"
+    )
+    arrival_station_id: Optional[str] = Field(
+        None, description="ID of the arrival station"
+    )
+
+    departure_time: Optional[str] = Field(
+        None,
+        description="Departure time of the trip option, e.g., '2025-11-23T13:10:00+01:00'",
+    )
+    arrival_time: Optional[str] = Field(
+        None,
+        description="Arrival time of the trip option, e.g., '2025-11-23T17:30:00+01:00'",
+    )
+
+    duration_hours: Optional[float] = Field(
+        None, description="Duration of the trip option in hours"
+    )
+    available_seats: Optional[int] = Field(
+        None, description="Number of available seats for the trip option"
+    )
+
+    price: float = Field(..., description="Price of the trip option")
+
+
+TripeOption._model_example = TripeOption(
+    departure_station_id="string",
+    arrival_station_id="string",
+    departure_time="2025-11-23T13:10:00+01:00",
+    arrival_time="2025-11-23T17:30:00+01:00",
+    duration_hours=4,
+    available_seats=10,
+    price=19.99,
+)
+
+TripeOption._model_json_schema = TripeOption._model_example.model_dump_json()
+
+
+class GetTripeOptionsListResult(BaseModel):
+    """
+    Contains a list of trip prices between two cities
+    """
+
+    options: list[TripeOption] = Field(
+        ..., description="List of trip options between two cities"
+    )
+
+    best_option: Optional[TripeOption] = Field(
+        None, description="Best trip option between two cities"
+    )
+
+    shop_url: Optional[str] = Field(
+        None, description="URL to shop for the best trip option"
+    )
+
+
+GetTripeOptionsListResult._model_example = GetTripeOptionsListResult(
+    options=[TripeOption._model_example],
+    best_option=TripeOption._model_example,
+    shop_url="string",
+)
+
+GetTripeOptionsListResult._model_json_schema = (
+    GetTripeOptionsListResult._model_example.model_dump_json()
+)

src/tools/schemas/{map_tool_schema.py → map_tools_schema.py}

@@ -1,6 +1,9 @@
 from pydantic import BaseModel, Field
 
 
+###################
+
+
 class MapWalkingDirectionResult(BaseModel):
     """
     Contains url of the map with walking direction route