summaryrefslogtreecommitdiff
path: root/mcp_server
diff options
context:
space:
mode:
Diffstat (limited to 'mcp_server')
-rw-r--r--mcp_server/__init__.py6
-rw-r--r--mcp_server/__main__.py4
-rw-r--r--mcp_server/client.py75
-rw-r--r--mcp_server/server.py168
-rw-r--r--mcp_server/tests.py85
5 files changed, 338 insertions, 0 deletions
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 <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/<secret>/ -> 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()
generated by cgit v1.2.3 (git 2.39.1) at 2026-07-16 11:37:11 +0000