From eed90569f7dafa7eb9c8358c152efac1a49f0405 Mon Sep 17 00:00:00 2001 From: Tom Flux Date: Tue, 23 Jun 2026 21:16:30 +0100 Subject: Phase 3b: FastMCP "ai service" for claude.ai cooking mode A standalone MCP server (no Django import) exposing 7 tools over Streamable HTTP, backed by the Django REST API via the caine token. - mcp_server/: client.py (httpx wrapper over /api/), server.py (the tools + FastMCP app + main), __main__.py, tests.py. - Tools: get_pantry, set_item_state, add_to_pantry, what_can_i_cook, get_recipes, log_cook (suggests, never mutates), create_meta_recipe (brainstorm -> commit). Each description says when to call it. - deploy/food-mcp.service: systemd unit (own process, runs .venv python -m mcp_server, loads /var/lib/food/.env). - deploy/food.tomflux.xyz.nginx: current config + an authless /mcp// location proxying to 127.0.0.1:8765 (the URL secret is the credential; SSE-friendly buffering/timeout). - pyproject: [dependency-groups] mcp = [fastmcp, httpx]; deploy with `uv sync --group mcp`. Verified: 7 tools register on fastmcp 3.x, run() accepts transport/ host/port/path, 6 FoodClient unit tests pass (httpx MockTransport). Co-Authored-By: Claude Opus 4.8 --- mcp_server/__init__.py | 6 ++ mcp_server/__main__.py | 4 ++ mcp_server/client.py | 75 ++++++++++++++++++++++ mcp_server/server.py | 168 +++++++++++++++++++++++++++++++++++++++++++++++++ mcp_server/tests.py | 85 +++++++++++++++++++++++++ 5 files changed, 338 insertions(+) create mode 100644 mcp_server/__init__.py create mode 100644 mcp_server/__main__.py create mode 100644 mcp_server/client.py create mode 100644 mcp_server/server.py create mode 100644 mcp_server/tests.py (limited to 'mcp_server') diff --git a/mcp_server/__init__.py b/mcp_server/__init__.py new file mode 100644 index 0000000..07b8054 --- /dev/null +++ b/mcp_server/__init__.py @@ -0,0 +1,6 @@ +"""FastMCP "ai service" for the Food pantry app. + +A standalone process (no Django import) that exposes a small set of MCP tools to +claude.ai cooking mode over Streamable HTTP, backed by the Food Django REST API. +See mcp.md for the full specification. +""" diff --git a/mcp_server/__main__.py b/mcp_server/__main__.py new file mode 100644 index 0000000..f5f6e40 --- /dev/null +++ b/mcp_server/__main__.py @@ -0,0 +1,4 @@ +from .server import main + +if __name__ == "__main__": + main() diff --git a/mcp_server/client.py b/mcp_server/client.py new file mode 100644 index 0000000..d2b9ff0 --- /dev/null +++ b/mcp_server/client.py @@ -0,0 +1,75 @@ +"""Thin HTTP client around the Food Django REST API. + +Pure HTTP — holds no Django import. Talks to the API over localhost using the +`caine` DRF token (Authorization: Token ). One method per endpoint the +MCP tools need; all errors surface as `FoodApiError` with a readable message. +""" +import os + +import httpx + + +class FoodApiError(Exception): + """The Django API returned a non-2xx response, or was unreachable.""" + + +class FoodClient: + def __init__(self, base=None, token=None, http=None): + base = (base or os.environ.get("FOOD_API_BASE", "http://127.0.0.1:8042/api")).rstrip("/") + token = token or os.environ.get("FOOD_API_TOKEN", "") + # `http` is injectable for tests (e.g. httpx.MockTransport). + self._http = http or httpx.Client( + base_url=base, + headers={"Authorization": f"Token {token}"}, + timeout=30.0, + ) + + # --- low level --- + + def _request(self, method, path, **kwargs): + try: + resp = self._http.request(method, path, **kwargs) + except httpx.HTTPError as e: + raise FoodApiError(f"could not reach the food API: {e}") from e + if resp.status_code >= 400: + raise FoodApiError(f"{resp.status_code}: {_safe_json(resp)}") + return _safe_json(resp) + + def _get(self, path, params=None): + return self._request("GET", path, params=params) + + def _post(self, path, json): + return self._request("POST", path, json=json) + + # --- endpoints (one per MCP tool need) --- + + def pantry(self): + return self._get("/pantry/") + + def set_state(self, ingredient, state, location=None): + body = {"ingredient": ingredient, "state": state} + if location: + body["location"] = location + return self._post("/pantry/set-state/", body) + + def bulk_add(self, items): + return self._post("/bulk-pantry-add/", {"items": items}) + + def what_can_i_cook(self, servings=2): + return self._get("/what-can-i-cook/", {"servings": servings}) + + def meta_recipes(self): + return self._get("/meta-recipes/") + + def log_cook(self, payload): + return self._post("/log-cook/", payload) + + def create_meta_recipe(self, payload): + return self._post("/create-meta-recipe/", payload) + + +def _safe_json(resp): + try: + return resp.json() + except ValueError: + return resp.text diff --git a/mcp_server/server.py b/mcp_server/server.py new file mode 100644 index 0000000..cc8fe4a --- /dev/null +++ b/mcp_server/server.py @@ -0,0 +1,168 @@ +"""FastMCP tool definitions for the Food pantry app. + +Each tool maps to a Django API call via FoodClient and returns compact, +Claude-friendly data (names + states, not raw API dumps). Tool descriptions +state *when* to call them — current models reach for tools conservatively. + +Run with `python -m mcp_server` (see __main__.py). FastMCP 2.x. +""" +import os + +from fastmcp import FastMCP + +from .client import FoodClient, FoodApiError + +mcp = FastMCP("food-pantry") +_api = FoodClient() + +VALID_STATES = ("in", "low", "out") + + +def _err(e): + return {"error": str(e)} + + +@mcp.tool +def get_pantry(location: str | None = None, include_out: bool = False) -> dict: + """List what's in the pantry with each item's state (in / low / out). Call + this before suggesting meals or a shopping list so advice is grounded in + what's actually in stock. `location` optionally filters to fridge, cupboard, + or freezer. 'out' items are omitted unless `include_out` is true.""" + try: + rows = _api.pantry() + except FoodApiError as e: + return _err(e) + + items = [] + counts = {"in": 0, "low": 0, "out": 0} + for r in rows: + state = r.get("state") + if state in counts: + counts[state] += 1 + if location and r.get("location") != location: + continue + if not include_out and state == "out": + continue + items.append({ + "name": r.get("ingredient_name"), + "state": state, + "location": r.get("location"), + "quantity": r.get("quantity"), + "unit": r.get("unit"), + "expiry": r.get("expiry_date"), + "staple": r.get("is_staple"), + }) + return {"items": items, "counts": counts} + + +@mcp.tool +def set_item_state(ingredient: str, state: str, location: str | None = None) -> dict: + """Mark an ingredient In stock / running Low / Out. Call this when the user + says they used up, ran low on, or restocked something. `state` must be one + of 'in', 'low', 'out'. `location` is optional (fridge/cupboard/freezer).""" + if state not in VALID_STATES: + return {"error": f"state must be one of {list(VALID_STATES)}"} + try: + return _api.set_state(ingredient, state, location) + except FoodApiError as e: + return _err(e) + + +@mcp.tool +def add_to_pantry(items: list[dict]) -> dict: + """Add items the user bought. Each item is {name, location?, quantity?, + unit?} — e.g. {"name": "pork mince", "location": "fridge"}. Existing items + are restocked (set back to In) rather than duplicated.""" + payload = [ + { + "ingredient_name": it.get("name") or it.get("ingredient_name"), + "location": it.get("location", "fridge"), + "quantity": it.get("quantity"), + "unit": it.get("unit"), + } + for it in items + ] + try: + return _api.bulk_add(payload) + except FoodApiError as e: + return _err(e) + + +@mcp.tool +def what_can_i_cook(servings: int = 2) -> dict: + """Match the pantry against the meta-recipe templates. Returns each recipe + as ready / partial / missing with per-slot availability — use it to suggest + what the user can make right now.""" + try: + return _api.what_can_i_cook(servings) + except FoodApiError as e: + return _err(e) + + +@mcp.tool +def get_recipes() -> dict: + """The meta-recipe templates with their swappable slot options. Use this to + suggest substitutions — each slot lists the ingredients that can fill it + (e.g. a protein slot accepts pork mince OR chicken).""" + try: + return {"meta_recipes": _api.meta_recipes()} + except FoodApiError as e: + return _err(e) + + +@mcp.tool +def log_cook( + meta_recipe_id: int | None = None, + recipe_id: int | None = None, + slot_choices: dict | None = None, + servings: int = 2, + rating: int | None = None, + notes: str = "", +) -> dict: + """Record a cooked meal (and an optional 1-5 rating). Provide exactly one of + `meta_recipe_id` or `recipe_id`. Returns `used_ingredients` as SUGGESTIONS + only — the pantry is NOT changed. Confirm with the user, then call + `set_item_state` for each ingredient they agree to mark low or out.""" + payload = { + "meta_recipe_id": meta_recipe_id, + "recipe_id": recipe_id, + "slot_choices": slot_choices or {}, + "servings": servings, + "notes": notes, + } + if rating is not None: + payload["rating"] = rating + try: + return _api.log_cook(payload) + except FoodApiError as e: + return _err(e) + + +@mcp.tool +def create_meta_recipe(recipe: dict) -> dict: + """Save a meta-recipe the user brainstormed in the chat. `recipe` is the full + template: + + {name, method, prep_time_mins?, cook_time_mins?, default_servings?, + gear_needed?, tags?, + slots: [{name, required?, max_choices?, + options: [{ingredient_name, quantity_per_serving, unit, tags?}]}], + base_ingredients: [{ingredient_name, quantity_per_serving, unit}]} + + Pass an `id` to update an existing template (it rebuilds slots/bases). + Unknown ingredients are auto-created.""" + try: + return _api.create_meta_recipe(recipe) + except FoodApiError as e: + return _err(e) + + +def main(): + host = os.environ.get("FOOD_MCP_HOST", "127.0.0.1") + port = int(os.environ.get("FOOD_MCP_PORT", "8765")) + # Streamable HTTP at the root path; nginx maps /mcp// -> here. + mcp.run(transport="http", host=host, port=port, path="/") + + +if __name__ == "__main__": + main() diff --git a/mcp_server/tests.py b/mcp_server/tests.py new file mode 100644 index 0000000..cd13383 --- /dev/null +++ b/mcp_server/tests.py @@ -0,0 +1,85 @@ +"""Tests for the FoodClient HTTP layer, using httpx.MockTransport (no network, +no Django, no fastmcp). Run with: python -m unittest mcp_server.tests +""" +import json +import unittest + +import httpx + +from mcp_server.client import FoodClient, FoodApiError + + +def make_client(handler): + transport = httpx.MockTransport(handler) + http = httpx.Client( + transport=transport, + base_url="http://api.test/api", + headers={"Authorization": "Token x"}, + ) + return FoodClient(http=http) + + +class FoodClientTests(unittest.TestCase): + def test_set_state_posts_expected_body(self): + captured = {} + + def handler(request): + captured["path"] = request.url.path + captured["body"] = json.loads(request.content) + return httpx.Response(200, json={"state": "out"}) + + out = make_client(handler).set_state("noodles", "out", location="cupboard") + self.assertEqual(captured["path"], "/api/pantry/set-state/") + self.assertEqual( + captured["body"], + {"ingredient": "noodles", "state": "out", "location": "cupboard"}, + ) + self.assertEqual(out["state"], "out") + + def test_set_state_omits_location_when_absent(self): + captured = {} + + def handler(request): + captured["body"] = json.loads(request.content) + return httpx.Response(200, json={}) + + make_client(handler).set_state("eggs", "in") + self.assertNotIn("location", captured["body"]) + + def test_bulk_add_wraps_items(self): + captured = {} + + def handler(request): + captured["body"] = json.loads(request.content) + return httpx.Response(201, json={"added": 1}) + + make_client(handler).bulk_add([{"ingredient_name": "eggs"}]) + self.assertEqual(captured["body"], {"items": [{"ingredient_name": "eggs"}]}) + + def test_what_can_i_cook_passes_servings(self): + captured = {} + + def handler(request): + captured["servings"] = request.url.params.get("servings") + return httpx.Response(200, json={"results": []}) + + make_client(handler).what_can_i_cook(servings=4) + self.assertEqual(captured["servings"], "4") + + def test_error_response_raises(self): + def handler(request): + return httpx.Response(404, json={"error": "nope"}) + + with self.assertRaises(FoodApiError): + make_client(handler).set_state("x", "out") + + def test_unreachable_api_raises(self): + def handler(request): + raise httpx.ConnectError("boom") + + with self.assertRaises(FoodApiError): + make_client(handler).pantry() + + +if __name__ == "__main__": + unittest.main() -- cgit v1.2.3