"""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 via this interface.** No tool returns the raw budget, the remaining-budget number, the stored body weight, the protein target, or the ``.budget`` file: an automated caller should not be handed the exact number to reason about or help game, even though the number itself is freely visible to the human via the CLI/GUI and the phone app. 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_for_log, 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 has been set 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 set (or a corrupt file): 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= to log it manually." ), } target_slot = slot if slot is not None else slot_for_log(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()