marcus
0e7c4da362
- Add get_wall_posts: read recent wall posts with like/comment counts - Add create_wall_post: publish new status posts to the wall - Add add_comment: add comments to wall posts and activities - like_post already supports both wall posts and activities (v1.2.0) - Update README.md with new Wall & Activities section - Update CLAUDE.md with v1.3.0 and tool reorganization - Update CHANGELOG.md with v1.3.0 release notes - Add wallpublish and walladdComment documentation to SPEC.md Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
3040 lines
106 KiB
Python
3040 lines
106 KiB
Python
"""MCP server for Family Wall — tools for circles, lists, tasks (read + write)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import contextlib
|
|
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
import sys
|
|
from typing import Any
|
|
|
|
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.circles import DEFAULT_INVITE_ROLE
|
|
from mcp_familywall.modules.lists import translate_name
|
|
from mcp_familywall.modules.recipes import (
|
|
build_create_params,
|
|
build_update_params,
|
|
parse_recipe_full,
|
|
parse_recipe_summary,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
mcp = FastMCP("familywall")
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _err(msg: str) -> str:
|
|
"""Return a JSON-encoded error response."""
|
|
return json.dumps({"error": msg}, ensure_ascii=False)
|
|
|
|
|
|
def _validate_date(date: str) -> str | None:
|
|
"""Return None when *date* is a valid ISO YYYY-MM-DD string, else an error message."""
|
|
if not re.fullmatch(r"\d{4}-\d{2}-\d{2}", date):
|
|
return f"Invalid date format {date!r}. Expected ISO YYYY-MM-DD (e.g. '2026-04-20')."
|
|
return None
|
|
|
|
|
|
def _accgetallfamily() -> dict[str, Any]:
|
|
"""Login, call accgetallfamily, logout and return the response body.
|
|
|
|
Raises:
|
|
RuntimeError: On credential or API errors.
|
|
"""
|
|
try:
|
|
email, password = get_credentials()
|
|
except RuntimeError as exc:
|
|
raise RuntimeError(str(exc)) from exc
|
|
|
|
try:
|
|
with FamilyWallClient() as client:
|
|
client.login(email, password)
|
|
data = client.call(
|
|
"accgetallfamily",
|
|
{"a01call": "taskcategorysync", "a02call": "tasksync"},
|
|
)
|
|
client.logout()
|
|
return data
|
|
except FamilyWallError as exc:
|
|
raise RuntimeError(f"Family Wall API error: {exc}") from exc
|
|
except Exception as exc:
|
|
raise RuntimeError(f"Connection error: {exc}") from exc
|
|
|
|
|
|
def _extract_lists(data: dict[str, Any]) -> list[dict[str, Any]]:
|
|
"""Extract task lists from an accgetallfamily response.
|
|
|
|
List IDs are derived from the sortingIndexByTaskList keys present in each
|
|
taskcategorysync entry (a01.r.r.updatedCreated[]). Each key is a unique
|
|
list ID of the form ``taskList/<id>``. Names and counters are not yet
|
|
available from this path and are left as None.
|
|
|
|
Args:
|
|
data: Raw response body from accgetallfamily.
|
|
|
|
Returns:
|
|
Deduplicated list of dicts with keys id, name, type, open, total.
|
|
"""
|
|
try:
|
|
categories = data["a01"]["r"]["r"]["updatedCreated"]
|
|
if not isinstance(categories, list):
|
|
return []
|
|
except (KeyError, TypeError):
|
|
return []
|
|
|
|
seen: set[str] = set()
|
|
result: list[dict[str, Any]] = []
|
|
for cat in categories:
|
|
sorting = cat.get("sortingIndexByTaskList")
|
|
if not isinstance(sorting, dict):
|
|
continue
|
|
for list_id in sorting:
|
|
if list_id in seen:
|
|
continue
|
|
seen.add(list_id)
|
|
result.append(
|
|
{
|
|
"id": list_id,
|
|
"name": list_id, # no name in sortingIndexByTaskList; get_lists has full data
|
|
"type": None,
|
|
"open": None,
|
|
"total": None,
|
|
}
|
|
)
|
|
|
|
logger.debug("Extracted %d unique list IDs from sortingIndexByTaskList", len(result))
|
|
return result
|
|
|
|
|
|
def _extract_tasks(data: dict[str, Any]) -> list[dict[str, Any]]:
|
|
"""Extract the list of tasks from an accgetallfamily response.
|
|
|
|
Tasks live under a02.r.r.updatedCreated[].
|
|
|
|
Args:
|
|
data: Raw response body from accgetallfamily.
|
|
|
|
Returns:
|
|
List of raw task dicts (may be empty).
|
|
"""
|
|
try:
|
|
items = data["a02"]["r"]["r"]["updatedCreated"]
|
|
if isinstance(items, list):
|
|
return items # type: ignore[return-value]
|
|
except (KeyError, TypeError):
|
|
pass
|
|
return []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tool: get_circles
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@mcp.tool()
|
|
def get_circles() -> str:
|
|
"""Return all Family Wall circles as JSON list of {id, name}."""
|
|
try:
|
|
raw_circles = _famlistfamily()
|
|
except RuntimeError as exc:
|
|
return _err(str(exc))
|
|
|
|
result = [{"id": c.get("metaId"), "name": c.get("name")} for c in raw_circles]
|
|
return json.dumps(result, ensure_ascii=False, indent=2)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helper: famlistfamily raw circles
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _circle_id_from_list_id(list_id: str) -> str | None:
|
|
"""Derive the circle metaId from a list metaId.
|
|
|
|
The list metaId format is ``taskList/<family_num>_<list_num>``.
|
|
This function returns ``"family/<family_num>"``.
|
|
|
|
Args:
|
|
list_id: List metaId (e.g. ``"taskList/23431854_29759623"``).
|
|
|
|
Returns:
|
|
Circle metaId (e.g. ``"family/23431854"``), or ``None`` when the
|
|
format cannot be parsed.
|
|
"""
|
|
bare = list_id.removeprefix("taskList/")
|
|
parts = bare.split("_", 1)
|
|
if len(parts) == 2 and parts[0].isdigit():
|
|
return f"family/{parts[0]}"
|
|
return None
|
|
|
|
|
|
def _famlistfamily() -> list[dict[str, Any]]:
|
|
"""Login, call famlistfamily, logout and return the raw circle list.
|
|
|
|
Raises:
|
|
RuntimeError: On credential or API errors.
|
|
"""
|
|
try:
|
|
email, password = get_credentials()
|
|
except RuntimeError as exc:
|
|
raise RuntimeError(str(exc)) from exc
|
|
|
|
try:
|
|
with FamilyWallClient() as client:
|
|
client.login(email, password)
|
|
data = client.call("famlistfamily")
|
|
client.logout()
|
|
except FamilyWallError as exc:
|
|
raise RuntimeError(f"Family Wall API error: {exc}") from exc
|
|
except Exception as exc:
|
|
raise RuntimeError(f"Connection error: {exc}") from exc
|
|
|
|
try:
|
|
raw = data["a00"]["r"]["r"]
|
|
if not isinstance(raw, list):
|
|
raise TypeError("a00.r.r is not a list")
|
|
return raw
|
|
except (KeyError, TypeError) as exc:
|
|
raise RuntimeError(f"Unexpected famlistfamily response structure: {exc}") from exc
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tool: get_members
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@mcp.tool()
|
|
def get_members(circle_id: str | None = None) -> str:
|
|
"""Return Family Wall circle members as JSON, optionally filtered by circle.
|
|
|
|
Args:
|
|
circle_id: Optional circle ID from get_circles (e.g. ``family/23431854``).
|
|
When omitted all members across all circles are returned.
|
|
|
|
Returns:
|
|
JSON list of member objects with keys id, name, email, role, right,
|
|
color, avatar, circle_id, circle_name.
|
|
"""
|
|
try:
|
|
raw_circles = _famlistfamily()
|
|
except RuntimeError as exc:
|
|
return _err(str(exc))
|
|
|
|
result: list[dict[str, Any]] = []
|
|
seen_ids: set[str] = set()
|
|
|
|
for circle in raw_circles:
|
|
c_id: str = circle.get("metaId", "")
|
|
c_name: str = circle.get("name", "")
|
|
|
|
if circle_id is not None and c_id != circle_id:
|
|
continue
|
|
|
|
for member in circle.get("members") or []:
|
|
account_id: str = member.get("accountId", "")
|
|
|
|
# Deduplicate members that appear in multiple circles when no
|
|
# circle_id filter is active (same account can be in several circles).
|
|
dedup_key = f"{c_id}:{account_id}"
|
|
if dedup_key in seen_ids:
|
|
continue
|
|
seen_ids.add(dedup_key)
|
|
|
|
# Extract avatar URL from the first media entry (may be a generated
|
|
# default avatar when pictureDefault is "true").
|
|
medias = member.get("medias") or []
|
|
avatar: str | None = medias[0].get("pictureUrl") if medias else None
|
|
|
|
# Prefer firstName as display name; fall back to the name field
|
|
# (which Family Wall sets to the e-mail address by default).
|
|
display_name: str = member.get("firstName") or member.get("name", "")
|
|
|
|
# Extract e-mail from identifiers list.
|
|
email_value: str | None = None
|
|
for ident in member.get("identifiers") or []:
|
|
if ident.get("type") == "Email":
|
|
email_value = ident.get("value")
|
|
break
|
|
|
|
result.append(
|
|
{
|
|
"id": account_id,
|
|
"name": display_name,
|
|
"email": email_value,
|
|
"role": member.get("role"),
|
|
"right": member.get("right"),
|
|
"color": member.get("color"),
|
|
"avatar": avatar,
|
|
"circle_id": c_id,
|
|
"circle_name": c_name,
|
|
}
|
|
)
|
|
|
|
return json.dumps(result, ensure_ascii=False, indent=2)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tool: get_lists
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@mcp.tool()
|
|
def get_lists(scope: str | None = None) -> str:
|
|
"""Return task lists as JSON, optionally filtered to a specific circle.
|
|
|
|
Each list object includes a ``circle_id`` field with the owning circle's
|
|
metaId (e.g. ``"family/23431854"``).
|
|
|
|
Args:
|
|
scope: Optional circle filter. Accepts either:
|
|
|
|
- A circle metaId (e.g. ``"family/23447378"``) — passed directly
|
|
to the API.
|
|
- A circle display name (e.g. ``"Test Kreis 2"``) — resolved to
|
|
the matching circle metaId via ``get_circles`` first.
|
|
|
|
When ``None`` (default) all lists from all circles are returned.
|
|
|
|
Returns:
|
|
JSON list of list objects with keys id, name, type, open, total,
|
|
emoji, color, circle_id.
|
|
"""
|
|
try:
|
|
email, password = get_credentials()
|
|
except RuntimeError as exc:
|
|
return _err(str(exc))
|
|
|
|
# Build the scope param for taskgettasklists.
|
|
# When scope is provided as a circle name (not a metaId), we need to
|
|
# resolve it via famlistfamily first — done in the same session.
|
|
api_scopes: list[str] = []
|
|
if scope:
|
|
if scope.startswith("family/"):
|
|
api_scopes = [scope]
|
|
else:
|
|
# Treat as circle name — look up the metaId.
|
|
try:
|
|
circles = _famlistfamily()
|
|
except RuntimeError as exc:
|
|
return _err(str(exc))
|
|
matched = next((c for c in circles if c.get("name") == scope), None)
|
|
if matched is None:
|
|
circle_names = [c.get("name") for c in circles]
|
|
return json.dumps(
|
|
{
|
|
"error": f"Circle not found: {scope!r}",
|
|
"available_circles": circle_names,
|
|
},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
api_scopes = [matched["metaId"]]
|
|
else:
|
|
# No scope filter: fetch all circles and iterate over them.
|
|
try:
|
|
circles = _famlistfamily()
|
|
api_scopes = [c["metaId"] for c in circles if "metaId" in c]
|
|
except RuntimeError as exc:
|
|
return _err(str(exc))
|
|
|
|
try:
|
|
with FamilyWallClient() as client:
|
|
client.login(email, password)
|
|
all_lists: list[dict[str, Any]] = []
|
|
for circle_scope in api_scopes:
|
|
params: dict[str, Any] = {"scope": circle_scope}
|
|
data = client.call("taskgettasklists", params)
|
|
raw_lists: list[dict[str, Any]] | None = None
|
|
try:
|
|
candidate = data["a00"]["r"]["r"]
|
|
if isinstance(candidate, list):
|
|
raw_lists = candidate
|
|
elif isinstance(candidate, dict) and isinstance(
|
|
candidate.get("updatedCreated"), list
|
|
):
|
|
raw_lists = candidate["updatedCreated"]
|
|
except (KeyError, TypeError):
|
|
pass
|
|
|
|
if raw_lists is not None:
|
|
all_lists.extend(raw_lists)
|
|
|
|
client.logout()
|
|
except FamilyWallError as exc:
|
|
return _err(str(exc))
|
|
except Exception as exc:
|
|
return _err(f"Connection error: {exc}")
|
|
|
|
result = []
|
|
for item in all_lists:
|
|
# emoji: API returns "" when unset — normalise to None for a clean JSON null.
|
|
# color: API omits the key entirely when unset — .get() returns None directly.
|
|
raw_emoji: str = item.get("emoji", "")
|
|
result.append(
|
|
{
|
|
"id": item.get("metaId"),
|
|
"name": translate_name(item.get("name", "")),
|
|
"type": item.get("taskListType"),
|
|
"open": item.get("remainingTaskNumber"),
|
|
"total": item.get("totalTaskNumber"),
|
|
"emoji": raw_emoji if raw_emoji else None,
|
|
"color": item.get("color") or None,
|
|
"circle_id": item.get("familyId"),
|
|
}
|
|
)
|
|
|
|
return json.dumps(result, ensure_ascii=False, indent=2)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tool: get_tasks
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@mcp.tool()
|
|
def get_tasks(list_id: str, only_open: bool = True) -> str:
|
|
"""Return tasks for a list as JSON. list_id from get_lists. only_open=True filters completed."""
|
|
try:
|
|
data = _accgetallfamily()
|
|
except RuntimeError as exc:
|
|
return _err(str(exc))
|
|
|
|
raw_tasks = _extract_tasks(data)
|
|
|
|
_fw_debug = os.environ.get("FW_DEBUG") == "1"
|
|
_known_fields = {
|
|
# output fields
|
|
"metaId",
|
|
"text",
|
|
"description",
|
|
"complete",
|
|
"taskCategoryId",
|
|
"dueDate",
|
|
"assigneeIds",
|
|
"taskListId",
|
|
# recurrency fields
|
|
"recurrency",
|
|
"recurrencyInterval",
|
|
"rrule",
|
|
"byDay",
|
|
"recurrencyDeletedOccurence",
|
|
"reminder",
|
|
# always-present housekeeping fields
|
|
"moodStarShortcut",
|
|
"bestMoment",
|
|
"lastAction",
|
|
"lastActionAuthor",
|
|
"lastActionDate",
|
|
"categories",
|
|
"moodMap",
|
|
"sortingIndex",
|
|
"comments",
|
|
"editable",
|
|
"creationDate",
|
|
"completedDate",
|
|
"familyId",
|
|
"toAll",
|
|
"accountId",
|
|
"medias",
|
|
"modifDate",
|
|
"assignee",
|
|
"taskId",
|
|
"clientOpId",
|
|
}
|
|
|
|
result = []
|
|
for task in raw_tasks:
|
|
if task.get("taskListId") != list_id:
|
|
continue
|
|
completed = str(task.get("complete", "false")).lower() == "true"
|
|
if only_open and completed:
|
|
continue
|
|
|
|
if _fw_debug:
|
|
unknown = {k: v for k, v in task.items() if k not in _known_fields}
|
|
if unknown:
|
|
print(
|
|
f"[FW_DEBUG] task {task.get('metaId')} unknown fields: "
|
|
+ json.dumps(unknown, ensure_ascii=False),
|
|
file=sys.stderr,
|
|
)
|
|
|
|
raw_recurrency = task.get("recurrency")
|
|
raw_interval = task.get("recurrencyInterval")
|
|
recurrency_interval = int(raw_interval) if raw_interval is not None else None
|
|
|
|
raw_reminder = task.get("reminder")
|
|
if raw_reminder and isinstance(raw_reminder, dict) and raw_reminder.get("reminderUnit"):
|
|
reminder = {
|
|
"unit": raw_reminder["reminderUnit"],
|
|
"value": int(raw_reminder.get("reminderValue", 0)),
|
|
}
|
|
else:
|
|
reminder = None
|
|
|
|
result.append(
|
|
{
|
|
"id": task.get("metaId"),
|
|
"text": task.get("text"),
|
|
"description": task.get("description"),
|
|
"completed": completed,
|
|
"category_id": task.get("taskCategoryId"),
|
|
"due_date": task.get("dueDate"),
|
|
"assignee_ids": task.get("assigneeIds") or [],
|
|
"recurrency": raw_recurrency,
|
|
"recurrency_interval": recurrency_interval,
|
|
"rrule": task.get("rrule"),
|
|
"reminder": reminder,
|
|
}
|
|
)
|
|
|
|
return json.dumps(result, ensure_ascii=False, indent=2)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tool: get_categories
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@mcp.tool()
|
|
def get_categories(list_id: str, locale: str = "de") -> str:
|
|
"""Return the task categories available for a list as JSON.
|
|
|
|
Only shopping lists (taskListType=SHOPPING_LIST) have categories. TODO
|
|
lists return an empty list. System categories are filtered by locale so only
|
|
the language-appropriate names are returned (default: German). Custom categories
|
|
have no locale field and are always displayed regardless of the locale parameter.
|
|
|
|
Use the returned ``id`` values as the ``category_id`` parameter in
|
|
``create_task`` and ``update_task``.
|
|
|
|
Args:
|
|
list_id: List ID from get_lists (e.g. ``taskList/23431854_29740942``).
|
|
locale: BCP-47 language code for category names (default ``"de"``).
|
|
Supported values seen in API: de, en, fr, es, it, nl, pt, sv,
|
|
ru, ko, ja.
|
|
|
|
Returns:
|
|
JSON list of {id, name, emoji} objects ordered by sortingIndex,
|
|
or an empty list when the list type has no categories.
|
|
"""
|
|
# Resolve the list's taskListType from taskgettasklists so we can match
|
|
# only the categories that belong to the same list type. Non-fatal: if
|
|
# the lookup fails we skip the type filter and return all locale matches.
|
|
list_type: str | None = None
|
|
try:
|
|
list_data = _authenticated_call("taskgettasklists", {})
|
|
raw_lists = list_data.get("a00", {}).get("r", {}).get("r", []) or []
|
|
for lst in raw_lists:
|
|
if lst.get("metaId") == list_id:
|
|
list_type = lst.get("taskListType")
|
|
break
|
|
except RuntimeError:
|
|
pass
|
|
|
|
try:
|
|
data = _accgetallfamily()
|
|
except RuntimeError as exc:
|
|
return _err(str(exc))
|
|
|
|
try:
|
|
raw_cats = data["a01"]["r"]["r"]["updatedCreated"]
|
|
if not isinstance(raw_cats, list):
|
|
raise TypeError("a01.r.r.updatedCreated is not a list")
|
|
except (KeyError, TypeError):
|
|
return json.dumps(
|
|
{"warning": "Unexpected taskcategorysync response structure", "raw": data.get("a01")},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
|
|
# Filter by locale (exact match) and taskListType (when known).
|
|
# sortingIndexByTaskList is NOT used for filtering: all categories are
|
|
# assigned to all lists regardless of type, making it an unreliable signal.
|
|
#
|
|
# Custom categories (rights.canDelete=true) bypass both filters: they have
|
|
# no locale or taskListType set by the API and must always be returned.
|
|
matched: list[tuple[int, dict[str, Any]]] = []
|
|
for cat in raw_cats:
|
|
is_custom = cat.get("rights", {}).get("canDelete") == "true"
|
|
if not is_custom:
|
|
if cat.get("locale") != locale:
|
|
continue
|
|
if list_type is not None and cat.get("taskListType") != list_type:
|
|
continue
|
|
sort_val = int(cat.get("initialSortingIndex") or 0)
|
|
matched.append((sort_val, cat))
|
|
|
|
matched.sort(key=lambda t: t[0])
|
|
result = [
|
|
{
|
|
"id": cat.get("metaId"),
|
|
"name": cat.get("name"),
|
|
"emoji": cat.get("emoji"),
|
|
# custom=True means the category was created by the family and can be
|
|
# deleted with delete_category. System categories have no canDelete right.
|
|
"custom": cat.get("rights", {}).get("canDelete") == "true",
|
|
}
|
|
for _, cat in matched
|
|
]
|
|
return json.dumps(result, ensure_ascii=False, indent=2)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tool: create_category
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@mcp.tool()
|
|
def create_category(list_id: str, name: str, icon: str | None = None) -> str:
|
|
"""Create a new custom category for a shopping list.
|
|
|
|
IMPORTANT: Ask the user for confirmation before calling this tool.
|
|
|
|
Custom categories appear alongside system categories when assigning
|
|
categories to tasks via ``create_task`` or ``update_task``. They can
|
|
later be removed with ``delete_category``.
|
|
|
|
Note: although ``list_id`` is accepted for context, the Family Wall API
|
|
assigns new categories to all lists in the family — there is no
|
|
per-list restriction.
|
|
|
|
Args:
|
|
list_id: Target list ID from get_lists (e.g. ``taskList/123_456``).
|
|
Only SHOPPING_LIST lists have categories; the parameter is
|
|
accepted for user context but does not restrict category scope.
|
|
name: Display name for the new category (e.g. ``"Bio-Produkte"``).
|
|
icon: Optional icon for the category. Pass a Unicode emoji character
|
|
(e.g. ``"🌿"``) or any short string identifier. When omitted the
|
|
category has no icon.
|
|
|
|
Returns:
|
|
JSON with the new category's ``id`` and ``name`` on success, or an
|
|
error message.
|
|
"""
|
|
params: dict[str, Any] = {"name": name}
|
|
if icon:
|
|
params["emoji"] = icon
|
|
|
|
try:
|
|
data = _authenticated_call("taskcategoryput", params)
|
|
except RuntimeError as exc:
|
|
return _err(str(exc))
|
|
|
|
try:
|
|
cat_obj = data["a00"]["r"]["r"]
|
|
meta_id: str = cat_obj["metaId"]
|
|
except (KeyError, TypeError):
|
|
return json.dumps(
|
|
{"warning": "Unexpected taskcategoryput response structure", "raw": data},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
|
|
return json.dumps(
|
|
{"created": True, "id": meta_id, "name": cat_obj.get("name", name)},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tool: delete_category
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@mcp.tool()
|
|
def delete_category(category_id: str) -> str:
|
|
"""Permanently delete a custom category.
|
|
|
|
IMPORTANT: Ask the user for confirmation before calling this tool.
|
|
|
|
Only custom (user-created) categories can be deleted. System categories
|
|
supplied by Family Wall (identified by ``custom=false`` in
|
|
``get_categories`` output) are protected and this tool will refuse to
|
|
delete them.
|
|
|
|
Args:
|
|
category_id: Category metaId from get_categories
|
|
(e.g. ``taskCategory/23431854_4956637``). Must be a custom
|
|
category (``custom=true`` in get_categories output).
|
|
|
|
Returns:
|
|
JSON success indicator or an error message.
|
|
"""
|
|
# Safety check + delete in a SINGLE session to minimise API round-trips.
|
|
# Previously two separate sessions were used (accgetallfamily + taskcategorydelete),
|
|
# causing 6 HTTP calls. One session = 4 HTTP calls (login, check, delete, logout).
|
|
try:
|
|
email, password = get_credentials()
|
|
except RuntimeError as exc:
|
|
return _err(str(exc))
|
|
|
|
cat_obj: dict[str, Any] | None = None
|
|
try:
|
|
with FamilyWallClient() as client:
|
|
client.login(email, password)
|
|
|
|
# Fetch categories and verify the target is custom (canDelete=true).
|
|
data = client.call(
|
|
"accgetallfamily",
|
|
{"a01call": "taskcategorysync", "a02call": "tasksync"},
|
|
)
|
|
try:
|
|
raw_cats: list[dict[str, Any]] = data["a01"]["r"]["r"]["updatedCreated"]
|
|
except (KeyError, TypeError):
|
|
raw_cats = []
|
|
|
|
cat_obj = next((c for c in raw_cats if c.get("metaId") == category_id), None)
|
|
if cat_obj is None:
|
|
client.logout()
|
|
return _err(f"Category '{category_id}' not found.")
|
|
|
|
can_delete: str | None = cat_obj.get("rights", {}).get("canDelete")
|
|
if can_delete != "true":
|
|
client.logout()
|
|
return json.dumps(
|
|
{
|
|
"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."
|
|
),
|
|
},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
|
|
# Verified — delete in the same session.
|
|
client.call("taskcategorydelete", {"id": category_id})
|
|
client.logout()
|
|
except FamilyWallError as exc:
|
|
return _err(f"Family Wall API error: {exc}")
|
|
except Exception as exc:
|
|
return _err(f"Connection error: {exc}")
|
|
|
|
return json.dumps(
|
|
{"deleted": True, "id": category_id, "name": cat_obj.get("name")},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tool: get_activities
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@mcp.tool()
|
|
def get_activities(limit: int = 20) -> str:
|
|
"""Return recent Family Wall wall activities as JSON. limit controls max number of results."""
|
|
try:
|
|
email, password = get_credentials()
|
|
except RuntimeError as exc:
|
|
return _err(str(exc))
|
|
|
|
# Load member data to resolve author IDs to display names.
|
|
# Non-fatal: fall back to raw account IDs when member lookup fails.
|
|
author_map: dict[str, str] = {}
|
|
try:
|
|
for circle in _famlistfamily():
|
|
for member in circle.get("members") or []:
|
|
acc_id: str = member.get("accountId", "")
|
|
display = member.get("firstName") or member.get("name") or acc_id
|
|
if acc_id:
|
|
author_map[acc_id] = display
|
|
except RuntimeError:
|
|
pass
|
|
|
|
try:
|
|
with FamilyWallClient() as client:
|
|
client.login(email, password)
|
|
data = client.call("wallget", {"nb": str(limit)})
|
|
client.logout()
|
|
except FamilyWallError as exc:
|
|
return _err(str(exc))
|
|
except Exception as exc:
|
|
return _err(f"Connection error: {exc}")
|
|
|
|
# Try known response patterns; fall back to raw JSON for verification.
|
|
raw_activities: list[dict[str, Any]] | None = None
|
|
try:
|
|
candidate = data["a00"]["r"]["r"]
|
|
if isinstance(candidate, list):
|
|
raw_activities = candidate
|
|
elif isinstance(candidate, dict) and isinstance(candidate.get("updatedCreated"), list):
|
|
raw_activities = candidate["updatedCreated"]
|
|
except (KeyError, TypeError):
|
|
pass
|
|
|
|
if raw_activities is None:
|
|
# Response structure not yet verified — return raw JSON for inspection.
|
|
return json.dumps(
|
|
{"warning": "Unexpected wallget response structure", "raw": data},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
|
|
result = []
|
|
for item in raw_activities:
|
|
raw_author: str = item.get("accountId", "")
|
|
result.append(
|
|
{
|
|
"id": item.get("metaId"),
|
|
"type": item.get("refType"),
|
|
"text": item.get("text"),
|
|
"date": item.get("creationDate"),
|
|
"author": author_map.get(raw_author, raw_author),
|
|
"author_id": raw_author,
|
|
}
|
|
)
|
|
|
|
return json.dumps(result, ensure_ascii=False, indent=2)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helper: authenticated single call
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
# Reminder units accepted by the Family Wall API (verified 2026-04-17 via
|
|
# taskupdate2 fuzz: WEEK is rejected, MINUTE/HOUR/DAY succeed).
|
|
_VALID_REMINDER_UNITS = ("MINUTE", "HOUR", "DAY")
|
|
|
|
|
|
def _build_reminder_params(
|
|
reminder_unit: str | None,
|
|
reminder_value: int | None,
|
|
clear_reminder: bool,
|
|
) -> tuple[dict[str, Any], str | None]:
|
|
"""Return (params, error_message). Params are the dot-notation reminder.* fields.
|
|
|
|
Encoding note: taskupdate2 only accepts reminders via dot-notation
|
|
(``reminder.reminderUnit``, ``reminder.reminderValue``, ``reminder.reminderType``).
|
|
Flat top-level keys, JSON strings, and bracket notation are silently ignored
|
|
by the server — verified 2026-04-17.
|
|
|
|
Clearing requires sending the full "inactive" block
|
|
(reminderType=NONE, reminderValue=0, reminderUnit=MINUTE, localId=0) —
|
|
partial updates return ``"task reminder invalid"``.
|
|
"""
|
|
if clear_reminder and (reminder_unit is not None or reminder_value is not None):
|
|
return {}, "'clear_reminder' cannot be combined with 'reminder_unit' or 'reminder_value'."
|
|
|
|
if clear_reminder:
|
|
return {
|
|
"reminder.reminderUnit": "MINUTE",
|
|
"reminder.reminderValue": "0",
|
|
"reminder.reminderType": "NONE",
|
|
"reminder.localId": "0",
|
|
}, None
|
|
|
|
if reminder_unit is None and reminder_value is None:
|
|
return {}, None
|
|
|
|
if reminder_unit is None or reminder_value is None:
|
|
return {}, "'reminder_unit' and 'reminder_value' must be provided together."
|
|
|
|
if reminder_unit not in _VALID_REMINDER_UNITS:
|
|
allowed = ", ".join(_VALID_REMINDER_UNITS)
|
|
return {}, f"Invalid reminder_unit {reminder_unit!r}. Allowed: {allowed}."
|
|
|
|
if reminder_value < 0:
|
|
return {}, "'reminder_value' must be non-negative."
|
|
|
|
return {
|
|
"reminder.reminderUnit": reminder_unit,
|
|
"reminder.reminderValue": str(reminder_value),
|
|
"reminder.reminderType": "SNOOZE",
|
|
"reminder.localId": "0",
|
|
}, None
|
|
|
|
|
|
def _authenticated_call(endpoint: str, params: dict[str, Any]) -> dict[str, Any]:
|
|
"""Login, call *endpoint* with *params*, logout, and return the response body.
|
|
|
|
Args:
|
|
endpoint: Family Wall API endpoint name.
|
|
params: Form parameters to send.
|
|
|
|
Raises:
|
|
RuntimeError: On credential or API errors.
|
|
"""
|
|
try:
|
|
email, password = get_credentials()
|
|
except RuntimeError as exc:
|
|
raise RuntimeError(str(exc)) from exc
|
|
|
|
try:
|
|
with FamilyWallClient() as client:
|
|
client.login(email, password)
|
|
data = client.call(endpoint, params)
|
|
client.logout()
|
|
return data
|
|
except FamilyWallError as exc:
|
|
raise RuntimeError(f"Family Wall API error: {exc}") from exc
|
|
except Exception as exc:
|
|
raise RuntimeError(f"Connection error: {exc}") from exc
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tool: create_task
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@mcp.tool()
|
|
def create_task(
|
|
list_id: str,
|
|
text: str,
|
|
description: str | None = None,
|
|
category_id: str | None = None,
|
|
due_date: str | None = None,
|
|
assignee_ids: list[str] | None = None,
|
|
reminder_unit: str | None = None,
|
|
reminder_value: int | None = None,
|
|
) -> str:
|
|
"""Create a new task in the given list.
|
|
|
|
IMPORTANT: Ask the user for confirmation before calling this tool.
|
|
|
|
Args:
|
|
list_id: Target list ID from get_lists (e.g. ``taskList/123_456``).
|
|
text: Task title / main text.
|
|
For quantities use the format "Äpfel (5x)" — item name first,
|
|
quantity in parentheses at the end. Examples:
|
|
"Äpfel (5x)", "Hackfleisch (500g)", "Joghurt (Erdbeere, 2x)".
|
|
description: Optional longer description.
|
|
category_id: Optional category metaId from get_categories
|
|
(e.g. ``taskCategory/23431854_200``).
|
|
For shopping lists: ALWAYS call get_categories first and assign
|
|
the most fitting category to each item. Never leave category_id
|
|
empty for shopping list tasks — uncategorized items are harder
|
|
to find in the store.
|
|
Example mapping (German category names from get_categories):
|
|
|
|
- Fleisch, Wurst, Speck, Kasseler → "Fleisch & Fisch"
|
|
- Obst, Gemüse, Kraut, Zwiebeln → "Obst & Gemüse"
|
|
- Bier, Wein, Saft, Wasser → "Getränke"
|
|
- Senf, Honig, Gewürze, Öl → "Zutaten & Gewürze"
|
|
- Brot, Brötchen → "Brot & Gebäck"
|
|
- Milch, Käse, Joghurt, Eier → "Milch & Käse"
|
|
|
|
For TODO lists: ignored.
|
|
due_date: Optional due date in ISO 8601 format (e.g. ``"2026-04-30T18:00:00"``).
|
|
assignee_ids: Optional list of member IDs from get_members to assign the task
|
|
(e.g. ``["23431898"]``). Empty list assigns to nobody.
|
|
reminder_unit: Reminder time unit — ``"MINUTE"``, ``"HOUR"``, or ``"DAY"``.
|
|
Must be provided together with *reminder_value*. Requires Family Wall Premium.
|
|
reminder_value: Reminder offset before the due date (e.g. ``30`` for
|
|
"30 minutes before"). Must be provided together with *reminder_unit*.
|
|
|
|
Returns:
|
|
JSON with the new task's metaId on success, or an error message.
|
|
"""
|
|
reminder_params, reminder_error = _build_reminder_params(
|
|
reminder_unit, reminder_value, clear_reminder=False
|
|
)
|
|
if reminder_error:
|
|
return _err(reminder_error)
|
|
|
|
params: dict[str, Any] = {"taskListId": list_id, "text": text}
|
|
if description:
|
|
params["description"] = description
|
|
if category_id:
|
|
params["taskCategoryId"] = category_id
|
|
if due_date is not None:
|
|
params["dueDate"] = due_date
|
|
if assignee_ids is not None:
|
|
params["assignee"] = assignee_ids if assignee_ids else ""
|
|
params.update(reminder_params)
|
|
|
|
try:
|
|
data = _authenticated_call("taskcreate2", params)
|
|
except RuntimeError as exc:
|
|
return _err(str(exc))
|
|
|
|
# Try to extract the new task's metaId from the response.
|
|
try:
|
|
meta_id = data["a00"]["r"]["r"]["metaId"]
|
|
except (KeyError, TypeError):
|
|
# Return raw response so the caller can inspect the actual structure.
|
|
return json.dumps(
|
|
{"warning": "Unexpected taskcreate2 response structure", "raw": data},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
|
|
return json.dumps({"created": True, "id": meta_id}, ensure_ascii=False, indent=2)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tool: update_task
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@mcp.tool()
|
|
def update_task(
|
|
task_id: str,
|
|
text: str | None = None,
|
|
description: str | None = None,
|
|
category_id: str | None = None,
|
|
due_date: str | None = None,
|
|
clear_due_date: bool = False,
|
|
assignee_ids: list[str] | None = None,
|
|
list_id: str | None = None,
|
|
recurrency: str | None = None,
|
|
recurrency_interval: int | None = None,
|
|
rrule: str | None = None,
|
|
clear_recurrency: bool = False,
|
|
reminder_unit: str | None = None,
|
|
reminder_value: int | None = None,
|
|
clear_reminder: bool = False,
|
|
) -> str:
|
|
"""Update an existing task's fields.
|
|
|
|
IMPORTANT: Ask the user for confirmation before calling this tool.
|
|
At least one parameter besides *task_id* must be provided.
|
|
|
|
Args:
|
|
task_id: Task metaId from get_tasks.
|
|
text: New title text (omit to leave unchanged).
|
|
description: New description (omit to leave unchanged).
|
|
category_id: New category metaId from get_categories
|
|
(e.g. ``taskCategory/23431854_200``). Only meaningful for
|
|
shopping lists; ignored for TODO lists.
|
|
due_date: New due date in ISO 8601 format (e.g. ``"2026-04-30T18:00:00"``).
|
|
Omit to leave unchanged. Cannot be used together with *clear_due_date*.
|
|
clear_due_date: Set to ``True`` to remove the due date from the task.
|
|
Cannot be used together with *due_date*.
|
|
assignee_ids: New list of member IDs from get_members (e.g. ``["23431898"]``).
|
|
Pass an empty list to remove all assignees. Omit to leave unchanged.
|
|
list_id: Move the task to a different list by providing the target list ID
|
|
from get_lists (e.g. ``"taskList/23431854_29740942"``). Omit to keep in
|
|
current list.
|
|
recurrency: Recurrence frequency — ``"DAILY"``, ``"WEEKLY"``, ``"MONTHLY"``,
|
|
``"YEARLY"``, or ``"NONE"``. Examples: daily → ``recurrency="DAILY",
|
|
recurrency_interval=1``; every 2 weeks on Friday →
|
|
``recurrency="WEEKLY", recurrency_interval=2,
|
|
rrule="FREQ=WEEKLY;INTERVAL=2;BYDAY=FR"``.
|
|
Cannot be used together with *clear_recurrency*.
|
|
recurrency_interval: Repeat every N units (e.g. ``2`` for "every 2 weeks").
|
|
Only meaningful when *recurrency* is set.
|
|
rrule: Full iCalendar RRULE string (e.g. ``"FREQ=WEEKLY;INTERVAL=2;BYDAY=FR"``).
|
|
Overrides simple recurrency fields when set.
|
|
clear_recurrency: Set to ``True`` to remove the recurrence rule from the task.
|
|
Cannot be used together with *recurrency*.
|
|
reminder_unit: Reminder time unit — ``"MINUTE"``, ``"HOUR"``, or ``"DAY"``.
|
|
Must be provided together with *reminder_value*. Requires Family Wall
|
|
Premium. Cannot be used together with *clear_reminder*.
|
|
reminder_value: Reminder offset before the due date (e.g. ``30`` for
|
|
"30 minutes before"). Must be provided together with *reminder_unit*.
|
|
clear_reminder: Set to ``True`` to remove the reminder from the task.
|
|
Cannot be used together with *reminder_unit* or *reminder_value*.
|
|
|
|
Returns:
|
|
JSON success indicator or an error message.
|
|
"""
|
|
if clear_due_date and due_date is not None:
|
|
return _err("'clear_due_date' and 'due_date' cannot be used together.")
|
|
if clear_recurrency and recurrency is not None:
|
|
return _err("'clear_recurrency' and 'recurrency' cannot be used together.")
|
|
|
|
reminder_params, reminder_error = _build_reminder_params(
|
|
reminder_unit, reminder_value, clear_reminder
|
|
)
|
|
if reminder_error:
|
|
return _err(reminder_error)
|
|
|
|
if (
|
|
text is None
|
|
and description is None
|
|
and category_id is None
|
|
and due_date is None
|
|
and not clear_due_date
|
|
and assignee_ids is None
|
|
and list_id is None
|
|
and recurrency is None
|
|
and not clear_recurrency
|
|
and not reminder_params
|
|
):
|
|
return _err(
|
|
"At least one of 'text', 'description', 'category_id', 'due_date',"
|
|
" 'clear_due_date', 'assignee_ids', 'list_id', 'recurrency',"
|
|
" 'clear_recurrency', 'reminder_unit'/'reminder_value', or"
|
|
" 'clear_reminder' must be provided."
|
|
)
|
|
|
|
params: dict[str, Any] = {"metaId": task_id}
|
|
if text is not None:
|
|
params["text"] = text
|
|
if description is not None:
|
|
params["description"] = description
|
|
if category_id is not None:
|
|
params["taskCategoryId"] = category_id
|
|
if clear_due_date:
|
|
# The FiZ framework uses "$empty" as a sentinel to clear optional date fields.
|
|
params["dueDate"] = "$empty"
|
|
elif due_date is not None:
|
|
params["dueDate"] = due_date
|
|
if assignee_ids is not None:
|
|
params["assignee"] = assignee_ids if assignee_ids else ""
|
|
if list_id is not None:
|
|
params["taskListId"] = list_id
|
|
|
|
# Recurrency fields are sent flat at the top level (verified: FiZ Ai() encoder).
|
|
if clear_recurrency:
|
|
params["recurrency"] = "NONE"
|
|
elif recurrency is not None:
|
|
params["recurrency"] = recurrency
|
|
if recurrency_interval is not None:
|
|
params["recurrencyInterval"] = recurrency_interval
|
|
if rrule is not None:
|
|
params["rrule"] = rrule
|
|
|
|
# Reminder fields use dot-notation `reminder.*` (verified: flat keys ignored).
|
|
params.update(reminder_params)
|
|
|
|
try:
|
|
_authenticated_call("taskupdate2", params)
|
|
except RuntimeError as exc:
|
|
return _err(str(exc))
|
|
|
|
# A response without 'ex'/'un' keys is treated as success by fw_client.
|
|
return json.dumps({"updated": True, "id": task_id}, ensure_ascii=False, indent=2)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tool: toggle_task
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@mcp.tool()
|
|
def toggle_task(task_id: str, complete: bool) -> str:
|
|
"""Mark a task as complete or incomplete.
|
|
|
|
IMPORTANT: Ask the user for confirmation before calling this tool.
|
|
|
|
Args:
|
|
task_id: Task metaId from get_tasks.
|
|
complete: ``True`` to mark done, ``False`` to reopen.
|
|
|
|
Returns:
|
|
JSON success indicator or an error message.
|
|
"""
|
|
params: dict[str, Any] = {
|
|
"taskId": task_id, # verified: taskmark uses 'taskId', not 'metaId'
|
|
"complete": "true" if complete else "false",
|
|
}
|
|
|
|
try:
|
|
_authenticated_call("taskmark", params)
|
|
except RuntimeError as exc:
|
|
return _err(str(exc))
|
|
|
|
return json.dumps(
|
|
{"toggled": True, "id": task_id, "complete": complete},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tool: delete_task
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@mcp.tool()
|
|
def delete_task(task_id: str) -> str:
|
|
"""Permanently delete a task. This action cannot be undone.
|
|
|
|
IMPORTANT: Ask the user for confirmation before calling this tool.
|
|
|
|
Args:
|
|
task_id: Task metaId from get_tasks.
|
|
|
|
Returns:
|
|
JSON success indicator or an error message.
|
|
"""
|
|
try:
|
|
_authenticated_call(
|
|
"metadelete", {"id": task_id}
|
|
) # verified: metadelete uses 'id', not 'metaId'
|
|
except RuntimeError as exc:
|
|
return _err(str(exc))
|
|
|
|
return json.dumps({"deleted": True, "id": task_id}, ensure_ascii=False, indent=2)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tool: clear_list
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@mcp.tool()
|
|
def clear_list(list_id: str, only_open: bool = False) -> str:
|
|
"""Delete all tasks in a list within a single authenticated session.
|
|
|
|
IMPORTANT: Ask the user for confirmation before calling this tool.
|
|
|
|
Significantly faster than deleting tasks one by one because all
|
|
delete calls share a single login/logout session.
|
|
|
|
Args:
|
|
list_id: List metaId from get_lists
|
|
(e.g. ``"taskList/16282169_29775360"``).
|
|
only_open: When ``True`` only incomplete tasks are deleted;
|
|
completed tasks are kept. Default ``False`` deletes all.
|
|
|
|
Returns:
|
|
JSON with ``deleted_count`` and ``list_id`` on success,
|
|
or an error message.
|
|
"""
|
|
if not list_id.startswith("taskList/"):
|
|
return _err("list_id must start with 'taskList/'")
|
|
|
|
try:
|
|
email, password = get_credentials()
|
|
except RuntimeError as exc:
|
|
return _err(str(exc))
|
|
|
|
try:
|
|
with FamilyWallClient() as client:
|
|
client.login(email, password)
|
|
|
|
data = client.call(
|
|
"accgetallfamily",
|
|
{"a01call": "taskcategorysync", "a02call": "tasksync"},
|
|
)
|
|
raw_tasks = _extract_tasks(data)
|
|
|
|
tasks_to_delete = [
|
|
t
|
|
for t in raw_tasks
|
|
if t.get("taskListId") == list_id
|
|
and (not only_open or str(t.get("complete", "false")).lower() != "true")
|
|
]
|
|
|
|
deleted_ids: list[str] = []
|
|
failed_ids: list[str] = []
|
|
for task in tasks_to_delete:
|
|
meta_id: str = task["metaId"]
|
|
try:
|
|
client.call("metadelete", {"id": meta_id})
|
|
deleted_ids.append(meta_id)
|
|
except Exception as e:
|
|
failed_ids.append(meta_id)
|
|
logger.warning(f"Failed to delete {meta_id}: {e}")
|
|
|
|
client.logout()
|
|
except FamilyWallError as exc:
|
|
return _err(f"Family Wall API error: {exc}")
|
|
except Exception as exc:
|
|
return _err(f"Connection error: {exc}")
|
|
|
|
result: dict[str, Any] = {
|
|
"deleted_count": len(deleted_ids),
|
|
"list_id": list_id,
|
|
}
|
|
if failed_ids:
|
|
result["failed_count"] = len(failed_ids)
|
|
result["failed_ids"] = failed_ids
|
|
return json.dumps(result, ensure_ascii=False, indent=2)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tool: create_list
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@mcp.tool()
|
|
def create_list(
|
|
name: str,
|
|
list_type: str,
|
|
shared_to_all: bool = True,
|
|
color: str | None = None,
|
|
emoji: str | None = None,
|
|
circle_id: str | None = None,
|
|
) -> str:
|
|
"""Create a new task list in Family Wall.
|
|
|
|
IMPORTANT: Ask the user for confirmation before calling this tool.
|
|
|
|
Args:
|
|
name: Display name for the new list (max 200 characters).
|
|
list_type: List type — ``"SHOPPING_LIST"``, ``"TODOS"``, or ``"OTHER"``.
|
|
shared_to_all: When ``True`` (default) the list is shared with all
|
|
circle members. When ``False`` it is private to the creator.
|
|
color: Optional background colour as a hex string (e.g. ``"#4784EC"``).
|
|
emoji: Optional Unicode emoji to use as the list icon (e.g. ``"🛒"``).
|
|
circle_id: Optional circle metaId to create the list in
|
|
(e.g. ``"family/23447378"``). When ``None`` (default) the list
|
|
is created in the primary circle. Use ``get_circles`` to
|
|
retrieve available circle IDs.
|
|
|
|
Returns:
|
|
JSON with the new list object on success, or an error message.
|
|
Includes ``circle_id`` field showing which circle the list was
|
|
created in.
|
|
"""
|
|
if list_type not in ("SHOPPING_LIST", "TODOS", "OTHER"):
|
|
return _err("list_type must be 'SHOPPING_LIST', 'TODOS', or 'OTHER'.")
|
|
if len(name) > 200:
|
|
return _err("name must not exceed 200 characters.")
|
|
|
|
params: dict[str, Any] = {
|
|
"name": name,
|
|
"taskListType": list_type,
|
|
"sharedToAll": "true" if shared_to_all else "false",
|
|
}
|
|
if color:
|
|
params["color"] = color
|
|
if emoji:
|
|
params["emoji"] = emoji
|
|
if circle_id:
|
|
# The API uses the 'scope' parameter to specify the target circle.
|
|
params["scope"] = circle_id
|
|
|
|
try:
|
|
data = _authenticated_call("taskcreatelist", params)
|
|
except RuntimeError as exc:
|
|
return _err(str(exc))
|
|
|
|
try:
|
|
list_obj = data["a00"]["r"]["r"]
|
|
meta_id: str = list_obj["metaId"]
|
|
except (KeyError, TypeError):
|
|
return json.dumps(
|
|
{"warning": "Unexpected taskcreatelist response structure", "raw": data},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
|
|
raw_emoji: str = list_obj.get("emoji", "")
|
|
return json.dumps(
|
|
{
|
|
"created": True,
|
|
"id": meta_id,
|
|
"name": list_obj.get("name", name),
|
|
"type": list_obj.get("taskListType"),
|
|
"shared_to_all": list_obj.get("sharedToAll") == "true",
|
|
"emoji": raw_emoji if raw_emoji else None,
|
|
"color": list_obj.get("color") or None,
|
|
"circle_id": list_obj.get("familyId") or circle_id,
|
|
},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tool: delete_list
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@mcp.tool()
|
|
def delete_list(list_id: str) -> str:
|
|
"""Permanently delete a task list and all its tasks.
|
|
|
|
IMPORTANT: Ask the user for confirmation before calling this tool.
|
|
|
|
This action cannot be undone. All tasks inside the list are also deleted.
|
|
Only lists with ``rights.canDelete="true"`` (user-created lists) can be
|
|
deleted. System lists are protected and this tool will refuse to delete them.
|
|
|
|
Args:
|
|
list_id: List metaId from get_lists
|
|
(e.g. ``"taskList/23431854_29759623"``). Must be a custom list
|
|
(``rights.canDelete="true"``).
|
|
|
|
Returns:
|
|
JSON success indicator or an error message.
|
|
"""
|
|
# Verify + delete in a single session to minimise round-trips.
|
|
try:
|
|
email, password = get_credentials()
|
|
except RuntimeError as exc:
|
|
return _err(str(exc))
|
|
|
|
# Derive the owning circle from the list metaId so that secondary-circle
|
|
# lists can be queried and deleted with the correct scope parameter.
|
|
# Format: taskList/<family_num>_<list_num> → family/<family_num>
|
|
circle_scope = _circle_id_from_list_id(list_id)
|
|
|
|
list_obj: dict[str, Any] | None = None
|
|
try:
|
|
with FamilyWallClient() as client:
|
|
client.login(email, password)
|
|
|
|
# Fetch lists scoped to the correct circle and verify deletion rights.
|
|
get_params: dict[str, Any] = {}
|
|
if circle_scope:
|
|
get_params["scope"] = circle_scope
|
|
raw = client.call("taskgettasklists", get_params)
|
|
try:
|
|
raw_lists: list[dict[str, Any]] = raw["a00"]["r"]["r"]
|
|
if not isinstance(raw_lists, list):
|
|
raw_lists = []
|
|
except (KeyError, TypeError):
|
|
raw_lists = []
|
|
|
|
list_obj = next((lst for lst in raw_lists if lst.get("metaId") == list_id), None)
|
|
if list_obj is None:
|
|
client.logout()
|
|
return _err(f"List '{list_id}' not found.")
|
|
|
|
can_delete: str | None = (list_obj.get("rights") or {}).get("canDelete")
|
|
if can_delete != "true":
|
|
client.logout()
|
|
return json.dumps(
|
|
{
|
|
"error": "System lists cannot be deleted.",
|
|
"id": list_id,
|
|
"name": list_obj.get("name"),
|
|
"hint": "Only user-created lists (rights.canDelete=true) can be deleted.",
|
|
},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
|
|
# Verified — delete in the same session.
|
|
# For secondary circles the 'scope' parameter is required.
|
|
del_params: dict[str, Any] = {"id": list_id}
|
|
if circle_scope:
|
|
del_params["scope"] = circle_scope
|
|
client.call("taskdeletelist", del_params)
|
|
client.logout()
|
|
except FamilyWallError as exc:
|
|
return _err(f"Family Wall API error: {exc}")
|
|
except Exception as exc:
|
|
return _err(f"Connection error: {exc}")
|
|
|
|
return json.dumps(
|
|
{"deleted": True, "id": list_id, "name": list_obj.get("name")},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tool: update_list
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@mcp.tool()
|
|
def update_list(
|
|
list_id: str,
|
|
name: str | None = None,
|
|
color: str | None = None,
|
|
emoji: str | None = None,
|
|
) -> str:
|
|
"""Update a task list's name, color, or emoji.
|
|
|
|
IMPORTANT: Ask the user for confirmation before calling this tool.
|
|
|
|
Performs a partial update — only the fields you provide are changed.
|
|
The current values for any omitted fields are preserved on the server
|
|
(the server keeps them; no need to fetch and re-send them).
|
|
|
|
Only user-created lists with ``rights.canUpdate="true"`` can be updated.
|
|
System lists are protected and this tool will refuse to update them.
|
|
|
|
Args:
|
|
list_id: List metaId from get_lists
|
|
(e.g. ``"taskList/23431854_29759623"``).
|
|
name: New display name (omit to keep existing).
|
|
color: New background colour as a hex string (e.g. ``"#E53935"``).
|
|
Omit to keep existing.
|
|
emoji: New Unicode emoji icon (e.g. ``"🧪"``).
|
|
Omit to keep existing.
|
|
|
|
Returns:
|
|
JSON with the updated list object on success, or an error message.
|
|
"""
|
|
if name is None and color is None and emoji is None:
|
|
return _err("At least one of 'name', 'color', or 'emoji' must be provided.")
|
|
|
|
try:
|
|
email, password = get_credentials()
|
|
except RuntimeError as exc:
|
|
return _err(str(exc))
|
|
|
|
# Derive the owning circle from the list metaId (same as delete_list).
|
|
circle_scope = _circle_id_from_list_id(list_id)
|
|
|
|
list_obj: dict[str, Any] | None = None
|
|
try:
|
|
with FamilyWallClient() as client:
|
|
client.login(email, password)
|
|
|
|
# Fetch list to verify rights.canUpdate before updating.
|
|
get_params: dict[str, Any] = {}
|
|
if circle_scope:
|
|
get_params["scope"] = circle_scope
|
|
raw = client.call("taskgettasklists", get_params)
|
|
try:
|
|
raw_lists: list[dict[str, Any]] = raw["a00"]["r"]["r"]
|
|
if not isinstance(raw_lists, list):
|
|
raw_lists = []
|
|
except (KeyError, TypeError):
|
|
raw_lists = []
|
|
|
|
list_obj = next((lst for lst in raw_lists if lst.get("metaId") == list_id), None)
|
|
if list_obj is None:
|
|
client.logout()
|
|
return _err(f"List '{list_id}' not found.")
|
|
|
|
can_update: str | None = (list_obj.get("rights") or {}).get("canUpdate")
|
|
if can_update != "true":
|
|
client.logout()
|
|
return json.dumps(
|
|
{
|
|
"error": "System lists cannot be updated.",
|
|
"id": list_id,
|
|
"name": list_obj.get("name"),
|
|
"hint": "Only user-created lists (rights.canUpdate=true) can be updated.",
|
|
},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
|
|
# Build update params — only include provided fields.
|
|
# Verified: taskupdatelist uses 'metaId' (not 'id') as the list identifier.
|
|
upd_params: dict[str, Any] = {"metaId": list_id}
|
|
if name is not None:
|
|
upd_params["name"] = name
|
|
if color is not None:
|
|
upd_params["color"] = color
|
|
if emoji is not None:
|
|
upd_params["emoji"] = emoji
|
|
|
|
resp = client.call("taskupdatelist", upd_params)
|
|
client.logout()
|
|
except FamilyWallError as exc:
|
|
return _err(f"Family Wall API error: {exc}")
|
|
except Exception as exc:
|
|
return _err(f"Connection error: {exc}")
|
|
|
|
try:
|
|
updated_obj: dict[str, Any] = resp["a00"]["r"]["r"]
|
|
if not isinstance(updated_obj, dict) or "metaId" not in updated_obj:
|
|
raise TypeError("unexpected shape")
|
|
except (KeyError, TypeError):
|
|
return json.dumps(
|
|
{"warning": "Unexpected taskupdatelist response structure", "raw": resp},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
|
|
raw_emoji: str = updated_obj.get("emoji", "")
|
|
return json.dumps(
|
|
{
|
|
"updated": True,
|
|
"id": updated_obj.get("metaId"),
|
|
"name": translate_name(updated_obj.get("name", "")),
|
|
"type": updated_obj.get("taskListType"),
|
|
"emoji": raw_emoji if raw_emoji else None,
|
|
"color": updated_obj.get("color") or None,
|
|
"circle_id": updated_obj.get("familyId"),
|
|
},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tool: create_circle
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@mcp.tool()
|
|
def create_circle(name: str) -> str:
|
|
"""Create a new Family Wall circle (group).
|
|
|
|
IMPORTANT: Ask the user for confirmation before calling this tool.
|
|
|
|
A circle is a group of people who share content on Family Wall (e.g. a
|
|
family, a group of friends). After creation the caller is automatically
|
|
added as the circle owner (SuperAdmin).
|
|
|
|
Note: Circle deletion is not supported by the Family Wall API. Test
|
|
circles must be deleted manually in the Family Wall app settings.
|
|
|
|
Args:
|
|
name: Display name for the new circle (e.g. ``"Van Elst"``).
|
|
|
|
Returns:
|
|
JSON with ``{created, id, name}`` on success, or an error message.
|
|
The server may capitalise the first letter of the name.
|
|
"""
|
|
try:
|
|
email, password = get_credentials()
|
|
except RuntimeError as exc:
|
|
return _err(str(exc))
|
|
|
|
try:
|
|
with FamilyWallClient() as client:
|
|
client.login(email, password)
|
|
|
|
# acccreatefamily returns only the numeric family ID.
|
|
data = client.call("acccreatefamily", {"name": name})
|
|
try:
|
|
numeric_id: str = data["a00"]["r"]["r"]
|
|
if not isinstance(numeric_id, str) or not numeric_id.isdigit():
|
|
raise TypeError(f"expected numeric ID, got {numeric_id!r}")
|
|
except (KeyError, TypeError):
|
|
client.logout()
|
|
return json.dumps(
|
|
{"warning": "Unexpected acccreatefamily response", "raw": data},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
|
|
circle_id = f"family/{numeric_id}"
|
|
|
|
# Read back the circle to obtain the server-stored (possibly
|
|
# capitalised) name in the same session.
|
|
canonical_name = name
|
|
try:
|
|
circles_data = client.call("famlistfamily")
|
|
raw_circles = circles_data.get("a00", {}).get("r", {}).get("r", []) or []
|
|
for c in raw_circles:
|
|
if c.get("metaId") == circle_id:
|
|
canonical_name = c.get("name", name)
|
|
break
|
|
except FamilyWallError:
|
|
pass # fallback: use caller-provided name
|
|
|
|
client.logout()
|
|
except FamilyWallError as exc:
|
|
return _err(f"Family Wall API error: {exc}")
|
|
except Exception as exc:
|
|
return _err(f"Connection error: {exc}")
|
|
|
|
return json.dumps(
|
|
{"created": True, "id": circle_id, "name": canonical_name},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tool: update_circle
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@mcp.tool()
|
|
def update_circle(circle_id: str, name: str) -> str:
|
|
"""Rename a Family Wall circle.
|
|
|
|
IMPORTANT: Ask the user for confirmation before calling this tool.
|
|
|
|
Only the circle name can be changed via the API. Note that the server
|
|
always capitalises the first letter of the new name.
|
|
|
|
Args:
|
|
circle_id: Circle metaId from ``get_circles``
|
|
(e.g. ``"family/23431854"``).
|
|
name: New display name for the circle.
|
|
|
|
Returns:
|
|
JSON with ``{updated, id, name}`` on success, or an error message.
|
|
"""
|
|
if not name or not name.strip():
|
|
return _err("'name' must not be empty.")
|
|
|
|
try:
|
|
email, password = get_credentials()
|
|
except RuntimeError as exc:
|
|
return _err(str(exc))
|
|
|
|
try:
|
|
with FamilyWallClient() as client:
|
|
client.login(email, password)
|
|
|
|
# Verify the circle exists before attempting the update.
|
|
circles_data = client.call("famlistfamily")
|
|
try:
|
|
raw_circles: list[dict[str, Any]] = circles_data["a00"]["r"]["r"]
|
|
if not isinstance(raw_circles, list):
|
|
raise TypeError("a00.r.r is not a list")
|
|
except (KeyError, TypeError):
|
|
client.logout()
|
|
return _err("Unexpected famlistfamily response structure.")
|
|
|
|
target = next((c for c in raw_circles if c.get("metaId") == circle_id), None)
|
|
if target is None:
|
|
client.logout()
|
|
available = [c.get("metaId") for c in raw_circles]
|
|
return json.dumps(
|
|
{
|
|
"error": f"Circle not found: {circle_id!r}",
|
|
"available_circles": available,
|
|
},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
|
|
if target.get("isFirstFamily") == "true":
|
|
client.logout()
|
|
return json.dumps(
|
|
{
|
|
"error": "Cannot rename the primary circle.",
|
|
"id": circle_id,
|
|
"name": target.get("name"),
|
|
"hint": (
|
|
"The primary (first) circle cannot be renamed via the API. "
|
|
"Use the Family Wall app settings to manage it."
|
|
),
|
|
},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
|
|
# Rename the circle.
|
|
# Verified: accupdatefamily with scope=<circle_metaId> targets any circle,
|
|
# both primary and secondary. The server capitalises the first letter.
|
|
resp = client.call("accupdatefamily", {"scope": circle_id, "name": name})
|
|
client.logout()
|
|
except FamilyWallError as exc:
|
|
return _err(f"Family Wall API error: {exc}")
|
|
except Exception as exc:
|
|
return _err(f"Connection error: {exc}")
|
|
|
|
# Response: a00.r.r = full circle object
|
|
try:
|
|
circle_obj: dict[str, Any] = resp["a00"]["r"]["r"]
|
|
if not isinstance(circle_obj, dict) or "metaId" not in circle_obj:
|
|
raise TypeError("unexpected shape")
|
|
except (KeyError, TypeError):
|
|
return json.dumps(
|
|
{"warning": "Unexpected accupdatefamily response structure", "raw": resp},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
|
|
return json.dumps(
|
|
{
|
|
"updated": True,
|
|
"id": circle_obj.get("metaId"),
|
|
"name": circle_obj.get("name"),
|
|
},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tool: add_member_to_circle
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@mcp.tool()
|
|
def add_member_to_circle(
|
|
circle_id: str,
|
|
email: str,
|
|
firstname: str | None = None,
|
|
) -> str:
|
|
"""Invite a person to a Family Wall circle by e-mail.
|
|
|
|
IMPORTANT: Ask the user for confirmation before calling this tool.
|
|
|
|
Sends an invitation to the given e-mail address. The recipient will
|
|
receive an e-mail with a link to join the circle. If they already have a
|
|
Family Wall account they must accept the invitation via the app; if they
|
|
do not have an account yet one will be created during the acceptance flow.
|
|
|
|
Note: The Family Wall API ``accinvite`` endpoint is limited to inviting
|
|
people who do not yet have an active Family Wall account. Inviting an
|
|
existing account holder returns an error from the server.
|
|
|
|
Args:
|
|
circle_id: Target circle metaId from ``get_circles``
|
|
(e.g. ``"family/23431854"``).
|
|
email: E-mail address of the person to invite.
|
|
firstname: First name of the invitee (used in the invitation e-mail).
|
|
When omitted the local part of the e-mail address is used as a
|
|
fallback (e.g. ``"john"`` from ``john.doe@example.com``).
|
|
|
|
Returns:
|
|
JSON success indicator or an error message.
|
|
"""
|
|
if not email or "@" not in email:
|
|
return _err("'email' must be a valid e-mail address.")
|
|
|
|
# Derive a sensible first-name fallback from the email local part.
|
|
if firstname is None:
|
|
local_part = email.split("@")[0]
|
|
firstname = local_part.split(".")[0].capitalize()
|
|
|
|
params: dict[str, Any] = {
|
|
"familyId": circle_id,
|
|
"identifier": email,
|
|
"role": DEFAULT_INVITE_ROLE,
|
|
"firstname": firstname,
|
|
}
|
|
|
|
try:
|
|
data = _authenticated_call("accinvite", params)
|
|
except RuntimeError as exc:
|
|
return _err(str(exc))
|
|
|
|
# On success the server returns the invitation object under a00.r.r.
|
|
try:
|
|
result_obj = data.get("a00", {}).get("r", {}).get("r")
|
|
except (AttributeError, TypeError):
|
|
result_obj = None
|
|
|
|
return json.dumps(
|
|
{
|
|
"invited": True,
|
|
"circle_id": circle_id,
|
|
"email": email,
|
|
"result": result_obj,
|
|
},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tool: delete_circle
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@mcp.tool()
|
|
def delete_circle(circle_id: str) -> str:
|
|
"""Permanently delete a Family Wall circle (group) and all its content.
|
|
|
|
IMPORTANT: Ask the user for confirmation before calling this tool.
|
|
|
|
This action cannot be undone. All lists, tasks, recipes, and wall posts
|
|
inside the circle are deleted along with it. The primary (first) circle
|
|
is protected and cannot be deleted.
|
|
|
|
Args:
|
|
circle_id: Circle metaId from ``get_circles``
|
|
(e.g. ``"family/23447378"``). Must not be the primary circle.
|
|
|
|
Returns:
|
|
JSON success indicator or an error message.
|
|
"""
|
|
try:
|
|
email, password = get_credentials()
|
|
except RuntimeError as exc:
|
|
return _err(str(exc))
|
|
|
|
circle_name: str = circle_id
|
|
try:
|
|
with FamilyWallClient() as client:
|
|
client.login(email, password)
|
|
|
|
# Fetch the circle list to verify the target exists and is not primary.
|
|
circles_data = client.call("famlistfamily")
|
|
try:
|
|
raw_circles: list[dict[str, Any]] = circles_data["a00"]["r"]["r"]
|
|
if not isinstance(raw_circles, list):
|
|
raise TypeError("a00.r.r is not a list")
|
|
except (KeyError, TypeError):
|
|
client.logout()
|
|
return _err("Unexpected famlistfamily response structure.")
|
|
|
|
target = next((c for c in raw_circles if c.get("metaId") == circle_id), None)
|
|
if target is None:
|
|
client.logout()
|
|
available = [c.get("metaId") for c in raw_circles]
|
|
return json.dumps(
|
|
{
|
|
"error": f"Circle not found: {circle_id!r}",
|
|
"available_circles": available,
|
|
},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
|
|
if target.get("isFirstFamily") == "true":
|
|
client.logout()
|
|
return json.dumps(
|
|
{
|
|
"error": "Cannot delete the primary circle.",
|
|
"id": circle_id,
|
|
"name": target.get("name"),
|
|
"hint": (
|
|
"The primary (first) circle cannot be deleted via the API. "
|
|
"Use the Family Wall app settings to manage it."
|
|
),
|
|
},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
|
|
circle_name = target.get("name", circle_id)
|
|
|
|
# Verified — delete in the same session.
|
|
# adminwipefamily uses scope=<circle_metaId> and returns a00.r.r="true".
|
|
client.call("adminwipefamily", {"scope": circle_id})
|
|
client.logout()
|
|
except FamilyWallError as exc:
|
|
return _err(f"Family Wall API error: {exc}")
|
|
except Exception as exc:
|
|
return _err(f"Connection error: {exc}")
|
|
|
|
return json.dumps(
|
|
{"deleted": True, "id": circle_id, "name": circle_name},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tool: like_post
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@mcp.tool()
|
|
def like_post(post_id: str, like: bool = True, mood: str = "STAR") -> str:
|
|
"""Like or unlike a wall post/activity.
|
|
|
|
IMPORTANT: Ask the user for confirmation before calling this tool.
|
|
|
|
Args:
|
|
post_id: Wall post ID from get_activities (e.g. ``wall/23431854_31119189``).
|
|
like: ``True`` to add a like (default), ``False`` to remove it.
|
|
mood: Mood type to set or remove. Default is ``"STAR"``.
|
|
|
|
Returns:
|
|
JSON with the resulting like state or an error message.
|
|
"""
|
|
# Like: add=STAR, remove=$empty (idempotent set)
|
|
# Unlike: add=$empty, remove.0=STAR (array dot-notation, verified via Network Interceptor)
|
|
if like:
|
|
params: dict[str, Any] = {
|
|
"wall_message_id": post_id,
|
|
"add": mood,
|
|
"remove": "$empty",
|
|
}
|
|
else:
|
|
params = {
|
|
"wall_message_id": post_id,
|
|
"add": "$empty",
|
|
"remove.0": mood,
|
|
}
|
|
|
|
try:
|
|
data = _authenticated_call("wallmood", params)
|
|
except RuntimeError as exc:
|
|
return _err(str(exc))
|
|
|
|
# Extract moodMap from the response to confirm the like was recorded.
|
|
try:
|
|
wall_obj = data["a00"]["r"]["r"]
|
|
if not isinstance(wall_obj, dict):
|
|
raise TypeError("a00.r.r is not a dict")
|
|
mood_map: dict[str, Any] = wall_obj.get("moodMap") or {}
|
|
account_id: str = wall_obj.get("postAccountId", "")
|
|
except (KeyError, TypeError):
|
|
return json.dumps(
|
|
{"warning": "Unexpected wallmood response structure", "raw": data},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
|
|
# Two complementary indicators for the like state:
|
|
# - moodStarShortcut: direct boolean per-user flag on the post object (primary)
|
|
# - moodMap: dict of accountId → [mood types]; contains "STAR" when liked (secondary)
|
|
# Use both so either storage path is covered.
|
|
star_shortcut = wall_obj.get("moodStarShortcut") == "true"
|
|
star_in_map = any("STAR" in moods for moods in mood_map.values())
|
|
now_liked = star_shortcut or star_in_map
|
|
|
|
result: dict[str, Any] = {"liked": now_liked, "id": post_id, "author": account_id}
|
|
|
|
if now_liked != like:
|
|
if like:
|
|
result["warning"] = (
|
|
"Like may not have been applied. "
|
|
"Possible causes: rate limit, unsupported post type (e.g. FAMILY_CREATED), "
|
|
"or self-like restriction."
|
|
)
|
|
else:
|
|
result["warning"] = (
|
|
"Unlike may not have been applied. "
|
|
"Possible causes: post was not liked, rate limit, or API restriction."
|
|
)
|
|
|
|
return json.dumps(result, ensure_ascii=False, indent=2)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tool: get_wall_posts
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@mcp.tool()
|
|
def get_wall_posts(limit: int = 20) -> str:
|
|
"""Return recent Family Wall posts as JSON.
|
|
|
|
Returns posts from the primary circle's wall, sorted by date descending.
|
|
Includes status posts, activity entries, comments, and like information.
|
|
|
|
Args:
|
|
limit: Maximum number of posts to return (default 20).
|
|
|
|
Returns:
|
|
JSON list of post objects with keys:
|
|
id, type, text, date, author, author_id,
|
|
liked_by_me, like_count, comment_count.
|
|
"""
|
|
try:
|
|
email, password = get_credentials()
|
|
except RuntimeError as exc:
|
|
return _err(str(exc))
|
|
|
|
author_map: dict[str, str] = {}
|
|
try:
|
|
for circle in _famlistfamily():
|
|
for member in circle.get("members") or []:
|
|
acc_id: str = member.get("accountId", "")
|
|
display = member.get("firstName") or member.get("name") or acc_id
|
|
if acc_id:
|
|
author_map[acc_id] = display
|
|
except RuntimeError:
|
|
pass
|
|
|
|
try:
|
|
with FamilyWallClient() as client:
|
|
client.login(email, password)
|
|
data = client.call("wallget", {"nb": str(limit)})
|
|
client.logout()
|
|
except FamilyWallError as exc:
|
|
return _err(str(exc))
|
|
except Exception as exc:
|
|
return _err(f"Connection error: {exc}")
|
|
|
|
raw_posts: list[dict[str, Any]] | None = None
|
|
try:
|
|
candidate = data["a00"]["r"]["r"]
|
|
if isinstance(candidate, list):
|
|
raw_posts = candidate
|
|
elif isinstance(candidate, dict) and isinstance(candidate.get("updatedCreated"), list):
|
|
raw_posts = candidate["updatedCreated"]
|
|
except (KeyError, TypeError):
|
|
pass
|
|
|
|
if raw_posts is None:
|
|
return json.dumps(
|
|
{"warning": "Unexpected wallget response structure", "raw": data},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
|
|
result = []
|
|
for item in raw_posts:
|
|
raw_author: str = item.get("accountId", "")
|
|
mood_map: dict[str, Any] = item.get("moodMap") or {}
|
|
|
|
liked_by_me = (
|
|
item.get("moodStarShortcut") == "true"
|
|
or any("STAR" in moods for moods in mood_map.values())
|
|
)
|
|
|
|
like_count = sum(len(moods) for moods in mood_map.values() if isinstance(moods, list))
|
|
|
|
comments: list[dict[str, Any]] = item.get("comments") or []
|
|
comment_count = len(comments)
|
|
|
|
result.append(
|
|
{
|
|
"id": item.get("metaId"),
|
|
"type": item.get("refType"),
|
|
"text": item.get("text") or item.get("tagline"),
|
|
"date": item.get("creationDate"),
|
|
"author": author_map.get(raw_author, raw_author),
|
|
"author_id": raw_author,
|
|
"liked_by_me": liked_by_me,
|
|
"like_count": like_count,
|
|
"comment_count": comment_count,
|
|
}
|
|
)
|
|
|
|
return json.dumps(result, ensure_ascii=False, indent=2)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tool: create_wall_post
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@mcp.tool()
|
|
def create_wall_post(text: str) -> str:
|
|
"""Create a new status post on the Family Wall.
|
|
|
|
IMPORTANT: Ask the user for confirmation before calling this tool.
|
|
|
|
Args:
|
|
text: Text content of the post.
|
|
|
|
Returns:
|
|
JSON with the new post's metaId on success, or an error message.
|
|
"""
|
|
try:
|
|
email, password = get_credentials()
|
|
except RuntimeError as exc:
|
|
return _err(str(exc))
|
|
|
|
try:
|
|
with FamilyWallClient() as client:
|
|
client.login(email, password)
|
|
data = client.call("wallpublish", {"tagline": text})
|
|
client.logout()
|
|
except FamilyWallError as exc:
|
|
return _err(str(exc))
|
|
except Exception as exc:
|
|
return _err(f"Connection error: {exc}")
|
|
|
|
try:
|
|
post = data["a00"]["r"]["r"]
|
|
if not isinstance(post, dict) or "metaId" not in post:
|
|
raise TypeError("unexpected shape")
|
|
except (KeyError, TypeError):
|
|
return json.dumps(
|
|
{"warning": "Unexpected wallpublish response structure", "raw": data},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
|
|
return json.dumps(
|
|
{
|
|
"id": post.get("metaId"),
|
|
"text": post.get("tagline") or post.get("text"),
|
|
"date": post.get("creationDate"),
|
|
},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tool: add_comment
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@mcp.tool()
|
|
def add_comment(post_id: str, comment: str) -> str:
|
|
"""Add a comment to a wall post or activity.
|
|
|
|
IMPORTANT: Ask the user for confirmation before calling this tool.
|
|
|
|
Args:
|
|
post_id: Post metaId from get_wall_posts
|
|
(e.g. ``"wall/16282169_31119189"``).
|
|
comment: Comment text.
|
|
|
|
Returns:
|
|
JSON success indicator or an error message.
|
|
"""
|
|
try:
|
|
email, password = get_credentials()
|
|
except RuntimeError as exc:
|
|
return _err(str(exc))
|
|
|
|
try:
|
|
with FamilyWallClient() as client:
|
|
client.login(email, password)
|
|
data = client.call(
|
|
"walladdComment",
|
|
{
|
|
"wall_message_id": post_id,
|
|
"comment": comment,
|
|
},
|
|
)
|
|
client.logout()
|
|
except FamilyWallError as exc:
|
|
return _err(str(exc))
|
|
except Exception as exc:
|
|
return _err(f"Connection error: {exc}")
|
|
|
|
try:
|
|
comment_obj = data["a00"]["r"]["r"]
|
|
if not isinstance(comment_obj, dict) or "metaId" not in comment_obj:
|
|
raise TypeError("unexpected shape")
|
|
except (KeyError, TypeError):
|
|
return json.dumps(
|
|
{"warning": "Unexpected walladdComment response structure", "raw": data},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
|
|
return json.dumps(
|
|
{
|
|
"id": comment_obj.get("metaId"),
|
|
"post_id": post_id,
|
|
"text": comment_obj.get("text") or comment_obj.get("comment"),
|
|
"date": comment_obj.get("creationDate"),
|
|
},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helper: fetch all raw recipes
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _get_family_id() -> str:
|
|
"""Extract the primary family ID from famlistfamily response.
|
|
|
|
Returns:
|
|
The family ID as a string (e.g. "23431854").
|
|
|
|
Raises:
|
|
RuntimeError: On credential or API errors.
|
|
"""
|
|
try:
|
|
families = _famlistfamily()
|
|
if families and isinstance(families, list) and len(families) > 0:
|
|
primary = families[0]
|
|
metaid = primary.get("metaId")
|
|
if isinstance(metaid, str) and metaid.startswith("family/"):
|
|
return metaid.split("/", 1)[1]
|
|
except RuntimeError:
|
|
pass
|
|
raise RuntimeError("Could not determine family ID")
|
|
|
|
|
|
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_recipe_categories
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@mcp.tool()
|
|
def get_recipe_categories() -> str:
|
|
"""Return all available recipe categories for the family.
|
|
|
|
Returns:
|
|
JSON list of category IDs (e.g. ["category/23431854_2", ...]).
|
|
Always includes the 5 free-tier standard categories if family ID is available.
|
|
Returns an error message string on failure.
|
|
"""
|
|
# Get the family ID to construct standard category IDs
|
|
family_id: str | None = None
|
|
with contextlib.suppress(RuntimeError):
|
|
family_id = _get_family_id()
|
|
|
|
# Collect all category IDs: standard categories + categories found in recipes
|
|
seen_ids: set[str] = set()
|
|
category_ids: list[str] = []
|
|
|
|
# Add the 5 free-tier standard categories if we have the family ID
|
|
if family_id:
|
|
standard_cat_ids = [
|
|
f"category/{family_id}_2", # Bei Kindern beliebt (KIDS_LOVE)
|
|
f"category/{family_id}_3", # Wirklich einfach (EASY)
|
|
f"category/{family_id}_4", # Nachspeisen (DESSERT)
|
|
f"category/{family_id}_5", # Schmeckt toll (DELICIOUS)
|
|
f"category/{family_id}_6", # Gemüse (VEGETABLES)
|
|
]
|
|
for cat_id in standard_cat_ids:
|
|
if cat_id not in seen_ids:
|
|
seen_ids.add(cat_id)
|
|
category_ids.append(cat_id)
|
|
|
|
# Add any additional categories found in recipes (e.g., premium categories)
|
|
try:
|
|
raw_recipes = _get_raw_recipes()
|
|
except RuntimeError:
|
|
raw_recipes = []
|
|
|
|
for recipe in raw_recipes:
|
|
if not isinstance(recipe, dict):
|
|
continue
|
|
recipe_cat_ids = recipe.get("recipeCategoryIdList")
|
|
if not isinstance(recipe_cat_ids, list):
|
|
continue
|
|
for cat_id in recipe_cat_ids:
|
|
if isinstance(cat_id, str) and cat_id and cat_id not in seen_ids:
|
|
seen_ids.add(cat_id)
|
|
category_ids.append(cat_id)
|
|
|
|
return json.dumps(category_ids, ensure_ascii=False, indent=2)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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 _err(str(exc))
|
|
|
|
result = [parse_recipe_summary(r) for r in raw_recipes]
|
|
return json.dumps(result, ensure_ascii=False, indent=2)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tool: get_recipe_box
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@mcp.tool()
|
|
def get_recipe_box() -> str:
|
|
"""Return only real recipes from the recipe box (isRecipe=true).
|
|
|
|
Excludes stub recipes that are auto-created when free-text meal plan
|
|
entries are added. Use this for meal planning and ingredient lookups.
|
|
For the full list including stubs use get_recipes.
|
|
|
|
Returns:
|
|
JSON list of recipe summary objects (same format as get_recipes).
|
|
Returns an error message string on failure.
|
|
"""
|
|
try:
|
|
raw_recipes = _get_raw_recipes()
|
|
except RuntimeError as exc:
|
|
return _err(str(exc))
|
|
|
|
result = [parse_recipe_summary(r) for r in raw_recipes if r.get("isRecipe") == "true"]
|
|
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 _err(str(exc))
|
|
|
|
raw = next((r for r in raw_recipes if r.get("metaId") == recipe_id), None)
|
|
if raw is None:
|
|
return _err(f"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,
|
|
category_ids: list[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).
|
|
category_ids: Optional list of recipe category IDs
|
|
(e.g. ``["category/23431854_2"]``). Use get_recipe_categories to find IDs.
|
|
|
|
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,
|
|
category_ids=category_ids,
|
|
)
|
|
|
|
try:
|
|
data = _authenticated_call("mprecipeput", params)
|
|
except RuntimeError as exc:
|
|
return _err(str(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 _err(str(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 _err(f"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 _err(f"Family Wall API error: {exc}")
|
|
except Exception as exc:
|
|
return _err(f"Connection error: {exc}")
|
|
|
|
return json.dumps(
|
|
{"deleted": True, "id": recipe_id, "name": recipe_obj.get("name")},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tool: update_recipe
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@mcp.tool()
|
|
def update_recipe(
|
|
recipe_id: str,
|
|
name: str | None = None,
|
|
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,
|
|
category_ids: list[str] | None = None,
|
|
) -> str:
|
|
"""Update an existing recipe in the Family Wall recipe box.
|
|
|
|
IMPORTANT: Ask the user for confirmation before calling this tool.
|
|
|
|
At least one field besides *recipe_id* must be provided. Fields that are
|
|
omitted are left unchanged on the server.
|
|
|
|
Args:
|
|
recipe_id: Recipe metaId from get_recipes
|
|
(e.g. ``"recipe/23431854_10968866"``).
|
|
name: New recipe title (omit to keep existing).
|
|
description: New description (omit to keep existing).
|
|
ingredients: New ingredient list as free text (omit to keep existing).
|
|
Use newlines (``\\n``) to separate items.
|
|
Example: ``"200g Mehl\\n3 Eier\\n100ml Milch"``
|
|
instructions: New cooking instructions as free text (omit to keep
|
|
existing). Use newlines (``\\n``) to separate steps.
|
|
prep_time_minutes: New preparation time in minutes (omit to keep
|
|
existing).
|
|
cook_time_minutes: New cooking/baking time in minutes (omit to keep
|
|
existing).
|
|
serves: New number of servings (omit to keep existing).
|
|
url: New external URL (omit to keep existing).
|
|
category_ids: New list of recipe category IDs (omit to keep existing).
|
|
Pass empty list to remove all categories.
|
|
|
|
Returns:
|
|
JSON with the updated recipe's full fields on success, or an error
|
|
message.
|
|
"""
|
|
if all(
|
|
v is None
|
|
for v in (
|
|
name,
|
|
description,
|
|
ingredients,
|
|
instructions,
|
|
prep_time_minutes,
|
|
cook_time_minutes,
|
|
serves,
|
|
url,
|
|
category_ids,
|
|
)
|
|
):
|
|
return (
|
|
"Error: At least one of 'name', 'description', 'ingredients', 'instructions',"
|
|
" 'prep_time_minutes', 'cook_time_minutes', 'serves', 'url', or 'category_ids'"
|
|
" must be provided."
|
|
)
|
|
|
|
# Single session: fetch current recipe (verify + get name fallback) → update.
|
|
try:
|
|
email, password = get_credentials()
|
|
except RuntimeError as exc:
|
|
return _err(str(exc))
|
|
|
|
try:
|
|
with FamilyWallClient() as client:
|
|
client.login(email, password)
|
|
|
|
# Fetch all recipes to verify the target exists and is updatable.
|
|
raw_data = client.call("metasync", {"id": "recipe"})
|
|
try:
|
|
items: list[dict[str, Any]] = raw_data["a00"]["r"]["r"]["updatedCreated"]
|
|
except (KeyError, TypeError):
|
|
items = []
|
|
|
|
current = next((r for r in items if r.get("metaId") == recipe_id), None)
|
|
if current is None:
|
|
client.logout()
|
|
return _err(f"Recipe '{recipe_id}' not found.")
|
|
|
|
can_update: str | None = (current.get("rights") or {}).get("canUpdate")
|
|
if can_update != "true":
|
|
client.logout()
|
|
return json.dumps(
|
|
{
|
|
"error": "Recipe cannot be updated.",
|
|
"id": recipe_id,
|
|
"name": current.get("name"),
|
|
"hint": "Only recipes you created (can_update=true) can be updated.",
|
|
},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
|
|
# Build params — current name is used as fallback when caller omits name.
|
|
params = build_update_params(
|
|
recipe_id=recipe_id,
|
|
current_name=current.get("name", ""),
|
|
name=name,
|
|
description=description,
|
|
ingredients=ingredients,
|
|
instructions=instructions,
|
|
prep_time_minutes=prep_time_minutes,
|
|
cook_time_minutes=cook_time_minutes,
|
|
serves=serves,
|
|
url=url,
|
|
category_ids=category_ids,
|
|
)
|
|
|
|
resp = client.call("mprecipeput", params)
|
|
client.logout()
|
|
except FamilyWallError as exc:
|
|
return _err(f"Family Wall API error: {exc}")
|
|
except Exception as exc:
|
|
return _err(f"Connection error: {exc}")
|
|
|
|
try:
|
|
recipe_obj = resp["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": resp},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
|
|
result = parse_recipe_full(recipe_obj)
|
|
result["updated"] = True
|
|
return json.dumps(result, ensure_ascii=False, indent=2)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tool: get_meal_plan
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@mcp.tool()
|
|
def get_meal_plan(date_from: str, date_to: str) -> str:
|
|
"""Return meal plan entries for a date range.
|
|
|
|
The response merges two API lists:
|
|
|
|
* ``dish`` entries — planned meals, optionally linked to a recipe.
|
|
* ``meal`` entries — free-text notes with an optional serving count.
|
|
|
|
Args:
|
|
date_from: Start date in ISO format (e.g. ``"2026-04-13"``).
|
|
date_to: End date in ISO format (e.g. ``"2026-04-19"``).
|
|
|
|
Returns:
|
|
JSON list of meal plan entries sorted by date then meal type
|
|
(BREAKFAST → LUNCH → SNACK → DINNER), with keys:
|
|
|
|
- ``id`` — metaId (``dish/…`` or ``meal/…``)
|
|
- ``date`` — ISO date string
|
|
- ``type`` — ``BREAKFAST``, ``LUNCH``, ``SNACK``, or ``DINNER``
|
|
- ``name`` — dish name, or ``null`` for meal entries
|
|
- ``recipe_id`` — linked recipe metaId, or ``null``
|
|
- ``is_from_recipe_box`` — ``true`` if the linked recipe was pulled from the
|
|
recipe box; ``false`` if it is a free-text stub (name only, no ingredients);
|
|
``null`` for meal entries or when no recipe is linked
|
|
- ``note`` — free-text note (meal entries only), or ``null``
|
|
- ``serves`` — number of servings as int (meal entries only), or ``null``
|
|
- ``can_update`` — whether the entry can be updated
|
|
- ``can_delete`` — whether the entry can be deleted
|
|
|
|
Returns an error message string on failure.
|
|
"""
|
|
if date_err := _validate_date(date_from):
|
|
return _err(date_err)
|
|
if date_err := _validate_date(date_to):
|
|
return _err(date_err)
|
|
try:
|
|
data = _authenticated_call("mplistinterval", {"from": date_from, "to": date_to})
|
|
except RuntimeError as exc:
|
|
return _err(str(exc))
|
|
|
|
try:
|
|
payload = data["a00"]["r"]["r"]
|
|
if not isinstance(payload, dict):
|
|
raise TypeError("a00.r.r is not a dict")
|
|
except (KeyError, TypeError):
|
|
return json.dumps(
|
|
{"warning": "Unexpected mplistinterval response structure", "raw": data},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
|
|
raw_dishes: list[dict[str, Any]] = payload.get("list") or []
|
|
raw_meals: list[dict[str, Any]] = payload.get("mealList") or []
|
|
raw_recipe_list: list[dict[str, Any]] = payload.get("recipeList") or []
|
|
|
|
# Build lookup: recipeId → is_from_recipe_box
|
|
# recipeList[] contains one entry per linked recipe; isRecipe="true" means it is
|
|
# a real recipe from the recipe box, "false" means a free-text stub (name only).
|
|
recipe_lookup: dict[str, bool] = {
|
|
r["metaId"]: r.get("isRecipe") == "true"
|
|
for r in raw_recipe_list
|
|
if isinstance(r, dict) and "metaId" in r
|
|
}
|
|
|
|
_type_order = {"BREAKFAST": 0, "LUNCH": 1, "SNACK": 2, "DINNER": 3}
|
|
|
|
result: list[dict[str, Any]] = []
|
|
|
|
for dish in raw_dishes:
|
|
rights = dish.get("rights") or {}
|
|
recipe_id: str | None = dish.get("recipeId") or None
|
|
result.append(
|
|
{
|
|
"id": dish.get("metaId"),
|
|
"date": dish.get("date"),
|
|
"type": dish.get("type"),
|
|
"name": dish.get("name"),
|
|
"recipe_id": recipe_id,
|
|
"is_from_recipe_box": recipe_lookup.get(recipe_id) if recipe_id else None,
|
|
"note": None,
|
|
"serves": None,
|
|
"can_update": rights.get("canUpdate") == "true",
|
|
"can_delete": rights.get("canDelete") == "true",
|
|
}
|
|
)
|
|
|
|
for meal in raw_meals:
|
|
rights = meal.get("rights") or {}
|
|
raw_serves = meal.get("serves")
|
|
result.append(
|
|
{
|
|
"id": meal.get("metaId"),
|
|
"date": meal.get("date"),
|
|
"type": meal.get("type"),
|
|
"name": None,
|
|
"recipe_id": None,
|
|
"is_from_recipe_box": None,
|
|
"note": meal.get("note") or None,
|
|
"serves": int(raw_serves) if raw_serves is not None else None,
|
|
"can_update": rights.get("canUpdate") == "true",
|
|
"can_delete": rights.get("canDelete") == "true",
|
|
}
|
|
)
|
|
|
|
result.sort(key=lambda e: (e.get("date") or "", _type_order.get(e.get("type") or "", 99)))
|
|
|
|
return json.dumps(result, ensure_ascii=False, indent=2)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tool: add_recipe_to_meal_plan
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@mcp.tool()
|
|
def add_recipe_to_meal_plan(
|
|
recipe_id: str,
|
|
date: str,
|
|
meal_type: str,
|
|
) -> str:
|
|
"""Add a recipe from the recipe box to the meal plan.
|
|
|
|
IMPORTANT: Ask the user for confirmation before calling this tool.
|
|
|
|
Args:
|
|
recipe_id: Recipe metaId from get_recipes
|
|
(e.g. ``"recipe/16282169_7932720"``).
|
|
date: Target date in ISO format (e.g. ``"2026-04-20"``).
|
|
meal_type: Meal slot — one of ``"BREAKFAST"``, ``"LUNCH"``,
|
|
``"SNACK"``, or ``"DINNER"``.
|
|
|
|
Returns:
|
|
JSON with the new dish entry on success, or an error message.
|
|
"""
|
|
if meal_type not in ("BREAKFAST", "LUNCH", "SNACK", "DINNER"):
|
|
return _err("meal_type must be one of 'BREAKFAST', 'LUNCH', 'SNACK', 'DINNER'.")
|
|
if date_err := _validate_date(date):
|
|
return _err(date_err)
|
|
|
|
params: dict[str, Any] = {
|
|
"recipeId": recipe_id,
|
|
"date": date,
|
|
"type": meal_type,
|
|
}
|
|
|
|
try:
|
|
data = _authenticated_call("mpcreateByRecipeId", params)
|
|
except RuntimeError as exc:
|
|
return _err(str(exc))
|
|
|
|
try:
|
|
dish = data["a00"]["r"]["r"]
|
|
if not isinstance(dish, dict) or "metaId" not in dish:
|
|
raise TypeError("unexpected shape")
|
|
except (KeyError, TypeError):
|
|
return json.dumps(
|
|
{"warning": "Unexpected mpcreateByRecipeId response structure", "raw": data},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
|
|
rights = dish.get("rights") or {}
|
|
result: dict[str, Any] = {
|
|
"id": dish.get("metaId"),
|
|
"date": dish.get("date"),
|
|
"type": dish.get("type"),
|
|
"name": dish.get("name"),
|
|
"recipe_id": dish.get("recipeId") or None,
|
|
# mpcreateByRecipeId always creates entries from the recipe box.
|
|
"is_from_recipe_box": True,
|
|
"note": None,
|
|
"serves": None,
|
|
"can_update": rights.get("canUpdate") == "true",
|
|
"can_delete": rights.get("canDelete") == "true",
|
|
}
|
|
return json.dumps(result, ensure_ascii=False, indent=2)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tool: add_meal_to_meal_plan
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@mcp.tool()
|
|
def add_meal_to_meal_plan(
|
|
name: str,
|
|
date: str,
|
|
meal_type: str,
|
|
) -> str:
|
|
"""Add a free-text meal entry to the meal plan.
|
|
|
|
IMPORTANT: Ask the user for confirmation before calling this tool.
|
|
|
|
Use this when the meal has no recipe in the recipe box.
|
|
For meals from the recipe box use ``add_recipe_to_meal_plan`` instead.
|
|
|
|
Args:
|
|
name: Display name of the meal (e.g. ``"Pfannkuchen"``).
|
|
date: Target date in ISO format (e.g. ``"2026-04-20"``).
|
|
meal_type: Meal slot — one of ``"BREAKFAST"``, ``"LUNCH"``,
|
|
``"SNACK"``, or ``"DINNER"``.
|
|
|
|
Returns:
|
|
JSON with the new dish entry on success, or an error message.
|
|
"""
|
|
if meal_type not in ("BREAKFAST", "LUNCH", "SNACK", "DINNER"):
|
|
return _err("meal_type must be one of 'BREAKFAST', 'LUNCH', 'SNACK', 'DINNER'.")
|
|
if not name or not name.strip():
|
|
return _err("'name' must not be empty.")
|
|
if date_err := _validate_date(date):
|
|
return _err(date_err)
|
|
|
|
params: dict[str, Any] = {
|
|
"name": name,
|
|
"date": date,
|
|
"type": meal_type,
|
|
}
|
|
|
|
try:
|
|
data = _authenticated_call("mpcreate", params)
|
|
except RuntimeError as exc:
|
|
return _err(str(exc))
|
|
|
|
# NOTE: mpcreate returns a00.r.r as an *array*, unlike mpcreateByRecipeId
|
|
# which returns a plain object. Take the first (and only) element.
|
|
try:
|
|
items = data["a00"]["r"]["r"]
|
|
if not isinstance(items, list) or not items:
|
|
raise TypeError("a00.r.r is not a non-empty list")
|
|
dish = items[0]
|
|
if not isinstance(dish, dict) or "metaId" not in dish:
|
|
raise TypeError("unexpected dish shape")
|
|
except (KeyError, TypeError):
|
|
return json.dumps(
|
|
{"warning": "Unexpected mpcreate response structure", "raw": data},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
|
|
rights = dish.get("rights") or {}
|
|
result: dict[str, Any] = {
|
|
"id": dish.get("metaId"),
|
|
"date": dish.get("date"),
|
|
"type": dish.get("type"),
|
|
"name": dish.get("name"),
|
|
"recipe_id": dish.get("recipeId") or None,
|
|
# mpcreate always creates free-text entries (not from the recipe box).
|
|
"is_from_recipe_box": False,
|
|
"note": None,
|
|
"serves": None,
|
|
"can_update": rights.get("canUpdate") == "true",
|
|
"can_delete": rights.get("canDelete") == "true",
|
|
}
|
|
return json.dumps(result, ensure_ascii=False, indent=2)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tool: delete_meal_plan_entry
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@mcp.tool()
|
|
def delete_meal_plan_entry(entry_id: str) -> str:
|
|
"""Permanently delete a meal plan entry (dish or meal).
|
|
|
|
IMPORTANT: Ask the user for confirmation before calling this tool.
|
|
|
|
Works for both ``dish/...`` entries (planned meals/recipes) and
|
|
``meal/...`` entries (notes/servings). Use ``get_meal_plan`` to
|
|
retrieve entry IDs.
|
|
|
|
Args:
|
|
entry_id: Entry metaId from get_meal_plan
|
|
(e.g. ``"dish/16282169_20009811"`` or ``"meal/16282169_1620659"``).
|
|
|
|
Returns:
|
|
JSON success indicator or an error message.
|
|
"""
|
|
if not entry_id.startswith(("dish/", "meal/")):
|
|
return _err(
|
|
"entry_id must be a dish or meal metaId "
|
|
"(e.g. 'dish/16282169_20009811' or 'meal/16282169_1620659')."
|
|
)
|
|
|
|
try:
|
|
_authenticated_call("metadelete", {"id": entry_id})
|
|
except RuntimeError as exc:
|
|
return _err(str(exc))
|
|
|
|
return json.dumps({"deleted": True, "id": entry_id}, ensure_ascii=False, indent=2)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tool: add_meal_note
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@mcp.tool()
|
|
def add_meal_note(
|
|
date: str,
|
|
meal_type: str,
|
|
note: str | None = None,
|
|
serves: int | None = None,
|
|
) -> str:
|
|
"""Add a note and/or serving count to a meal plan slot.
|
|
|
|
IMPORTANT: Ask the user for confirmation before calling this tool.
|
|
|
|
Creates a ``meal/`` entry for the given date and meal type.
|
|
At least one of ``note`` or ``serves`` must be provided.
|
|
|
|
Args:
|
|
date: Target date in ISO format (e.g. ``"2026-04-20"``).
|
|
meal_type: Meal slot — one of ``"BREAKFAST"``, ``"LUNCH"``,
|
|
``"SNACK"``, or ``"DINNER"``.
|
|
note: Optional free-text note (e.g. ``"Bitte ohne Zwiebeln"``).
|
|
serves: Optional number of servings as integer (e.g. ``4``).
|
|
|
|
Returns:
|
|
JSON with the new meal entry on success, or an error message.
|
|
"""
|
|
if meal_type not in ("BREAKFAST", "LUNCH", "SNACK", "DINNER"):
|
|
return _err("meal_type must be one of 'BREAKFAST', 'LUNCH', 'SNACK', 'DINNER'.")
|
|
if note is None and serves is None:
|
|
return _err("At least one of 'note' or 'serves' must be provided.")
|
|
if date_err := _validate_date(date):
|
|
return _err(date_err)
|
|
|
|
params: dict[str, Any] = {"date": date, "type": meal_type}
|
|
if note is not None:
|
|
params["note"] = note
|
|
if serves is not None:
|
|
params["serves"] = str(serves) # API expects string (int→string)
|
|
|
|
try:
|
|
data = _authenticated_call("mpmealput", params)
|
|
except RuntimeError as exc:
|
|
return _err(str(exc))
|
|
|
|
try:
|
|
meal = data["a00"]["r"]["r"]
|
|
if not isinstance(meal, dict) or "metaId" not in meal:
|
|
raise TypeError("unexpected shape")
|
|
except (KeyError, TypeError):
|
|
return json.dumps(
|
|
{"warning": "Unexpected mpmealput response structure", "raw": data},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
|
|
rights = meal.get("rights") or {}
|
|
raw_serves = meal.get("serves")
|
|
result: dict[str, Any] = {
|
|
"id": meal.get("metaId"),
|
|
"date": meal.get("date"),
|
|
"type": meal.get("type"),
|
|
"name": None,
|
|
"recipe_id": None,
|
|
"is_from_recipe_box": None,
|
|
"note": meal.get("note") or None,
|
|
"serves": int(raw_serves) if raw_serves is not None else None,
|
|
"can_update": rights.get("canUpdate") == "true",
|
|
"can_delete": rights.get("canDelete") == "true",
|
|
}
|
|
return json.dumps(result, ensure_ascii=False, indent=2)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Factory
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def create_server() -> FastMCP:
|
|
"""Return the configured Family Wall MCP server instance.
|
|
|
|
Returns:
|
|
FastMCP instance with all tools registered.
|
|
"""
|
|
return mcp
|