[MCP] Add Google Flights price intelligence MCP App dashboard

src/server.py

@@ -4,7 +4,7 @@
 import re
 import time
 from collections import Counter
-from datetime import datetime
+from datetime import UTC, datetime
 from pathlib import Path
 from typing import Any
 from urllib.parse import urlparse
@@ -24,6 +24,7 @@
     Alert,
     AlertDescription,
     AlertTitle,
+    Badge,
     Card,
     CardContent,
     CardHeader,
@@ -34,10 +35,11 @@
     If,
     Link,
     Metric,
+    Row,
     Small,
     Text,
 )
-from prefab_ui.components.charts import PieChart
+from prefab_ui.components.charts import AreaChart, ChartSeries, PieChart
 from prefab_ui.rx import STATE, Rx
 from starlette.middleware import Middleware
 from starlette.middleware.base import BaseHTTPMiddleware
@@ -551,6 +553,212 @@ def build_dashboard_app(data: dict[str, Any]) -> PrefabApp:
     return app
 
 
+# ---------------------------------------------------------------------------
+# Flights-specific App builder
+# ---------------------------------------------------------------------------
+
+
+def flights_rows(data: dict[str, Any]) -> list[dict[str, Any]]:
+    """Flatten best_flights + other_flights into table-ready rows."""
+    rows: list[dict[str, Any]] = []
+    for section in ("best_flights", "other_flights"):
+        for itinerary in data.get(section) or []:
+            segments = itinerary.get("flights") or []
+            airlines = sorted({seg.get("airline", "") for seg in segments} - {""})
+            departure = segments[0] if segments else {}
+            arrival = segments[-1] if segments else {}
+            dep_airport = departure.get("departure_airport") or {}
+            arr_airport = arrival.get("arrival_airport") or {}
+            stops = len(itinerary.get("layovers") or [])
+            carbon = itinerary.get("carbon_emissions") or {}
+            carbon_pct = carbon.get("difference_percent")
+            rows.append(
+                {
+                    "airline": ", ".join(airlines) or "—",
+                    "route": f"{dep_airport.get('id', '?')} → {arr_airport.get('id', '?')}",
+                    "departure": dep_airport.get("time", ""),
+                    "arrival": arr_airport.get("time", ""),
+                    "duration": _format_duration(itinerary.get("total_duration")),
+                    "stops": "Direct"
+                    if stops == 0
+                    else f"{stops} stop{'s' if stops > 1 else ''}",
+                    "price": itinerary.get("price") or 0,
+                    "price_fmt": f"${itinerary['price']:,}"
+                    if itinerary.get("price")
+                    else "—",
+                    "carbon_delta": carbon_pct,
+                    "carbon_fmt": f"{carbon_pct:+d}% vs typical"
+                    if isinstance(carbon_pct, int)
+                    else "—",
+                    "type": itinerary.get("type", ""),
+                }
+            )
+    return rows
+
+
+def _format_duration(minutes: int | None) -> str:
+    if not minutes:
+        return "—"
+    h, m = divmod(int(minutes), 60)
+    return f"{h}h {m}m" if h else f"{m}m"
+
+
+def price_history_points(data: dict[str, Any]) -> list[dict[str, Any]]:
+    """Convert price_insights.price_history into chart-ready [{date, price}]."""
+    insights = data.get("price_insights") or {}
+    history = insights.get("price_history") or []
+    points: list[dict[str, Any]] = []
+    for entry in history:
+        if isinstance(entry, list) and len(entry) >= 2:
+            ts, price = entry[0], entry[1]
+            points.append(
+                {
+                    "date": datetime.fromtimestamp(ts, tz=UTC).strftime("%b %d"),
+                    "price": price,
+                }
+            )
+    return points
+
+
+def flights_price_insights(data: dict[str, Any]) -> dict[str, Any]:
+    """Extract price intelligence metrics from a flights response."""
+    insights = data.get("price_insights") or {}
+    typical = insights.get("typical_price_range") or []
+    return {
+        "lowest_price": insights.get("lowest_price"),
+        "price_level": insights.get("price_level", "unknown"),
+        "typical_low": typical[0] if len(typical) >= 1 else None,
+        "typical_high": typical[1] if len(typical) >= 2 else None,
+    }
+
+
+_PRICE_LEVEL_VARIANTS = {
+    "low": "success",
+    "typical": "secondary",
+    "high": "warning",
+    "very high": "destructive",
+}
+
+_FLIGHTS_COLUMNS = [
+    DataTableColumn(key="airline", header="Airline", sortable=True),
+    DataTableColumn(key="route", header="Route", sortable=True),
+    DataTableColumn(key="departure", header="Departs", sortable=True),
+    DataTableColumn(key="arrival", header="Arrives", sortable=True),
+    DataTableColumn(key="duration", header="Duration", sortable=True),
+    DataTableColumn(key="stops", header="Stops", sortable=True),
+    DataTableColumn(key="price", header="Price", sortable=True, format="currency"),
+]
+
+
+def build_flights_app(data: dict[str, Any]) -> PrefabApp:
+    """Compose the flights price intelligence dashboard."""
+    insights = flights_price_insights(data)
+    rows = flights_rows(data)
+    history = price_history_points(data)
+
+    lowest = insights["lowest_price"]
+    level = insights["price_level"]
+    typical_low = insights["typical_low"]
+    typical_high = insights["typical_high"]
+
+    title = "Flights dashboard"
+    params = data.get("search_parameters") or {}
+    dep = params.get("departure_id", "")
+    arr = params.get("arrival_id", "")
+    if dep and arr:
+        title = f"Flights: {dep} → {arr}"
+
+    with PrefabApp(title=title, state={"selected": None}) as app:
+        with Column(gap=4, css_class="p-4"):
+            # Metrics row
+            with Grid(columns=[1, 1, 1, 1], gap=4):
+                Metric(
+                    label="Lowest price",
+                    value=f"${lowest:,}" if lowest else "—",
+                )
+                Metric(
+                    label="Typical range",
+                    value=(
+                        f"${typical_low:,}–${typical_high:,}"
+                        if typical_low and typical_high
+                        else "—"
+                    ),
+                )
+                Metric(
+                    label="Flights found",
+                    value=str(len(rows)),
+                )
+                with Column(gap=1):
+                    Text(content="Price level")
+                    Badge(
+                        label=level.capitalize(),
+                        variant=_PRICE_LEVEL_VARIANTS.get(level, "outline"),
+                    )
+
+            # Price history chart
+            if history:
+                AreaChart(
+                    data=history,
+                    series=[ChartSeries(data_key="price", label="Price ($)")],
+                    x_axis="date",
+                    height=280,
+                    curve="smooth",
+                    show_dots=False,
+                )
+
+            # Flights table
+            DataTable(
+                columns=_FLIGHTS_COLUMNS,
+                rows=rows,
+                search=True,
+                paginated=True,
+                page_size=15,
+                on_row_click=SetState("selected", Rx("$event")),
+            )
+
+            # Detail panel
+            with If(STATE.selected):
+                with Card():
+                    with CardHeader():
+                        with Row(gap=2):
+                            H3(Rx("selected.airline"))
+                            Badge(label=Rx("selected.stops"), variant="secondary")
+                            Badge(label=Rx("selected.carbon_fmt"), variant="outline")
+                    with CardContent():
+                        with Grid(columns=[1, 1, 1, 1], gap=4):
+                            with Column(gap=1):
+                                Small(content="Route")
+                                Text(content=Rx("selected.route"))
+                            with Column(gap=1):
+                                Small(content="Departure")
+                                Text(content=Rx("selected.departure"))
+                            with Column(gap=1):
+                                Small(content="Arrival")
+                                Text(content=Rx("selected.arrival"))
+                            with Column(gap=1):
+                                Small(content="Duration")
+                                Text(content=Rx("selected.duration"))
+                        with Grid(columns=[1, 1, 1, 1], gap=4, css_class="mt-2"):
+                            with Column(gap=1):
+                                Small(content="Price")
+                                Text(content=Rx("selected.price_fmt"))
+                            with Column(gap=1):
+                                Small(content="Carbon emissions")
+                                Text(content=Rx("selected.carbon_fmt"))
+                            with Column(gap=1):
+                                Small(content="Trip type")
+                                Text(content=Rx("selected.type"))
+
+    return app
+
+
+# Engine-specific app dispatch: maps engine names to their dedicated builders.
+# Falls back to the generic dashboard for unregistered engines.
+ENGINE_APP_BUILDERS: dict[str, Any] = {
+    "google_flights": build_flights_app,
+}
+
+
 @mcp.tool(
     app=True,
     description=(
@@ -583,7 +791,9 @@ async def search_table(params: dict[str, Any] = None) -> PrefabApp:
         "Interactive dashboard variant of `search`: returns summary metrics, a "
         "source breakdown chart, and a results table with a click-to-expand "
         "detail panel, all rendered in the conversation. Same params as "
-        "`search`. Use for a richer visual overview of a query's results."
+        "`search`. Use for a richer visual overview of a query's results. "
+        "Automatically selects an engine-specific dashboard when available "
+        "(e.g. google_flights gets price intelligence charting)."
     ),
     annotations=ToolAnnotations(
         title="SerpApi search (dashboard)",
@@ -600,7 +810,9 @@ async def search_dashboard(params: dict[str, Any] = None) -> PrefabApp:
         return _error_app(
             str(exc) if isinstance(exc, RuntimeError) else map_search_error(exc)
         )
-    return build_dashboard_app(data)
+    engine = (params or {}).get("engine", "google_light")
+    builder = ENGINE_APP_BUILDERS.get(engine, build_dashboard_app)
+    return builder(data)
 
 
 async def healthcheck_handler(request):