nearle_daily/catalogue_backend
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: nearle_daily:2493b86ed85dafaca81ae0a8e4dfdb2b8ec3f0ac
Branches Tags
nearle_daily:master
..
compare: nearle_daily:ad7bb9250bdbaef9d18761fd4217539250eaa08c
Branches Tags
nearle_daily:master

2 Commits
2493b86ed8 ... ad7bb9250b

Author SHA1 Message Date
Suriyakumarvijayanayagam
ad7bb9250b Expose the catalog to AI clients over MCP at /mcp 
Adds a FastMCP server mounted onto the existing FastAPI app, so it ships in the
same container and answers on the same host rather than needing a process of
its own.

Fifteen 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. Each wraps a service function the REST
API already reaches through a GET. None of the write or compute endpoints are
exposed, because a tool list is chosen from by a model rather than by a person,
and catalog generation or model retraining is not something to leave one tool
call away.

The tools are written by hand rather than generated from the OpenAPI schema.
Mirroring all 63 routes would work, but a model picks a tool by reading its
description, and 63 near-identical generated entries is a worse thing to choose
from than a dozen written to be told apart.

Authentication reuses the access token from POST /api/auth/login - no separate
MCP credential, the same Principal and expiry as the REST API. The check lives
in one middleware rather than at the top of each tool, so a tool added later
cannot be left unguarded by forgetting a line. Note that this requires passing
include={"authorization"} to get_http_headers(), which strips that header by
default to avoid forwarding it downstream; without it the header is invisible
and every request looks unauthenticated, valid ones included.

Mounting a sub-app does not run its lifespan - only the outermost app's is
executed - so the MCP app's lifespan is chained through the FastAPI one. Without
that the endpoint accepts a connection and then fails on the first message with
a session manager that was never started.

Adds GET /api/mcp/info and POST /api/mcp/tools/{name} for the admin UI. The MCP
endpoint speaks streamable HTTP with session handling, so rendering a tool list
in the browser would otherwise mean shipping a full MCP client in React.

fastmcp needs Python 3.10+. The image is 3.11; on anything older the import
fails and the REST API starts without the MCP endpoint instead of not starting.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
 2026-08-13 13:14:00 +05:30
Suriyakumarvijayanayagam
e47edd2bb7 Serve the API on 3000 and 8000 at once, matching the frontend image 
Dokploy routes the domain to port 3000, but the container only bound 8000, so
the proxy had nothing to talk to and the domain returned 502 with a perfectly
healthy process behind it.

The frontend image already solved this by answering on both 80 and 3000
(`listen 80; listen 3000;` in nginx.conf). Do the same here rather than swap one
guess for another: 3000 is what the platform routes to, and 8000 is what the
README, the vite dev proxy and docker-compose all target, so binding both means
the container works whichever one it is pointed at.

uvicorn's CLI takes a single --port, but Server.run() accepts pre-bound
sockets, so serve.py binds each port and hands the list to one uvicorn - no
extra worker or second process to supervise. PORT still pins a single port for
anyone who wants one; PORTS changes the pair.

A port that cannot be bound is logged and skipped rather than being fatal,
since losing one of the two should not take down a service the platform only
routes to on the other. It exits non-zero only when nothing is listening at
all, so a genuinely dead container is still reported as failed.

The healthcheck moves into the same file and passes if either port answers,
which keeps it from drifting out of sync with what is actually bound.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
 2026-08-13 12:38:35 +05:30
9 changed files with 988 additions and 27 deletions
Expand all files Collapse all files

13
.env.example
View File

@@ -16,6 +16,19 @@
# localhost URL points the backend at its own empty ports. Containers reach
# each other by service name over the compose network instead.
# --- Ports (container only) -----------------------------------------------
# The image serves 3000 and 8000 at once - 3000 because that is what Dokploy
# routes a domain to, 8000 because the README, the vite dev proxy and
# docker-compose all use it. Serving both means the container works whichever
# one the platform points at.
#
# PORTS the comma-separated pair to bind. PORT pins a single port instead and
# takes precedence, e.g. PORT=8080 serves only 8080.
#
# Neither affects running uvicorn directly for local development.
# PORTS=3000,8000
# PORT=8080
# --- Persistence (IMPORTANT in Docker) ------------------------------------
# The three directories the app WRITES to at runtime. The defaults point inside
# the repo/image and are right for local development; in a container each one

47
Dockerfile
View File

@@ -44,6 +44,7 @@ COPY app ./app
COPY cli ./cli
COPY scripts ./scripts
COPY data ./data
COPY serve.py .
# Pristine copies of everything the app also WRITES to, kept at a path that is
# never mounted over.
@@ -68,31 +69,29 @@ RUN mkdir -p /app/.bundled \
# file) name them properly; this is the floor, not the recommended setup.
VOLUME ["/app/data", "/app/app/intelligence/artifacts"]
# The port uvicorn binds. Overridable because Dokploy assigns the container
# port per service - the frontend image answers on both 80 and 3000 for the
# same reason. A single process cannot listen twice, so this is the knob:
# PORT=3000 in the service's environment, if you standardise on 3000.
ENV PORT=8000
EXPOSE 8000
# Liveness only. /api/health always answers 200 - it reports Postgres and
# Ollama in the body as "degraded" rather than failing - which is deliberate
# here: a healthcheck that went red whenever Postgres blinked would have
# Dokploy restart a perfectly healthy API in a loop.
# Answers on BOTH ports, the same way the frontend image does (nginx.conf has
# `listen 80; listen 3000;`). 3000 is what Dokploy routes a domain to; 8000 is
# what this project's README, the vite dev proxy and docker-compose all use.
# Serving both means the container works whichever one the platform is pointed
# at, instead of returning 502 from a perfectly healthy process.
#
# The 10s timeout is not padding: the handler probes Ollama over HTTP with a
# 3s timeout of its own, so an unreachable Ollama makes every check take ~3s.
# serve.py binds both sockets and hands them to one uvicorn - see the note
# there. To pin a single port, set PORT (PORT=8080 serves only 8080); to change
# the pair, set PORTS.
ENV PORTS=3000,8000
EXPOSE 3000 8000
# Liveness only, and passes if EITHER port answers. /api/health always returns
# 200 - it reports Postgres and Ollama in the body as "degraded" rather than
# failing - which is deliberate: a check that went red whenever Postgres blinked
# would have Dokploy restart a perfectly healthy API in a loop.
#
# The 10s timeout is not padding: the handler probes Ollama over HTTP with a 3s
# timeout of its own, so an unreachable Ollama makes every check take ~3s.
# start-period covers first boot, where the venv is still cold.
HEALTHCHECK --interval=30s --timeout=10s --start-period=40s --retries=3 \
CMD python -c "import os,urllib.request;urllib.request.urlopen('http://127.0.0.1:'+os.environ.get('PORT','8000')+'/api/health',timeout=8)" || exit 1
CMD ["python", "serve.py", "--healthcheck"]
# `exec` matters: without it the shell stays PID 1 and Docker's SIGTERM never
# reaches uvicorn, so every deploy waits out the 10s kill timeout instead of
# shutting down cleanly. sh is only here to expand $PORT.
#
# --proxy-headers/--forwarded-allow-ips: this container is never reached
# directly - nginx (and, in prod, Dokploy's Traefik) sit in front of it.
# Without these, uvicorn ignores X-Forwarded-Proto and reports every request as
# plain http, so any redirect or generated absolute URL would downgrade an
# https request.
CMD ["sh", "-c", "exec uvicorn app.main:app --host 0.0.0.0 --port ${PORT:-8000} --proxy-headers --forwarded-allow-ips '*'"]
# Exec form: python is PID 1, so Docker's SIGTERM reaches it directly and a
# redeploy shuts down cleanly instead of waiting out the 10s kill timeout.
CMD ["python", "serve.py"]

62
README.md
View File

@@ -69,6 +69,68 @@ 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 <token>` 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 <token>" }
}
}
}
```
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
frontend image answers on 80 and 3000. 3000 is what Dokploy routes a domain to;
8000 is what this README, the vite dev proxy and `docker-compose.yml` use. Both
being live means the deployment works whichever one it is pointed at, instead of
returning 502 from a healthy container.
`serve.py` is what makes that possible - uvicorn's CLI binds a single `--port`,
but `Server.run()` accepts a list of pre-bound sockets, so it is still one
process. If one port is unavailable it logs and carries on with the other; it
exits non-zero only when nothing is listening.
```bash
python serve.py # 3000 and 8000
PORT=8080 python serve.py # only 8080 (PORT pins a single port)
PORTS=80,3000 python serve.py # a different pair
```
For local development `uvicorn app.main:app --reload --port 8000` is still the
normal thing to run - one port is all you need, and it gives you autoreload.
## Persistence: the two volumes a deployment needs
Most state lives in Postgres, but three things are written to the filesystem,

167
app/api/routers/mcp_info.py Normal file
View File

@@ -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

40
app/main.py
View File

@@ -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():

522
app/mcp_server.py Normal file
View File

@@ -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 <token> 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)

3
docker-compose.yml
View File

@@ -61,6 +61,9 @@ services:
OLLAMA_BASE_URL: http://host.docker.internal:11434
extra_hosts:
- "host.docker.internal:host-gateway"
# The container listens on 3000 and 8000 at once (see serve.py), so either
# side of this mapping can change without touching the image. 8000 is
# published because vite.config.js proxies /api to 127.0.0.1:8000.
ports:
- "8000:8000"
volumes:

7
requirements.txt
View File

@@ -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

154
serve.py Normal file
View File

@@ -0,0 +1,154 @@
"""
Container entry point: serve the API on every port the platform might route to.
The frontend image answers on both 80 and 3000 (``listen 80; listen 3000;`` in
nginx.conf) so it works whatever port the deployment is configured to hit. This
does the same for the API, which otherwise has to guess: Dokploy routes a domain
to one container port, this project's own README, vite.config.js proxy and
docker-compose mapping all say 8000, and picking wrong produces a 502 with a
perfectly healthy container behind it.
uvicorn's CLI binds a single ``--port``, but ``Server.run()`` accepts a list of
already-bound sockets, so one process can listen on several. That is what this
does - no extra worker, no second process to supervise.
Usage::
python serve.py # binds PORTS (default "3000,8000")
PORT=8080 python serve.py # binds only 8080
python serve.py --healthcheck # probe mode, used by HEALTHCHECK
Run it directly rather than through ``uvicorn app.main:app`` when you need the
multi-port behaviour; the plain uvicorn command still works for local
development where one port is all anyone wants.
"""
from __future__ import annotations
import logging
import os
import socket
import sys
import urllib.error
import urllib.request
logger = logging.getLogger("serve")
# Both of the ports this project actually uses anywhere: 3000 because that is
# what the platform routes a domain to by default, 8000 because the README,
# the vite dev proxy and docker-compose all target it.
DEFAULT_PORTS = "3000,8000"
HOST = os.getenv("HOST", "0.0.0.0")
def configured_ports() -> list[int]:
"""
The ports to bind, in order.
``PORT`` wins over ``PORTS`` and is treated as an explicit single choice:
setting it means "serve here", not "serve here as well". ``PORTS`` takes a
comma-separated list for the both-at-once case.
"""
raw = os.getenv("PORT") or os.getenv("PORTS") or DEFAULT_PORTS
ports: list[int] = []
for chunk in raw.split(","):
chunk = chunk.strip()
if not chunk:
continue
try:
port = int(chunk)
except ValueError:
logger.warning("Ignoring non-numeric port %r in %r", chunk, raw)
continue
if not 1 <= port <= 65535:
logger.warning("Ignoring out-of-range port %d", port)
continue
if port not in ports: # binding the same port twice would fail
ports.append(port)
if not ports:
logger.error("No usable port in %r - falling back to %s", raw, DEFAULT_PORTS)
return [int(p) for p in DEFAULT_PORTS.split(",")]
return ports
def _bind(port: int) -> socket.socket | None:
"""Bind one listening socket, or return None with the reason logged."""
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
try:
sock.bind((HOST, port))
except OSError as exc:
# Not fatal on its own. If the platform only routes to one of these,
# losing the other (already in use, not permitted) should not take the
# service down - _run() fails only when nothing at all is listening.
logger.warning("Could not bind %s:%d - %s", HOST, port, exc)
sock.close()
return None
sock.listen(2048)
sock.set_inheritable(True)
return sock
def _run() -> int:
import uvicorn
ports = configured_ports()
sockets = [s for s in (_bind(p) for p in ports) if s is not None]
if not sockets:
logger.error(
"Could not bind any of %s. The API is not listening; exiting so the "
"platform restarts or reports the container as failed.",
", ".join(str(p) for p in ports),
)
return 1
bound = [s.getsockname()[1] for s in sockets]
logger.info("Serving on %s port(s): %s", HOST, ", ".join(str(p) for p in bound))
config = uvicorn.Config(
"app.main:app",
# proxy_headers/forwarded_allow_ips: this container is never reached
# directly - nginx, and Dokploy's Traefik, sit in front of it. Without
# them uvicorn ignores X-Forwarded-Proto and reports every request as
# plain http, so any redirect or generated absolute URL would downgrade
# an https request.
proxy_headers=True,
forwarded_allow_ips="*",
)
uvicorn.Server(config).run(sockets=sockets)
return 0
def _healthcheck() -> int:
"""
Probe the API, passing if ANY configured port answers.
Shares configured_ports() with the server so the check cannot drift from
what is actually bound - the reason this lives here rather than being a
python -c one-liner in the Dockerfile.
"""
ports = configured_ports()
for port in ports:
try:
with urllib.request.urlopen(
f"http://127.0.0.1:{port}/api/health", timeout=8
) as resp:
if 200 <= resp.status < 400:
return 0
except (urllib.error.URLError, OSError, ValueError):
continue
print(
f"health: no response from any of {', '.join(str(p) for p in ports)}",
file=sys.stderr,
)
return 1
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")
if "--healthcheck" in sys.argv:
sys.exit(_healthcheck())
sys.exit(_run())