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>
94 lines
4.0 KiB
Python
94 lines
4.0 KiB
Python
"""MCP endpoint (app/mcp_server.py): the tool surface, and that tools return the
|
|
same catalogue data as the REST API. Database tests are skipped when the local
|
|
Postgres container is not running."""
|
|
from __future__ import annotations
|
|
|
|
from decimal import Decimal
|
|
|
|
import anyio
|
|
from fastmcp import Client
|
|
|
|
from app.mcp_server import mcp
|
|
|
|
|
|
def _call(name: str, args: dict):
|
|
async def go():
|
|
async with Client(mcp) as c:
|
|
return (await c.call_tool(name, args)).data
|
|
return anyio.run(go)
|
|
|
|
|
|
def test_only_read_only_catalogue_tools_are_exposed():
|
|
async def go():
|
|
async with Client(mcp) as c:
|
|
return {t.name: set(t.input_schema.get("properties", {})) for t in await c.list_tools()}
|
|
tools = anyio.run(go)
|
|
assert set(tools) == {"list_categories", "search_products", "get_product", "price_history"}
|
|
assert tools["search_products"] == {"query", "category", "brand", "max_price", "min_price", "limit"}
|
|
# Nothing that can start a run, log in, or change data.
|
|
assert not any(w in name for name in tools for w in ("admin", "run", "login", "probe", "review"))
|
|
|
|
|
|
def test_mcp_endpoint_answers_an_initialize_handshake():
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.main import app
|
|
|
|
body = {"jsonrpc": "2.0", "id": 1, "method": "initialize",
|
|
"params": {"protocolVersion": "2025-06-18", "capabilities": {},
|
|
"clientInfo": {"name": "test", "version": "1"}}}
|
|
headers = {"Accept": "application/json, text/event-stream"}
|
|
with TestClient(app) as c: # `with` runs the lifespan that starts MCP sessions
|
|
r = c.post("/mcp/", json=body, headers=headers)
|
|
assert r.status_code == 200, r.text
|
|
assert "Electronics Catalog" in r.text # server name in the initialize result
|
|
|
|
|
|
def test_tools_return_the_catalogue(db):
|
|
from app.electronics.collector import Collector, RunOptions, RunStats
|
|
from app.electronics.db import repository as repo
|
|
from app.electronics.models import Listing
|
|
from app.electronics.normalise.title_parser import parse_title, variant_key
|
|
|
|
def listing(site, sku, price):
|
|
title = "Samsung Galaxy S24 5G (8GB RAM, 256GB)"
|
|
p = parse_title(title, "mobiles")
|
|
l = Listing(site_domain=site, source_sku=sku, source_url=f"https://www.{site}/p/{sku}",
|
|
source_type="search_snippet", brand_slug="samsung", category="mobiles", title=title,
|
|
evidence_text=f"{title} ₹{price}", confidence=0.5, parser="test",
|
|
model=p.model, ram_gb=p.ram_gb, storage_gb=p.storage_gb, price=Decimal(price))
|
|
l.model_norm, l.variant_key = p.model_norm, variant_key(p, "mobiles")
|
|
return l
|
|
|
|
c = Collector.__new__(Collector)
|
|
c.opt = RunOptions(category="mobiles", brands=["samsung"])
|
|
c.ids, c.run_id, c._touched_products, c.stats = repo.id_maps(), None, {}, RunStats()
|
|
c.store(listing("amazon.in", "B0CS5XW6TN", 74999))
|
|
c.store(listing("poorvika.com", "samsung-galaxy-s24", 73999))
|
|
repo.refresh_verification()
|
|
|
|
cats = {x["slug"]: x["product_count"] for x in _call("list_categories", {})}
|
|
assert cats["mobiles"] == 1
|
|
|
|
found = _call("search_products", {"query": "galaxy", "category": "mobiles", "max_price": 80000})
|
|
assert found["total"] == 1
|
|
hit = found["products"][0]
|
|
assert hit["best_price"] == "73999.00" and hit["best_price_site"] == "Poorvika"
|
|
assert _call("search_products", {"max_price": 1000})["total"] == 0
|
|
|
|
detail = _call("get_product", {"product_id": hit["product_id"]})
|
|
assert {o["site"] for o in detail["offers"]} == {"Amazon.in", "Poorvika"}
|
|
assert all(o["source_url"].startswith("https://") for o in detail["offers"])
|
|
assert detail["image_urls"] == [] and detail["rating"] is None
|
|
|
|
history = _call("price_history", {"product_id": hit["product_id"]})
|
|
assert {h["price"] for h in history} == {"74999.00", "73999.00"}
|
|
|
|
|
|
def test_unknown_product_is_a_tool_error(db):
|
|
import pytest
|
|
from fastmcp.exceptions import ToolError
|
|
|
|
with pytest.raises(ToolError, match="not found"):
|
|
_call("get_product", {"product_id": 999999})
|