"""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 = [] for it in items: entry = { "ingredient_name": it.get("name") or it.get("ingredient_name"), "location": it.get("location", "fridge"), } # Only forward optional fields when set, so the API applies its defaults. if it.get("quantity") is not None: entry["quantity"] = it["quantity"] if it.get("unit"): entry["unit"] = it["unit"] payload.append(entry) 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()