"""Small, confirmation-friendly MCP facade over the Mealie API.""" from __future__ import annotations import asyncio import html 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, food_matches_line, has_extraction_failure, ingredient_display_lines, line_remainder, looks_mangled, match_unit, normalize_for_parser, same_line, scale_line, unit_token, ) 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") READ_ONLY_TOOL_NAMES = frozenset( { "check_auth", "find_by_source_url", "get_recipe", "list_organizers", "resolve_foods", "scale_ingredients", "search_recipes", "suggest_recipes", "verify_recipe", } ) IMPORT_TOOL_NAMES = READ_ONLY_TOOL_NAMES | frozenset( { "import_recipe_url", "import_recipe_text", "import_recipe_image", "patch_recipe", "parse_ingredients", "set_cover_image", } ) 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. """ required = ("Titel", "Ingredienser", "Gör så här") missing = [ heading for heading in required if not re.search(rf"(?im)^\s*{re.escape(heading)}\s*:", text) ] if missing: raise ValueError( "Text import requires explicit Swedish section headings; missing: " + ", ".join(missing) ) 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. The stable recipe id is captured before the PATCH because changing ``name`` also changes Mealie's slug, making the old slug invalid for read-back. """ before = get_recipe(slug_or_id) stable_id = before.get("id") or slug_or_id _request("PATCH", f"/api/recipes/{slug_or_id}", json=patch) return get_recipe(stable_id) @mcp.tool() def delete_recipe(slug_or_id: str, confirm_slug: str) -> dict[str, Any]: """Delete exactly one recipe, with explicit slug confirmation and read-back proof. This is primarily the rollback path for failed imports and marked test recipes. The caller must first read the recipe and pass its exact stored slug as ``confirm_slug``. Names and ids are not accepted as confirmation. """ recipe = get_recipe(slug_or_id) stored_slug = recipe.get("slug") or "" if confirm_slug != stored_slug: raise ValueError( f"Deletion confirmation does not match stored slug {stored_slug!r}" ) with _client() as client: response = client.delete(f"/api/recipes/{stored_slug}") _raise_for_status(response) probe = client.get(f"/api/recipes/{stored_slug}") if probe.status_code != 404: raise RuntimeError( f"Delete returned {response.status_code}, but read-back returned " f"{probe.status_code} instead of 404" ) return { "deleted": True, "slug": stored_slug, "name": recipe.get("name"), "delete_status": response.status_code, "readback_status": probe.status_code, } def _unit_records() -> list[dict[str, Any]]: """The instance's own unit table — the authority for what `dl` and `msk` mean.""" data = _request("GET", "/api/units", params={"perPage": 200}) return data.get("items", data if isinstance(data, list) else []) def _vet_structure(line: str, ingredient: dict[str, Any], units: list[dict[str, Any]]) -> tuple[ Any, dict[str, Any] | None, dict[str, Any] | None, list[str] ]: """Keep only parser output the Swedish line actually supports. Verified against the live instance: the parser resolves `3 dl` to the `liter` record, swaps `kycklinglårfilé` for the existing food `kycklingfilé`, and invents foods with ``id: null`` — the last of which makes the recipe PATCH fail with a 500. Unsupported values are dropped, never guessed at, and every drop is reported so the caller can see what was not structured. """ quantity = ingredient.get("quantity") unit = ingredient.get("unit") food = ingredient.get("food") notes: list[str] = [] if looks_mangled(line): return None, None, None, ["stored line is already parser-damaged; left unstructured"] if unit: written = unit_token(line) confirmed = match_unit(written, units) if confirmed and confirmed.get("id") != unit.get("id"): notes.append( f"unit {unit.get('name')!r} corrected to {confirmed.get('name')!r} " f"from the written token {written!r}" ) unit = confirmed elif not confirmed: notes.append( f"unit {unit.get('name')!r} is not what the line writes " f"({written!r}); dropped" ) unit = None if food and not food.get("id"): # A food with no id makes Mealie 500 on the recipe patch (ValueError). notes.append(f"food {food.get('name')!r} does not exist in Mealie; dropped") food = None elif food and not food_matches_line(food.get("name") or "", line): notes.append(f"food {food.get('name')!r} is not the ingredient this line names; dropped") food = None return quantity, unit, food, notes def _unstructured_item(base: dict[str, Any], line: str) -> dict[str, Any]: """A row carrying only the human line — Mealie renders ``note`` verbatim.""" item = dict(base) item["quantity"] = 0 item["unit"] = None item["food"] = None item["note"] = line return item def _structured_item( base: dict[str, Any], line: str, quantity: Any, unit: Any, food: Any ) -> dict[str, Any]: """A row holding structure, with ``note`` reduced to the leftover text.""" item = dict(base) item["quantity"] = quantity item["unit"] = unit item["food"] = food item["note"] = line_remainder(line, unit_token(line) if unit else None, (food or {}).get("name") if food else None) return item @mcp.tool() def parse_ingredients(slug_or_id: str) -> dict[str, Any]: """Add structured food/unit/quantity data without changing any visible Swedish line. ``display`` is computed by Mealie from ``quantity unit food note`` and cannot be set through the API — writing it is silently ignored. Structuring a row therefore rewrites how it reads, and Mealie's rendering is not always the Swedish the recipe was written in (``800 g`` renders as ``800 gram``, ``0,25`` as ``¹/₄``, and plural foods render singular). So structure is applied, read back, and kept only for rows Mealie renders identically to the original line. Any row it would have reworded is restored unstructured, and reported in ``left_unstructured`` with the text Mealie wanted to show. The visible line is never changed by this tool. """ 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"}, ) units = _unit_records() base = list(recipe.get("recipeIngredient") or []) candidates = [_unstructured_item(item, line) for item, line in zip(base, canonical)] corrections: list[dict[str, Any]] = [] for index, result in enumerate(parsed or []): if index >= len(base): break line = canonical[index] quantity, unit, food, notes = _vet_structure( line, (result or {}).get("ingredient") or {}, units ) if notes: corrections.append({"line": line, "notes": notes}) if unit or food: candidates[index] = _structured_item(base[index], line, quantity, unit, food) rejected: list[int] = [] def apply(items: list[dict[str, Any]]) -> list[dict[str, Any]] | None: try: _request("PATCH", f"/api/recipes/{slug_or_id}", json={"recipeIngredient": items}) except httpx.HTTPStatusError as exc: rejected.append(exc.response.status_code) return None return get_recipe(slug_or_id).get("recipeIngredient") or [] stored = apply(candidates) if stored is None: # A rejected patch writes nothing, so the readable import is already intact. return { "parsed": False, "reason": f"Structured patch rejected by Mealie ({rejected[0]}); the recipe " "was not modified", "canonical_lines": canonical, } # Mealie is the only authority on how a row renders, so compare what it stored. reverted: list[dict[str, Any]] = [] final = list(candidates) for index, line in enumerate(canonical): if index >= len(stored): break shown = stored[index].get("display") or "" if not same_line(shown, line): final[index] = _unstructured_item(base[index], line) reverted.append({"line": line, "mealie_would_show": shown}) if reverted: # The first patch succeeded, so these rows are stored reworded right now and # the revert is a repair, not a precaution — say so plainly if it fails. stored = apply(final) if stored is None: return { "parsed": False, "reason": f"Rows were reworded by Mealie and the repair patch failed " f"({rejected[-1]}); these lines are still reworded in Mealie " f"and need fixing by hand", "reworded_rows": reverted, } structured = [line for index, line in enumerate(canonical) if final[index].get("unit") or final[index].get("food")] return { "parsed": True, "structured_rows": structured, "left_unstructured": reverted, "corrections": corrections, "lines_unchanged": all(same_line((stored[i].get("display") or ""), line) for i, line in enumerate(canonical) if i < len(stored)), "recipe": get_recipe(slug_or_id), } #: Hero image declared by a recipe page, in the order worth trusting. _HERO_META = ( re.compile(r']+property=["\']og:image["\'][^>]+content=["\']([^"\']+)', re.I), re.compile(r']+content=["\']([^"\']+)["\'][^>]+property=["\']og:image["\']', re.I), re.compile(r']+name=["\']twitter:image["\'][^>]+content=["\']([^"\']+)', re.I), ) def _hero_image_url(page_url: str) -> str | None: """Find the image a recipe page advertises as its own. Mealie's cover endpoint downloads the URL it is given and rejects anything that is not an image, so a page has to be reduced to a real image URL first. Fetched with a browser user-agent because recipe sites refuse default ones. """ try: response = httpx.get( page_url, headers={"User-Agent": "Mozilla/5.0 (X11; Linux x86_64)"}, timeout=30, follow_redirects=True, ) response.raise_for_status() except httpx.HTTPError: return None for pattern in _HERO_META: match = pattern.search(response.text) if match: return html.unescape(match.group(1)) return None @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 from a URL or a local file. ``source_url`` may be a recipe page or a direct image URL. Mealie's endpoint only accepts the latter — handing it a page answers ``400 {"message": "Url is not an image"}`` — so a page is resolved to its ``og:image`` first and that is what gets sent. Local uploads must send the multipart ``extension`` field or the upload fails validation silently on this instance. """ 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: used_url = source_url method = "scrape" try: _request("POST", f"/api/recipes/{slug_or_id}/image", json={"url": used_url}) except httpx.HTTPStatusError as exc: if exc.response.status_code != 400 or "not an image" not in exc.response.text: raise used_url = _hero_image_url(source_url) if not used_url: raise ValueError( f"{source_url} is not an image and no og:image was found on the page; " "pass a direct image URL or use image_path" ) from exc _request("POST", f"/api/recipes/{slug_or_id}/image", json={"url": used_url}) method = "scrape (resolved og:image)" recipe = get_recipe(slug_or_id) result = {"method": method, "image": recipe.get("image"), "slug": recipe.get("slug")} if not image_path and used_url != source_url: result["resolved_image_url"] = used_url return result # --- 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() def _configure_read_only( server: MCPServer, allowed_tool_names: frozenset[str] = READ_ONLY_TOOL_NAMES, ) -> None: """Expose only explicitly approved tools to a read-only client.""" registered_tool_names = {tool.name for tool in asyncio.run(server.list_tools())} for name in sorted(registered_tool_names - allowed_tool_names): server.remove_tool(name) def main_read_only() -> None: """Run the same server without import, patch, delete, parse, image, or HA writes.""" _configure_read_only(mcp) mcp.run() def main_import() -> None: """Run the server with the import pipeline but without delete or HA writes. Adding a recipe is recoverable through the Mealie UI; removing one is not, so `delete_recipe` stays out even though this profile writes. """ _configure_read_only(mcp, IMPORT_TOOL_NAMES) mcp.run()