Add MCP server exposing diet status + a gated log_meal
Some checks are pending
pre-commit / pre-commit (push) Waiting to run
Tests / test (3.10) (push) Waiting to run
Tests / test (3.11) (push) Waiting to run
Tests / test (3.12) (push) Waiting to run

Expose diet_guard's read surface and one gated write action as typed MCP
tools (diet_guard._mcp) so Claude Code and its subagents can query today's
intake and -- with explicit confirmation -- log a meal without shelling out
to the interactive gate CLI.

- _mcp.py: FastMCP stdio server. Read tools get_status, list_today,
  get_slots expose today's CONSUMED kcal/macros and the qualitative
  consumption band only -- never the raw daily budget, remaining number,
  body weight, or protein target (those would back out the deliberately
  hidden budget). Gated write log_meal previews unless confirm=True and
  degrades gracefully on write failure. Nutrition is grouped into a Macros
  model (kcal + optional protein/carbs/fat) to keep the tool under the
  arg-count limits with no lint suppression, mirroring the CLI's grouping.
  Logs to stderr; no secret (HMAC key / sync token) is ever returned.
- .mcp.json + scripts/setup_mcp.sh: registration via a dedicated
  ~/.venvs/diet-guard-mcp interpreter; mcp is an optional-dep, kept out of
  the systemd/CLI system-python path.
- 100% coverage of the new module; verified end-to-end over stdio.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LQGQj8Ne6TriBExv91nG8v
This commit is contained in:
Krzysztof Rudnicki 2026-07-10 21:36:46 +02:00
parent a121da211d
commit 06999274d7
6 changed files with 575 additions and 0 deletions

10
.mcp.json Normal file
View File

@ -0,0 +1,10 @@
{
"mcpServers": {
"diet-guard": {
"type": "stdio",
"command": "${HOME}/.venvs/diet-guard-mcp/bin/python",
"args": ["-m", "diet_guard._mcp"],
"env": {}
}
}
}

View File

@ -25,6 +25,41 @@ python -m diet_guard gate --demo # test the lock window (safe, closeable)
The timer runs the gate automatically every ~30 minutes; no manual The timer runs the gate automatically every ~30 minutes; no manual
invocation is needed once installed. invocation is needed once installed.
## MCP server
An optional [Model Context Protocol](https://modelcontextprotocol.io) server
lets an MCP client (Claude Code and its subagents) query today's intake and log
meals as typed tools instead of shelling out to the CLI. It runs in its own
venv (`~/.venvs/diet-guard-mcp`) so the `mcp` SDK never touches the
dependency-light system/CLI python path.
Tools:
- `get_status` — today's consumed kcal + macros, the consumption *band*
(`on track` / `approaching limit` / `OVER BUDGET`, or none when no budget is
sealed), plus due / logged / current meal slots.
- `list_today` — today's logged meals (description, kcal, macros, source, slot).
- `get_slots` — the day's fixed meal slots and the current one.
- `log_meal(description, grams=, kcal=, protein=, carbs=, fat=, slot=,
confirm=False)` — **gated write**: with `confirm=False` it previews the
resolved nutrition and target slot and mutates nothing; only `confirm=True`
appends the entry. The gate is intrinsic — never add this tool to a
permission allowlist, since a client's auto-approve mode would otherwise
bypass the human confirmation.
The read tools deliberately never expose the raw daily budget number (only the
qualitative band), nor the sealed `.budget` file, the shared HMAC key, or the
sync token — the budget stays hidden by design.
Setup:
```bash
bash scripts/setup_mcp.sh # create the venv and install diet_guard[mcp]
```
Then restart Claude Code in this repo and approve the project MCP server
(registered in `.mcp.json`).
## Development ## Development
```bash ```bash

284
diet_guard/_mcp.py Normal file
View File

@ -0,0 +1,284 @@
"""MCP (Model Context Protocol) server for diet_guard.
Exposes diet_guard's read surface and one *gated* write action as typed MCP
tools, so an MCP client (Claude Code and its subagents) can query today's
intake and -- with explicit confirmation -- log a meal without shelling out to
the interactive CLI.
Run via the dedicated venv that has the ``mcp`` extra installed::
~/.venvs/diet-guard-mcp/bin/python -m diet_guard._mcp
(see ``scripts/setup_mcp.sh`` and the repo-root ``.mcp.json``).
Safety invariants (do not break when adding tools):
* **stdout is the JSON-RPC channel.** This module and every function a tool
calls must never write to stdout. All logging is routed to STDERR below,
and tools call only stdout-free leaf helpers (never the ``_cmd_*`` /
``main`` CLI handlers, which write to stdout, read stdin, and ``sys.exit``).
* **The daily budget number never leaves.** No tool returns the raw budget,
the remaining-budget number, the stored body weight, the protein target, or
the sealed ``.budget`` file: those would let a caller back out the
deliberately-hidden budget. Read tools expose only today's *consumed* kcal /
macros and the qualitative :func:`consumption_band` string.
* **No secret ever leaves.** There is no tool that reads the shared HMAC key,
the sync token, or any file under ``/etc``.
* **Writes are gated.** The write tool defaults to a dry-run preview and
mutates only when ``confirm=True``. It must never be added to a permission
allowlist (a subagent could then bypass the human), and it never raises --
a failed write degrades to ``{"ok": false, ...}``.
"""
from __future__ import annotations
import logging
import sys
from typing import Any
from mcp.server.fastmcp import FastMCP
from pydantic import BaseModel
from diet_guard._budget import BudgetError
from diet_guard._gate import due_slots
from diet_guard._resolve import ManualMacros, resolve_nutrition
from diet_guard._slots import current_slot, day_slots, slot_label
from diet_guard._state import (
consumption_band,
logged_slots_today,
now_local,
today_entries,
today_total_kcal,
today_total_macros,
)
from diet_guard._state import (
log_meal as record_meal,
)
# Log to STDERR only -- STDOUT carries the MCP JSON-RPC protocol frames, so a
# single stray stdout write would corrupt the stream and kill the session.
logging.basicConfig(
level=logging.INFO,
stream=sys.stderr,
format="%(asctime)s [%(levelname)s] diet-guard-mcp: %(message)s",
)
logger = logging.getLogger(__name__)
mcp = FastMCP("diet-guard")
def _entry_view(entry: dict[str, object]) -> dict[str, Any]:
"""Project a stored log entry to a safe, JSON-friendly summary.
Deliberately drops the ``hmac`` field and any unknown keys: the signature
is internal integrity metadata, not something a client needs.
Args:
entry: A stored food-log entry.
Returns:
A flat dict of the display-relevant fields.
"""
return {
"time": entry.get("time"),
"desc": entry.get("desc"),
"kcal": entry.get("kcal"),
"protein_g": entry.get("protein_g"),
"carbs_g": entry.get("carbs_g"),
"fat_g": entry.get("fat_g"),
"grams": entry.get("grams"),
"source": entry.get("source"),
"slot": entry.get("slot"),
}
# ──────────────────────────────────────────────────────────────
# Read tools (consumed intake + qualitative band only; NEVER the budget number)
# ──────────────────────────────────────────────────────────────
@mcp.tool()
def get_status() -> dict[str, Any]:
"""Return today's intake status without ever revealing the daily budget.
Reports the calories and macros *consumed* so far, the qualitative
:func:`consumption_band` (``"on track"`` / ``"approaching limit"`` /
``"OVER BUDGET"``, or ``None`` when no budget is sealed yet), and the
meal-slot picture (which slots are due, which are already logged, and the
current slot). The raw budget number is intentionally withheld -- only the
band is exposed, mirroring how the CLI status shows a label to an automated
caller rather than the anchor number.
"""
try:
band: str | None = consumption_band()
except BudgetError:
# No budget sealed (or a broken seal): surface the absence, not a number.
band = None
return {
"consumed_kcal": today_total_kcal(),
"consumed_macros_g": _macros_dict(today_total_macros()),
"consumption_band": band,
"budget_initialized": band is not None,
"due_slots": [slot_label(slot) for slot in due_slots()],
"logged_slots": sorted(logged_slots_today()),
"current_slot": current_slot(now_local()),
}
@mcp.tool()
def list_today() -> dict[str, Any]:
"""List today's logged meals (valid entries only), newest last.
Returns the per-entry description, calories, macros, portion, source, and
slot -- the same data the CLI ``status`` listing renders, minus the internal
HMAC signature.
"""
entries = today_entries()
return {
"count": len(entries),
"entries": [_entry_view(entry) for entry in entries],
}
@mcp.tool()
def get_slots() -> dict[str, Any]:
"""Return the day's fixed meal slots and which one is current.
Pure schedule information (08:00 / 12:00 / 16:00 / 20:00 by default) with no
budget or intake data attached.
"""
return {
"day_slots": [
{"hour": slot, "label": slot_label(slot)} for slot in day_slots()
],
"current_slot": current_slot(now_local()),
}
# ──────────────────────────────────────────────────────────────
# Gated write tool (preview unless confirm=True; NEVER allowlist this)
# ──────────────────────────────────────────────────────────────
class Macros(BaseModel):
"""Manually-entered nutrition for one meal.
Bundled into a single parameter so ``log_meal`` stays under the arg-count
limits without any lint suppression (mirrors the ``_ManualMacros`` grouping
the CLI uses for the same reason). ``kcal`` is required; when this is
supplied to ``log_meal``, the food-bank / Open Food Facts lookups are
skipped and the meal is logged exactly as given (fully offline).
"""
kcal: float
protein: float = 0.0
carbs: float = 0.0
fat: float = 0.0
@mcp.tool()
def log_meal(
description: str,
grams: float | None = None,
macros: Macros | None = None,
slot: int | None = None,
*,
confirm: bool = False,
) -> dict[str, Any]:
"""Resolve and (on confirm) log a meal to today's food log (gated write).
With ``confirm=False`` (the default) this performs **no** mutation: it
resolves the nutrition and reports the slot the entry would satisfy, so the
caller can review before applying. Call again with ``confirm=True`` to
actually append the entry. Nutrition is resolved exactly as the CLI's
``ate`` does -- manual ``macros`` first (when given), then the local food
bank / staple table, then Open Food Facts -- so passing ``macros`` keeps the
write fully offline and deterministic.
The write degrades gracefully: if the food log cannot be written it returns
``{"ok": false, ...}`` rather than raising, so the stdio server survives.
A missing shared HMAC key is not an error -- :func:`record_meal` then stores
the entry unsigned (and it is still accepted on read on a keyless system).
Args:
description: Free-text meal description, e.g. ``"big mac"``.
grams: Portion size in grams (rescales every nutrition source).
macros: Manually-entered nutrition (``kcal`` plus optional
protein/carbs/fat). When given, lookups are skipped and the meal is
logged exactly as specified.
slot: The meal-slot hour to satisfy; defaults to the current elapsed
slot (may be ``None`` early in the day, logging calories only).
confirm: Set ``True`` to actually append the entry; otherwise preview.
Returns:
A preview or applied result dict; ``{"ok": false, "reason": ...}`` when
the food cannot be resolved or the log cannot be written.
"""
manual_macros = (
ManualMacros(
kcal=macros.kcal,
protein=macros.protein,
carbs=macros.carbs,
fat=macros.fat,
)
if macros is not None
else None
)
nutrition = resolve_nutrition(description, grams=grams, manual_macros=manual_macros)
if nutrition is None:
return {
"ok": False,
"reason": (
f'could not resolve "{description}" from the food bank, staples, '
"or Open Food Facts. Pass kcal=<number> to log it manually."
),
}
target_slot = slot if slot is not None else current_slot(now_local())
resolved = {
"kcal": nutrition.kcal,
"protein_g": nutrition.protein_g,
"carbs_g": nutrition.carbs_g,
"fat_g": nutrition.fat_g,
"grams": nutrition.grams,
"source": nutrition.source,
}
if not confirm:
return {
"ok": True,
"preview": True,
"action": "log_meal",
"description": description,
"resolved": resolved,
"target_slot": target_slot,
"confirm_required": True,
}
try:
entry = record_meal(description, nutrition, target_slot)
except OSError as exc: # never crash the stdio server on a write failure
logger.warning("log_meal write failed: %s", exc)
return {"ok": False, "reason": "could not write the food log."}
logger.info("log_meal applied: %s (%g kcal)", description, nutrition.kcal)
return {
"ok": True,
"applied": True,
"action": "log_meal",
"description": description,
"logged": resolved,
"target_slot": target_slot,
"signed": "hmac" in entry,
}
def _macros_dict(macros: tuple[float, float, float]) -> dict[str, float]:
"""Return a ``(protein, carbs, fat)`` triple as a labelled dict."""
protein, carbs, fat = macros
return {"protein": protein, "carbs": carbs, "fat": fat}
def main() -> None:
"""Run the MCP server over stdio (STDOUT = JSON-RPC, STDERR = logs)."""
logger.info("Starting diet-guard MCP server (python=%s)", sys.executable)
mcp.run() # pragma: no cover
if __name__ == "__main__":
main()

View File

@ -0,0 +1,195 @@
"""Tests for the MCP server tools in ``_mcp``.
Leaf helpers are patched at the ``_mcp`` module namespace (where they are
imported), keeping each tool's own logic under test while isolating the
already-tested state/resolution functions. No test touches the real food log,
budget seal, or shared HMAC key.
"""
from __future__ import annotations
from unittest.mock import patch
from diet_guard import _mcp
from diet_guard._budget import BudgetNotInitializedError
from diet_guard._estimator import Nutrition
def _nutrition(kcal: float = 250.0) -> Nutrition:
return Nutrition(
kcal=kcal,
protein_g=10.0,
carbs_g=20.0,
fat_g=5.0,
grams=150.0,
source="manual",
)
class TestGetStatus:
def test_band_and_intake(self) -> None:
with (
patch.object(_mcp, "today_total_kcal", return_value=640.0),
patch.object(_mcp, "today_total_macros", return_value=(30.0, 50.0, 12.0)),
patch.object(_mcp, "consumption_band", return_value="on track"),
patch.object(_mcp, "due_slots", return_value=(12,)),
patch.object(_mcp, "logged_slots_today", return_value={8}),
patch.object(_mcp, "current_slot", return_value=12),
patch.object(_mcp, "now_local"),
patch.object(_mcp, "slot_label", return_value="12:00"),
):
out = _mcp.get_status()
assert out["consumed_kcal"] == 640.0
assert out["consumed_macros_g"] == {"protein": 30.0, "carbs": 50.0, "fat": 12.0}
assert out["consumption_band"] == "on track"
assert out["budget_initialized"] is True
assert out["due_slots"] == ["12:00"]
assert out["logged_slots"] == [8]
assert out["current_slot"] == 12
def test_no_budget_hides_band(self) -> None:
with (
patch.object(_mcp, "today_total_kcal", return_value=0.0),
patch.object(_mcp, "today_total_macros", return_value=(0.0, 0.0, 0.0)),
patch.object(
_mcp,
"consumption_band",
side_effect=BudgetNotInitializedError(),
),
patch.object(_mcp, "due_slots", return_value=()),
patch.object(_mcp, "logged_slots_today", return_value=set()),
patch.object(_mcp, "current_slot", return_value=None),
patch.object(_mcp, "now_local"),
):
out = _mcp.get_status()
assert out["consumption_band"] is None
assert out["budget_initialized"] is False
assert out["due_slots"] == []
assert out["logged_slots"] == []
class TestListToday:
def test_projects_entries_and_drops_hmac(self) -> None:
entries = [
{
"time": "2026-07-10T12:30:00",
"desc": "big mac",
"kcal": 550.0,
"protein_g": 25.0,
"carbs_g": 45.0,
"fat_g": 30.0,
"grams": 215.0,
"source": "openfoodfacts: Big Mac",
"slot": 12,
"hmac": "deadbeef",
},
]
with patch.object(_mcp, "today_entries", return_value=entries):
out = _mcp.list_today()
assert out["count"] == 1
view = out["entries"][0]
assert view["desc"] == "big mac"
assert view["kcal"] == 550.0
assert view["slot"] == 12
assert "hmac" not in view
def test_empty(self) -> None:
with patch.object(_mcp, "today_entries", return_value=[]):
out = _mcp.list_today()
assert out == {"count": 0, "entries": []}
class TestGetSlots:
def test_lists_slots_and_current(self) -> None:
with (
patch.object(_mcp, "day_slots", return_value=(8, 12, 16, 20)),
patch.object(_mcp, "current_slot", return_value=16),
patch.object(_mcp, "now_local"),
patch.object(_mcp, "slot_label", side_effect=lambda s: f"{s:02d}:00"),
):
out = _mcp.get_slots()
assert out["current_slot"] == 16
assert out["day_slots"][0] == {"hour": 8, "label": "08:00"}
assert len(out["day_slots"]) == 4
class TestLogMealGate:
def test_unresolvable_returns_reason(self) -> None:
with patch.object(_mcp, "resolve_nutrition", return_value=None):
out = _mcp.log_meal("mystery stew")
assert out["ok"] is False
assert "could not resolve" in out["reason"]
def test_preview_manual_macros_does_not_mutate(self) -> None:
with (
patch.object(
_mcp, "resolve_nutrition", return_value=_nutrition()
) as resolve,
patch.object(_mcp, "ManualMacros") as manual,
patch.object(_mcp, "record_meal") as record,
):
out = _mcp.log_meal(
"protein shake",
macros=_mcp.Macros(kcal=180.0, protein=30.0, carbs=5.0, fat=2.0),
slot=12,
)
manual.assert_called_once_with(kcal=180.0, protein=30.0, carbs=5.0, fat=2.0)
resolve.assert_called_once_with(
"protein shake", grams=None, manual_macros=manual.return_value
)
record.assert_not_called()
assert out["preview"] is True
assert out["target_slot"] == 12
assert out["resolved"]["kcal"] == 250.0
assert out["confirm_required"] is True
def test_preview_no_manual_defaults_slot_to_current(self) -> None:
with (
patch.object(_mcp, "resolve_nutrition", return_value=_nutrition()),
patch.object(_mcp, "ManualMacros") as manual,
patch.object(_mcp, "current_slot", return_value=8),
patch.object(_mcp, "now_local"),
patch.object(_mcp, "record_meal") as record,
):
out = _mcp.log_meal("apple")
manual.assert_not_called()
record.assert_not_called()
assert out["preview"] is True
assert out["target_slot"] == 8
def test_confirm_applies_signed(self) -> None:
entry = {"desc": "apple", "hmac": "sig"}
with (
patch.object(_mcp, "resolve_nutrition", return_value=_nutrition()),
patch.object(_mcp, "record_meal", return_value=entry) as record,
):
out = _mcp.log_meal("apple", slot=8, confirm=True)
record.assert_called_once()
assert out["applied"] is True
assert out["signed"] is True
assert out["target_slot"] == 8
def test_confirm_applies_unsigned(self) -> None:
entry = {"desc": "apple"}
with (
patch.object(_mcp, "resolve_nutrition", return_value=_nutrition()),
patch.object(_mcp, "record_meal", return_value=entry),
):
out = _mcp.log_meal("apple", slot=8, confirm=True)
assert out["applied"] is True
assert out["signed"] is False
def test_confirm_write_failure_returns_gracefully(self) -> None:
with (
patch.object(_mcp, "resolve_nutrition", return_value=_nutrition()),
patch.object(_mcp, "record_meal", side_effect=OSError("disk full")),
):
out = _mcp.log_meal("apple", slot=8, confirm=True)
assert out["ok"] is False
assert "could not write" in out["reason"]
def test_main_runs_stdio_server() -> None:
with patch.object(_mcp.mcp, "run") as run:
_mcp.main()
run.assert_called_once_with()

View File

@ -8,6 +8,14 @@ dependencies = [
"requests>=2.31.0", "requests>=2.31.0",
] ]
# Optional extras. The `mcp` group is ONLY needed by the MCP server
# (diet_guard._mcp), which Claude Code spawns via a dedicated venv (see
# scripts/setup_mcp.sh / .mcp.json). It is deliberately NOT a core dependency
# so the systemd/CLI system-python path stays dependency-light and never has
# to import the MCP SDK.
[project.optional-dependencies]
mcp = ["mcp>=1.9"]
# Without this, setuptools' flat-layout auto-discovery treats the Flutter # Without this, setuptools' flat-layout auto-discovery treats the Flutter
# app/ directory as a second top-level package and refuses to build. # app/ directory as a second top-level package and refuses to build.
[tool.setuptools.packages.find] [tool.setuptools.packages.find]

43
scripts/setup_mcp.sh Executable file
View File

@ -0,0 +1,43 @@
#!/bin/bash
# ============================================================================
# Set up the dedicated virtualenv that hosts the diet_guard MCP server.
# Claude Code spawns this interpreter (see the repo-root .mcp.json) to run
# `python -m diet_guard._mcp` over stdio.
#
# The MCP SDK (`mcp`) lives ONLY in this venv -- it is an optional dependency,
# deliberately kept out of the system/CLI/systemd python path (see CLAUDE.md's
# "Production dependency installation" note: the gate service runs
# /usr/bin/python directly and must stay dependency-light). Both `mcp` and
# `diet_guard` must be importable by this one interpreter or the MCP server
# fails to start silently.
#
# Idempotent: safe to re-run to pick up dependency changes.
# ============================================================================
set -euo pipefail
readonly SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
readonly REPO_DIR="$(cd "${SCRIPT_DIR}/.." && pwd)"
readonly VENV_DIR="${HOME}/.venvs/diet-guard-mcp"
main() {
echo "== Setting up MCP venv at ${VENV_DIR} =="
if [[ ! -d "${VENV_DIR}" ]]; then
python3 -m venv "${VENV_DIR}"
fi
"${VENV_DIR}/bin/pip" install --quiet --upgrade pip
"${VENV_DIR}/bin/pip" install --quiet -e "${REPO_DIR}[mcp]"
echo "== Verifying imports =="
"${VENV_DIR}/bin/python" -c \
"import mcp, diet_guard; print('mcp + diet_guard import OK')"
echo
echo "Done. The server is registered via ${REPO_DIR}/.mcp.json"
echo "Restart Claude Code in this repo and approve the project MCP server prompt."
}
main "$@"