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/<secret>/ 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 <noreply@anthropic.com>
This commit is contained in:
Tom Flux
2026-06-23 21:16:30 +01:00
co-authored by Claude Opus 4.8
parent 7255c9302c
commit 44de784f70
9 changed files with 449 additions and 8 deletions
+18
View File
@@ -0,0 +1,18 @@
[Unit]
Description=Food MCP service (FastMCP — claude.ai connector)
After=network.target food.service
[Service]
Type=simple
User=openclaw
Group=openclaw
WorkingDirectory=/var/lib/food
Environment="PATH=/var/lib/food/.venv/bin:/usr/bin"
# FOOD_API_TOKEN (the caine DRF token), optional FOOD_API_BASE / FOOD_MCP_PORT.
EnvironmentFile=-/var/lib/food/.env
ExecStart=/var/lib/food/.venv/bin/python -m mcp_server
Restart=on-failure
RestartSec=5
[Install]
WantedBy=multi-user.target
+71
View File
@@ -0,0 +1,71 @@
# nginx config for food.tomflux.xyz
# Adds the MCP location to the existing web-app server block.
#
# The MCP connector is "authless" from claude.ai's side — the long secret in the
# URL path IS the credential. Only that exact prefix is proxied to the FastMCP
# service; everything else under /mcp/ falls through to the catch-all 301.
# Generate the secret once and keep it out of git, e.g.:
# python -c "import secrets; print(secrets.token_urlsafe(32))"
# then replace REPLACE_WITH_LONG_SECRET below and in the claude.ai connector URL:
# https://food.tomflux.xyz/mcp/REPLACE_WITH_LONG_SECRET/
server {
server_name food.tomflux.xyz;
location /app/ {
proxy_pass http://127.0.0.1:8042;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
}
location /accounts/ {
proxy_pass http://127.0.0.1:8042;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
}
location /static/ {
proxy_pass http://127.0.0.1:8042;
}
# --- MCP server (claude.ai cooking-mode connector) ---
# Trailing slashes on both sides strip the secret prefix, so the FastMCP
# service (serving at "/") sees clean paths.
location /mcp/REPLACE_WITH_LONG_SECRET/ {
proxy_pass http://127.0.0.1:8765/;
proxy_http_version 1.1;
proxy_set_header Host $host;
proxy_set_header X-Forwarded-Proto $scheme;
proxy_buffering off; # MCP Streamable HTTP / SSE
proxy_read_timeout 3600s;
}
location / {
return 301 /app/;
}
# Block API and admin from external access (MCP reaches the API over
# localhost, so this does not affect it).
location /api/ { return 404; }
location /admin/ { return 404; }
listen 443 ssl; # managed by Certbot
ssl_certificate /etc/letsencrypt/live/dav.jihakuz.xyz/fullchain.pem; # managed by Certbot
ssl_certificate_key /etc/letsencrypt/live/dav.jihakuz.xyz/privkey.pem; # managed by Certbot
include /etc/letsencrypt/options-ssl-nginx.conf; # managed by Certbot
ssl_dhparam /etc/letsencrypt/ssl-dhparams.pem; # managed by Certbot
}
server {
if ($host = food.tomflux.xyz) {
return 301 https://$host$request_uri;
} # managed by Certbot
server_name food.tomflux.xyz;
listen 80;
return 404; # managed by Certbot
}
+14 -8
View File
@@ -8,7 +8,7 @@
> Decisions locked in `plan.md §3` and `requirements.md §7`: a **separate
> FastMCP "ai service"** (a pattern Tom already runs elsewhere) that talks to
> the existing Django REST API, hosted on the same box at
> **`food.jihakuz.xyz/mcp`**, behind nginx, **bearer-token** auth.
> **`food.tomflux.xyz/mcp`**, behind nginx, **bearer-token** auth.
---
@@ -43,7 +43,7 @@ blocks — now grounded instead of guessed.
claude.ai (cooking mode)
│ Streamable HTTP, secret in the URL path
nginx food.jihakuz.xyz/mcp (TLS, public)
nginx food.tomflux.xyz/mcp (TLS, public)
│ proxy_pass 127.0.0.1:8765
FastMCP "ai service" (food-mcp.service, localhost only)
@@ -71,7 +71,7 @@ localhost Django**. It holds no database connection of its own.
- **Transport:** Streamable HTTP (the remote-MCP transport claude.ai connectors
use). FastMCP served over HTTP, bound to `127.0.0.1:8765`.
- **Public route:** nginx on `food.jihakuz.xyz`, `location /mcp`
- **Public route:** nginx on `food.tomflux.xyz`, `location /mcp`
`proxy_pass http://127.0.0.1:8765`. TLS via the existing Let's Encrypt setup.
Note SSE/streaming needs `proxy_buffering off;` and a long
`proxy_read_timeout` on that location.
@@ -90,7 +90,7 @@ header**, so the original bearer-header plan isn't directly supported. Two ways
to honour the "tight secret, single user" intent:
- **Recommended — authless connector + secret in the URL path.** Register the
connector URL as `https://food.jihakuz.xyz/mcp/<long-random-secret>/`. To
connector URL as `https://food.tomflux.xyz/mcp/<long-random-secret>/`. To
claude.ai it's an authless server; in reality nginx proxies *only* that exact
secret prefix to the FastMCP service and 404s everything else. The URL **is**
the bearer-equivalent — it matches Tom's "a tight enough secret, I'm the only
@@ -178,11 +178,17 @@ A small package in this repo, its own process — not bolted into Django:
```
mcp_server/
__init__.py
__main__.py # FastMCP app; runs streamable-HTTP on 127.0.0.1:8765
__main__.py # entrypoint: from .server import main; main()
server.py # FastMCP app + the @mcp.tool definitions (§5) + main()
client.py # thin httpx client around the Django API (caine token)
tools.py # the @mcp.tool definitions from §5
tests.py # FoodClient tests via httpx.MockTransport (no network)
```
`main()` runs streamable-HTTP at path `/` on `127.0.0.1:8765`
(`mcp.run(transport="http", host, port, path="/")`); nginx maps
`/mcp/<secret>/` → that root. Client tests run with
`python -m unittest mcp_server.tests` (needs the `mcp` dep group).
- Deps via a uv group so they only install where needed:
`[dependency-groups] mcp = ["fastmcp", "httpx"]`. Deploy with
`uv sync --group mcp` on the box that runs the service.
@@ -204,10 +210,10 @@ All out of git; the systemd unit loads the same `.env` the Django service uses.
1. `uv sync --group mcp` on the box.
2. Add the three env vars to `/var/lib/food/.env`; generate the bearer token.
3. Install `deploy/food-mcp.service`, `daemon-reload`, start, enable.
4. Add the `location /mcp` block to the `food.jihakuz.xyz` nginx config
4. Add the `location /mcp` block to the `food.tomflux.xyz` nginx config
(`proxy_buffering off`, long read timeout), `nginx -t`, reload.
5. In claude.ai, add a custom connector pointing at the **secret URL**
`https://food.jihakuz.xyz/mcp/<secret>/` (authless connector — see §4).
`https://food.tomflux.xyz/mcp/<secret>/` (authless connector — see §4).
6. Smoke test: in cooking mode, "what's in my pantry?" → Claude calls
`get_pantry` and lists real items.
+6
View File
@@ -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.
"""
+4
View File
@@ -0,0 +1,4 @@
from .server import main
if __name__ == "__main__":
main()
+75
View File
@@ -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
+168
View File
@@ -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()
+85
View File
@@ -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()
+8
View File
@@ -14,3 +14,11 @@ dependencies = [
# No [build-system]: this is an application, not an installable package, so uv
# treats it as a virtual project and won't try to build/install it.
# Optional deps for the MCP "ai service" (deploy with `uv sync --group mcp`).
# Kept out of the default set so the Django service stays lean.
[dependency-groups]
mcp = [
"fastmcp>=2",
"httpx>=0.27",
]