food/mcp_server/server.py

"""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/<secret>/ -> here.
    mcp.run(transport="http", host=host, port=port, path="/")


if __name__ == "__main__":
    main()