diff --git a/src/mcp_components/__init__.py b/src/mcp_components/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/mcp_components/apps.py b/src/mcp_components/apps.py new file mode 100644 index 0000000..7858e7b --- /dev/null +++ b/src/mcp_components/apps.py @@ -0,0 +1,802 @@ +from collections import Counter +from datetime import UTC, datetime +from typing import Any +from urllib.parse import urlparse + +from fastmcp.tools import tool +from mcp.types import ToolAnnotations +from prefab_ui.actions import SetState +from prefab_ui.app import PrefabApp +from prefab_ui.components import ( + H3, + Alert, + AlertDescription, + AlertTitle, + Badge, + Card, + CardContent, + CardHeader, + Column, + DataTable, + DataTableColumn, + Grid, + If, + Link, + Metric, + Row, + Small, + Text, +) +from prefab_ui.components.charts import AreaChart, BarChart, ChartSeries, PieChart +from prefab_ui.rx import STATE, Rx + +from src.mcp_components.tools import fetch_search_data, map_search_error + + +# --------------------------------------------------------------------------- +# MCP Apps (SEP-1865): interactive UI variants of `search`. +# +# These are opt-in: the plain-text `search` tool above is unchanged and stays +# the default. App-aware hosts can call `search_table` / `search_dashboard` +# to get an interactive UI rendered in the conversation; the bulk SERP JSON +# never enters the model context window. Hosts that don't support the Apps +# extension simply ignore these tools. +# --------------------------------------------------------------------------- + + +def _result_source(result: dict[str, Any]) -> str: + """Best-effort source label for an organic result (explicit source or host).""" + source = result.get("source") + if source: + return str(source) + host = urlparse(result.get("link", "") or "").netloc + return host[4:] if host.startswith("www.") else host + + +def organic_rows(data: dict[str, Any]) -> list[dict[str, Any]]: + """Flatten SerpApi organic_results into compact, table-ready rows.""" + rows: list[dict[str, Any]] = [] + for index, result in enumerate(data.get("organic_results") or [], start=1): + rows.append( + { + "position": result.get("position", index), + "title": result.get("title", ""), + "link": result.get("link", ""), + "source": _result_source(result), + "snippet": result.get("snippet", ""), + } + ) + return rows + + +def source_breakdown( + rows: list[dict[str, Any]], limit: int = 8 +) -> list[dict[str, Any]]: + """Count results per source for the dashboard pie chart (top `limit`).""" + counts = Counter(row["source"] for row in rows if row["source"]) + return [ + {"source": source, "count": count} + for source, count in counts.most_common(limit) + ] + + +def dashboard_summary(data: dict[str, Any]) -> dict[str, Any]: + """Derive the dashboard view-model from a SerpApi response.""" + params = data.get("search_parameters") or {} + info = data.get("search_information") or {} + rows = organic_rows(data) + return { + "query": params.get("q", ""), + "engine": params.get("engine", ""), + "total_results": info.get("total_results"), + "result_count": len(rows), + "rows": rows, + "sources": source_breakdown(rows), + } + + +def _error_app(message: str) -> PrefabApp: + """Render an upstream/search error as an Apps alert instead of raw text.""" + with PrefabApp(title="Search error") as app: + with Alert(variant="destructive"): + AlertTitle(content="Search failed") + AlertDescription(content=message) + return app + + +_ORGANIC_COLUMNS = [ + DataTableColumn(key="position", header="#", sortable=True, width="64px"), + DataTableColumn(key="title", header="Title", sortable=True), + DataTableColumn(key="source", header="Source", sortable=True), + DataTableColumn(key="snippet", header="Snippet"), +] + + +def build_table_app(data: dict[str, Any]) -> PrefabApp: + """Compose the results-table UI from a SerpApi response.""" + with PrefabApp(title="Search results") as app: + with Column(gap=4, css_class="p-4"): + DataTable( + columns=_ORGANIC_COLUMNS, + rows=organic_rows(data), + search=True, + paginated=True, + page_size=10, + ) + return app + + +def build_dashboard_app(data: dict[str, Any]) -> PrefabApp: + """Compose the dashboard UI (metrics + chart + table + detail) from a response.""" + summary = dashboard_summary(data) + total = summary["total_results"] + + with PrefabApp(title="Search dashboard", state={"selected": None}) as app: + with Column(gap=4, css_class="p-4"): + with Grid(columns=[1, 1, 1], gap=4): + Metric(label="Query", value=summary["query"] or "—") + Metric(label="Engine", value=summary["engine"] or "—") + Metric( + label="Results shown", + value=str(summary["result_count"]), + description=( + f"of ~{total:,} total" if isinstance(total, int) else None + ), + ) + + with Grid(columns=[1, 2], gap=4): + if summary["sources"]: + PieChart( + data=summary["sources"], + data_key="count", + name_key="source", + show_legend=True, + height=260, + ) + DataTable( + columns=_ORGANIC_COLUMNS, + rows=summary["rows"], + search=True, + on_row_click=SetState("selected", Rx("$event")), + ) + + with If(STATE.selected): + with Card(): + with CardHeader(): + H3(Rx("selected.title")) + Small(content=Rx("selected.source")) + with CardContent(): + with Column(gap=2): + Text(content=Rx("selected.snippet")) + Link( + content=Rx("selected.link"), + href=Rx("selected.link"), + target="_blank", + ) + return app + + +# --------------------------------------------------------------------------- +# Currency helpers +# --------------------------------------------------------------------------- + +_CURRENCY_SYMBOLS: dict[str, str] = { + "USD": "$", + "EUR": "€", + "GBP": "£", + "JPY": "¥", + "CNY": "¥", + "INR": "₹", + "KRW": "₩", + "BRL": "R$", + "AUD": "A$", + "CAD": "C$", +} + + +def _currency_symbol(data: dict[str, Any]) -> str: + """Extract currency symbol from a SerpApi response's search_parameters.""" + code = (data.get("search_parameters") or {}).get("currency", "USD") + return _CURRENCY_SYMBOLS.get(code, code + " ") + + +def _fmt_price(amount: int | float | None, symbol: str) -> str: + """Format a numeric price with the given currency symbol.""" + if not amount: + return "—" + return f"{symbol}{amount:,.0f}" + + +# --------------------------------------------------------------------------- +# Flights-specific App builder +# --------------------------------------------------------------------------- + + +def flights_rows(data: dict[str, Any]) -> list[dict[str, Any]]: + """Flatten best_flights + other_flights into table-ready rows.""" + symbol = _currency_symbol(data) + 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": _fmt_price(itinerary.get("price"), symbol), + "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) + symbol = _currency_symbol(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=_fmt_price(lowest, symbol), + ) + Metric( + label="Typical range", + value=( + f"{_fmt_price(typical_low, symbol)}–{_fmt_price(typical_high, symbol)}" + 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 + + +# --------------------------------------------------------------------------- +# Jobs-specific App builder +# --------------------------------------------------------------------------- + +# Benefits detected from extensions that get badge treatment. +_JOB_BENEFIT_LABELS = { + "Health insurance", + "Dental insurance", + "Paid time off", + "401(k)", + "Vision insurance", + "Life insurance", + "Disability insurance", + "Commuter benefits", + "Tuition reimbursement", +} + + +def jobs_rows(data: dict[str, Any]) -> list[dict[str, Any]]: + """Flatten jobs_results into table-ready rows with structured metadata.""" + rows: list[dict[str, Any]] = [] + for job in data.get("jobs_results") or []: + ext = job.get("detected_extensions") or {} + extensions = job.get("extensions") or [] + benefits = [e for e in extensions if e in _JOB_BENEFIT_LABELS] + rows.append( + { + "title": job.get("title", ""), + "company": job.get("company_name", ""), + "location": job.get("location", ""), + "salary": ext.get("salary", ""), + "schedule": ext.get("schedule_type", ""), + "posted": ext.get("posted_at", ""), + "qualifications": ext.get("qualifications", ""), + "work_from_home": ext.get("work_from_home", False), + "benefits": benefits, + "benefits_fmt": ", ".join(benefits) if benefits else "—", + "via": job.get("via", ""), + "description": (job.get("description") or "")[:300], + "highlights": job.get("job_highlights") or [], + "apply_options": job.get("apply_options") or [], + "source_link": job.get("source_link", ""), + } + ) + return rows + + +def jobs_summary(data: dict[str, Any]) -> dict[str, Any]: + """Derive summary metrics from a jobs response.""" + rows = jobs_rows(data) + total = len(rows) + with_salary = sum(1 for r in rows if r["salary"]) + remote = sum(1 for r in rows if r["work_from_home"]) + return { + "total": total, + "with_salary": with_salary, + "remote": remote, + "salary_pct": f"{with_salary * 100 // total}%" if total else "—", + "remote_pct": f"{remote * 100 // total}%" if total else "—", + "rows": rows, + "schedule_breakdown": jobs_schedule_breakdown(rows), + } + + +def jobs_schedule_breakdown(rows: list[dict[str, Any]]) -> list[dict[str, Any]]: + """Count jobs per schedule type for the pie chart.""" + counts = Counter(r["schedule"] or "Unspecified" for r in rows) + return [ + {"schedule": schedule, "count": count} + for schedule, count in counts.most_common() + ] + + +_JOBS_COLUMNS = [ + DataTableColumn(key="title", header="Title", sortable=True), + DataTableColumn(key="company", header="Company", sortable=True), + DataTableColumn(key="location", header="Location", sortable=True), + DataTableColumn(key="salary", header="Salary", sortable=True), + DataTableColumn(key="schedule", header="Type", sortable=True), + DataTableColumn(key="posted", header="Posted", sortable=True), + DataTableColumn(key="benefits_fmt", header="Benefits"), +] + + +def build_jobs_app(data: dict[str, Any]) -> PrefabApp: + """Compose the jobs explorer dashboard.""" + summary = jobs_summary(data) + rows = summary["rows"] + params = data.get("search_parameters") or {} + query = params.get("q", "") + + title = f"Jobs: {query}" if query else "Jobs dashboard" + + 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="Jobs found", value=str(summary["total"])) + Metric( + label="With salary", + value=str(summary["with_salary"]), + description=summary["salary_pct"], + ) + Metric( + label="Remote", + value=str(summary["remote"]), + description=summary["remote_pct"], + ) + Metric( + label="Query", + value=query or "—", + ) + + # Schedule type breakdown + if summary["schedule_breakdown"]: + PieChart( + data=summary["schedule_breakdown"], + data_key="count", + name_key="schedule", + show_legend=True, + height=220, + ) + + # Jobs table + DataTable( + columns=_JOBS_COLUMNS, + rows=rows, + search=True, + paginated=True, + page_size=10, + on_row_click=SetState("selected", Rx("$event")), + ) + + # Detail panel + with If(STATE.selected): + with Card(): + with CardHeader(): + H3(Rx("selected.title")) + with Row(gap=2): + Small(content=Rx("selected.company")) + Text(content="·") + Small(content=Rx("selected.location")) + with Row(gap=2, css_class="mt-2"): + with If(Rx("selected.salary")): + Badge(label=Rx("selected.salary"), variant="default") + with If(Rx("selected.schedule")): + Badge( + label=Rx("selected.schedule"), + variant="secondary", + ) + with If(Rx("selected.work_from_home")): + Badge(label="Remote", variant="success") + with CardContent(): + with Column(gap=3): + Text(content=Rx("selected.description")) + with If(Rx("selected.source_link")): + Link( + content="View full listing →", + href=Rx("selected.source_link"), + target="_blank", + ) + + return app + + +# --------------------------------------------------------------------------- +# Shopping-specific App builder +# --------------------------------------------------------------------------- + + +def shopping_rows(data: dict[str, Any]) -> list[dict[str, Any]]: + """Flatten shopping_results into table-ready rows.""" + rows: list[dict[str, Any]] = [] + for item in data.get("shopping_results") or []: + price = item.get("extracted_price") + extensions = item.get("extensions") or [] + discount_tag = next((e for e in extensions if "OFF" in e), "") + rows.append( + { + "title": item.get("title", ""), + "source": item.get("source", ""), + "price": price or 0, + "price_fmt": item.get("price", "—"), + "old_price_fmt": item.get("old_price", ""), + "discount": discount_tag, + "rating": item.get("rating") or 0, + "reviews": item.get("reviews") or 0, + "snippet": item.get("snippet", ""), + "product_link": item.get("product_link", ""), + } + ) + return rows + + +def _extract_currency_prefix(data: dict[str, Any]) -> str: + """Extract the currency symbol from the first shopping result's price string.""" + for item in data.get("shopping_results") or []: + price_str = item.get("price", "") + if price_str: + prefix = "" + for ch in price_str: + if ch.isdigit() or ch in ".,": + break + prefix += ch + if prefix: + return prefix + return "$" + + +def shopping_summary(data: dict[str, Any]) -> dict[str, Any]: + """Derive summary metrics and price-by-source chart data.""" + rows = shopping_rows(data) + prices = [r["price"] for r in rows if r["price"] > 0] + on_sale = sum(1 for r in rows if r["old_price_fmt"]) + avg_rating = ( + sum(r["rating"] for r in rows if r["rating"]) + / max(1, sum(1 for r in rows if r["rating"])) + if rows + else 0 + ) + + # Price by source (top 10 cheapest for the bar chart) + priced = sorted([r for r in rows if r["price"] > 0], key=lambda r: r["price"]) + price_chart = [{"source": r["source"], "price": r["price"]} for r in priced[:10]] + + symbol = _extract_currency_prefix(data) + + return { + "total": len(rows), + "price_min": min(prices) if prices else 0, + "price_max": max(prices) if prices else 0, + "on_sale": on_sale, + "avg_rating": round(avg_rating, 1), + "rows": rows, + "price_chart": price_chart, + "currency_symbol": symbol, + } + + +_SHOPPING_COLUMNS = [ + DataTableColumn(key="title", header="Product", sortable=True), + DataTableColumn(key="source", header="Seller", sortable=True), + DataTableColumn(key="price", header="Price", sortable=True, format="currency"), + DataTableColumn(key="old_price_fmt", header="Was"), + DataTableColumn(key="discount", header="Discount"), + DataTableColumn(key="rating", header="Rating", sortable=True), + DataTableColumn(key="reviews", header="Reviews", sortable=True), +] + + +def build_shopping_app(data: dict[str, Any]) -> PrefabApp: + """Compose the shopping price comparison dashboard.""" + summary = shopping_summary(data) + rows = summary["rows"] + params = data.get("search_parameters") or {} + query = params.get("q", "") + sym = summary["currency_symbol"] + + title = f"Shopping: {query}" if query else "Shopping dashboard" + + 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="Products", value=str(summary["total"])) + Metric( + label="Price range", + value=( + f"{sym}{summary['price_min']:,.0f}–{sym}{summary['price_max']:,.0f}" + if summary["price_min"] + else "—" + ), + ) + Metric(label="On sale", value=str(summary["on_sale"])) + Metric( + label="Avg rating", + value=str(summary["avg_rating"]) if summary["avg_rating"] else "—", + ) + + # Price comparison bar chart (top 10 cheapest sellers) + if summary["price_chart"]: + BarChart( + data=summary["price_chart"], + series=[ChartSeries(data_key="price", label="Price ($)")], + x_axis="source", + height=240, + horizontal=True, + ) + + # Products table + DataTable( + columns=_SHOPPING_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(): + H3(Rx("selected.title")) + with Row(gap=2): + Small(content=Rx("selected.source")) + with If(Rx("selected.discount")): + Badge( + label=Rx("selected.discount"), + variant="destructive", + ) + with CardContent(): + with Column(gap=2): + with Row(gap=4): + Text(content=Rx("selected.price_fmt")) + with If(Rx("selected.old_price_fmt")): + Small(content=Rx("selected.old_price_fmt")) + with If(Rx("selected.snippet")): + Text(content=Rx("selected.snippet")) + with If(Rx("selected.product_link")): + Link( + content="View on Google Shopping →", + href=Rx("selected.product_link"), + target="_blank", + ) + + 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, + "google_jobs": build_jobs_app, + "google_shopping": build_shopping_app, +} + + +@tool( + meta={"ui": True}, + description=( + "Interactive UI variant of `search`: returns organic results as a " + "sortable, searchable table rendered in the conversation. Same params " + "as `search`. Use when the host supports MCP Apps and the user wants " + "to browse results visually rather than read JSON." + ), + annotations=ToolAnnotations( + title="SerpApi search (table)", + readOnlyHint=True, + destructiveHint=False, + idempotentHint=False, + openWorldHint=True, + ), +) +async def search_table(params: dict[str, Any] = None) -> PrefabApp: + try: + data = fetch_search_data(params) + except Exception as exc: + return _error_app( + str(exc) if isinstance(exc, RuntimeError) else map_search_error(exc) + ) + return build_table_app(data) + + +@tool( + meta={"ui": True}, + description=( + "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. " + "Automatically selects an engine-specific dashboard when available " + "(e.g. google_flights gets price intelligence charting)." + ), + annotations=ToolAnnotations( + title="SerpApi search (dashboard)", + readOnlyHint=True, + destructiveHint=False, + idempotentHint=False, + openWorldHint=True, + ), +) +async def search_dashboard(params: dict[str, Any] = None) -> PrefabApp: + try: + data = fetch_search_data(params) + except Exception as exc: + return _error_app( + str(exc) if isinstance(exc, RuntimeError) else map_search_error(exc) + ) + engine = (params or {}).get("engine", "google_light") + builder = ENGINE_APP_BUILDERS.get(engine, build_dashboard_app) + return builder(data) diff --git a/src/mcp_components/resources.py b/src/mcp_components/resources.py new file mode 100644 index 0000000..e10c209 --- /dev/null +++ b/src/mcp_components/resources.py @@ -0,0 +1,88 @@ +import json +import logging +import re +from pathlib import Path + +from fastmcp.exceptions import NotFoundError +from fastmcp.resources import ResourceContent, ResourceResult, resource +from mcp.types import Annotations + + +logger = logging.getLogger(__name__) + +ENGINES_DIR = Path(__file__).resolve().parents[2] / "engines" + + +def _get_engine_files() -> list[Path]: + if not ENGINES_DIR.exists(): + logger.warning("Engines directory not found: %s", ENGINES_DIR) + return [] + return sorted(ENGINES_DIR.glob("*.json")) + + +@resource( + "serpapi://engines", + name="serpapi-engines-index", + description="Index of available SerpApi engines and their resource URIs.", + mime_type="application/json", + annotations=Annotations( + audience=["assistant"], + priority=0.3, + ), +) +def engines_index() -> ResourceResult: + engine_files = _get_engine_files() + engines = [path.stem for path in engine_files] + resource_content = json.dumps( + { + "count": len(engines), + "engines": engines, + "resources": [f"serpapi://engines/{engine}" for engine in engines], + "schema": { + "note": "Each engine resource uses a flat schema: params are engine-specific; common_params are shared SerpApi parameters.", + "params_key": "params", + "common_params_key": "common_params", + }, + } + ) + return ResourceResult( + contents=[ + ResourceContent(content=resource_content, mime_type="application/json"), + ] + ) + + +@resource( + "serpapi://engines/{engine_name}", + name="serpapi-engine", + description=( + "SerpApi engine specification. The URI parameter {engine_name} " + "is the engine identifier (e.g. 'google', 'bing', 'walmart'). " + "Use serpapi://engines to list valid values." + ), + mime_type="application/json", + annotations=Annotations( + audience=["assistant"], + priority=0.3, + ), +) +def get_engine_schema(engine_name: str) -> ResourceResult: + if not re.fullmatch(r"[a-z0-9_]+", engine_name): + raise NotFoundError( + f"Invalid engine name: {engine_name!r}. Expected [a-z0-9_]+." + ) + engine_path = ENGINES_DIR / f"{engine_name}.json" + if not engine_path.exists(): + raise NotFoundError( + f"Unknown engine: {engine_name!r}. See serpapi://engines for the full list." + ) + return ResourceResult( + contents=[ + # The json dump and load chain looks redundant - but it will help remove newlines from the file at `engine_path`, + # making the response context efficient for LLMs + ResourceContent( + content=json.dumps(json.loads(engine_path.read_text())), + mime_type="application/json", + ), + ] + ) diff --git a/src/mcp_components/tools.py b/src/mcp_components/tools.py new file mode 100644 index 0000000..fae4915 --- /dev/null +++ b/src/mcp_components/tools.py @@ -0,0 +1,194 @@ +import json +from typing import Any + +import serpapi +from fastmcp.server.dependencies import get_http_request +from fastmcp.tools import tool +from mcp.types import ToolAnnotations + + +def extract_error_response(exception) -> str: + """ + Helper function to extract meaningful error information from nested exceptions. + + Traverses exception.args[0] chain until it finds a valid .response object, + then attempts to extract JSON from response.json(). Falls back to str(e). + + Args: + exception: The exception to process + + Returns: + str: Formatted error message with response data if available + """ + current = exception + max_depth = 10 + depth = 0 + + while depth < max_depth: + if hasattr(current, "response") and current.response is not None: + try: + response_data = current.response.json() + return json.dumps(response_data, indent=2) + except (ValueError, AttributeError, TypeError): + try: + return current.response.text + except (AttributeError, TypeError): + pass + + if hasattr(current, "args") and current.args and len(current.args) > 0: + current = current.args[0] + depth += 1 + else: + break + + # Fallback + return str(exception) + + +def map_search_error(exception) -> str: + """Map a SerpApi/transport exception to a user-facing 'Error: ...' string. + + Shared by the text `search` tool and the App tools so all entry points + surface identical messages for the same upstream failure. + """ + if isinstance(exception, serpapi.exceptions.HTTPError): + text = str(exception) + if "429" in text: + return "Error: Rate limit exceeded. Please try again later." + if "401" in text: + return ( + "Error: Invalid SerpApi API key. " + "Check your API key in the path or Authorization header." + ) + if "403" in text: + return ( + "Error: SerpApi API key forbidden. " + "Verify your subscription and key validity." + ) + return f"Error: {extract_error_response(exception)}" + + +search_tool_description = """Universal search tool supporting all SerpApi engines and result types. + + When to use: + - Any query needing live, structured SERP data: web results, news, product listings, job postings, local businesses, flight/hotel prices, video results, images, stock/weather cards, knowledge graph entities. + + Engine discovery via MCP resources: + - serpapi://engines lists all engines supported by this tool. + - serpapi://engines/ provides engine-specific parameters and supported options. + - Example: serpapi://engines/google_news + + Input schema: + params: JSON object containing SerpApi engine parameters. + Common parameters: + - q: Search query. Required for most engines. + - engine: SerpApi engine name. Defaults to "google_light". + - location: Optional geographic location for localized results. + - num: Optional number of results to return. + + Engine-specific parameters are available via MCP resources: + - serpapi://engines lists all supported engines. + - serpapi://engines/ provides parameters and options for one engine. + + mode: Response mode. Defaults to "complete". + - "complete": Return the full SerpApi JSON response. + - "compact": Return a reduced response with metadata removed. + + Output schema: + JSON string containing search results, structured engine output, or an error message. + + Examples: + Weather: {"params": {"q": "weather in London", "engine": "google"}, "mode": "complete"} + Stock: {"params": {"q": "AAPL stock", "engine": "google"}, "mode": "complete"} + General: {"params": {"q": "coffee shops", "engine": "google_light", "location": "Austin, TX"}, "mode": "complete"} + Compact: {"params": {"q": "news"}, "mode": "compact"} + + Supported engines include (not limited to): + - google + - google_light + - google_flights + - google_hotels + - google_images + - google_news + - google_local + - google_shopping + - google_jobs + - bing + - yahoo + - duckduckgo + - youtube_search + - baidu + - ebay + """ + + +@tool( + description=search_tool_description, + annotations=ToolAnnotations( + title="SerpApi search", + readOnlyHint=True, # search is read-only; no state mutation + destructiveHint=False, # nothing deleted or modified + idempotentHint=False, # SERP can change between calls; cache is 1h + openWorldHint=True, # talks to external search engines + ), +) +async def search(params: dict[str, Any] = None, mode: str = "complete") -> str: + """Universal search tool supporting all SerpApi engines and result types. + + Args: + params: Dictionary of SerpApi engine-specific parameters. Common parameters include: + - q: Search query (required for most engines) + - engine: Search engine to use (default: "google_light") + - location: Geographic location filter + - num: Number of results to return + + mode: Response mode (default: "complete") + - "complete": Returns full JSON response with all fields + - "compact": Returns JSON response with metadata fields removed + + Returns: + A JSON string containing search results or an error message. + """ + + # Validate mode parameter + if mode not in ["complete", "compact"]: + return "Error: Invalid mode. Must be 'complete' or 'compact'" + + try: + data = fetch_search_data(params) + + # Apply mode-specific filtering + if mode == "compact": + # Remove specified fields for compact mode + fields_to_remove = [ + "search_metadata", + "search_parameters", + "search_information", + "pagination", + "serpapi_pagination", + ] + for field in fields_to_remove: + data.pop(field, None) + + # Return JSON response for both modes + return json.dumps(data, indent=2, ensure_ascii=False) + + except RuntimeError as e: + return str(e) + except Exception as e: + return map_search_error(e) + + +def fetch_search_data(params: dict[str, Any] | None) -> dict[str, Any]: + """Run a SerpApi search using the request's API key. Raises on failure.""" + request = get_http_request() + api_key = getattr(getattr(request, "state", None), "api_key", None) + if not api_key: + raise RuntimeError("Error: Unable to access API key from request context") + + search_params = { + "api_key": api_key, + "engine": "google_light", + **(params or {}), + } + return serpapi.search(search_params).as_dict() diff --git a/src/server.py b/src/server.py index 8cbda3e..4fec55c 100644 --- a/src/server.py +++ b/src/server.py @@ -1,134 +1,29 @@ import json import logging import os -import re import time -from collections import Counter from datetime import UTC, datetime from pathlib import Path -from typing import Any -from urllib.parse import urlparse -import serpapi import uvicorn from dotenv import load_dotenv -from fastmcp import FastMCP -from fastmcp.exceptions import NotFoundError -from fastmcp.resources import ResourceContent, ResourceResult -from fastmcp.server.dependencies import get_http_request -from mcp.types import Annotations, ToolAnnotations -from prefab_ui.actions import SetState -from prefab_ui.app import PrefabApp -from prefab_ui.components import ( - H3, - Alert, - AlertDescription, - AlertTitle, - Badge, - Card, - CardContent, - CardHeader, - Column, - DataTable, - DataTableColumn, - Grid, - If, - Link, - Metric, - Row, - Small, - Text, -) -from prefab_ui.components.charts import AreaChart, BarChart, ChartSeries, PieChart -from prefab_ui.rx import STATE, Rx from starlette.middleware import Middleware from starlette.middleware.base import BaseHTTPMiddleware from starlette.middleware.cors import CORSMiddleware from starlette.requests import Request from starlette.responses import JSONResponse -load_dotenv() - -mcp = FastMCP("SerpApi MCP Server") -logger = logging.getLogger(__name__) -logger.setLevel(logging.INFO) - -ENGINES_DIR = Path(__file__).resolve().parents[1] / "engines" - - -def _get_engine_files() -> list[Path]: - if not ENGINES_DIR.exists(): - logger.warning("Engines directory not found: %s", ENGINES_DIR) - return [] - return sorted(ENGINES_DIR.glob("*.json")) +from fastmcp import FastMCP +from fastmcp.server.providers import FileSystemProvider +COMPONENTS_DIR = Path(__file__).parent / "mcp_components" -@mcp.resource( - "serpapi://engines", - name="serpapi-engines-index", - description="Index of available SerpApi engines and their resource URIs.", - mime_type="application/json", - annotations=Annotations( - audience=["assistant"], - priority=0.3, - ), -) -def engines_index() -> ResourceResult: - engine_files = _get_engine_files() - engines = [path.stem for path in engine_files] - resource_content = json.dumps( - { - "count": len(engines), - "engines": engines, - "resources": [f"serpapi://engines/{engine}" for engine in engines], - "schema": { - "note": "Each engine resource uses a flat schema: params are engine-specific; common_params are shared SerpApi parameters.", - "params_key": "params", - "common_params_key": "common_params", - }, - } - ) - return ResourceResult( - contents=[ - ResourceContent(content=resource_content, mime_type="application/json"), - ] - ) +mcp = FastMCP("SerpApi MCP Server", providers=[FileSystemProvider(COMPONENTS_DIR)]) +load_dotenv() -@mcp.resource( - "serpapi://engines/{engine_name}", - name="serpapi-engine", - description=( - "SerpApi engine specification. The URI parameter {engine_name} " - "is the engine identifier (e.g. 'google', 'bing', 'walmart'). " - "Use serpapi://engines to list valid values." - ), - mime_type="application/json", - annotations=Annotations( - audience=["assistant"], - priority=0.3, - ), -) -def get_engine_schema(engine_name: str) -> ResourceResult: - if not re.fullmatch(r"[a-z0-9_]+", engine_name): - raise NotFoundError( - f"Invalid engine name: {engine_name!r}. Expected [a-z0-9_]+." - ) - engine_path = ENGINES_DIR / f"{engine_name}.json" - if not engine_path.exists(): - raise NotFoundError( - f"Unknown engine: {engine_name!r}. See serpapi://engines for the full list." - ) - return ResourceResult( - contents=[ - # The json dump and load chain looks redundant - but it will help remove newlines from the file at `engine_path`, - # making the response context efficient for LLMs - ResourceContent( - content=json.dumps(json.loads(engine_path.read_text())), - mime_type="application/json", - ), - ] - ) +logger = logging.getLogger(__name__) +logger.setLevel(logging.INFO) def emit_metric(namespace: str, metrics: dict, dimensions: dict = {}): @@ -153,67 +48,6 @@ def emit_metric(namespace: str, metrics: dict, dimensions: dict = {}): logger.info(json.dumps(emf_event)) -def extract_error_response(exception) -> str: - """ - Helper function to extract meaningful error information from nested exceptions. - - Traverses exception.args[0] chain until it finds a valid .response object, - then attempts to extract JSON from response.json(). Falls back to str(e). - - Args: - exception: The exception to process - - Returns: - str: Formatted error message with response data if available - """ - current = exception - max_depth = 10 - depth = 0 - - while depth < max_depth: - if hasattr(current, "response") and current.response is not None: - try: - response_data = current.response.json() - return json.dumps(response_data, indent=2) - except (ValueError, AttributeError, TypeError): - try: - return current.response.text - except (AttributeError, TypeError): - pass - - if hasattr(current, "args") and current.args and len(current.args) > 0: - current = current.args[0] - depth += 1 - else: - break - - # Fallback - return str(exception) - - -def map_search_error(exception) -> str: - """Map a SerpApi/transport exception to a user-facing 'Error: ...' string. - - Shared by the text `search` tool and the App tools so all entry points - surface identical messages for the same upstream failure. - """ - if isinstance(exception, serpapi.exceptions.HTTPError): - text = str(exception) - if "429" in text: - return "Error: Rate limit exceeded. Please try again later." - if "401" in text: - return ( - "Error: Invalid SerpApi API key. " - "Check your API key in the path or Authorization header." - ) - if "403" in text: - return ( - "Error: SerpApi API key forbidden. " - "Verify your subscription and key validity." - ) - return f"Error: {extract_error_response(exception)}" - - class ApiKeyMiddleware(BaseHTTPMiddleware): async def dispatch(self, request: Request, call_next): # Skip authentication for healthcheck endpoint @@ -272,915 +106,11 @@ async def dispatch(self, request: Request, call_next): return response -search_tool_description = """Universal search tool supporting all SerpApi engines and result types. - - When to use: - - Any query needing live, structured SERP data: web results, news, product listings, job postings, local businesses, flight/hotel prices, video results, images, stock/weather cards, knowledge graph entities. - - Engine discovery via MCP resources: - - serpapi://engines lists all engines supported by this tool. - - serpapi://engines/ provides engine-specific parameters and supported options. - - Example: serpapi://engines/google_news - - Input schema: - params: JSON object containing SerpApi engine parameters. - Common parameters: - - q: Search query. Required for most engines. - - engine: SerpApi engine name. Defaults to "google_light". - - location: Optional geographic location for localized results. - - num: Optional number of results to return. - - Engine-specific parameters are available via MCP resources: - - serpapi://engines lists all supported engines. - - serpapi://engines/ provides parameters and options for one engine. - - mode: Response mode. Defaults to "complete". - - "complete": Return the full SerpApi JSON response. - - "compact": Return a reduced response with metadata removed. - - Output schema: - JSON string containing search results, structured engine output, or an error message. - - Examples: - Weather: {"params": {"q": "weather in London", "engine": "google"}, "mode": "complete"} - Stock: {"params": {"q": "AAPL stock", "engine": "google"}, "mode": "complete"} - General: {"params": {"q": "coffee shops", "engine": "google_light", "location": "Austin, TX"}, "mode": "complete"} - Compact: {"params": {"q": "news"}, "mode": "compact"} - - Supported engines include (not limited to): - - google - - google_light - - google_flights - - google_hotels - - google_images - - google_news - - google_local - - google_shopping - - google_jobs - - bing - - yahoo - - duckduckgo - - youtube_search - - baidu - - ebay - """ - - -@mcp.tool( - description=search_tool_description, - annotations=ToolAnnotations( - title="SerpApi search", - readOnlyHint=True, # search is read-only; no state mutation - destructiveHint=False, # nothing deleted or modified - idempotentHint=False, # SERP can change between calls; cache is 1h - openWorldHint=True, # talks to external search engines - ), -) -async def search(params: dict[str, Any] = None, mode: str = "complete") -> str: - """Universal search tool supporting all SerpApi engines and result types. - - Args: - params: Dictionary of SerpApi engine-specific parameters. Common parameters include: - - q: Search query (required for most engines) - - engine: Search engine to use (default: "google_light") - - location: Geographic location filter - - num: Number of results to return - - mode: Response mode (default: "complete") - - "complete": Returns full JSON response with all fields - - "compact": Returns JSON response with metadata fields removed - - Returns: - A JSON string containing search results or an error message. - """ - - # Validate mode parameter - if mode not in ["complete", "compact"]: - return "Error: Invalid mode. Must be 'complete' or 'compact'" - - if params is None: - params = {} - - request = get_http_request() - api_key = getattr(getattr(request, "state", None), "api_key", None) - if not api_key: - return "Error: Unable to access API key from request context" - - search_params = { - "api_key": api_key, - "engine": "google_light", # Fastest engine by default - **params, # Include any additional parameters - } - - try: - data = serpapi.search(search_params).as_dict() - - # Apply mode-specific filtering - if mode == "compact": - # Remove specified fields for compact mode - fields_to_remove = [ - "search_metadata", - "search_parameters", - "search_information", - "pagination", - "serpapi_pagination", - ] - for field in fields_to_remove: - data.pop(field, None) - - # Return JSON response for both modes - return json.dumps(data, indent=2, ensure_ascii=False) - - except Exception as e: - return map_search_error(e) - - -# --------------------------------------------------------------------------- -# MCP Apps (SEP-1865): interactive UI variants of `search`. -# -# These are opt-in: the plain-text `search` tool above is unchanged and stays -# the default. App-aware hosts can call `search_table` / `search_dashboard` -# to get an interactive UI rendered in the conversation; the bulk SERP JSON -# never enters the model context window. Hosts that don't support the Apps -# extension simply ignore these tools. -# --------------------------------------------------------------------------- - - -def _result_source(result: dict[str, Any]) -> str: - """Best-effort source label for an organic result (explicit source or host).""" - source = result.get("source") - if source: - return str(source) - host = urlparse(result.get("link", "") or "").netloc - return host[4:] if host.startswith("www.") else host - - -def organic_rows(data: dict[str, Any]) -> list[dict[str, Any]]: - """Flatten SerpApi organic_results into compact, table-ready rows.""" - rows: list[dict[str, Any]] = [] - for index, result in enumerate(data.get("organic_results") or [], start=1): - rows.append( - { - "position": result.get("position", index), - "title": result.get("title", ""), - "link": result.get("link", ""), - "source": _result_source(result), - "snippet": result.get("snippet", ""), - } - ) - return rows - - -def source_breakdown( - rows: list[dict[str, Any]], limit: int = 8 -) -> list[dict[str, Any]]: - """Count results per source for the dashboard pie chart (top `limit`).""" - counts = Counter(row["source"] for row in rows if row["source"]) - return [ - {"source": source, "count": count} - for source, count in counts.most_common(limit) - ] - - -def dashboard_summary(data: dict[str, Any]) -> dict[str, Any]: - """Derive the dashboard view-model from a SerpApi response.""" - params = data.get("search_parameters") or {} - info = data.get("search_information") or {} - rows = organic_rows(data) - return { - "query": params.get("q", ""), - "engine": params.get("engine", ""), - "total_results": info.get("total_results"), - "result_count": len(rows), - "rows": rows, - "sources": source_breakdown(rows), - } - - -def fetch_search_data(params: dict[str, Any] | None) -> dict[str, Any]: - """Run a SerpApi search using the request's API key. Raises on failure.""" - request = get_http_request() - api_key = getattr(getattr(request, "state", None), "api_key", None) - if not api_key: - raise RuntimeError("Error: Unable to access API key from request context") - - search_params = { - "api_key": api_key, - "engine": "google_light", - **(params or {}), - } - return serpapi.search(search_params).as_dict() - - -def _error_app(message: str) -> PrefabApp: - """Render an upstream/search error as an Apps alert instead of raw text.""" - with PrefabApp(title="Search error") as app: - with Alert(variant="destructive"): - AlertTitle(content="Search failed") - AlertDescription(content=message) - return app - - -_ORGANIC_COLUMNS = [ - DataTableColumn(key="position", header="#", sortable=True, width="64px"), - DataTableColumn(key="title", header="Title", sortable=True), - DataTableColumn(key="source", header="Source", sortable=True), - DataTableColumn(key="snippet", header="Snippet"), -] - - -def build_table_app(data: dict[str, Any]) -> PrefabApp: - """Compose the results-table UI from a SerpApi response.""" - with PrefabApp(title="Search results") as app: - with Column(gap=4, css_class="p-4"): - DataTable( - columns=_ORGANIC_COLUMNS, - rows=organic_rows(data), - search=True, - paginated=True, - page_size=10, - ) - return app - - -def build_dashboard_app(data: dict[str, Any]) -> PrefabApp: - """Compose the dashboard UI (metrics + chart + table + detail) from a response.""" - summary = dashboard_summary(data) - total = summary["total_results"] - - with PrefabApp(title="Search dashboard", state={"selected": None}) as app: - with Column(gap=4, css_class="p-4"): - with Grid(columns=[1, 1, 1], gap=4): - Metric(label="Query", value=summary["query"] or "—") - Metric(label="Engine", value=summary["engine"] or "—") - Metric( - label="Results shown", - value=str(summary["result_count"]), - description=( - f"of ~{total:,} total" if isinstance(total, int) else None - ), - ) - - with Grid(columns=[1, 2], gap=4): - if summary["sources"]: - PieChart( - data=summary["sources"], - data_key="count", - name_key="source", - show_legend=True, - height=260, - ) - DataTable( - columns=_ORGANIC_COLUMNS, - rows=summary["rows"], - search=True, - on_row_click=SetState("selected", Rx("$event")), - ) - - with If(STATE.selected): - with Card(): - with CardHeader(): - H3(Rx("selected.title")) - Small(content=Rx("selected.source")) - with CardContent(): - with Column(gap=2): - Text(content=Rx("selected.snippet")) - Link( - content=Rx("selected.link"), - href=Rx("selected.link"), - target="_blank", - ) - return app - - -# --------------------------------------------------------------------------- -# Currency helpers -# --------------------------------------------------------------------------- - -_CURRENCY_SYMBOLS: dict[str, str] = { - "USD": "$", - "EUR": "€", - "GBP": "£", - "JPY": "¥", - "CNY": "¥", - "INR": "₹", - "KRW": "₩", - "BRL": "R$", - "AUD": "A$", - "CAD": "C$", -} - - -def _currency_symbol(data: dict[str, Any]) -> str: - """Extract currency symbol from a SerpApi response's search_parameters.""" - code = (data.get("search_parameters") or {}).get("currency", "USD") - return _CURRENCY_SYMBOLS.get(code, code + " ") - - -def _fmt_price(amount: int | float | None, symbol: str) -> str: - """Format a numeric price with the given currency symbol.""" - if not amount: - return "—" - return f"{symbol}{amount:,.0f}" - - -# --------------------------------------------------------------------------- -# Flights-specific App builder -# --------------------------------------------------------------------------- - - -def flights_rows(data: dict[str, Any]) -> list[dict[str, Any]]: - """Flatten best_flights + other_flights into table-ready rows.""" - symbol = _currency_symbol(data) - 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": _fmt_price(itinerary.get("price"), symbol), - "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) - symbol = _currency_symbol(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=_fmt_price(lowest, symbol), - ) - Metric( - label="Typical range", - value=( - f"{_fmt_price(typical_low, symbol)}–{_fmt_price(typical_high, symbol)}" - 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 - - -# --------------------------------------------------------------------------- -# Jobs-specific App builder -# --------------------------------------------------------------------------- - -# Benefits detected from extensions that get badge treatment. -_JOB_BENEFIT_LABELS = { - "Health insurance", - "Dental insurance", - "Paid time off", - "401(k)", - "Vision insurance", - "Life insurance", - "Disability insurance", - "Commuter benefits", - "Tuition reimbursement", -} - - -def jobs_rows(data: dict[str, Any]) -> list[dict[str, Any]]: - """Flatten jobs_results into table-ready rows with structured metadata.""" - rows: list[dict[str, Any]] = [] - for job in data.get("jobs_results") or []: - ext = job.get("detected_extensions") or {} - extensions = job.get("extensions") or [] - benefits = [e for e in extensions if e in _JOB_BENEFIT_LABELS] - rows.append( - { - "title": job.get("title", ""), - "company": job.get("company_name", ""), - "location": job.get("location", ""), - "salary": ext.get("salary", ""), - "schedule": ext.get("schedule_type", ""), - "posted": ext.get("posted_at", ""), - "qualifications": ext.get("qualifications", ""), - "work_from_home": ext.get("work_from_home", False), - "benefits": benefits, - "benefits_fmt": ", ".join(benefits) if benefits else "—", - "via": job.get("via", ""), - "description": (job.get("description") or "")[:300], - "highlights": job.get("job_highlights") or [], - "apply_options": job.get("apply_options") or [], - "source_link": job.get("source_link", ""), - } - ) - return rows - - -def jobs_summary(data: dict[str, Any]) -> dict[str, Any]: - """Derive summary metrics from a jobs response.""" - rows = jobs_rows(data) - total = len(rows) - with_salary = sum(1 for r in rows if r["salary"]) - remote = sum(1 for r in rows if r["work_from_home"]) - return { - "total": total, - "with_salary": with_salary, - "remote": remote, - "salary_pct": f"{with_salary * 100 // total}%" if total else "—", - "remote_pct": f"{remote * 100 // total}%" if total else "—", - "rows": rows, - "schedule_breakdown": jobs_schedule_breakdown(rows), - } - - -def jobs_schedule_breakdown(rows: list[dict[str, Any]]) -> list[dict[str, Any]]: - """Count jobs per schedule type for the pie chart.""" - counts = Counter(r["schedule"] or "Unspecified" for r in rows) - return [ - {"schedule": schedule, "count": count} - for schedule, count in counts.most_common() - ] - - -_JOBS_COLUMNS = [ - DataTableColumn(key="title", header="Title", sortable=True), - DataTableColumn(key="company", header="Company", sortable=True), - DataTableColumn(key="location", header="Location", sortable=True), - DataTableColumn(key="salary", header="Salary", sortable=True), - DataTableColumn(key="schedule", header="Type", sortable=True), - DataTableColumn(key="posted", header="Posted", sortable=True), - DataTableColumn(key="benefits_fmt", header="Benefits"), -] - - -def build_jobs_app(data: dict[str, Any]) -> PrefabApp: - """Compose the jobs explorer dashboard.""" - summary = jobs_summary(data) - rows = summary["rows"] - params = data.get("search_parameters") or {} - query = params.get("q", "") - - title = f"Jobs: {query}" if query else "Jobs dashboard" - - 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="Jobs found", value=str(summary["total"])) - Metric( - label="With salary", - value=str(summary["with_salary"]), - description=summary["salary_pct"], - ) - Metric( - label="Remote", - value=str(summary["remote"]), - description=summary["remote_pct"], - ) - Metric( - label="Query", - value=query or "—", - ) - - # Schedule type breakdown - if summary["schedule_breakdown"]: - PieChart( - data=summary["schedule_breakdown"], - data_key="count", - name_key="schedule", - show_legend=True, - height=220, - ) - - # Jobs table - DataTable( - columns=_JOBS_COLUMNS, - rows=rows, - search=True, - paginated=True, - page_size=10, - on_row_click=SetState("selected", Rx("$event")), - ) - - # Detail panel - with If(STATE.selected): - with Card(): - with CardHeader(): - H3(Rx("selected.title")) - with Row(gap=2): - Small(content=Rx("selected.company")) - Text(content="·") - Small(content=Rx("selected.location")) - with Row(gap=2, css_class="mt-2"): - with If(Rx("selected.salary")): - Badge(label=Rx("selected.salary"), variant="default") - with If(Rx("selected.schedule")): - Badge( - label=Rx("selected.schedule"), - variant="secondary", - ) - with If(Rx("selected.work_from_home")): - Badge(label="Remote", variant="success") - with CardContent(): - with Column(gap=3): - Text(content=Rx("selected.description")) - with If(Rx("selected.source_link")): - Link( - content="View full listing →", - href=Rx("selected.source_link"), - target="_blank", - ) - - return app - - -# --------------------------------------------------------------------------- -# Shopping-specific App builder -# --------------------------------------------------------------------------- - - -def shopping_rows(data: dict[str, Any]) -> list[dict[str, Any]]: - """Flatten shopping_results into table-ready rows.""" - rows: list[dict[str, Any]] = [] - for item in data.get("shopping_results") or []: - price = item.get("extracted_price") - old_price = item.get("extracted_old_price") - extensions = item.get("extensions") or [] - discount_tag = next((e for e in extensions if "OFF" in e), "") - rows.append( - { - "title": item.get("title", ""), - "source": item.get("source", ""), - "price": price or 0, - "price_fmt": item.get("price", "—"), - "old_price_fmt": item.get("old_price", ""), - "discount": discount_tag, - "rating": item.get("rating") or 0, - "reviews": item.get("reviews") or 0, - "snippet": item.get("snippet", ""), - "product_link": item.get("product_link", ""), - } - ) - return rows - - -def _extract_currency_prefix(data: dict[str, Any]) -> str: - """Extract the currency symbol from the first shopping result's price string.""" - for item in data.get("shopping_results") or []: - price_str = item.get("price", "") - if price_str: - prefix = "" - for ch in price_str: - if ch.isdigit() or ch in ".,": - break - prefix += ch - if prefix: - return prefix - return "$" - - -def shopping_summary(data: dict[str, Any]) -> dict[str, Any]: - """Derive summary metrics and price-by-source chart data.""" - rows = shopping_rows(data) - prices = [r["price"] for r in rows if r["price"] > 0] - on_sale = sum(1 for r in rows if r["old_price_fmt"]) - avg_rating = ( - sum(r["rating"] for r in rows if r["rating"]) - / max(1, sum(1 for r in rows if r["rating"])) - if rows - else 0 - ) - - # Price by source (top 10 cheapest for the bar chart) - priced = sorted([r for r in rows if r["price"] > 0], key=lambda r: r["price"]) - price_chart = [{"source": r["source"], "price": r["price"]} for r in priced[:10]] - - symbol = _extract_currency_prefix(data) - - return { - "total": len(rows), - "price_min": min(prices) if prices else 0, - "price_max": max(prices) if prices else 0, - "on_sale": on_sale, - "avg_rating": round(avg_rating, 1), - "rows": rows, - "price_chart": price_chart, - "currency_symbol": symbol, - } - - -_SHOPPING_COLUMNS = [ - DataTableColumn(key="title", header="Product", sortable=True), - DataTableColumn(key="source", header="Seller", sortable=True), - DataTableColumn(key="price", header="Price", sortable=True, format="currency"), - DataTableColumn(key="old_price_fmt", header="Was"), - DataTableColumn(key="discount", header="Discount"), - DataTableColumn(key="rating", header="Rating", sortable=True), - DataTableColumn(key="reviews", header="Reviews", sortable=True), -] - - -def build_shopping_app(data: dict[str, Any]) -> PrefabApp: - """Compose the shopping price comparison dashboard.""" - summary = shopping_summary(data) - rows = summary["rows"] - params = data.get("search_parameters") or {} - query = params.get("q", "") - sym = summary["currency_symbol"] - - title = f"Shopping: {query}" if query else "Shopping dashboard" - - 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="Products", value=str(summary["total"])) - Metric( - label="Price range", - value=f"{sym}{summary['price_min']:,.0f}–{sym}{summary['price_max']:,.0f}" - if summary["price_min"] - else "—", - ) - Metric(label="On sale", value=str(summary["on_sale"])) - Metric( - label="Avg rating", - value=str(summary["avg_rating"]) if summary["avg_rating"] else "—", - ) - - # Price comparison bar chart (top 10 cheapest sellers) - if summary["price_chart"]: - BarChart( - data=summary["price_chart"], - series=[ChartSeries(data_key="price", label="Price ($)")], - x_axis="source", - height=240, - horizontal=True, - ) - - # Products table - DataTable( - columns=_SHOPPING_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(): - H3(Rx("selected.title")) - with Row(gap=2): - Small(content=Rx("selected.source")) - with If(Rx("selected.discount")): - Badge( - label=Rx("selected.discount"), - variant="destructive", - ) - with CardContent(): - with Column(gap=2): - with Row(gap=4): - Text(content=Rx("selected.price_fmt")) - with If(Rx("selected.old_price_fmt")): - Small(content=Rx("selected.old_price_fmt")) - with If(Rx("selected.snippet")): - Text(content=Rx("selected.snippet")) - with If(Rx("selected.product_link")): - Link( - content="View on Google Shopping →", - href=Rx("selected.product_link"), - target="_blank", - ) - - 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, - "google_jobs": build_jobs_app, - "google_shopping": build_shopping_app, -} - - -@mcp.tool( - app=True, - description=( - "Interactive UI variant of `search`: returns organic results as a " - "sortable, searchable table rendered in the conversation. Same params " - "as `search`. Use when the host supports MCP Apps and the user wants " - "to browse results visually rather than read JSON." - ), - annotations=ToolAnnotations( - title="SerpApi search (table)", - readOnlyHint=True, - destructiveHint=False, - idempotentHint=False, - openWorldHint=True, - ), -) -async def search_table(params: dict[str, Any] = None) -> PrefabApp: - try: - data = fetch_search_data(params) - except Exception as exc: - return _error_app( - str(exc) if isinstance(exc, RuntimeError) else map_search_error(exc) - ) - return build_table_app(data) - - -@mcp.tool( - app=True, - description=( - "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. " - "Automatically selects an engine-specific dashboard when available " - "(e.g. google_flights gets price intelligence charting)." - ), - annotations=ToolAnnotations( - title="SerpApi search (dashboard)", - readOnlyHint=True, - destructiveHint=False, - idempotentHint=False, - openWorldHint=True, - ), -) -async def search_dashboard(params: dict[str, Any] = None) -> PrefabApp: - try: - data = fetch_search_data(params) - except Exception as exc: - return _error_app( - str(exc) if isinstance(exc, RuntimeError) else map_search_error(exc) - ) - engine = (params or {}).get("engine", "google_light") - builder = ENGINE_APP_BUILDERS.get(engine, build_dashboard_app) - return builder(data) - - async def healthcheck_handler(request): return JSONResponse( { "status": "healthy", - "timestamp": datetime.utcnow().isoformat() + "Z", + "timestamp": datetime.now(UTC).isoformat().replace("+00:00", "Z"), "service": "SerpApi MCP Server", } ) diff --git a/tests/test_server.py b/tests/test_server.py index f899905..7a8fbfb 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -14,6 +14,9 @@ from serpapi.models import SerpResults from starlette.requests import Request +import src.mcp_components.apps as mcp_apps +import src.mcp_components.resources as mcp_resources +import src.mcp_components.tools as mcp_tools import src.server as server @@ -58,11 +61,46 @@ def serp_results(payload): def use_request(monkeypatch, request): - monkeypatch.setattr(server, "get_http_request", lambda: request) + monkeypatch.setattr(mcp_tools, "get_http_request", lambda: request) def use_search(monkeypatch, fn): - monkeypatch.setattr(server.serpapi, "search", fn) + monkeypatch.setattr(mcp_tools.serpapi, "search", fn) + + +async def test_filesystem_provider_registers_tools_apps_and_resources(): + tools = {tool.name: tool for tool in await server.mcp.list_tools()} + resources = {str(resource.uri) for resource in await server.mcp.list_resources()} + templates = { + template.uri_template for template in await server.mcp.list_resource_templates() + } + + assert {"search", "search_table", "search_dashboard"} <= set(tools) + assert tools["search"].meta is None + assert ( + tools["search_table"].meta["ui"]["resourceUri"].startswith("ui://prefab/tool/") + ) + assert ( + tools["search_dashboard"] + .meta["ui"]["resourceUri"] + .startswith("ui://prefab/tool/") + ) + assert "serpapi://engines" in resources + assert "serpapi://engines/{engine_name}" in templates + + +def test_engines_dir_resolves_to_repo_engines_directory(): + assert mcp_resources.ENGINES_DIR.exists() + assert (mcp_resources.ENGINES_DIR / "google_light.json").exists() + + +async def test_engines_index_resource_reads_engine_files(): + result = await server.mcp.read_resource("serpapi://engines") + body = json.loads(result.contents[0].content) + + assert body["count"] == len(list(mcp_resources.ENGINES_DIR.glob("*.json"))) + assert "google_light" in body["engines"] + assert "serpapi://engines/google_light" in body["resources"] def raiser(exc): @@ -106,7 +144,7 @@ def test_extract_error_response_reads_json_body_from_wrapped_request_error(): assert ( err.response is None ) # the wrapper has no response; the body is one level down - assert json.loads(server.extract_error_response(err)) == { + assert json.loads(mcp_tools.extract_error_response(err)) == { "error": "Invalid API key." } @@ -120,11 +158,13 @@ def test_extract_error_response_falls_back_to_response_text_when_not_json(): resp.raise_for_status() except requests.exceptions.HTTPError as exc: err = serpapi.exceptions.HTTPError(exc) - assert server.extract_error_response(err) == "upstream boom" + assert mcp_tools.extract_error_response(err) == "upstream boom" def test_extract_error_response_falls_back_to_str(): - assert server.extract_error_response(ValueError("plain message")) == "plain message" + assert ( + mcp_tools.extract_error_response(ValueError("plain message")) == "plain message" + ) def test_extract_error_response_terminates_and_returns_innermost_message(): @@ -133,27 +173,27 @@ def test_extract_error_response_terminates_and_returns_innermost_message(): err = ValueError(err) # 20 levels deep with no .response anywhere: the walk must terminate (not # hang) and fall back to the chain's message string. - assert server.extract_error_response(err) == "deepest" + assert mcp_tools.extract_error_response(err) == "deepest" def test_extract_error_response_finds_response_at_depth_cap_boundary(): leaf = _WithResponse(_Resp({"error": "deep"})) # index 9 is the last position the depth cap (10) still inspects. err = nest(9, leaf) - assert json.loads(server.extract_error_response(err)) == {"error": "deep"} + assert json.loads(mcp_tools.extract_error_response(err)) == {"error": "deep"} def test_extract_error_response_stops_one_past_the_depth_cap(): leaf = _WithResponse(_Resp({"error": "too deep"})) # index 10 is one past the cap: the body must never be reached. err = nest(10, leaf) - out = server.extract_error_response(err) + out = mcp_tools.extract_error_response(err) assert "too deep" not in out # cap enforced, not just "returns a string" assert out == "boom" # falls back to str() of the chain async def test_search_rejects_invalid_mode(): - out = await server.search(params={"q": "x"}, mode="bogus") + out = await mcp_tools.search(params={"q": "x"}, mode="bogus") assert out == "Error: Invalid mode. Must be 'complete' or 'compact'" @@ -161,7 +201,7 @@ async def test_search_without_api_key_returns_graceful_error(monkeypatch): # A real starlette Request with empty state: request.state.api_key would raise # AttributeError, so the guard must use getattr, not attribute access. use_request(monkeypatch, real_request(state={})) - out = await server.search(params={"q": "x"}) + out = await mcp_tools.search(params={"q": "x"}) assert out == "Error: Unable to access API key from request context" @@ -169,7 +209,7 @@ async def test_search_complete_returns_full_payload(monkeypatch): payload = {"search_metadata": {"id": "1"}, "organic_results": [{"title": "hit"}]} use_request(monkeypatch, real_request(state={"api_key": "KEY"})) use_search(monkeypatch, lambda params: serp_results(payload)) - assert json.loads(await server.search(params={"q": "x"})) == payload + assert json.loads(await mcp_tools.search(params={"q": "x"})) == payload async def test_search_compact_strips_serpapi_metadata(monkeypatch): @@ -183,7 +223,7 @@ async def test_search_compact_strips_serpapi_metadata(monkeypatch): } use_request(monkeypatch, real_request(state={"api_key": "KEY"})) use_search(monkeypatch, lambda params: serp_results(payload)) - out = json.loads(await server.search(params={"q": "x"}, mode="compact")) + out = json.loads(await mcp_tools.search(params={"q": "x"}, mode="compact")) assert out == {"organic_results": [{"title": "hit"}]} @@ -192,7 +232,7 @@ async def test_search_compact_does_not_mutate_the_live_result(monkeypatch): results = serp_results(payload) use_request(monkeypatch, real_request(state={"api_key": "KEY"})) use_search(monkeypatch, lambda params: results) - await server.search(params={"q": "x"}, mode="compact") + await mcp_tools.search(params={"q": "x"}, mode="compact") assert "search_metadata" in results.as_dict() @@ -205,7 +245,7 @@ def capture(params): use_request(monkeypatch, real_request(state={"api_key": "KEY"})) use_search(monkeypatch, capture) - await server.search(params={"q": "x"}) + await mcp_tools.search(params={"q": "x"}) assert captured["api_key"] == "KEY" assert captured["engine"] == "google_light" assert captured["q"] == "x" @@ -220,7 +260,7 @@ def capture(params): use_request(monkeypatch, real_request(state={"api_key": "KEY"})) use_search(monkeypatch, capture) - await server.search(params={"q": "x", "engine": "google_news"}) + await mcp_tools.search(params={"q": "x", "engine": "google_news"}) assert captured["engine"] == "google_news" @@ -235,7 +275,7 @@ def capture(params): async def test_search_maps_real_http_errors(monkeypatch, status, fragment): use_request(monkeypatch, real_request(state={"api_key": "KEY"})) use_search(monkeypatch, raiser(make_serpapi_http_error(status, {"error": "x"}))) - out = await server.search(params={"q": "x"}) + out = await mcp_tools.search(params={"q": "x"}) assert out.startswith("Error:") assert fragment in out @@ -245,7 +285,7 @@ async def test_search_unmapped_http_error_returns_json_body(monkeypatch): use_search( monkeypatch, raiser(make_serpapi_http_error(500, {"error": "server boom"})) ) - out = await server.search(params={"q": "x"}) + out = await mcp_tools.search(params={"q": "x"}) assert out.startswith("Error:") assert "server boom" in out @@ -253,7 +293,7 @@ async def test_search_unmapped_http_error_returns_json_body(monkeypatch): async def test_search_generic_exception_uses_extractor(monkeypatch): use_request(monkeypatch, real_request(state={"api_key": "KEY"})) use_search(monkeypatch, raiser(ValueError("weird failure"))) - assert await server.search(params={"q": "x"}) == "Error: weird failure" + assert await mcp_tools.search(params={"q": "x"}) == "Error: weird failure" async def passthrough(request): @@ -316,20 +356,20 @@ async def test_healthcheck_returns_healthy_with_utc_timestamp(): ], ) def test_map_search_error_maps_known_statuses(status, fragment): - out = server.map_search_error(make_serpapi_http_error(status, {"error": "x"})) + out = mcp_tools.map_search_error(make_serpapi_http_error(status, {"error": "x"})) assert out.startswith("Error:") assert fragment in out def test_map_search_error_falls_back_to_json_body(): - out = server.map_search_error( + out = mcp_tools.map_search_error( make_serpapi_http_error(500, {"error": "server boom"}) ) assert "server boom" in out def test_map_search_error_handles_generic_exception(): - assert server.map_search_error(ValueError("weird")) == "Error: weird" + assert mcp_tools.map_search_error(ValueError("weird")) == "Error: weird" # --- MCP Apps: pure view-model helpers ------------------------------------- @@ -348,7 +388,7 @@ def test_organic_rows_flattens_results(): {"position": 2, "title": "B", "link": "https://b.com/y", "snippet": "s2"}, ] } - rows = server.organic_rows(data) + rows = mcp_apps.organic_rows(data) assert rows[0] == { "position": 1, "title": "A", @@ -362,17 +402,17 @@ def test_organic_rows_flattens_results(): def test_organic_rows_strips_www_from_derived_source(): data = {"organic_results": [{"title": "x", "link": "https://www.example.com/p"}]} - assert server.organic_rows(data)[0]["source"] == "example.com" + assert mcp_apps.organic_rows(data)[0]["source"] == "example.com" def test_organic_rows_empty_without_results(): - assert server.organic_rows({}) == [] - assert server.organic_rows({"organic_results": None}) == [] + assert mcp_apps.organic_rows({}) == [] + assert mcp_apps.organic_rows({"organic_results": None}) == [] def test_source_breakdown_counts_and_limits(): rows = [{"source": "a"}, {"source": "a"}, {"source": "b"}, {"source": ""}] - breakdown = server.source_breakdown(rows, limit=1) + breakdown = mcp_apps.source_breakdown(rows, limit=1) assert breakdown == [{"source": "a", "count": 2}] @@ -382,7 +422,7 @@ def test_dashboard_summary_shape(): "search_information": {"total_results": 999}, "organic_results": [{"title": "A", "link": "https://a.com", "source": "A"}], } - summary = server.dashboard_summary(data) + summary = mcp_apps.dashboard_summary(data) assert summary["query"] == "coffee" assert summary["engine"] == "google_light" assert summary["total_results"] == 999 @@ -415,7 +455,7 @@ def ui_json(app): async def test_search_table_returns_results_app(monkeypatch): use_request(monkeypatch, real_request(state={"api_key": "KEY"})) use_search(monkeypatch, lambda params: serp_results(_SAMPLE_PAYLOAD)) - app = await server.search_table(params={"q": "coffee"}) + app = await mcp_apps.search_table(params={"q": "coffee"}) assert app.title == "Search results" body = ui_json(app) assert "DataTable" in body @@ -425,7 +465,7 @@ async def test_search_table_returns_results_app(monkeypatch): async def test_search_dashboard_returns_dashboard_app(monkeypatch): use_request(monkeypatch, real_request(state={"api_key": "KEY"})) use_search(monkeypatch, lambda params: serp_results(_SAMPLE_PAYLOAD)) - app = await server.search_dashboard(params={"q": "coffee"}) + app = await mcp_apps.search_dashboard(params={"q": "coffee"}) assert app.title == "Search dashboard" # click-to-expand detail panel starts collapsed. assert app.state == {"selected": None} @@ -436,7 +476,7 @@ async def test_search_dashboard_returns_dashboard_app(monkeypatch): async def test_search_table_without_api_key_renders_error_app(monkeypatch): use_request(monkeypatch, real_request(state={})) - app = await server.search_table(params={"q": "x"}) + app = await mcp_apps.search_table(params={"q": "x"}) assert app.title == "Search error" assert "Unable to access API key" in ui_json(app) @@ -444,7 +484,7 @@ async def test_search_table_without_api_key_renders_error_app(monkeypatch): async def test_search_dashboard_maps_http_error_to_error_app(monkeypatch): use_request(monkeypatch, real_request(state={"api_key": "KEY"})) use_search(monkeypatch, raiser(make_serpapi_http_error(429, {"error": "x"}))) - app = await server.search_dashboard(params={"q": "x"}) + app = await mcp_apps.search_dashboard(params={"q": "x"}) assert app.title == "Search error" assert "Rate limit exceeded" in ui_json(app) @@ -571,7 +611,7 @@ async def test_search_dashboard_maps_http_error_to_error_app(monkeypatch): def test_flights_rows_extracts_all_flights(): - rows = server.flights_rows(_SAMPLE_FLIGHTS_PAYLOAD) + rows = mcp_apps.flights_rows(_SAMPLE_FLIGHTS_PAYLOAD) assert len(rows) == 3 # Direct flight assert rows[0]["airline"] == "United" @@ -598,8 +638,8 @@ def test_flights_rows_extracts_all_flights(): def test_flights_rows_handles_empty_data(): - assert server.flights_rows({}) == [] - assert server.flights_rows({"best_flights": None, "other_flights": None}) == [] + assert mcp_apps.flights_rows({}) == [] + assert mcp_apps.flights_rows({"best_flights": None, "other_flights": None}) == [] def test_flights_rows_handles_missing_airports(): @@ -608,7 +648,7 @@ def test_flights_rows_handles_missing_airports(): {"flights": [], "layovers": [], "total_duration": 0, "price": 100} ] } - rows = server.flights_rows(data) + rows = mcp_apps.flights_rows(data) assert len(rows) == 1 assert rows[0]["route"] == "? → ?" assert rows[0]["price"] == 100 @@ -618,21 +658,21 @@ def test_flights_rows_handles_missing_airports(): def test_flights_rows_zero_price_defaults(): data = {"best_flights": [{"flights": [], "layovers": [], "total_duration": 120}]} - rows = server.flights_rows(data) + rows = mcp_apps.flights_rows(data) assert rows[0]["price"] == 0 assert rows[0]["price_fmt"] == "—" def test_format_duration(): - assert server._format_duration(330) == "5h 30m" - assert server._format_duration(45) == "45m" - assert server._format_duration(60) == "1h 0m" - assert server._format_duration(None) == "—" - assert server._format_duration(0) == "—" + assert mcp_apps._format_duration(330) == "5h 30m" + assert mcp_apps._format_duration(45) == "45m" + assert mcp_apps._format_duration(60) == "1h 0m" + assert mcp_apps._format_duration(None) == "—" + assert mcp_apps._format_duration(0) == "—" def test_price_history_points_converts_timestamps(): - points = server.price_history_points(_SAMPLE_FLIGHTS_PAYLOAD) + points = mcp_apps.price_history_points(_SAMPLE_FLIGHTS_PAYLOAD) assert len(points) == 5 assert points[0]["price"] == 310 assert "date" in points[0] @@ -641,22 +681,24 @@ def test_price_history_points_converts_timestamps(): def test_price_history_points_handles_empty(): - assert server.price_history_points({}) == [] - assert server.price_history_points({"price_insights": {}}) == [] - assert server.price_history_points({"price_insights": {"price_history": []}}) == [] + assert mcp_apps.price_history_points({}) == [] + assert mcp_apps.price_history_points({"price_insights": {}}) == [] + assert ( + mcp_apps.price_history_points({"price_insights": {"price_history": []}}) == [] + ) def test_price_history_points_skips_malformed_entries(): data = { "price_insights": {"price_history": [[1719792000], "bad", [1719878400, 300]]} } - points = server.price_history_points(data) + points = mcp_apps.price_history_points(data) assert len(points) == 1 assert points[0]["price"] == 300 def test_flights_price_insights_extracts_metrics(): - insights = server.flights_price_insights(_SAMPLE_FLIGHTS_PAYLOAD) + insights = mcp_apps.flights_price_insights(_SAMPLE_FLIGHTS_PAYLOAD) assert insights["lowest_price"] == 199 assert insights["price_level"] == "low" assert insights["typical_low"] == 250 @@ -664,7 +706,7 @@ def test_flights_price_insights_extracts_metrics(): def test_flights_price_insights_handles_missing(): - insights = server.flights_price_insights({}) + insights = mcp_apps.flights_price_insights({}) assert insights["lowest_price"] is None assert insights["price_level"] == "unknown" assert insights["typical_low"] is None @@ -672,7 +714,7 @@ def test_flights_price_insights_handles_missing(): def test_build_flights_app_produces_valid_app(): - app = server.build_flights_app(_SAMPLE_FLIGHTS_PAYLOAD) + app = mcp_apps.build_flights_app(_SAMPLE_FLIGHTS_PAYLOAD) assert "SFO → JFK" in app.title assert app.state == {"selected": None} body = ui_json(app) @@ -710,7 +752,7 @@ def test_build_flights_app_without_price_history(): ], "price_insights": {}, } - app = server.build_flights_app(data) + app = mcp_apps.build_flights_app(data) body = ui_json(app) # Should still render table without crashing, just no chart assert "DataTable" in body @@ -719,7 +761,7 @@ def test_build_flights_app_without_price_history(): def test_build_flights_app_generic_title_without_route(): data = {"search_parameters": {"engine": "google_flights"}, "best_flights": []} - app = server.build_flights_app(data) + app = mcp_apps.build_flights_app(data) assert app.title == "Flights dashboard" @@ -752,9 +794,9 @@ def test_flights_currency_inr(): "price_level": "typical", }, } - rows = server.flights_rows(data) + rows = mcp_apps.flights_rows(data) assert rows[0]["price_fmt"] == "₹35,906" - app = server.build_flights_app(data) + app = mcp_apps.build_flights_app(data) body = ui_json(app) assert "₹35,758" in body assert "₹20,500" in body @@ -767,7 +809,7 @@ def test_flights_currency_inr(): async def test_search_dashboard_dispatches_to_flights(monkeypatch): use_request(monkeypatch, real_request(state={"api_key": "KEY"})) use_search(monkeypatch, lambda params: serp_results(_SAMPLE_FLIGHTS_PAYLOAD)) - app = await server.search_dashboard( + app = await mcp_apps.search_dashboard( params={"engine": "google_flights", "departure_id": "SFO", "arrival_id": "JFK"} ) assert "SFO → JFK" in app.title @@ -778,7 +820,7 @@ async def test_search_dashboard_dispatches_to_flights(monkeypatch): async def test_search_dashboard_falls_back_to_generic(monkeypatch): use_request(monkeypatch, real_request(state={"api_key": "KEY"})) use_search(monkeypatch, lambda params: serp_results(_SAMPLE_PAYLOAD)) - app = await server.search_dashboard(params={"q": "coffee"}) + app = await mcp_apps.search_dashboard(params={"q": "coffee"}) assert app.title == "Search dashboard" @@ -859,7 +901,7 @@ async def test_search_dashboard_falls_back_to_generic(monkeypatch): def test_jobs_rows_extracts_all_jobs(): - rows = server.jobs_rows(_SAMPLE_JOBS_PAYLOAD) + rows = mcp_apps.jobs_rows(_SAMPLE_JOBS_PAYLOAD) assert len(rows) == 3 # Rich job with salary and benefits @@ -892,8 +934,8 @@ def test_jobs_rows_extracts_all_jobs(): def test_jobs_rows_handles_empty_data(): - assert server.jobs_rows({}) == [] - assert server.jobs_rows({"jobs_results": None}) == [] + assert mcp_apps.jobs_rows({}) == [] + assert mcp_apps.jobs_rows({"jobs_results": None}) == [] def test_jobs_rows_handles_missing_extensions(): @@ -906,7 +948,7 @@ def test_jobs_rows_handles_missing_extensions(): } ] } - rows = server.jobs_rows(data) + rows = mcp_apps.jobs_rows(data) assert len(rows) == 1 assert rows[0]["salary"] == "" assert rows[0]["schedule"] == "" @@ -916,7 +958,7 @@ def test_jobs_rows_handles_missing_extensions(): def test_jobs_summary_computes_metrics(): - summary = server.jobs_summary(_SAMPLE_JOBS_PAYLOAD) + summary = mcp_apps.jobs_summary(_SAMPLE_JOBS_PAYLOAD) assert summary["total"] == 3 assert summary["with_salary"] == 1 assert summary["remote"] == 1 @@ -931,7 +973,7 @@ def test_jobs_summary_computes_metrics(): def test_jobs_summary_handles_empty(): - summary = server.jobs_summary({}) + summary = mcp_apps.jobs_summary({}) assert summary["total"] == 0 assert summary["salary_pct"] == "—" assert summary["remote_pct"] == "—" @@ -940,13 +982,13 @@ def test_jobs_summary_handles_empty(): def test_jobs_schedule_breakdown_groups_unspecified(): rows = [{"schedule": ""}, {"schedule": ""}, {"schedule": "Full-time"}] - breakdown = server.jobs_schedule_breakdown(rows) + breakdown = mcp_apps.jobs_schedule_breakdown(rows) assert {"schedule": "Unspecified", "count": 2} in breakdown assert {"schedule": "Full-time", "count": 1} in breakdown def test_build_jobs_app_produces_valid_app(): - app = server.build_jobs_app(_SAMPLE_JOBS_PAYLOAD) + app = mcp_apps.build_jobs_app(_SAMPLE_JOBS_PAYLOAD) assert app.title == "Jobs: software engineer" assert app.state == {"selected": None} body = ui_json(app) @@ -963,7 +1005,7 @@ def test_build_jobs_app_produces_valid_app(): def test_build_jobs_app_without_query(): data = {"search_parameters": {"engine": "google_jobs"}, "jobs_results": []} - app = server.build_jobs_app(data) + app = mcp_apps.build_jobs_app(data) assert app.title == "Jobs dashboard" @@ -980,15 +1022,14 @@ def test_build_jobs_app_description_truncated(): } ], } - app = server.build_jobs_app(data) - rows = server.jobs_rows(data) + rows = mcp_apps.jobs_rows(data) assert len(rows[0]["description"]) == 300 async def test_search_dashboard_dispatches_to_jobs(monkeypatch): use_request(monkeypatch, real_request(state={"api_key": "KEY"})) use_search(monkeypatch, lambda params: serp_results(_SAMPLE_JOBS_PAYLOAD)) - app = await server.search_dashboard( + app = await mcp_apps.search_dashboard( params={"engine": "google_jobs", "q": "software engineer"} ) assert "software engineer" in app.title @@ -1058,7 +1099,7 @@ async def test_search_dashboard_dispatches_to_jobs(monkeypatch): def test_shopping_rows_extracts_all_products(): - rows = server.shopping_rows(_SAMPLE_SHOPPING_PAYLOAD) + rows = mcp_apps.shopping_rows(_SAMPLE_SHOPPING_PAYLOAD) assert len(rows) == 4 # Product with discount @@ -1083,13 +1124,13 @@ def test_shopping_rows_extracts_all_products(): def test_shopping_rows_handles_empty(): - assert server.shopping_rows({}) == [] - assert server.shopping_rows({"shopping_results": None}) == [] + assert mcp_apps.shopping_rows({}) == [] + assert mcp_apps.shopping_rows({"shopping_results": None}) == [] def test_shopping_rows_handles_missing_fields(): data = {"shopping_results": [{"title": "Widget", "source": "Store"}]} - rows = server.shopping_rows(data) + rows = mcp_apps.shopping_rows(data) assert rows[0]["price"] == 0 assert rows[0]["price_fmt"] == "—" assert rows[0]["rating"] == 0 @@ -1097,7 +1138,7 @@ def test_shopping_rows_handles_missing_fields(): def test_shopping_summary_computes_metrics(): - summary = server.shopping_summary(_SAMPLE_SHOPPING_PAYLOAD) + summary = mcp_apps.shopping_summary(_SAMPLE_SHOPPING_PAYLOAD) assert summary["total"] == 4 assert summary["price_min"] == 189.0 assert summary["price_max"] == 298.0 @@ -1110,7 +1151,7 @@ def test_shopping_summary_computes_metrics(): def test_shopping_summary_handles_empty(): - summary = server.shopping_summary({}) + summary = mcp_apps.shopping_summary({}) assert summary["total"] == 0 assert summary["price_min"] == 0 assert summary["price_max"] == 0 @@ -1118,7 +1159,7 @@ def test_shopping_summary_handles_empty(): def test_build_shopping_app_produces_valid_app(): - app = server.build_shopping_app(_SAMPLE_SHOPPING_PAYLOAD) + app = mcp_apps.build_shopping_app(_SAMPLE_SHOPPING_PAYLOAD) assert app.title == "Shopping: Sony WH-1000XM5" assert app.state == {"selected": None} body = ui_json(app) @@ -1133,7 +1174,7 @@ def test_build_shopping_app_produces_valid_app(): def test_build_shopping_app_without_query(): data = {"search_parameters": {"engine": "google_shopping"}, "shopping_results": []} - app = server.build_shopping_app(data) + app = mcp_apps.build_shopping_app(data) assert app.title == "Shopping dashboard" @@ -1142,7 +1183,7 @@ def test_build_shopping_app_no_chart_without_prices(): "search_parameters": {"q": "test"}, "shopping_results": [{"title": "Free thing", "source": "Store"}], } - app = server.build_shopping_app(data) + app = mcp_apps.build_shopping_app(data) body = ui_json(app) assert "BarChart" not in body assert "DataTable" in body @@ -1171,10 +1212,10 @@ def test_shopping_currency_inr(): }, ], } - summary = server.shopping_summary(data) + summary = mcp_apps.shopping_summary(data) assert summary["currency_symbol"] == "₹" - app = server.build_shopping_app(data) + app = mcp_apps.build_shopping_app(data) body = ui_json(app) assert "₹19,990" in body assert "₹24,990" in body @@ -1185,28 +1226,28 @@ def test_shopping_currency_inr(): def test_extract_currency_prefix_various(): assert ( - server._extract_currency_prefix({"shopping_results": [{"price": "$99.00"}]}) + mcp_apps._extract_currency_prefix({"shopping_results": [{"price": "$99.00"}]}) == "$" ) assert ( - server._extract_currency_prefix({"shopping_results": [{"price": "₹6,999"}]}) + mcp_apps._extract_currency_prefix({"shopping_results": [{"price": "₹6,999"}]}) == "₹" ) assert ( - server._extract_currency_prefix({"shopping_results": [{"price": "€49.99"}]}) + mcp_apps._extract_currency_prefix({"shopping_results": [{"price": "€49.99"}]}) == "€" ) assert ( - server._extract_currency_prefix({"shopping_results": [{"price": "R$150"}]}) + mcp_apps._extract_currency_prefix({"shopping_results": [{"price": "R$150"}]}) == "R$" ) - assert server._extract_currency_prefix({}) == "$" + assert mcp_apps._extract_currency_prefix({}) == "$" async def test_search_dashboard_dispatches_to_shopping(monkeypatch): use_request(monkeypatch, real_request(state={"api_key": "KEY"})) use_search(monkeypatch, lambda params: serp_results(_SAMPLE_SHOPPING_PAYLOAD)) - app = await server.search_dashboard( + app = await mcp_apps.search_dashboard( params={"engine": "google_shopping", "q": "Sony WH-1000XM5"} ) assert "Sony WH-1000XM5" in app.title