Compare commits
8 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 242bb15641 | |||
| c8775ac713 | |||
| 141b84bae1 | |||
| e45156f11b | |||
| 620d3b4fe2 | |||
| 1f10bd321c | |||
| ddc0fb1a5e | |||
| f2d233b430 |
@@ -42,6 +42,57 @@ Create the Mealie token at `/user/profile/api-tokens`.
|
||||
Home Assistant is optional and isolated: without `HA_BASE_URL` and `HA_TOKEN`
|
||||
everything except `shopping_list_add` works normally.
|
||||
|
||||
## Client integration
|
||||
|
||||
### Hermes
|
||||
|
||||
Hermes runs this server over stdio. The credential stays in
|
||||
`~/.hermes/.env`; `config.yaml` contains only environment placeholders.
|
||||
|
||||
```bash
|
||||
hermes mcp add mealie \
|
||||
--command /home/fredrik/.local/bin/uv \
|
||||
--connect-timeout 60 \
|
||||
--env 'MEALIE_API_TOKEN=${MEALIE_API_TOKEN}' \
|
||||
'MEALIE_BASE_URL=${MEALIE_BASE_URL}' \
|
||||
'MEALIE_USER_AGENT=Mealie-MCP/0.1' \
|
||||
--args --directory /home/fredrik/.buzz/REPOS/mealie-mcp run mealie-mcp
|
||||
|
||||
hermes config set mcp_discovery_timeout 10
|
||||
hermes mcp test mealie
|
||||
```
|
||||
|
||||
The normal `hermes chat` client and the long-running gateway wait for MCP
|
||||
discovery. On the Hermes version used for the P1 acceptance test,
|
||||
`hermes -z` could snapshot its tools before a slower stdio server finished
|
||||
discovery; use the normal chat/gateway path for this server until that
|
||||
one-shot startup issue is fixed.
|
||||
|
||||
### Buzz managed agent
|
||||
|
||||
Buzz's managed-agent harness accepts one per-agent MCP executable. Set the
|
||||
dedicated Recept agent's MCP command to this read-only launcher:
|
||||
|
||||
```text
|
||||
/home/fredrik/.buzz/REPOS/mealie-mcp/scripts/run-mealie-mcp-read-only-for-buzz
|
||||
```
|
||||
|
||||
It contains search, read, scaling, organizer, duplicate-check, suggestion, and
|
||||
verification tools, but no import, patch, delete, ingredient-parse, image, or
|
||||
shopping-list tools. Switch to the full `mealie-mcp` entry point only after the
|
||||
separate write acceptance test has passed.
|
||||
|
||||
The launcher reads only `MEALIE_API_TOKEN`, `MEALIE_BASE_URL`, and
|
||||
`MEALIE_USER_AGENT` from `~/.hermes/.env`, then starts the same stdio server
|
||||
with a clean environment. It does not place the token in the agent prompt,
|
||||
agent definition, command line, or Codex configuration.
|
||||
|
||||
Keep this configuration agent-specific. Adding Mealie globally to
|
||||
`~/.codex/config.toml` would make the tools available to every Codex-based
|
||||
Buzz agent on the host, which is broader access than the private Recept agent
|
||||
needs. No HTTP transport is required while the agent and server run on the
|
||||
same machine.
|
||||
|
||||
## Tools
|
||||
|
||||
**Read**
|
||||
@@ -63,6 +114,7 @@ campaign junk in the title. A response is not proof of a good import.
|
||||
|
||||
**Write**
|
||||
- `patch_recipe(slug_or_id, patch)`
|
||||
- `delete_recipe(slug_or_id, confirm_slug)` — exact-slug confirmation plus 404 read-back proof
|
||||
- `parse_ingredients(slug_or_id)`
|
||||
- `set_cover_image(slug_or_id, source_url=None, image_path=None)`
|
||||
- `shopping_list_add(items)`
|
||||
@@ -104,7 +156,7 @@ file tools; folding it in would make this server two things at once.
|
||||
## Tests
|
||||
|
||||
```bash
|
||||
python -m pytest
|
||||
uv run --extra dev python -m pytest
|
||||
```
|
||||
|
||||
The suite covers the verified failure modes: Cloudflare 1010 vs 401, decimal-comma
|
||||
|
||||
@@ -151,6 +151,86 @@ def scale_line(line: str, factor: Decimal) -> tuple[str, bool]:
|
||||
return line[: match.start("low")] + replacement + line[match.end() :], True
|
||||
|
||||
|
||||
#: The word directly after a leading amount — the unit candidate in a Swedish line.
|
||||
_UNIT_TOKEN = re.compile(
|
||||
rf"^\s*(?:{_AMOUNT})(?:\s*[-–]\s*(?:{_AMOUNT}))?\s+(?P<token>[^\W\d_]+)",
|
||||
flags=re.UNICODE,
|
||||
)
|
||||
|
||||
|
||||
def unit_token(line: str) -> str | None:
|
||||
"""Return the word following the leading amount, or None if the line has no amount.
|
||||
|
||||
This is the unit as *written* (``dl``, ``msk``, ``g``, ``klyfta``). It is the
|
||||
authority for which unit a row means, because the Mealie NLP parser resolves
|
||||
``3 dl`` to the ``liter`` record — a verified 10x error on a live instance.
|
||||
Not every match is a unit (``0,5 citron`` yields ``citron``); callers confirm
|
||||
the token against the instance's own unit table.
|
||||
"""
|
||||
match = _UNIT_TOKEN.match(line)
|
||||
return match.group("token") if match else None
|
||||
|
||||
|
||||
def match_unit(token: str | None, units: list[dict]) -> dict | None:
|
||||
"""Find the unit record a written token refers to, by name or abbreviation.
|
||||
|
||||
Matches singular and plural forms of both. Returns None when the token is not
|
||||
a unit in this Mealie instance, which is the signal to store no unit at all
|
||||
rather than the parser's guess.
|
||||
"""
|
||||
if not token:
|
||||
return None
|
||||
needle = token.strip().lower()
|
||||
fields = ("name", "pluralName", "abbreviation", "pluralAbbreviation")
|
||||
for unit in units:
|
||||
if any((unit.get(field) or "").strip().lower() == needle for field in fields):
|
||||
return unit
|
||||
return None
|
||||
|
||||
|
||||
def food_matches_line(food_name: str, line: str) -> bool:
|
||||
"""True when the parser's food is actually the ingredient the line names.
|
||||
|
||||
The parser substitutes near neighbours from the existing food table —
|
||||
``800 g kycklinglårfilé`` came back as the food ``kycklingfilé``. A food is
|
||||
accepted only when its name appears in the line outright (``koncentrerad
|
||||
kycklingfond``) or begins a word in it, which keeps the singular record
|
||||
``körsbärstomat`` for ``körsbärstomater`` while rejecting the lår/filé swap.
|
||||
"""
|
||||
name = (food_name or "").strip().lower()
|
||||
if not name:
|
||||
return False
|
||||
# The name must begin a word, so `lök` does not match inside `vitlök`, while a
|
||||
# trailing plural (`körsbärstomat` in `körsbärstomater`) is still allowed.
|
||||
return re.search(rf"(?<![^\W\d_]){re.escape(name)}", line.lower()) is not None
|
||||
|
||||
|
||||
def line_remainder(line: str, written_unit: str | None, food_name: str | None) -> str:
|
||||
"""Return what is left of a line once amount, unit, and food are held as structure.
|
||||
|
||||
Mealie composes the visible row as ``quantity unit food note``, so ``note`` must
|
||||
hold only the leftover text. Passing the whole line duplicates it
|
||||
(``3 dl vispgrädde 3 dl vispgrädde``). The food match is widened to the end of
|
||||
the word it starts, so the record ``körsbärstomat`` consumes
|
||||
``körsbärstomater`` and leaves ``(i olika färger)``.
|
||||
"""
|
||||
rest = _LEADING_AMOUNT.sub("", line, count=1)
|
||||
if written_unit:
|
||||
rest = re.sub(rf"^\s*{re.escape(written_unit)}\b", "", rest, count=1, flags=re.IGNORECASE)
|
||||
if food_name:
|
||||
match = re.search(
|
||||
rf"(?<![^\W\d_]){re.escape(food_name.strip().lower())}[^\W\d_]*", rest.lower()
|
||||
)
|
||||
if match:
|
||||
rest = rest[: match.start()] + rest[match.end() :]
|
||||
return _WHITESPACE.sub(" ", rest).strip()
|
||||
|
||||
|
||||
def same_line(left: str, right: str) -> bool:
|
||||
"""Compare two ingredient rows ignoring only whitespace differences."""
|
||||
return _WHITESPACE.sub(" ", left or "").strip() == _WHITESPACE.sub(" ", right or "").strip()
|
||||
|
||||
|
||||
def looks_mangled(line: str) -> bool:
|
||||
"""Heuristic for parser-damaged display text.
|
||||
|
||||
|
||||
+284
-35
@@ -1,6 +1,7 @@
|
||||
"""Small, confirmation-friendly MCP facade over the Mealie API."""
|
||||
from __future__ import annotations
|
||||
|
||||
import html
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
@@ -13,11 +14,16 @@ 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
|
||||
|
||||
@@ -32,6 +38,19 @@ RESOURCE_DIR = Path(__file__).parent / "resources"
|
||||
|
||||
mcp = MCPServer("mealie", version="0.1.0")
|
||||
|
||||
WRITE_TOOL_NAMES = frozenset(
|
||||
{
|
||||
"import_recipe_url",
|
||||
"import_recipe_text",
|
||||
"import_recipe_image",
|
||||
"patch_recipe",
|
||||
"delete_recipe",
|
||||
"parse_ingredients",
|
||||
"set_cover_image",
|
||||
"shopping_list_add",
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
class MealieAuthError(RuntimeError):
|
||||
"""The Mealie token is missing, expired, or wrong (HTTP 401)."""
|
||||
@@ -283,6 +302,17 @@ def import_recipe_text(text: str, source_url: str | None = None) -> dict[str, An
|
||||
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
|
||||
@@ -313,20 +343,138 @@ 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.
|
||||
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(slug_or_id)
|
||||
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 while keeping the Swedish display text.
|
||||
"""Add structured food/unit/quantity data without changing any visible Swedish line.
|
||||
|
||||
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.
|
||||
``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)
|
||||
@@ -338,57 +486,127 @@ def parse_ingredients(slug_or_id: str) -> dict[str, Any]:
|
||||
"/api/parser/ingredients",
|
||||
json={"ingredients": [normalize_for_parser(line) for line in canonical], "parser": "nlp"},
|
||||
)
|
||||
units = _unit_records()
|
||||
|
||||
ingredients = list(recipe.get("recipeIngredient") or [])
|
||||
degraded: list[str] = []
|
||||
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(ingredients):
|
||||
if index >= len(base):
|
||||
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
|
||||
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": ingredients})
|
||||
_request("PATCH", f"/api/recipes/{slug_or_id}", json={"recipeIngredient": items})
|
||||
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.
|
||||
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 ({exc.response.status_code}); "
|
||||
"readable Swedish lines were left intact",
|
||||
"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),
|
||||
"readability_fallback_rows": degraded,
|
||||
}
|
||||
|
||||
|
||||
#: Hero image declared by a recipe page, in the order worth trusting.
|
||||
_HERO_META = (
|
||||
re.compile(r'<meta[^>]+property=["\']og:image["\'][^>]+content=["\']([^"\']+)', re.I),
|
||||
re.compile(r'<meta[^>]+content=["\']([^"\']+)["\'][^>]+property=["\']og:image["\']', re.I),
|
||||
re.compile(r'<meta[^>]+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 by scraping a URL or uploading a local file.
|
||||
"""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. When a scrape does not produce a visible
|
||||
cover (observed on Allrecipes), download the hero image and upload it instead.
|
||||
validation silently on this instance.
|
||||
"""
|
||||
if not source_url and not image_path:
|
||||
raise ValueError("Provide either source_url or image_path")
|
||||
@@ -407,11 +625,27 @@ def set_cover_image(
|
||||
_raise_for_status(response)
|
||||
method = f"upload ({extension})"
|
||||
else:
|
||||
_request("POST", f"/api/recipes/{slug_or_id}/image", json={"url": source_url})
|
||||
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)
|
||||
return {"method": method, "image": recipe.get("image"), "slug": recipe.get("slug")}
|
||||
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 -----------------------------------------------------------------
|
||||
@@ -524,3 +758,18 @@ instructions, and units (`msk`, `tsk`, `dl`, `krm`, `st`, `klyfta`).
|
||||
|
||||
def main() -> None:
|
||||
mcp.run()
|
||||
|
||||
|
||||
def _configure_read_only(
|
||||
server: MCPServer,
|
||||
write_tool_names: frozenset[str] = WRITE_TOOL_NAMES,
|
||||
) -> None:
|
||||
"""Remove mutating tools before exposing the server to a read-only client."""
|
||||
for name in sorted(write_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()
|
||||
|
||||
@@ -10,6 +10,7 @@ dev = ["pytest>=8.0"]
|
||||
|
||||
[project.scripts]
|
||||
mealie-mcp = "mealie_mcp.server:main"
|
||||
mealie-mcp-read-only = "mealie_mcp.server:main_read_only"
|
||||
|
||||
[build-system]
|
||||
requires = ["hatchling"]
|
||||
|
||||
Executable
+79
@@ -0,0 +1,79 @@
|
||||
#!/usr/bin/env bash
|
||||
set -euo pipefail
|
||||
|
||||
# Buzz's managed-agent harness accepts one MCP executable per agent. Keep the
|
||||
# Mealie credential out of the agent definition by loading only the three
|
||||
# variables this server understands from the existing Hermes dotenv file.
|
||||
mealie_env_file="${MEALIE_ENV_FILE:-${HOME}/.hermes/.env}"
|
||||
mealie_repo_dir="${MEALIE_MCP_REPO_DIR:-/home/fredrik/.buzz/REPOS/mealie-mcp}"
|
||||
mealie_uv_bin="${MEALIE_UV_BIN:-/home/fredrik/.local/bin/uv}"
|
||||
|
||||
mealie_token="${MEALIE_API_TOKEN:-}"
|
||||
mealie_base_url="${MEALIE_BASE_URL:-}"
|
||||
mealie_user_agent="${MEALIE_USER_AGENT:-}"
|
||||
|
||||
clean_dotenv_value() {
|
||||
local value="$1"
|
||||
local first
|
||||
local last
|
||||
|
||||
value="${value%$'\r'}"
|
||||
if (( ${#value} >= 2 )); then
|
||||
first="${value:0:1}"
|
||||
last="${value: -1}"
|
||||
if [[ "$first" == "$last" && ( "$first" == "'" || "$first" == '"' ) ]]; then
|
||||
value="${value:1:${#value}-2}"
|
||||
fi
|
||||
fi
|
||||
REPLY="$value"
|
||||
}
|
||||
|
||||
if [[ -r "$mealie_env_file" ]]; then
|
||||
while IFS= read -r line || [[ -n "$line" ]]; do
|
||||
case "$line" in
|
||||
MEALIE_API_TOKEN=*)
|
||||
if [[ -z "$mealie_token" ]]; then
|
||||
clean_dotenv_value "${line#*=}"
|
||||
mealie_token="$REPLY"
|
||||
fi
|
||||
;;
|
||||
MEALIE_BASE_URL=*)
|
||||
if [[ -z "$mealie_base_url" ]]; then
|
||||
clean_dotenv_value "${line#*=}"
|
||||
mealie_base_url="$REPLY"
|
||||
fi
|
||||
;;
|
||||
MEALIE_USER_AGENT=*)
|
||||
if [[ -z "$mealie_user_agent" ]]; then
|
||||
clean_dotenv_value "${line#*=}"
|
||||
mealie_user_agent="$REPLY"
|
||||
fi
|
||||
;;
|
||||
esac
|
||||
done < "$mealie_env_file"
|
||||
fi
|
||||
|
||||
if [[ -z "$mealie_token" ]]; then
|
||||
printf 'MEALIE_API_TOKEN is not set and was not found in %s\n' "$mealie_env_file" >&2
|
||||
exit 1
|
||||
fi
|
||||
|
||||
mealie_base_url="${mealie_base_url:-https://recept.famfallman.com}"
|
||||
mealie_user_agent="${mealie_user_agent:-Mealie-MCP/0.1}"
|
||||
|
||||
# Do not forward the rest of the Desktop/Hermes environment to the MCP
|
||||
# subprocess. In particular, unrelated provider and platform tokens must not
|
||||
# become visible to this server. Export after sanitizing so the token is never
|
||||
# passed as a command-line argument to an intermediate `env` process.
|
||||
while IFS= read -r variable_name; do
|
||||
case "$variable_name" in
|
||||
HOME | PATH | LANG) ;;
|
||||
*) unset "$variable_name" ;;
|
||||
esac
|
||||
done < <(compgen -e)
|
||||
|
||||
export MEALIE_API_TOKEN="$mealie_token"
|
||||
export MEALIE_BASE_URL="$mealie_base_url"
|
||||
export MEALIE_USER_AGENT="$mealie_user_agent"
|
||||
|
||||
exec "$mealie_uv_bin" --directory "$mealie_repo_dir" run mealie-mcp-read-only
|
||||
@@ -0,0 +1,79 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
LAUNCHER = (
|
||||
Path(__file__).parents[1] / "scripts" / "run-mealie-mcp-read-only-for-buzz"
|
||||
)
|
||||
|
||||
|
||||
def _fake_uv(tmp_path: Path) -> Path:
|
||||
executable = tmp_path / "fake-uv"
|
||||
executable.write_text(
|
||||
"#!/usr/bin/env bash\n"
|
||||
"printf '%s\\n' \"${MEALIE_API_TOKEN}|${MEALIE_BASE_URL}|"
|
||||
"${MEALIE_USER_AGENT}|${UNRELATED_SECRET-unset}|$*\"\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
executable.chmod(0o755)
|
||||
return executable
|
||||
|
||||
|
||||
def test_launcher_reads_only_mealie_values_and_clears_unrelated_secrets(
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
env_file = tmp_path / "hermes.env"
|
||||
env_file.write_text(
|
||||
"OTHER_PROVIDER_TOKEN=must-not-leak\n"
|
||||
"MEALIE_API_TOKEN='test-token'\n"
|
||||
"MEALIE_BASE_URL=https://mealie.example\n"
|
||||
'MEALIE_USER_AGENT="Test-Agent/1.0"\n',
|
||||
encoding="utf-8",
|
||||
)
|
||||
env = {
|
||||
**os.environ,
|
||||
"MEALIE_ENV_FILE": str(env_file),
|
||||
"MEALIE_MCP_REPO_DIR": "/tmp/test-repo",
|
||||
"MEALIE_UV_BIN": str(_fake_uv(tmp_path)),
|
||||
"UNRELATED_SECRET": "must-not-leak",
|
||||
}
|
||||
|
||||
result = subprocess.run(
|
||||
[str(LAUNCHER)],
|
||||
check=False,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
env=env,
|
||||
)
|
||||
|
||||
assert result.returncode == 0
|
||||
assert result.stdout.strip() == (
|
||||
"test-token|https://mealie.example|Test-Agent/1.0|unset|"
|
||||
"--directory /tmp/test-repo run mealie-mcp-read-only"
|
||||
)
|
||||
assert "must-not-leak" not in result.stdout
|
||||
|
||||
|
||||
def test_launcher_fails_without_mealie_token(tmp_path: Path) -> None:
|
||||
env_file = tmp_path / "empty.env"
|
||||
env_file.write_text("MEALIE_BASE_URL=https://mealie.example\n", encoding="utf-8")
|
||||
env = {
|
||||
"HOME": str(tmp_path),
|
||||
"PATH": os.environ["PATH"],
|
||||
"MEALIE_ENV_FILE": str(env_file),
|
||||
"MEALIE_UV_BIN": str(_fake_uv(tmp_path)),
|
||||
}
|
||||
|
||||
result = subprocess.run(
|
||||
[str(LAUNCHER)],
|
||||
check=False,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
env=env,
|
||||
)
|
||||
|
||||
assert result.returncode == 1
|
||||
assert "MEALIE_API_TOKEN is not set" in result.stderr
|
||||
@@ -6,11 +6,16 @@ import pytest
|
||||
from mealie_mcp.normalize import (
|
||||
clean_title,
|
||||
find_english_leftovers,
|
||||
food_matches_line,
|
||||
has_extraction_failure,
|
||||
ingredient_display_lines,
|
||||
line_remainder,
|
||||
looks_mangled,
|
||||
match_unit,
|
||||
normalize_for_parser,
|
||||
same_line,
|
||||
slugify,
|
||||
unit_token,
|
||||
)
|
||||
|
||||
|
||||
@@ -97,3 +102,33 @@ class TestIngredientDisplayLines:
|
||||
]
|
||||
}
|
||||
assert ingredient_display_lines(recipe) == ["2 dl grädde", "1 gul lök", "salt"]
|
||||
|
||||
|
||||
class TestParserStructureSafety:
|
||||
def test_reads_the_written_unit_token(self):
|
||||
assert unit_token("4,7 dl basmatiris") == "dl"
|
||||
assert unit_token("1-1,5 msk fond") == "msk"
|
||||
assert unit_token("salt") is None
|
||||
|
||||
def test_matches_instance_unit_by_abbreviation(self):
|
||||
gram = {"id": "g", "name": "gram", "abbreviation": "g"}
|
||||
assert match_unit("g", [gram]) == gram
|
||||
assert match_unit("dl", [gram]) is None
|
||||
|
||||
def test_rejects_a_nearby_but_different_food(self):
|
||||
assert food_matches_line("kycklinglårfilé", "800 g kycklinglårfilé")
|
||||
assert not food_matches_line("kycklingfilé", "800 g kycklinglårfilé")
|
||||
|
||||
def test_builds_note_from_only_the_unstructured_remainder(self):
|
||||
assert (
|
||||
line_remainder(
|
||||
"200 g körsbärstomater (i olika färger)",
|
||||
"g",
|
||||
"körsbärstomat",
|
||||
)
|
||||
== "(i olika färger)"
|
||||
)
|
||||
|
||||
def test_same_line_ignores_only_whitespace(self):
|
||||
assert same_line("3 dl grädde", " 3 dl grädde ")
|
||||
assert not same_line("3 dl grädde", "3 liter grädde")
|
||||
|
||||
+196
-17
@@ -96,46 +96,136 @@ class TestImportReport:
|
||||
assert server._import_report(recipe)["suggested_title"] is None
|
||||
|
||||
|
||||
class TestTextImport:
|
||||
def test_rejects_text_without_explicit_sections_before_writing(self, monkeypatch):
|
||||
monkeypatch.setattr(
|
||||
server,
|
||||
"_request",
|
||||
lambda *_args, **_kwargs: pytest.fail("must not create a junk recipe"),
|
||||
)
|
||||
|
||||
with pytest.raises(ValueError, match="Titel"):
|
||||
server.import_recipe_text(
|
||||
"Namnlöst recept\n\nIngredienser:\n1 dl vatten\n\nGör så här:\nBlanda."
|
||||
)
|
||||
|
||||
def test_accepts_the_documented_swedish_section_format(self, monkeypatch):
|
||||
text = (
|
||||
"Titel: Testsoppa\n\nPortioner: 2\n\nIngredienser:\n1 dl vatten\n\n"
|
||||
"Gör så här:\nBlanda."
|
||||
)
|
||||
recipe = {
|
||||
"slug": "testsoppa",
|
||||
"name": "Testsoppa",
|
||||
"recipeIngredient": [{"display": "1 dl vatten"}],
|
||||
"recipeInstructions": [{"text": "Blanda."}],
|
||||
}
|
||||
monkeypatch.setattr(server, "_request", lambda *_args, **_kwargs: "testsoppa")
|
||||
monkeypatch.setattr(server, "get_recipe", lambda _slug: recipe)
|
||||
|
||||
assert server.import_recipe_text(text)["recipe"] == recipe
|
||||
|
||||
|
||||
class TestDeleteRecipe:
|
||||
def test_requires_the_exact_stored_slug(self, monkeypatch):
|
||||
monkeypatch.setattr(
|
||||
server,
|
||||
"get_recipe",
|
||||
lambda _slug: {"slug": "stored-slug", "name": "Test"},
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
server,
|
||||
"_client",
|
||||
lambda: pytest.fail("must not delete without exact confirmation"),
|
||||
)
|
||||
|
||||
with pytest.raises(ValueError, match="stored-slug"):
|
||||
server.delete_recipe("recipe-id", "wrong-slug")
|
||||
|
||||
def test_deletes_and_proves_the_recipe_is_gone(self, mock_mealie):
|
||||
recipe = {"slug": "zzz-test", "name": "ZZZ Test"}
|
||||
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
if request.method == "DELETE":
|
||||
return httpx.Response(200, json={})
|
||||
if request.url.path == "/api/recipes/zzz-test":
|
||||
# First GET is get_recipe; second GET is the deletion proof.
|
||||
if getattr(handler, "read_once", False):
|
||||
return httpx.Response(404, json={"detail": "Not Found"})
|
||||
handler.read_once = True
|
||||
return httpx.Response(200, json=recipe)
|
||||
return httpx.Response(500)
|
||||
|
||||
mock_mealie(handler)
|
||||
result = server.delete_recipe("zzz-test", "zzz-test")
|
||||
|
||||
assert result == {
|
||||
"deleted": True,
|
||||
"slug": "zzz-test",
|
||||
"name": "ZZZ Test",
|
||||
"delete_status": 200,
|
||||
"readback_status": 404,
|
||||
}
|
||||
|
||||
|
||||
class TestParseIngredients:
|
||||
def test_swedish_display_text_survives_the_parser(self, mock_mealie):
|
||||
def test_reverts_structure_when_mealie_rewords_the_swedish_line(self, mock_mealie):
|
||||
canonical = "4,7 dl basmatiris"
|
||||
recipe = {
|
||||
"slug": "ris",
|
||||
"recipeIngredient": [{"display": canonical, "note": canonical}],
|
||||
}
|
||||
seen: dict[str, object] = {}
|
||||
patches: list[list[dict]] = []
|
||||
reads = 0
|
||||
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
import json as _json
|
||||
|
||||
nonlocal reads
|
||||
if request.url.path == "/api/parser/ingredients":
|
||||
seen["sent"] = _json.loads(request.content)["ingredients"]
|
||||
# The parser returns mangled display text; it must not be stored.
|
||||
return httpx.Response(200, json=[{
|
||||
"ingredient": {
|
||||
"quantity": 4.7,
|
||||
"unit": {"id": "u", "name": "dl"},
|
||||
"unit": {"id": "liter", "name": "liter"},
|
||||
"food": {"id": "f", "name": "basmatiris"},
|
||||
"display": "47⁄10 liter basmatiris",
|
||||
}
|
||||
}])
|
||||
if request.url.path == "/api/units":
|
||||
return httpx.Response(
|
||||
200,
|
||||
json={"items": [{"id": "dl", "name": "dl", "abbreviation": "dl"}]},
|
||||
)
|
||||
if request.method == "PATCH":
|
||||
seen["patched"] = _json.loads(request.content)["recipeIngredient"]
|
||||
patches.append(_json.loads(request.content)["recipeIngredient"])
|
||||
return httpx.Response(200, json={})
|
||||
reads += 1
|
||||
if reads == 1:
|
||||
return httpx.Response(200, json=recipe)
|
||||
if len(patches) == 1:
|
||||
# This is what Mealie rendered from the first structured patch.
|
||||
return httpx.Response(200, json={
|
||||
"slug": "ris",
|
||||
"recipeIngredient": [{
|
||||
**patches[0][0],
|
||||
"display": "47⁄10 dl basmatiris",
|
||||
}],
|
||||
})
|
||||
return httpx.Response(200, json=recipe)
|
||||
|
||||
mock_mealie(handler)
|
||||
server.parse_ingredients("ris")
|
||||
result = server.parse_ingredients("ris")
|
||||
|
||||
# Decimal comma is normalized for the parser only...
|
||||
assert seen["sent"] == ["4.7 dl basmatiris"]
|
||||
patched = seen["patched"][0]
|
||||
# ...while the stored human-facing line stays the original Swedish text.
|
||||
assert patched["display"] == canonical
|
||||
assert patched["note"] == canonical
|
||||
# Structured data from the parser is still applied.
|
||||
assert patched["quantity"] == 4.7
|
||||
assert patched["food"]["name"] == "basmatiris"
|
||||
assert len(patches) == 2
|
||||
assert patches[0][0]["unit"]["id"] == "dl"
|
||||
assert patches[1][0]["quantity"] == 0
|
||||
assert patches[1][0]["unit"] is None
|
||||
assert patches[1][0]["food"] is None
|
||||
assert patches[1][0]["note"] == canonical
|
||||
assert result["lines_unchanged"] is True
|
||||
assert result["left_unstructured"] == [{
|
||||
"line": canonical,
|
||||
"mealie_would_show": "47⁄10 dl basmatiris",
|
||||
}]
|
||||
|
||||
def test_failed_structured_patch_preserves_the_readable_import(self, mock_mealie):
|
||||
recipe = {"slug": "ris", "recipeIngredient": [{"display": "2 dl grädde"}]}
|
||||
@@ -169,3 +259,92 @@ class TestCoverImage:
|
||||
def test_requires_a_source(self):
|
||||
with pytest.raises(ValueError):
|
||||
server.set_cover_image("x")
|
||||
|
||||
def test_a_recipe_page_is_resolved_to_its_hero_image(self, mock_mealie, monkeypatch):
|
||||
# Verified live: Mealie answers 400 "Url is not an image" for a recipe page,
|
||||
# and 200 for that same page's og:image.
|
||||
page = "https://www.ica.se/recept/kalsoppa-722069/"
|
||||
hero = "https://assets.icanet.se/imagevaultfiles/kalsoppa.jpg"
|
||||
posted: list[str] = []
|
||||
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
import json as _json
|
||||
|
||||
if request.method == "POST" and request.url.path.endswith("/image"):
|
||||
url = _json.loads(request.content)["url"]
|
||||
posted.append(url)
|
||||
if url == page:
|
||||
return httpx.Response(
|
||||
400, json={"detail": {"message": "Url is not an image", "error": True}}
|
||||
)
|
||||
return httpx.Response(200, json=None)
|
||||
return httpx.Response(200, json={"slug": "soppa", "image": "abcd"})
|
||||
|
||||
mock_mealie(handler)
|
||||
monkeypatch.setattr(
|
||||
server, "_hero_image_url", lambda url: hero if url == page else None
|
||||
)
|
||||
|
||||
result = server.set_cover_image("soppa", source_url=page)
|
||||
|
||||
# The page is tried first, then retried with the image it advertises.
|
||||
assert posted == [page, hero]
|
||||
assert result["method"] == "scrape (resolved og:image)"
|
||||
assert result["resolved_image_url"] == hero
|
||||
|
||||
def test_a_direct_image_url_is_sent_unchanged(self, mock_mealie, monkeypatch):
|
||||
recorded = mock_mealie(lambda r: httpx.Response(200, json={"slug": "x", "image": "img"}))
|
||||
monkeypatch.setattr(server, "_hero_image_url", lambda url: pytest.fail("not needed"))
|
||||
|
||||
result = server.set_cover_image("x", source_url="https://example.com/hero.jpg")
|
||||
|
||||
assert result["method"] == "scrape"
|
||||
assert "resolved_image_url" not in result
|
||||
assert any(r.method == "POST" for r in recorded)
|
||||
|
||||
def test_a_page_with_no_hero_image_fails_loudly(self, mock_mealie, monkeypatch):
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
if request.method == "POST" and request.url.path.endswith("/image"):
|
||||
return httpx.Response(400, json={"detail": {"message": "Url is not an image"}})
|
||||
return httpx.Response(200, json={"slug": "x"})
|
||||
|
||||
mock_mealie(handler)
|
||||
monkeypatch.setattr(server, "_hero_image_url", lambda url: None)
|
||||
|
||||
with pytest.raises(ValueError, match="no og:image"):
|
||||
server.set_cover_image("x", source_url="https://example.com/page")
|
||||
|
||||
|
||||
class TestHeroImageUrl:
|
||||
def test_reads_og_image_in_either_attribute_order(self, monkeypatch):
|
||||
pages = {
|
||||
"a": '<meta property="og:image" content="https://x.test/a.jpg">',
|
||||
"b": '<meta content="https://x.test/b.jpg" property="og:image">',
|
||||
"c": '<meta name="twitter:image" content="https://x.test/c.jpg">',
|
||||
"d": "<html><body>no image here</body></html>",
|
||||
}
|
||||
|
||||
def fake_get(url, **kwargs):
|
||||
return httpx.Response(200, text=pages[url], request=httpx.Request("GET", "http://t"))
|
||||
|
||||
monkeypatch.setattr(server.httpx, "get", fake_get)
|
||||
assert server._hero_image_url("a") == "https://x.test/a.jpg"
|
||||
assert server._hero_image_url("b") == "https://x.test/b.jpg"
|
||||
assert server._hero_image_url("c") == "https://x.test/c.jpg"
|
||||
assert server._hero_image_url("d") is None
|
||||
|
||||
def test_unescapes_entities_in_the_url(self, monkeypatch):
|
||||
# Query strings in og:image arrive HTML-escaped and must be decoded.
|
||||
page = '<meta property="og:image" content="https://x.test/a.jpg?w=1&h=2">'
|
||||
monkeypatch.setattr(
|
||||
server.httpx, "get",
|
||||
lambda url, **kw: httpx.Response(200, text=page, request=httpx.Request("GET", "http://t")),
|
||||
)
|
||||
assert server._hero_image_url("p") == "https://x.test/a.jpg?w=1&h=2"
|
||||
|
||||
def test_an_unreachable_page_is_not_an_error(self, monkeypatch):
|
||||
def boom(url, **kwargs):
|
||||
raise httpx.ConnectError("no route")
|
||||
|
||||
monkeypatch.setattr(server.httpx, "get", boom)
|
||||
assert server._hero_image_url("https://example.com/page") is None
|
||||
|
||||
@@ -1,8 +1,11 @@
|
||||
"""Boundary contracts for the verified Mealie instance quirks."""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
from mcp.server.mcpserver import MCPServer
|
||||
|
||||
from mealie_mcp import server
|
||||
|
||||
@@ -21,6 +24,26 @@ class TestAuthenticationFailures:
|
||||
server._raise_for_status(response(403, "error code: 1010"))
|
||||
|
||||
|
||||
def test_read_only_server_removes_all_mutating_tools():
|
||||
test_server = MCPServer("test")
|
||||
|
||||
@test_server.tool()
|
||||
def search_recipes(query: str) -> list[str]:
|
||||
return [query]
|
||||
|
||||
def mutating_tool(value: str) -> str:
|
||||
return value
|
||||
|
||||
for name in server.WRITE_TOOL_NAMES:
|
||||
test_server.tool(name=name)(mutating_tool)
|
||||
|
||||
server._configure_read_only(test_server)
|
||||
|
||||
tool_names = {tool.name for tool in asyncio.run(test_server.list_tools())}
|
||||
|
||||
assert tool_names == {"search_recipes"}
|
||||
|
||||
|
||||
class TestUrlImportResponse:
|
||||
def test_slug_only_import_response_is_resolved_before_reporting(self, monkeypatch):
|
||||
recipe = {"slug": "lax-med-citron", "name": "Lax med citron", "recipeIngredient": []}
|
||||
@@ -62,14 +85,24 @@ class TestUrlImportResponse:
|
||||
class TestPatchContract:
|
||||
def test_patch_recipe_reads_back_the_final_object(self, monkeypatch):
|
||||
seen = []
|
||||
updated = {"slug": "soppa", "name": "Soppa", "tags": [{"name": "📅 Vardag"}]}
|
||||
before = {"id": "stable-id", "slug": "soppa", "name": "Soppa"}
|
||||
updated = {
|
||||
"id": "stable-id",
|
||||
"slug": "ny-soppa",
|
||||
"name": "Ny soppa",
|
||||
"tags": [{"name": "📅 Vardag"}],
|
||||
}
|
||||
|
||||
def request(method, path, **kwargs):
|
||||
seen.append((method, path, kwargs))
|
||||
return None
|
||||
|
||||
monkeypatch.setattr(server, "_request", request)
|
||||
monkeypatch.setattr(server, "get_recipe", lambda slug: updated if slug == "soppa" else None)
|
||||
monkeypatch.setattr(
|
||||
server,
|
||||
"get_recipe",
|
||||
lambda key: before if key == "soppa" else updated if key == "stable-id" else None,
|
||||
)
|
||||
|
||||
assert server.patch_recipe("soppa", {"tags": updated["tags"]}) == updated
|
||||
assert seen == [("PATCH", "/api/recipes/soppa", {"json": {"tags": updated["tags"]}})]
|
||||
|
||||
Reference in New Issue
Block a user