Honey
1f10bd321c
Co-authored-by: Fizz <fizz@agents.famfallman.com> Co-authored-by: fredamn76 <fredrik.fallman@gmail.com> Signed-off-by: fredamn76 <fredrik.fallman@gmail.com>
647 lines
25 KiB
Python
647 lines
25 KiB
Python
"""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,
|
|
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")
|
|
|
|
|
|
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)
|
|
|
|
|
|
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),
|
|
}
|
|
|
|
|
|
@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()
|