"""MCP server (FastMCP) over the read-only catalogue. Mounted by app/main.py at /mcp/, in the same process as the REST API. Each tool calls the same function the matching /api/elec endpoint uses, so the two can never disagree. Only read-only catalogue data is exposed: no admin, no login, no collection runs. Images are returned as URLs (the retailer's own image address, as stored in elec.product_image); nothing is downloaded or re-hosted. """ from __future__ import annotations from decimal import Decimal from typing import Any, Dict, List, Optional import anyio from fastapi import HTTPException from fastmcp import FastMCP from fastmcp.exceptions import ToolError from app.api.routers import elec mcp = FastMCP( name="Electronics Catalog", instructions=( "Verified catalogue of mobiles and laptops sold in India (Tamil Nadu focus). " "Every product is confirmed by real listings on at least two retail platforms; " "prices, ratings and reviews come with the page they were read from. Prices are " "rupee strings. Use search_products to find products, then get_product for " "per-platform offers, specs, images, rating and reviews, and recommend_products " "for similar alternatives." ), ) _SEARCH_FIELDS = ( "product_id", "brand", "category", "display_name", "ram_gb", "storage_gb", "best_price", "best_price_site", "platform_count", "sold_by_tn_retailer", "image_url", "rating", "rating_count", ) async def _run(fn, *args): """The catalogue functions use blocking DB calls; keep them off the event loop.""" try: return await anyio.to_thread.run_sync(lambda: fn(*args)) except HTTPException as exc: raise ToolError(str(exc.detail)) from exc @mcp.tool async def list_categories() -> List[Dict[str, Any]]: """List product categories (e.g. mobiles, laptops) with how many verified products each has.""" return await _run(elec.categories) @mcp.tool async def search_products( query: Optional[str] = None, category: Optional[str] = None, brand: Optional[str] = None, max_price: Optional[float] = None, min_price: Optional[float] = None, limit: int = 20, ) -> Dict[str, Any]: """Search verified products. Args: query: Text to match in the product or brand name, e.g. "galaxy s25", "vivobook". category: Category slug: "mobiles" or "laptops". brand: Brand slug, e.g. "samsung", "xiaomi", "hp", "lenovo". max_price: Highest best price in rupees. min_price: Lowest best price in rupees. limit: Maximum products to return (1-100). Returns the total match count and, per product: id, name, variant, best price (rupee string) and the platform offering it, number of platforms, an image URL (or null), and the overall rating and rating count (null when no platform publishes a rating). """ limit = max(1, min(int(limit), 100)) result = await _run( elec.products, category, brand, (query or None), None if min_price is None else Decimal(str(min_price)), None if max_price is None else Decimal(str(max_price)), False, None, False, limit, 0, ) return { "total": result["total"], "products": [{k: p.get(k) for k in _SEARCH_FIELDS} for p in result["products"]], } @mcp.tool async def get_product(product_id: int) -> Dict[str, Any]: """Full details of one product by its product_id (from search_products). Returns per-platform offers (price, MRP, source URL, when seen), normalised specs, image URLs, the overall rating with per-platform sources and, when a platform publishes one, a 5-to-1 star breakdown (rating is null if no platform publishes a rating), and up to 10 real customer reviews with the site each came from (often empty). """ d = await _run(elec.product, int(product_id)) return { "product_id": d["product_id"], "brand": d.get("brand"), "category": d.get("category"), "display_name": d.get("display_name"), "best_price": d.get("best_price"), "best_price_site": d.get("best_price_site"), "specs": d.get("canonical_specs") or {}, "offers": [ {k: o.get(k) for k in ("site", "price", "mrp", "source_url", "observed_at", "price_outlier")} for o in d.get("offers", []) ], "image_urls": [i["url"] for i in d.get("images", [])], "rating": d.get("rating"), "reviews": d.get("reviews", []), } @mcp.tool async def recommend_products(product_id: int, kind: str = "similar", limit: int = 6) -> Dict[str, Any]: """Alternatives to suggest for one product (by its product_id). Args: product_id: The product to find alternatives for. kind: "similar" - closest specs, ranked by spec similarity, rating (weighted by how many people rated it) and price closeness; when few exist, the best-rated in the category fill the list. "better_rated" - products rated higher than this one by at least 5 people. limit: Maximum products to return (1-12). Always same category, in stock, within a similar price (+/-30% for similar, +/-20% for better_rated), with other variants of the same model left out. Each item has a short reason (e.g. "Similar specs · 4.5★ vs 4.1★"). The same model's other RAM/storage variants are listed separately under other_variants. """ if kind not in ("similar", "better_rated"): raise ToolError('kind must be "similar" or "better_rated"') d = await _run(elec.recommendations, int(product_id), kind, max(1, min(int(limit), 12)), False) return { "items": [ {k: i.get(k) for k in ("product_id", "brand", "display_name", "ram_gb", "storage_gb", "best_price", "best_price_site", "rating", "rating_count", "reason")} for i in d["items"] ], "other_variants": d["other_variants"], } @mcp.tool async def price_history(product_id: int) -> List[Dict[str, Any]]: """Every price observed for a product, per platform, oldest first (rupee strings, ISO times).""" rows = await _run(elec.price_history, int(product_id)) return [{k: r.get(k) for k in ("site", "price", "mrp", "observed_at")} for r in rows]