"""Small, confirmation-friendly MCP facade over the Mealie API.""" from __future__ import annotations import json import os import re from decimal import Decimal, ROUND_HALF_UP from pathlib import Path from typing import Any import httpx from mcp.server.mcpserver import MCPServer from .normalize import ( clean_title, has_extraction_failure, ingredient_display_lines, looks_mangled, normalize_for_parser, scale_line, ) from .verify import verify_recipe as run_verification BASE_URL = os.getenv("MEALIE_BASE_URL", "https://recept.famfallman.com").rstrip("/") TOKEN = os.getenv("MEALIE_API_TOKEN") UA = os.getenv("MEALIE_USER_AGENT", "Mealie-MCP/0.1") HA_BASE_URL = (os.getenv("HA_BASE_URL") or "").rstrip("/") HA_TOKEN = os.getenv("HA_TOKEN") SHOPPING_LIST_ENTITY = os.getenv("HA_SHOPPING_LIST_ENTITY", "todo.ourgroceries_shoppinglista") RESOURCE_DIR = Path(__file__).parent / "resources" mcp = MCPServer("mealie", version="0.1.0") class MealieAuthError(RuntimeError): """The Mealie token is missing, expired, or wrong (HTTP 401).""" class MealieBlockedError(RuntimeError): """The edge/WAF rejected the client fingerprint (HTTP 403, Cloudflare 1010).""" def _client() -> httpx.Client: if not TOKEN: raise RuntimeError("MEALIE_API_TOKEN is not set") return httpx.Client( base_url=BASE_URL, headers={"Authorization": f"Bearer {TOKEN}", "User-Agent": UA}, timeout=30, ) def _raise_for_status(response: httpx.Response) -> None: """Map Mealie/Cloudflare failures to distinguishable errors. A 403 with Cloudflare's ``error code: 1010`` is a blocked client fingerprint, not a bad token — conflating the two sends debugging in the wrong direction. """ if response.status_code == 401: raise MealieAuthError( "401 from Mealie: the API token is missing, expired, or wrong. " "Create a new one at /user/profile/api-tokens." ) if response.status_code == 403 and "1010" in response.text: raise MealieBlockedError( f"403 from the edge (Cloudflare error 1010) for User-Agent {UA!r}. " "This is a WAF rule on the client fingerprint, not a Mealie auth problem. " "Set MEALIE_USER_AGENT to a non-default value." ) response.raise_for_status() def _request(method: str, path: str, **kwargs: Any) -> Any: with _client() as client: response = client.request(method, path, **kwargs) _raise_for_status(response) if not response.content: return None return response.json() # --- Read ------------------------------------------------------------------- @mcp.tool() def check_auth() -> dict[str, Any]: """Verify the Mealie token and distinguish auth failures from WAF blocks.""" try: user = _request("GET", "/api/users/self") except MealieAuthError as exc: return {"ok": False, "reason": "auth", "detail": str(exc)} except MealieBlockedError as exc: return {"ok": False, "reason": "waf", "detail": str(exc)} return { "ok": True, "base_url": BASE_URL, "user": user.get("username"), "user_agent": UA, } @mcp.tool() def search_recipes(query: str, limit: int = 10) -> list[dict[str, Any]]: """Search Mealie recipes by text.""" data = _request("GET", "/api/recipes", params={"search": query, "perPage": limit}) return data.get("items", data if isinstance(data, list) else []) @mcp.tool() def resolve_foods(names: list[str]) -> dict[str, Any]: """Map food names to the Mealie food records they refer to. ``/api/recipes/suggestions`` takes food UUIDs, not names, so free-text has to be resolved first. Returns the best match per name plus the alternatives, so the caller can tell that `kyckling` resolved to `kycklingbröst`. """ resolved: dict[str, Any] = {} unresolved: list[str] = [] for name in names: data = _request("GET", "/api/foods", params={"search": name, "perPage": 5}) items = data.get("items", data if isinstance(data, list) else []) if not items: unresolved.append(name) continue exact = next((i for i in items if (i.get("name") or "").lower() == name.lower()), None) best = exact or items[0] resolved[name] = { "id": best.get("id"), "name": best.get("name"), "exact": exact is not None, "alternatives": [i.get("name") for i in items if i.get("id") != best.get("id")], } return {"resolved": resolved, "unresolved": unresolved} @mcp.tool() def suggest_recipes(foods: list[str], limit: int = 10, max_missing_foods: int = 5) -> dict[str, Any]: """Find recipes matching ingredients on hand, given plain food names. The endpoint expects food UUIDs as repeated query parameters; passing names or a comma-joined string returns 422. Names are resolved first, and anything that could not be resolved is reported rather than silently dropped. """ lookup = resolve_foods(foods) ids = [entry["id"] for entry in lookup["resolved"].values()] if not ids: return {"items": [], "matched_foods": {}, "unresolved_foods": lookup["unresolved"]} data = _request("GET", "/api/recipes/suggestions", params=[ *[("foods", food_id) for food_id in ids], ("limit", limit), ("maxMissingFoods", max_missing_foods), ("includeFoodsOnHand", "true"), ]) items = data.get("items", data if isinstance(data, list) else []) return { "items": items, "matched_foods": {name: e["name"] for name, e in lookup["resolved"].items()}, "unresolved_foods": lookup["unresolved"], } @mcp.tool() def get_recipe(slug_or_id: str) -> dict[str, Any]: """Read complete recipe details.""" return _request("GET", f"/api/recipes/{slug_or_id}") @mcp.tool() def list_organizers() -> dict[str, list[str]]: """List existing categories and tags so imports reuse the library's vocabulary. Read this before tagging anything: the library has settled naming (`Källa: Köket`, `📅 Vardag`, `💨 Airfryer`) and inventing parallel names fragments retrieval. """ def names(path: str) -> list[str]: data = _request("GET", path, params={"perPage": 200}) items = data.get("items", data if isinstance(data, list) else []) return sorted(item.get("name", "") for item in items) return { "categories": names("/api/organizers/categories"), "tags": names("/api/organizers/tags"), } @mcp.tool() def find_by_source_url(url: str) -> list[dict[str, Any]]: """Look for an existing recipe with this source URL, to avoid duplicate imports. ``orgURL`` is the reliable duplicate key; imported titles and slugs legitimately differ from the URL text, so matching on those produces false negatives. """ data = _request("GET", "/api/recipes", params={"perPage": 500}) items = data.get("items", data if isinstance(data, list) else []) return [item for item in items if (item.get("orgURL") or "").rstrip("/") == url.rstrip("/")] @mcp.tool() def scale_ingredients(slug_or_id: str, servings: int) -> dict[str, Any]: """Return readable ingredient lines scaled to a requested serving count. Amounts are read from the display text, not the structured ``quantity`` field: unparsed recipes store ``quantity: 0.0`` while the real amount lives in the line. Lines with no leading amount (`salt`, `smör (att steka i)`) are returned untouched and listed in ``unscaled`` — never silently presented as scaled. """ recipe = get_recipe(slug_or_id) original = recipe.get("recipeServings") or recipe.get("recipeYieldQuantity") if not original: # recipeYield is free text ("4 portioner"); take a leading number if present. match = re.match(r"\s*(\d+(?:[.,]\d+)?)", str(recipe.get("recipeYield") or "")) original = match.group(1).replace(",", ".") if match else None if not original or Decimal(str(original)) == 0: raise ValueError( "Recipe has no serving metadata (recipeServings, recipeYieldQuantity, " "or a numeric recipeYield); scaling would be a guess" ) factor = Decimal(servings) / Decimal(str(original)) scaled_lines: list[str] = [] unscaled: list[str] = [] for line in ingredient_display_lines(recipe): text, was_scaled = scale_line(line, factor) scaled_lines.append(text) if not was_scaled and line.strip(): unscaled.append(line) return { "recipe": recipe.get("name"), "original_servings": float(Decimal(str(original))), "servings": servings, "factor": float(round(factor, 3)), "ingredients": scaled_lines, "unscaled": unscaled, } # --- Import ----------------------------------------------------------------- def _import_report(recipe: dict[str, Any]) -> dict[str, Any]: """Flag extraction problems the caller must fix before claiming success.""" title = recipe.get("name") or "" cleaned = clean_title(title) return { "slug": recipe.get("slug"), "title": title, "suggested_title": cleaned if cleaned != title else None, "extraction_failed": has_extraction_failure(recipe), "has_image_field": bool(recipe.get("image")), "ingredient_count": len(recipe.get("recipeIngredient") or []), "instruction_count": len(recipe.get("recipeInstructions") or []), } @mcp.tool() def import_recipe_url(url: str, check_duplicates: bool = True) -> dict[str, Any]: """Import a recipe page through Mealie's URL scraper. Returns the created recipe plus an extraction report. A non-empty response is not success: check ``extraction_failed`` and ``suggested_title`` before reporting, and run the Swedish normalization pass afterwards. """ if check_duplicates: existing = find_by_source_url(url) if existing: return { "imported": False, "reason": "duplicate", "existing": [{"slug": r.get("slug"), "name": r.get("name")} for r in existing], } created = _request("POST", "/api/recipes/create/url", json={"url": url}) recipe = get_recipe(created) if isinstance(created, str) else created return {"imported": True, "recipe": recipe, "report": _import_report(recipe)} @mcp.tool() def import_recipe_text(text: str, source_url: str | None = None) -> dict[str, Any]: """Import a normalized recipe text block (OCR, transcript, or hand-built Swedish). Use this when the source is not a clean single-recipe page — an editorial roundup, a YouTube description, or a social caption. Build the text in Swedish with ``Titel`` / ``Portioner`` / ``Ingredienser`` / ``Gör så här`` sections. """ payload: dict[str, Any] = {"data": text} if source_url: payload["url"] = source_url created = _request("POST", "/api/recipes/create/html-or-json", json=payload) recipe = get_recipe(created) if isinstance(created, str) else created return {"imported": True, "recipe": recipe, "report": _import_report(recipe)} @mcp.tool() def import_recipe_image(image_path: str) -> dict[str, Any]: """Import a recipe from a photo or screenshot via Mealie's image parser.""" path = Path(image_path) if not path.is_file(): raise ValueError(f"No such image file: {image_path}") with _client() as client, path.open("rb") as handle: response = client.post("/api/recipes/create/image", files={"images": (path.name, handle)}) _raise_for_status(response) created = response.json() recipe = get_recipe(created) if isinstance(created, str) else created return {"imported": True, "recipe": recipe, "report": _import_report(recipe)} # --- Write ------------------------------------------------------------------ @mcp.tool() def patch_recipe(slug_or_id: str, patch: dict[str, Any]) -> dict[str, Any]: """Apply an explicit Mealie recipe patch, then read the recipe back. Use this for tags and categories too — the bulk-action endpoints return 500 on this instance. Patch ``recipeCategory`` and ``tags`` directly. """ _request("PATCH", f"/api/recipes/{slug_or_id}", json=patch) return get_recipe(slug_or_id) @mcp.tool() def parse_ingredients(slug_or_id: str) -> dict[str, Any]: """Add structured food/unit/quantity data while keeping the Swedish display text. The parser is a structured-data helper only. Its ``display`` output mangles Swedish rows (``4,7 dl basmatiris`` came back as ``47⁄10 liter ...``), so the canonical human line is preserved and restored after parsing. Rows the parser still mangles keep readable text with structure cleared — legibility wins. """ recipe = get_recipe(slug_or_id) canonical = ingredient_display_lines(recipe) if not canonical: return {"parsed": False, "reason": "Recipe has no ingredient lines"} parsed = _request( "POST", "/api/parser/ingredients", json={"ingredients": [normalize_for_parser(line) for line in canonical], "parser": "nlp"}, ) ingredients = list(recipe.get("recipeIngredient") or []) degraded: list[str] = [] for index, result in enumerate(parsed or []): if index >= len(ingredients): break item = dict(ingredients[index]) confidence = (result or {}).get("ingredient") or {} item["quantity"] = confidence.get("quantity") item["unit"] = confidence.get("unit") item["food"] = confidence.get("food") # The canonical Swedish line always wins over parser-generated display text. item["display"] = canonical[index] item["note"] = canonical[index] if looks_mangled(canonical[index]): item["quantity"] = None item["unit"] = None item["food"] = None degraded.append(canonical[index]) ingredients[index] = item try: _request("PATCH", f"/api/recipes/{slug_or_id}", json={"recipeIngredient": ingredients}) except httpx.HTTPStatusError as exc: # A full recipeIngredient patch with nested parser objects can 500 here. # Preserve the readable import rather than corrupting the recipe. return { "parsed": False, "reason": f"Structured patch rejected by Mealie ({exc.response.status_code}); " "readable Swedish lines were left intact", "canonical_lines": canonical, } return { "parsed": True, "recipe": get_recipe(slug_or_id), "readability_fallback_rows": degraded, } @mcp.tool() def set_cover_image( slug_or_id: str, source_url: str | None = None, image_path: str | None = None, ) -> dict[str, Any]: """Set a recipe cover image by scraping a URL or uploading a local file. Local uploads must send the multipart ``extension`` field or the upload fails validation silently on this instance. When a scrape does not produce a visible cover (observed on Allrecipes), download the hero image and upload it instead. """ if not source_url and not image_path: raise ValueError("Provide either source_url or image_path") if image_path: path = Path(image_path) if not path.is_file(): raise ValueError(f"No such image file: {image_path}") extension = path.suffix.lstrip(".").lower() or "jpg" with _client() as client, path.open("rb") as handle: response = client.put( f"/api/recipes/{slug_or_id}/image", files={"image": (path.name, handle)}, data={"extension": extension}, ) _raise_for_status(response) method = f"upload ({extension})" else: _request("POST", f"/api/recipes/{slug_or_id}/image", json={"url": source_url}) method = "scrape" recipe = get_recipe(slug_or_id) return {"method": method, "image": recipe.get("image"), "slug": recipe.get("slug")} # --- Verify ----------------------------------------------------------------- @mcp.tool() def verify_recipe( slug_or_id: str, image_verified: bool = False, taxonomy_skipped_reason: str | None = None, ) -> dict[str, Any]: """Run the finished-import checklist and return pass/fail per check. Set ``image_verified`` only if you actually ran an image step — a non-empty ``image`` field does not prove the UI shows a cover. """ recipe = get_recipe(slug_or_id) return run_verification( recipe, image_verified=image_verified, taxonomy_skipped_reason=taxonomy_skipped_reason, ) # --- Home Assistant --------------------------------------------------------- @mcp.tool() def shopping_list_add(items: list[str]) -> dict[str, Any]: """Add ingredient lines to the Home Assistant shopping list. Each line must carry its quantity — the todo entity has no structured quantity field, so `400 g kycklinglårfilé`, not `kycklinglårfilé`. Duplicates are never silently removed; consolidation is a separate, explicit operation. """ if not HA_BASE_URL or not HA_TOKEN: raise RuntimeError("HA_BASE_URL and HA_TOKEN must be set to write the shopping list") added: list[str] = [] with httpx.Client( base_url=HA_BASE_URL, headers={"Authorization": f"Bearer {HA_TOKEN}", "Content-Type": "application/json"}, timeout=30, ) as client: for item in items: response = client.post( "/api/services/todo/add_item", json={"entity_id": SHOPPING_LIST_ENTITY, "item": item}, ) response.raise_for_status() added.append(item) return {"entity_id": SHOPPING_LIST_ENTITY, "added": added, "count": len(added)} # --- Resources and prompts -------------------------------------------------- def _resource(filename: str) -> str: return (RESOURCE_DIR / filename).read_text(encoding="utf-8") @mcp.resource("mealie://reference/family-taste-profile") def family_taste_profile() -> str: """Fredrik's household taste profile, for judging whether a recipe is a good fit.""" return _resource("family-taste-profile.md") @mcp.resource("mealie://reference/import-pitfalls") def import_pitfalls() -> str: """Verified post-import failure modes: image, parser, bulk-action, and decimal issues.""" return _resource("import-pitfalls.md") @mcp.resource("mealie://reference/koket-attribution") def koket_attribution() -> str: """How to preserve Köket author/programme attribution after import.""" return _resource("koket-attribution.md") @mcp.resource("mealie://reference/cloudflare-hostname") def cloudflare_hostname() -> str: """Why the external hostname returns 403/1010 and how to avoid it.""" return _resource("cloudflare-hostname.md") @mcp.prompt() def import_pipeline(source: str) -> str: """The full import order, so no step gets skipped.""" return f"""Import this source into Mealie: {source} Follow this order exactly. Skipping a step is how imports end up half-finished. 1. `check_auth` if you have not already this session. 2. `find_by_source_url` — stop and report if it already exists. 3. Import with `import_recipe_url` (single recipe page) or `import_recipe_text` (editorial roundup, transcript, caption). Read the returned report: if `extraction_failed` is true, rebuild the recipe manually instead of patching. 4. `list_organizers` to see the existing category/tag vocabulary. 5. `patch_recipe` with Swedish title, description, instructions, and ingredient lines, plus `orgURL`, `extras.author`, `extras.source`, `tags`, and `recipeCategory`. Apply `suggested_title` if the report offered one. 6. `set_cover_image` explicitly — do not trust a non-empty `image` field. 7. `parse_ingredients` to add structure while keeping the Swedish display text. 8. `verify_recipe` with `image_verified=True`. Do not report success until `complete` is true, or state exactly which checks failed and why. All stored values must be Swedish: title, description, ingredient names, instructions, and units (`msk`, `tsk`, `dl`, `krm`, `st`, `klyfta`). """ def main() -> None: mcp.run()