homelab/mealie-mcp
3
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Initial commit: Mealie MCP server

Browse Source 
Ports a working Hermes agent skill to an MCP server. The skill's value was not
its Mealie endpoints but ~25 operational rules found by running imports against
the live instance; those are now code with tests rather than prompt text.

The server owns deterministic mechanics — auth, the non-default User-Agent
Cloudflare requires, the `extension` field on cover uploads, PATCH instead of
the 500-ing bulk-action endpoints, decimal-comma normalization, and restoring
Swedish display text after parsing. Language and taste judgement stay with the
model, fed by the four reference notes shipped as MCP resources.

verify_recipe runs the finished-import definition as code so an agent cannot
report success on a recipe with an English ingredient line, a missing cover, or
ingredients still in the "Click Parse" state.

Home Assistant is optional and isolated; the Obsidian meal plan is out of scope.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Fizz
2026-07-31 07:31:10 +02:00
committed by fredamn76
commit f2d233b430
17 changed files with 1643 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
mealie_mcp/server.py
+463
View File
@@ -0,0 +1,463 @@
"""Small, confirmation-friendly MCP facade over the Mealie API."""
from __future__ import annotations
import json
import os
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,
)
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 suggest_recipes(foods: list[str], limit: int = 10, max_missing_foods: int = 0) -> Any:
"""Find recipes matching ingredients on hand."""
return _request("GET", "/api/recipes/suggestions", params={
"foods": ",".join(foods), "limit": limit,
"maxMissingFoods": max_missing_foods, "includeFoodsOnHand": "true",
})
@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."""
recipe = get_recipe(slug_or_id)
original = recipe.get("recipeYield") or recipe.get("recipeServings")
if not original:
raise ValueError("Recipe has no serving/yield metadata; scaling is not exact")
try:
base = Decimal(str(original).split()[0])
factor = Decimal(servings) / base
except Exception as exc:
raise ValueError("Recipe serving metadata is not numeric") from exc
lines = []
for item in recipe.get("recipeIngredient", []):
text = item.get("display") or item.get("note") or item.get("food", {}).get("name", "")
quantity = item.get("quantity")
if quantity is not None:
scaled = (Decimal(str(quantity)) * factor).quantize(Decimal("0.1"), rounding=ROUND_HALF_UP)
text = text.replace(str(quantity), format(scaled, "f"), 1)
lines.append(text)
return {"recipe": recipe.get("name"), "servings": servings, "ingredients": lines}
# --- 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()