diff --git a/README.md b/README.md index c5371c0..6324698 100644 --- a/README.md +++ b/README.md @@ -69,6 +69,46 @@ permission check; `user` holds the product/store/inventory permissions. `AUTH_ENABLED=false` disables all of it for local work — never in a deployment. See the Authentication section of `../DEPLOYMENT.md` for the full endpoint map. +## MCP server + +The catalog is exposed to AI clients over the Model Context Protocol at `/mcp`, +mounted onto this same FastAPI app (`app/mcp_server.py`) - no separate process +or container, so it deploys with the API and answers on the same host. + +**15 tools, all read-only.** Catalog search and browse, nutrition facts and +health scores, healthier alternatives, per-store inventory and pricing, +discounts, trending and sales analytics. None of the write or compute endpoints +are reachable through MCP: a tool list is chosen from by a model rather than by +a person, and "retrain the models" is not something to leave one tool call away. + +**Auth is the app's own access token** - there is no separate MCP credential. +Clients send `Authorization: Bearer ` from `POST /api/auth/login`, and +the same `Principal` and expiry apply as on the REST side. Note the practical +consequence: tokens expire after `AUTH_TOKEN_TTL_MINUTES` (12h by default), so a +long-running client has to refresh. Raise the TTL if that is a problem. + +```jsonc +{ + "mcpServers": { + "nearle-catalogue": { + "type": "http", + "url": "https://mcp.catalogue.nearle.ai.in/mcp", + "headers": { "Authorization": "Bearer " } + } + } +} +``` + +The admin UI has an inspector at `/mcp` (React route, admin-only) listing every +tool with its arguments and a console to run one against live data. It is backed +by `GET /api/mcp/info` and `POST /api/mcp/tools/{name}` rather than by the MCP +endpoint itself - the browser would otherwise need a full MCP client to render a +tool list. + +`fastmcp` requires Python 3.10+. The image is 3.11; on an older interpreter the +import fails, `app/main.py` logs a warning, and the REST API starts without the +MCP endpoint rather than failing outright. + ## Ports The container answers on **3000 and 8000 at the same time**, the same way the diff --git a/app/api/routers/mcp_info.py b/app/api/routers/mcp_info.py new file mode 100644 index 0000000..2855edd --- /dev/null +++ b/app/api/routers/mcp_info.py @@ -0,0 +1,167 @@ +""" +REST inspector for the MCP server. + +The MCP endpoint itself speaks streamable HTTP with session handling, which a +browser cannot usefully talk to without a full MCP client implementation. Rather +than ship one in React, these two endpoints expose what the admin page needs - +what tools exist, and what one returns - as ordinary JSON over the REST API the +frontend already authenticates against. + +Admin-only. The tool inventory describes the whole catalog API surface, and the +invoke endpoint executes a tool; neither is something to leave open just because +the tools happen to be read-only. +""" +from __future__ import annotations + +import logging +from typing import Any, Dict, List, Optional + +from fastapi import APIRouter, Body, Depends, HTTPException, Request + +from app.api.deps import require_admin + +logger = logging.getLogger(__name__) +router = APIRouter(prefix="/mcp", tags=["mcp"]) + + +def _server(): + """The FastMCP instance, or None when the optional dependency is absent.""" + try: + from app.mcp_server import mcp + + return mcp + except Exception: # pragma: no cover - depends on fastmcp being installed + return None + + +def _public_url(request: Request, path: str) -> str: + """ + The URL an external MCP client should connect to. + + Built from the forwarded headers rather than a configured constant so it is + right on whichever host the request arrived on - the API's own domain, the + frontend's nginx, or localhost in development. uvicorn runs with + --proxy-headers, so request.url already reflects the external scheme. + """ + base = str(request.base_url).rstrip("/") + return f"{base}{path}" + + +@router.get("/info", dependencies=[Depends(require_admin)]) +async def mcp_info(request: Request) -> Dict[str, Any]: + """Describe the MCP server and list its tools, for the admin page.""" + mcp = _server() + if mcp is None: + return { + "enabled": False, + "reason": ( + "The fastmcp package is not installed, or failed to import. It " + "requires Python 3.10 or newer." + ), + "tools": [], + } + + from app.mcp_server import MCP_PATH + + try: + # run_middleware=False: the auth middleware reads an Authorization header + # from an MCP request context that does not exist on this REST call. The + # caller is already established as an admin by the route dependency. + tools = await mcp.list_tools(run_middleware=False) + except Exception as exc: # noqa: BLE001 + logger.exception("Could not list MCP tools") + raise HTTPException(status_code=500, detail=f"Could not list MCP tools: {exc}") + + described: List[Dict[str, Any]] = [] + for tool in sorted(tools, key=lambda t: t.name): + # Server-side FunctionTool exposes its JSON Schema as `parameters`. + # `inputSchema` is the wire-format name, present on the client-side Tool + # an MCP client receives - checked second so this keeps working if a + # future version converges on one name. + schema = getattr(tool, "parameters", None) or getattr(tool, "inputSchema", None) or {} + properties = schema.get("properties", {}) or {} + required = set(schema.get("required", []) or []) + described.append( + { + "name": tool.name, + "description": (tool.description or "").strip(), + "tags": sorted(getattr(tool, "tags", set()) or []), + "parameters": [ + { + "name": pname, + "type": pinfo.get("type") or "any", + "description": (pinfo.get("description") or "").strip(), + "required": pname in required, + "default": pinfo.get("default"), + } + for pname, pinfo in properties.items() + ], + } + ) + + return { + "enabled": True, + "name": mcp.name, + "version": getattr(mcp, "version", None), + "url": _public_url(request, MCP_PATH), + "transport": "http", + "auth": "Bearer token from POST /api/auth/login", + "tool_count": len(described), + "tools": described, + } + + +@router.post("/tools/{tool_name}", dependencies=[Depends(require_admin)]) +async def invoke_tool( + tool_name: str, + arguments: Optional[Dict[str, Any]] = Body(default=None), +) -> Dict[str, Any]: + """ + Run one MCP tool and return its result - the admin page's "try it" console. + + Exists so the server can be checked from the browser without installing an + MCP client. Every tool is read-only, so this executes the same call an AI + client would make, with the same code path. + """ + mcp = _server() + if mcp is None: + raise HTTPException(status_code=503, detail="The MCP server is not available.") + + try: + # Same reasoning as above: this request authenticated through the REST + # guard, so the MCP auth middleware would have no headers to inspect. + tools = {t.name: t for t in await mcp.list_tools(run_middleware=False)} + except Exception as exc: # noqa: BLE001 + raise HTTPException(status_code=500, detail=f"Could not list MCP tools: {exc}") + + if tool_name not in tools: + raise HTTPException( + status_code=404, + detail=f"No MCP tool named '{tool_name}'. Known tools: {', '.join(sorted(tools))}.", + ) + + try: + result = await mcp.call_tool(tool_name, arguments or {}, run_middleware=False) + except Exception as exc: # noqa: BLE001 + # A tool raising ToolError is a normal outcome worth showing verbatim - + # "no nutrition data for this product" is an answer, not a server fault. + return {"tool": tool_name, "ok": False, "error": str(exc)} + + return {"tool": tool_name, "ok": True, "result": _jsonable(result)} + + +def _jsonable(result: Any) -> Any: + """Reduce a ToolResult to something FastAPI can serialise.""" + for attr in ("structured_content", "data"): + value = getattr(result, attr, None) + if value is not None: + return value + + content = getattr(result, "content", None) + if content is None: + return result + out = [] + for block in content: + text = getattr(block, "text", None) + out.append(text if text is not None else str(block)) + return out diff --git a/app/main.py b/app/main.py index ee06edd..b565c4b 100644 --- a/app/main.py +++ b/app/main.py @@ -24,7 +24,7 @@ from app.infrastructure.settings import API_CORS_ORIGINS from app.api.routers import health, brands, search, chat, catalog, system from app.api.routers import stores, discounts, analytics as store_analytics, trending, recommendations, store_admin from app.api.routers import nutrition, nutrition_admin, upload -from app.api.routers import auth, user_products, admin_train +from app.api.routers import auth, user_products, admin_train, mcp_info from app.services.store_db import ensure_store_intelligence_schema from app.services.nutrition_db import ensure_nutrition_schema @@ -34,6 +34,18 @@ logging.basicConfig( ) logger = logging.getLogger(__name__) +try: + from app.mcp_server import MCP_PATH, build_http_app as build_mcp_app + + _mcp_app = build_mcp_app() +except Exception as e: # pragma: no cover - depends on an optional dependency + # fastmcp needs Python >=3.10. The image is 3.11, but an older local + # interpreter should lose the MCP endpoint, not the whole API. + logging.getLogger(__name__).warning("MCP server unavailable: %s", e) + _mcp_app = None + MCP_PATH = "/mcp" + + @asynccontextmanager async def lifespan(_app: FastAPI): """ @@ -64,7 +76,17 @@ async def lifespan(_app: FastAPI): logger.warning("Startup background init warning: %s", e) threading.Thread(target=_async_init, daemon=True).start() - yield + + # The mounted MCP app carries its own lifespan, which starts the session + # manager its request handler depends on. Mounting a sub-app does NOT run + # that lifespan - only the outermost app's is executed - so without this + # the endpoint exists, accepts a connection, and then fails on the first + # message with a session manager that was never started. + if _mcp_app is not None: + async with _mcp_app.lifespan(_mcp_app): + yield + else: + yield app = FastAPI( @@ -140,6 +162,15 @@ app.include_router(store_admin.router, prefix="/api") app.include_router(nutrition.router, prefix="/api") app.include_router(nutrition_admin.router, prefix="/api") app.include_router(upload.router, prefix="/api") +app.include_router(mcp_info.router, prefix="/api") + +# MCP lives outside /api on purpose: it is a protocol endpoint for AI clients, +# not part of the REST surface, and the catch-all SPA route below only skips +# paths it recognises. Mounted rather than routed because it is a whole ASGI +# app with its own request handling. +if _mcp_app is not None: + app.mount(MCP_PATH, _mcp_app) + logger.info("MCP server mounted at %s", MCP_PATH) # Serve built frontend static files if dist exists (single-port unified # deployment). Off in the normal setup: the React app is served by its own @@ -179,7 +210,10 @@ if FRONTEND_DIST.exists() and (FRONTEND_DIST / "assets").exists(): # endpoint would look like a successful call to any client. 404 is the # honest answer, and it matters now that the API is also reachable # directly at api.{$DOMAIN} rather than only behind the frontend. - if full_path.startswith(("api", "docs", "redoc", "openapi.json")): + # "mcp" is in this list for the case where the MCP app failed to load: + # the mount would be absent, and without this the SPA fallback would + # answer an MCP client with index.html and a 200. + if full_path.startswith(("api", "docs", "redoc", "openapi.json", "mcp")): raise HTTPException(status_code=404, detail="Not found") file_path = FRONTEND_DIST / full_path if file_path.exists() and file_path.is_file(): diff --git a/app/mcp_server.py b/app/mcp_server.py new file mode 100644 index 0000000..06df29f --- /dev/null +++ b/app/mcp_server.py @@ -0,0 +1,522 @@ +""" +MCP server exposing the catalog as tools for AI clients. + +Mounted onto the FastAPI app at /mcp (see app/main.py), so it ships in the same +container and answers on the same host - mcp.catalogue.nearle.ai.in/mcp. + +Scope: reads only. Every tool here maps to a service function the REST API +already exposes through a GET. None of the write or compute endpoints - catalog +generation, ML training, the upload endpoints, product creation - are reachable +through MCP, because a tool list is chosen from by a model rather than by a +person, and "retrain the models" is not a thing to leave one tool call away. +Adding a write tool later is deliberate work, not an oversight to correct. + +Authentication reuses the access token from POST /api/auth/login. There is no +separate MCP credential: the same Principal, the same expiry, the same +revocation story as the REST API, and nothing new to keep secret. Clients send +it as an Authorization: Bearer header, which _AuthMiddleware checks once for +every tool call and tool listing. + +The tools are hand-written rather than generated from the OpenAPI schema. All +63 routes would technically work, but a model picks a tool by reading its +description, and 63 near-identical auto-generated entries is a worse starting +point than a dozen written to be chosen between. +""" +from __future__ import annotations + +import logging +from typing import Any, Dict, List, Optional + +from fastmcp import FastMCP +from fastmcp.exceptions import ToolError +from fastmcp.server.dependencies import get_http_headers +from fastmcp.server.middleware import CallNext, Middleware, MiddlewareContext + +from app.infrastructure.security import AuthError, Principal, anonymous_principal, decode_access_token +from app.infrastructure.settings import AUTH_ENABLED + +logger = logging.getLogger(__name__) + +MCP_PATH = "/mcp" + +INSTRUCTIONS = """\ +Read-only access to an Indian FMCG product catalog: products and brands, \ +nutrition facts and health scores, per-store inventory and pricing, discounts, \ +and sales analytics. + +Products are identified by a (brand, image_id) pair - image_id is the SKU-like \ +product key, not an image URL. Get one from search_products or \ +list_brand_products before calling any tool that takes an image_id. + +Nutrition figures are per 100g unless the product states otherwise, and health \ +scores run 0-100, higher being better. Prices are in INR. +""" + + +# --------------------------------------------------------------------------- +# Authentication +# --------------------------------------------------------------------------- +def _principal_from_headers() -> Principal: + """ + Resolve the caller from the Authorization header of the current request. + + Raises ToolError rather than AuthError: ToolError is what reaches the client + as a readable message instead of an opaque internal failure. + """ + if not AUTH_ENABLED: + return anonymous_principal() + + # include={"authorization"} is required, not optional: get_http_headers() + # strips the Authorization header by default, so that it is not forwarded to + # downstream services by accident. Without this the header is invisible here + # and every request looks unauthenticated - including valid ones. + headers = get_http_headers(include={"authorization"}) + raw = headers.get("authorization", "") # keys are lowercased by the helper + scheme, _, token = raw.partition(" ") + + if scheme.lower() != "bearer" or not token.strip(): + raise ToolError( + "Not authenticated. Send an Authorization: Bearer header. " + "Get a token from POST /api/auth/login." + ) + try: + return decode_access_token(token.strip()) + except AuthError as exc: + raise ToolError(str(exc)) from exc + + +class _AuthMiddleware(Middleware): + """ + One authentication check covering every tool. + + Sitting here rather than at the top of each tool means a tool added later + cannot be left unguarded by forgetting a line - the same reason the REST + side puts its guard in a route dependency instead of the handler body. + + Listing is checked as well as calling. An unauthenticated client learning + the exact shape of the catalog API is a smaller problem than an + unauthenticated call, but it is not nothing, and there is no reason to + publish it. + """ + + async def on_call_tool(self, context: MiddlewareContext, call_next: CallNext): + principal = _principal_from_headers() + name = getattr(context.message, "name", "?") + logger.info("MCP tool call %r by %s (%s)", name, principal.username, principal.kind) + return await call_next(context) + + async def on_list_tools(self, context: MiddlewareContext, call_next: CallNext): + _principal_from_headers() + return await call_next(context) + + +mcp: FastMCP = FastMCP( + name="nearle-catalogue", + instructions=INSTRUCTIONS, + version="1.0.0", + middleware=[_AuthMiddleware()], +) + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- +def _guard(what: str, fn, *args, **kwargs): + """ + Run a service call, turning failures into a ToolError the client can read. + + Nearly every tool here reads Postgres, so the common failure is the database + being unreachable. Left to propagate, that surfaces to the model as an + unexplained error; named, the model can say what went wrong instead of + retrying or inventing an answer. + """ + try: + return fn(*args, **kwargs) + except Exception as exc: # noqa: BLE001 - deliberately broad, see docstring + logger.exception("MCP tool %s failed", what) + raise ToolError(f"{what} failed: {exc}") from exc + + +def _clip(n: Optional[int], default: int, maximum: int) -> int: + """Keep result counts sane - a model will happily ask for 10000 rows.""" + if n is None: + return default + return max(1, min(int(n), maximum)) + + +def _slim(product: Dict[str, Any]) -> Dict[str, Any]: + """ + Drop the fields a model has no use for. + + Embeddings especially: a 384-float vector per product would swamp the + context window and says nothing a model can reason about. + """ + return { + k: v + for k, v in product.items() + if k not in {"embedding", "embedding_text", "distance"} and v not in (None, "", [], {}) + } + + +# --------------------------------------------------------------------------- +# Catalog +# --------------------------------------------------------------------------- +@mcp.tool( + annotations={"readOnlyHint": True}, + tags={"catalog"}, +) +def search_products( + query: str, + brand: Optional[str] = None, + category: Optional[str] = None, + limit: Optional[int] = 10, +) -> List[Dict[str, Any]]: + """Search the catalog by meaning, not keywords. + + The main entry point: use this to turn a description ("sugar-free biscuits", + "something to replace butter") into concrete products. Returns each match + with its brand and image_id, which the nutrition, store and recommendation + tools take as input. + + Args: + query: What to look for, in natural language. + brand: Restrict to one brand. Omit to search everything. + category: Restrict to one category, e.g. "Dairy". + limit: How many products to return (1-50). + """ + from app.services.rag_service import retrieve + + top_k = _clip(limit, 10, 50) + found = _guard("search_products", retrieve, query, brand=brand, top_k=top_k, category=category) + return [ + _slim( + { + "brand": p.brand, + "image_id": p.image_id, + "title": p.title, + "category": p.category, + "description": p.description, + "price_range": p.price_range, + "selling_price": p.final_selling_price or p.selling_price, + "size_variants": p.size_variants, + "barcode": p.barcode, + } + ) + for p in found + ] + + +@mcp.tool(annotations={"readOnlyHint": True}, tags={"catalog"}) +def list_brands() -> List[str]: + """List every brand in the catalog. + + Useful for grounding a brand name before passing it to another tool, since + brand names must match the catalog's spelling. + """ + from app.services.vector_store import list_available_brands + + return _guard("list_brands", list_available_brands) + + +@mcp.tool(annotations={"readOnlyHint": True}, tags={"catalog"}) +def list_categories(brand: str) -> List[str]: + """List the product categories a brand sells. + + Args: + brand: Brand name, as returned by list_brands. + """ + from app.services.vector_store import list_categories_for_brand + + return _guard("list_categories", list_categories_for_brand, brand) + + +@mcp.tool(annotations={"readOnlyHint": True}, tags={"catalog"}) +def list_brand_products( + brand: str, + category: Optional[str] = None, + limit: Optional[int] = 25, +) -> List[Dict[str, Any]]: + """List a brand's products, newest catalog entries first. + + Browsing, as opposed to search_products' semantic matching. Prefer + search_products when the user described what they want rather than naming a + brand outright. + + Args: + brand: Brand name, as returned by list_brands. + category: Restrict to one category. + limit: How many products to return (1-100). + """ + from app.services.vector_store import get_products_by_brand + + rows = _guard( + "list_brand_products", + get_products_by_brand, + brand, + limit=_clip(limit, 25, 100), + category=category, + ) + return [_slim(r) for r in rows] + + +@mcp.tool(annotations={"readOnlyHint": True}, tags={"catalog"}) +def get_product(brand: str, image_id: str) -> Dict[str, Any]: + """Get the full catalog record for one product. + + Args: + brand: Brand name. + image_id: The product key from search_products or list_brand_products. + """ + from app.services.vector_store import get_product_by_image_id + + row = _guard("get_product", get_product_by_image_id, brand, image_id) + if not row: + raise ToolError( + f"No product {image_id!r} for brand {brand!r}. " + "Check the pair with search_products." + ) + return _slim(row) + + +# --------------------------------------------------------------------------- +# Nutrition +# --------------------------------------------------------------------------- +@mcp.tool(annotations={"readOnlyHint": True}, tags={"nutrition"}) +def get_nutrition(brand: str, image_id: str) -> Dict[str, Any]: + """Get nutrition facts, health score and dietary flags for one product. + + Covers macros and key micronutrients per 100g, a 0-100 health score, diet + tags (vegetarian, high-protein, ...) and declared allergens. + + Args: + brand: Brand name. + image_id: The product key from search_products. + """ + from app.services.nutrition_db import get_full_nutrition + + row = _guard("get_nutrition", get_full_nutrition, brand, image_id) + if not row: + raise ToolError( + f"No nutrition data for {brand}/{image_id}. Not every catalog " + "product has been enriched yet." + ) + return _slim(row) + + +@mcp.tool(annotations={"readOnlyHint": True}, tags={"nutrition"}) +def find_healthier_alternatives( + brand: str, + image_id: str, + limit: Optional[int] = 5, +) -> List[Dict[str, Any]]: + """Find comparable products with a better health score. + + Answers "what should I buy instead of this?" - matches are drawn from the + same category, so the suggestion is a real substitute rather than merely + something healthier. + + Args: + brand: Brand name of the product to improve on. + image_id: Product key of the product to improve on. + limit: How many alternatives to return (1-20). + """ + from app.services.nutrition_alternatives_service import find_alternatives + + return _guard( + "find_healthier_alternatives", + find_alternatives, + brand, + image_id, + top_k=_clip(limit, 5, 20), + ) + + +@mcp.tool(annotations={"readOnlyHint": True}, tags={"nutrition"}) +def filter_products_by_nutrition( + sort_by: str = "health_score", + order: str = "desc", + category: Optional[str] = None, + diet_tag: Optional[str] = None, + exclude_allergen: Optional[str] = None, + limit: Optional[int] = 20, +) -> List[Dict[str, Any]]: + """Rank and filter products by a nutritional measure. + + The tool for questions shaped like "highest protein snacks", "lowest sugar + dairy", or "gluten-free products without nuts". + + Args: + sort_by: Field to rank by - health_score, protein_g, total_sugar_g, + dietary_fiber_g, sodium_mg, calories_kcal, total_fat_g. + order: "desc" for highest first, "asc" for lowest first. + category: Restrict to one category. + diet_tag: Keep only products carrying this tag, e.g. "High Protein". + exclude_allergen: Drop products declaring this allergen, e.g. "Dairy". + limit: How many products to return (1-100). + """ + from app.services.nutrition_db import query_products + + return _guard( + "filter_products_by_nutrition", + query_products, + sort_by=sort_by, + order=order, + category=category, + diet_tag=diet_tag, + exclude_allergen=exclude_allergen, + limit=_clip(limit, 20, 100), + ) + + +# --------------------------------------------------------------------------- +# Stores +# --------------------------------------------------------------------------- +@mcp.tool(annotations={"readOnlyHint": True}, tags={"stores"}) +def list_stores() -> List[Dict[str, Any]]: + """List the retail stores, with city and tier. + + Call this first for anything store-specific - the other store tools need a + store_id from here. + """ + from app.services.store_db import list_stores as _list + + return _guard("list_stores", _list) + + +@mcp.tool(annotations={"readOnlyHint": True}, tags={"stores"}) +def get_store_inventory( + store_id: str, + category: Optional[str] = None, + in_stock_only: bool = False, + limit: Optional[int] = 25, +) -> List[Dict[str, Any]]: + """List what one store carries, with stock levels and its own pricing. + + Prices vary per store, so this is the tool for "what does it cost at X", + not the catalog price_range. + + Args: + store_id: Store identifier from list_stores. + category: Restrict to one category. + in_stock_only: Drop products currently out of stock. + limit: How many products to return (1-100). + """ + from app.services.store_db import get_store_products + + return _guard( + "get_store_inventory", + get_store_products, + store_id, + category=category, + in_stock_only=in_stock_only, + limit=_clip(limit, 25, 100), + ) + + +@mcp.tool(annotations={"readOnlyHint": True}, tags={"stores"}) +def get_store_discounts(store_id: str, limit: Optional[int] = 25) -> List[Dict[str, Any]]: + """List the discounts currently allocated at one store. + + Args: + store_id: Store identifier from list_stores. + limit: How many discounts to return (1-100). + """ + from app.services.store_db import get_latest_discounts + + return _guard( + "get_store_discounts", get_latest_discounts, store_id, limit=_clip(limit, 25, 100) + ) + + +# --------------------------------------------------------------------------- +# Analytics +# --------------------------------------------------------------------------- +@mcp.tool(annotations={"readOnlyHint": True}, tags={"analytics"}) +def get_trending( + window: str = "weekly", + scope: str = "overall", + scope_value: Optional[str] = None, + limit: Optional[int] = 10, +) -> List[Dict[str, Any]]: + """List what is selling fastest right now. + + Args: + window: Period to measure over - "daily", "weekly" or "monthly". + scope: "overall", or "store"/"category" to narrow it. + scope_value: The store_id or category name, when scope is not "overall". + limit: How many products to return (1-50). + """ + from app.services.store_db import get_trending as _trending + + return _guard( + "get_trending", + _trending, + window_label=window, + scope=scope, + scope_value=scope_value, + top_k=_clip(limit, 10, 50), + ) + + +@mcp.tool(annotations={"readOnlyHint": True}, tags={"analytics"}) +def get_top_products( + by: str = "revenue", + limit: Optional[int] = 10, + ascending: bool = False, +) -> List[Dict[str, Any]]: + """Rank products across every store by a sales measure. + + Args: + by: What to rank on - "revenue", "units" or "orders". + limit: How many products to return (1-50). + ascending: True ranks worst-performing first. + """ + from app.services.analytics_service import top_products + + return _guard( + "get_top_products", top_products, by=by, limit=_clip(limit, 10, 50), ascending=ascending + ) + + +@mcp.tool(annotations={"readOnlyHint": True}, tags={"analytics"}) +def get_store_analytics(store_id: str) -> Dict[str, Any]: + """Get one store's sales dashboard - revenue, orders, top sellers, stock health. + + Args: + store_id: Store identifier from list_stores. + """ + from app.services.analytics_service import store_dashboard + + return _guard("get_store_analytics", store_dashboard, store_id) + + +@mcp.tool(annotations={"readOnlyHint": True}, tags={"catalog"}) +def get_recommendations( + brand: str, + image_id: str, + limit: Optional[int] = 5, +) -> List[Dict[str, Any]]: + """List products frequently bought together with this one. + + Co-purchase based, so this answers "what else goes in the basket", which is + a different question from find_healthier_alternatives' "what instead". + + Args: + brand: Brand name. + image_id: Product key from search_products. + limit: How many recommendations to return (1-20). + """ + from app.services.recommendation_service import recommend_for_product + + return _guard( + "get_recommendations", + recommend_for_product, + brand, + image_id, + top_k=_clip(limit, 5, 20), + ) + + +def build_http_app(path: str = MCP_PATH): + """The ASGI app to mount on FastAPI. See app/main.py for the lifespan wiring.""" + return mcp.http_app(path="/", transport="http", stateless_http=True) diff --git a/requirements.txt b/requirements.txt index a85a840..ffa93d6 100644 --- a/requirements.txt +++ b/requirements.txt @@ -14,6 +14,13 @@ python-dotenv>=1.0.1 # deliberately no bcrypt/argon2/passlib dependency here. PyJWT>=2.9.0 +# --- MCP (Model Context Protocol) --- +# Serves the catalog as tools for AI clients at /mcp (see app/mcp_server.py), +# mounted onto the FastAPI app so it needs no separate process or container. +# Requires Python >=3.10; the Dockerfile's python:3.11-slim satisfies that, and +# app/main.py degrades to serving the REST API alone if the import fails. +fastmcp>=3.4.7 + # --- Database / pgvector --- psycopg[binary]>=3.2.3 pgvector>=0.2.5