feat(recipes): implement get_recipes, get_recipe, create_recipe, delete_recipe (v0.6.0)
Adds 4 new MCP tools for the Family Wall recipe box: - get_recipes: list all family recipes via metasync id='recipe' - get_recipe: fetch full recipe detail by id (filters from metasync response) - create_recipe: create a new recipe via mprecipeput (params use 'recipe.' prefix) - delete_recipe: delete a recipe via metadelete (same endpoint as tasks) Verified endpoints and parameter names via FW_DEBUG=1 probe scripts. All 4 tools pass the create → read → get_single → delete integration test. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -1 +1 @@
|
||||
__version__ = "0.5.2"
|
||||
__version__ = "0.6.0"
|
||||
|
||||
@@ -0,0 +1,120 @@
|
||||
"""Recipe helper functions for the Family Wall recipe box.
|
||||
|
||||
Verified endpoints (2026-04-16 via FW_DEBUG=1):
|
||||
- Create: POST mprecipeput — params use 'recipe.' prefix (recipe.name, etc.)
|
||||
- Read all: POST metasync with id='recipe' — response at a00.r.r.updatedCreated[]
|
||||
- Delete: POST metadelete with id=<recipe_metaId>
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
|
||||
def parse_recipe_summary(raw: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Extract a compact recipe summary from a raw API recipe object.
|
||||
|
||||
Args:
|
||||
raw: Raw recipe dict from the API (metasync or mprecipeput response).
|
||||
|
||||
Returns:
|
||||
Dict with keys: id, name, prep_time_minutes, cook_time_minutes, serves,
|
||||
description, can_delete.
|
||||
"""
|
||||
prep = raw.get("prepTime")
|
||||
cook = raw.get("cookTime")
|
||||
srv = raw.get("serves")
|
||||
return {
|
||||
"id": raw.get("metaId"),
|
||||
"name": raw.get("name"),
|
||||
"description": raw.get("description") or None,
|
||||
"prep_time_minutes": int(prep) if prep else None,
|
||||
"cook_time_minutes": int(cook) if cook else None,
|
||||
"serves": int(srv) if srv else None,
|
||||
"can_delete": raw.get("rights", {}).get("canDelete") == "true",
|
||||
}
|
||||
|
||||
|
||||
def parse_recipe_full(raw: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Extract the full recipe from a raw API recipe object.
|
||||
|
||||
Args:
|
||||
raw: Raw recipe dict from the API (metasync or mprecipeput response).
|
||||
|
||||
Returns:
|
||||
Dict with all recipe fields including ingredients, instructions, etc.
|
||||
"""
|
||||
prep = raw.get("prepTime")
|
||||
cook = raw.get("cookTime")
|
||||
srv = raw.get("serves")
|
||||
|
||||
# ingredientsList is auto-parsed by the server; normalise to a plain list of names.
|
||||
ingredients_list_raw: list[dict[str, Any]] = raw.get("ingredientsList") or []
|
||||
ingredients_parsed = [item.get("name") for item in ingredients_list_raw if item.get("name")]
|
||||
|
||||
return {
|
||||
"id": raw.get("metaId"),
|
||||
"name": raw.get("name"),
|
||||
"description": raw.get("description") or None,
|
||||
"ingredients": raw.get("ingredients") or None,
|
||||
"ingredients_parsed": ingredients_parsed,
|
||||
"instructions": raw.get("instructions") or None,
|
||||
"prep_time_minutes": int(prep) if prep else None,
|
||||
"cook_time_minutes": int(cook) if cook else None,
|
||||
"serves": int(srv) if srv else None,
|
||||
"url": raw.get("url") or None,
|
||||
"is_favorite": raw.get("isFavorite") == "true",
|
||||
"can_delete": raw.get("rights", {}).get("canDelete") == "true",
|
||||
"can_update": raw.get("rights", {}).get("canUpdate") == "true",
|
||||
"created_at": raw.get("creationDate"),
|
||||
"account_id": raw.get("accountId"),
|
||||
}
|
||||
|
||||
|
||||
def build_create_params(
|
||||
name: str,
|
||||
description: str | None = None,
|
||||
ingredients: str | None = None,
|
||||
instructions: str | None = None,
|
||||
prep_time_minutes: int | None = None,
|
||||
cook_time_minutes: int | None = None,
|
||||
serves: int | None = None,
|
||||
url: str | None = None,
|
||||
) -> dict[str, str]:
|
||||
"""Build the form parameters for a mprecipeput create call.
|
||||
|
||||
The Family Wall API requires the 'recipe.' prefix for all recipe fields.
|
||||
isRecipe='true' is always required.
|
||||
|
||||
Args:
|
||||
name: Recipe title (required).
|
||||
description: Optional description.
|
||||
ingredients: Optional free-text ingredients, lines separated by \\n.
|
||||
instructions: Optional free-text cooking instructions, lines separated by \\n.
|
||||
prep_time_minutes: Optional preparation time in minutes.
|
||||
cook_time_minutes: Optional cooking time in minutes.
|
||||
serves: Optional number of servings.
|
||||
url: Optional external URL (e.g. original recipe source).
|
||||
|
||||
Returns:
|
||||
Dict of form parameters ready to send to mprecipeput.
|
||||
"""
|
||||
params: dict[str, str] = {
|
||||
"recipe.name": name,
|
||||
"recipe.isRecipe": "true",
|
||||
}
|
||||
if description is not None:
|
||||
params["recipe.description"] = description
|
||||
if ingredients is not None:
|
||||
params["recipe.ingredients"] = ingredients
|
||||
if instructions is not None:
|
||||
params["recipe.instructions"] = instructions
|
||||
if prep_time_minutes is not None:
|
||||
params["recipe.prepTime"] = str(prep_time_minutes)
|
||||
if cook_time_minutes is not None:
|
||||
params["recipe.cookTime"] = str(cook_time_minutes)
|
||||
if serves is not None:
|
||||
params["recipe.serves"] = str(serves)
|
||||
if url is not None:
|
||||
params["recipe.url"] = url
|
||||
return params
|
||||
@@ -11,6 +11,11 @@ from mcp.server.fastmcp import FastMCP
|
||||
from mcp_familywall.auth import get_credentials
|
||||
from mcp_familywall.fw_client import FamilyWallClient, FamilyWallError
|
||||
from mcp_familywall.modules.lists import translate_name
|
||||
from mcp_familywall.modules.recipes import (
|
||||
build_create_params,
|
||||
parse_recipe_full,
|
||||
parse_recipe_summary,
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -532,9 +537,7 @@ def delete_category(category_id: str) -> str:
|
||||
except (KeyError, TypeError):
|
||||
raw_cats = []
|
||||
|
||||
cat_obj = next(
|
||||
(c for c in raw_cats if c.get("metaId") == category_id), None
|
||||
)
|
||||
cat_obj = next((c for c in raw_cats if c.get("metaId") == category_id), None)
|
||||
if cat_obj is None:
|
||||
client.logout()
|
||||
return f"Error: Category '{category_id}' not found."
|
||||
@@ -547,7 +550,9 @@ def delete_category(category_id: str) -> str:
|
||||
"error": "System categories cannot be deleted.",
|
||||
"id": category_id,
|
||||
"name": cat_obj.get("name"),
|
||||
"hint": "Only custom categories (custom=true in get_categories) can be deleted.",
|
||||
"hint": (
|
||||
"Only custom categories (custom=true in get_categories) can be deleted."
|
||||
),
|
||||
},
|
||||
ensure_ascii=False,
|
||||
indent=2,
|
||||
@@ -789,7 +794,10 @@ def update_task(
|
||||
and assignee_ids is None
|
||||
and list_id is None
|
||||
):
|
||||
return "Error: At least one of 'text', 'description', 'category_id', 'due_date', 'clear_due_date', 'assignee_ids', or 'list_id' must be provided."
|
||||
return (
|
||||
"Error: At least one of 'text', 'description', 'category_id', 'due_date',"
|
||||
" 'clear_due_date', 'assignee_ids', or 'list_id' must be provided."
|
||||
)
|
||||
|
||||
params: dict[str, Any] = {"metaId": task_id}
|
||||
if text is not None:
|
||||
@@ -994,9 +1002,7 @@ def delete_list(list_id: str) -> str:
|
||||
except (KeyError, TypeError):
|
||||
raw_lists = []
|
||||
|
||||
list_obj = next(
|
||||
(lst for lst in raw_lists if lst.get("metaId") == list_id), None
|
||||
)
|
||||
list_obj = next((lst for lst in raw_lists if lst.get("metaId") == list_id), None)
|
||||
if list_obj is None:
|
||||
client.logout()
|
||||
return f"Error: List '{list_id}' not found."
|
||||
@@ -1115,6 +1121,222 @@ def like_post(post_id: str, like: bool = True) -> str:
|
||||
return json.dumps(result, ensure_ascii=False, indent=2)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helper: fetch all raw recipes
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _get_raw_recipes() -> list[dict[str, Any]]:
|
||||
"""Login, call metasync with id='recipe', logout and return the raw recipe list.
|
||||
|
||||
Raises:
|
||||
RuntimeError: On credential or API errors.
|
||||
"""
|
||||
data = _authenticated_call("metasync", {"id": "recipe"})
|
||||
try:
|
||||
items = data["a00"]["r"]["r"]["updatedCreated"]
|
||||
if isinstance(items, list):
|
||||
return items # type: ignore[return-value]
|
||||
except (KeyError, TypeError):
|
||||
pass
|
||||
return []
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tool: get_recipes
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def get_recipes() -> str:
|
||||
"""Return all family recipes as a compact JSON list.
|
||||
|
||||
Returns:
|
||||
JSON list of recipe summary objects with keys: id, name, description,
|
||||
prep_time_minutes, cook_time_minutes, serves, can_delete.
|
||||
Returns an error message string on failure.
|
||||
"""
|
||||
try:
|
||||
raw_recipes = _get_raw_recipes()
|
||||
except RuntimeError as exc:
|
||||
return f"Error: {exc}"
|
||||
|
||||
result = [parse_recipe_summary(r) for r in raw_recipes]
|
||||
return json.dumps(result, ensure_ascii=False, indent=2)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tool: get_recipe
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def get_recipe(recipe_id: str) -> str:
|
||||
"""Return a single recipe in full detail.
|
||||
|
||||
Args:
|
||||
recipe_id: Recipe metaId from get_recipes
|
||||
(e.g. ``"recipe/23431854_10968866"``).
|
||||
|
||||
Returns:
|
||||
JSON object with full recipe fields including ingredients, instructions,
|
||||
ingredients_parsed, url, is_favorite, can_delete, can_update, created_at.
|
||||
Returns an error message string on failure or when not found.
|
||||
"""
|
||||
try:
|
||||
raw_recipes = _get_raw_recipes()
|
||||
except RuntimeError as exc:
|
||||
return f"Error: {exc}"
|
||||
|
||||
raw = next((r for r in raw_recipes if r.get("metaId") == recipe_id), None)
|
||||
if raw is None:
|
||||
return f"Error: Recipe '{recipe_id}' not found."
|
||||
|
||||
return json.dumps(parse_recipe_full(raw), ensure_ascii=False, indent=2)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tool: create_recipe
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def create_recipe(
|
||||
name: str,
|
||||
description: str | None = None,
|
||||
ingredients: str | None = None,
|
||||
instructions: str | None = None,
|
||||
prep_time_minutes: int | None = None,
|
||||
cook_time_minutes: int | None = None,
|
||||
serves: int | None = None,
|
||||
url: str | None = None,
|
||||
) -> str:
|
||||
"""Create a new recipe in the Family Wall recipe box.
|
||||
|
||||
IMPORTANT: Ask the user for confirmation before calling this tool.
|
||||
|
||||
Args:
|
||||
name: Recipe title (required).
|
||||
description: Optional short description or teaser text.
|
||||
ingredients: Optional ingredient list as free text.
|
||||
Use newlines (``\\n``) to separate items.
|
||||
Example: ``"200g Mehl\\n3 Eier\\n100ml Milch"``
|
||||
The server auto-parses this into a structured list.
|
||||
instructions: Optional cooking instructions as free text.
|
||||
Use newlines (``\\n``) to separate steps.
|
||||
prep_time_minutes: Optional preparation time in minutes.
|
||||
cook_time_minutes: Optional cooking/baking time in minutes.
|
||||
serves: Optional number of servings.
|
||||
url: Optional external URL (e.g. original recipe source).
|
||||
|
||||
Returns:
|
||||
JSON with the new recipe's full fields on success, or an error message.
|
||||
"""
|
||||
params = build_create_params(
|
||||
name=name,
|
||||
description=description,
|
||||
ingredients=ingredients,
|
||||
instructions=instructions,
|
||||
prep_time_minutes=prep_time_minutes,
|
||||
cook_time_minutes=cook_time_minutes,
|
||||
serves=serves,
|
||||
url=url,
|
||||
)
|
||||
|
||||
try:
|
||||
data = _authenticated_call("mprecipeput", params)
|
||||
except RuntimeError as exc:
|
||||
return f"Error: {exc}"
|
||||
|
||||
try:
|
||||
recipe_obj = data["a00"]["r"]["r"]
|
||||
if not isinstance(recipe_obj, dict) or "metaId" not in recipe_obj:
|
||||
raise TypeError("unexpected shape")
|
||||
except (KeyError, TypeError):
|
||||
return json.dumps(
|
||||
{"warning": "Unexpected mprecipeput response structure", "raw": data},
|
||||
ensure_ascii=False,
|
||||
indent=2,
|
||||
)
|
||||
|
||||
result = parse_recipe_full(recipe_obj)
|
||||
result["created"] = True
|
||||
return json.dumps(result, ensure_ascii=False, indent=2)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tool: delete_recipe
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def delete_recipe(recipe_id: str) -> str:
|
||||
"""Permanently delete a recipe from the Family Wall recipe box.
|
||||
|
||||
IMPORTANT: Ask the user for confirmation before calling this tool.
|
||||
|
||||
Only recipes created by the current account (can_delete=true in
|
||||
get_recipes output) can be deleted. The action cannot be undone.
|
||||
|
||||
Args:
|
||||
recipe_id: Recipe metaId from get_recipes
|
||||
(e.g. ``"recipe/23431854_10968866"``).
|
||||
|
||||
Returns:
|
||||
JSON success indicator or an error message.
|
||||
"""
|
||||
# Verify the recipe exists and is deletable, then delete — single session.
|
||||
try:
|
||||
email, password = get_credentials()
|
||||
except RuntimeError as exc:
|
||||
return f"Error: {exc}"
|
||||
|
||||
recipe_obj: dict[str, Any] | None = None
|
||||
try:
|
||||
with FamilyWallClient() as client:
|
||||
client.login(email, password)
|
||||
|
||||
# Fetch all recipes and verify the target can be deleted.
|
||||
raw_data = client.call("metasync", {"id": "recipe"})
|
||||
try:
|
||||
items: list[dict[str, Any]] = raw_data["a00"]["r"]["r"]["updatedCreated"]
|
||||
except (KeyError, TypeError):
|
||||
items = []
|
||||
|
||||
recipe_obj = next((r for r in items if r.get("metaId") == recipe_id), None)
|
||||
if recipe_obj is None:
|
||||
client.logout()
|
||||
return f"Error: Recipe '{recipe_id}' not found."
|
||||
|
||||
can_delete: str | None = (recipe_obj.get("rights") or {}).get("canDelete")
|
||||
if can_delete != "true":
|
||||
client.logout()
|
||||
return json.dumps(
|
||||
{
|
||||
"error": "Recipe cannot be deleted.",
|
||||
"id": recipe_id,
|
||||
"name": recipe_obj.get("name"),
|
||||
"hint": "Only recipes you created (can_delete=true) can be deleted.",
|
||||
},
|
||||
ensure_ascii=False,
|
||||
indent=2,
|
||||
)
|
||||
|
||||
# Verified — delete in the same session.
|
||||
client.call("metadelete", {"id": recipe_id})
|
||||
client.logout()
|
||||
except FamilyWallError as exc:
|
||||
return f"Error: Family Wall API error: {exc}"
|
||||
except Exception as exc:
|
||||
return f"Error: Connection error: {exc}"
|
||||
|
||||
return json.dumps(
|
||||
{"deleted": True, "id": recipe_id, "name": recipe_obj.get("name")},
|
||||
ensure_ascii=False,
|
||||
indent=2,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Factory
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
Reference in New Issue
Block a user