1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
|
"""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()
|
