c7e4d59188
Verified catalogue of mobiles and laptops sold in India, collected from real retail listings (FastAPI backend, React frontend, Postgres/pgvector). - REST API under /api/elec (read-only catalogue; admin endpoints need login) - MCP server (FastMCP) at /mcp/ with list_categories, search_products, get_product and price_history tools - Real ratings and reviews read from product pages and search results - Production Dockerfile (requirements-api.txt, no PyTorch) and .env.production.example; remote database only via an explicit ELEC_ALLOW_REMOTE_DB host/name allowlist - docs/API.md: endpoint and MCP reference with live examples Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
122 lines
4.5 KiB
Python
122 lines
4.5 KiB
Python
"""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."
|
|
),
|
|
)
|
|
|
|
_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",
|
|
)
|
|
|
|
|
|
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, and an image URL (or null).
|
|
"""
|
|
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 (null if none is published),
|
|
and up to 10 real customer reviews (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 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]
|