marcus
7abe58dee2
Documents the preferred format for quantities in shopping list tasks: item name first, quantity in parentheses at the end. Examples: "Äpfel (5x)", "Hackfleisch (500g)", "Joghurt (Erdbeere, 2x)". No functional code changes. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
1130 lines
39 KiB
Python
1130 lines
39 KiB
Python
"""MCP server for Family Wall — tools for circles, lists, tasks (read + write)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
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.lists import translate_name
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
mcp = FastMCP("familywall")
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
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, # real name unknown — TODO once field identified
|
|
"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():
|
|
"""Return all Family Wall circles as JSON list of {id, name}."""
|
|
try:
|
|
raw_circles = _famlistfamily()
|
|
except RuntimeError as exc:
|
|
return f"Error: {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 _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 f"Error: {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):
|
|
"""Return task lists as JSON, optionally filtered by circle name (scope)."""
|
|
try:
|
|
email, password = get_credentials()
|
|
except RuntimeError as exc:
|
|
return f"Error: {exc}"
|
|
|
|
try:
|
|
with FamilyWallClient() as client:
|
|
client.login(email, password)
|
|
data = client.call("taskgettasklists", {})
|
|
client.logout()
|
|
except FamilyWallError as exc:
|
|
return f"Error: {exc}"
|
|
except Exception as exc:
|
|
return f"Connection error: {exc}"
|
|
|
|
# Try known response patterns; fall back to raw JSON for verification.
|
|
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 None:
|
|
# Response structure not yet verified — return raw JSON for inspection.
|
|
return json.dumps(
|
|
{"warning": "Unexpected taskgettasklists response structure", "raw": data},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
|
|
result = []
|
|
for item in raw_lists:
|
|
# TODO: apply scope filtering once the circle field is identified.
|
|
# 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,
|
|
}
|
|
)
|
|
|
|
return json.dumps(result, ensure_ascii=False, indent=2)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tool: get_tasks
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@mcp.tool()
|
|
def get_tasks(list_id: str, only_open: bool = True):
|
|
"""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 f"Error: {exc}"
|
|
|
|
raw_tasks = _extract_tasks(data)
|
|
|
|
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
|
|
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 [],
|
|
}
|
|
)
|
|
|
|
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. Categories are filtered by locale so only
|
|
the language-appropriate names are returned (default: German).
|
|
|
|
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 f"Error: {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 f"Error: {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 f"Error: {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 f"Error: 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 f"Error: Family Wall API error: {exc}"
|
|
except Exception as exc:
|
|
return f"Error: 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):
|
|
"""Return recent Family Wall wall activities as JSON. limit controls max number of results."""
|
|
try:
|
|
email, password = get_credentials()
|
|
except RuntimeError as exc:
|
|
return f"Error: {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 f"Error: {exc}"
|
|
except Exception as exc:
|
|
return 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
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
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,
|
|
) -> 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``). Only meaningful for
|
|
shopping lists; ignored for TODO lists.
|
|
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.
|
|
|
|
Returns:
|
|
JSON with the new task's metaId on success, or an error message.
|
|
"""
|
|
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 ""
|
|
|
|
try:
|
|
data = _authenticated_call("taskcreate2", params)
|
|
except RuntimeError as exc:
|
|
return f"Error: {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,
|
|
) -> 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.
|
|
|
|
Returns:
|
|
JSON success indicator or an error message.
|
|
"""
|
|
if clear_due_date and due_date is not None:
|
|
return "Error: 'clear_due_date' and 'due_date' cannot be used together."
|
|
|
|
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
|
|
):
|
|
return "Error: At least one of 'text', 'description', 'category_id', 'due_date', 'clear_due_date', 'assignee_ids', or 'list_id' must be provided."
|
|
|
|
params: dict[str, Any] = {"metaId": task_id}
|
|
if text is not None:
|
|
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
|
|
|
|
try:
|
|
_authenticated_call("taskupdate2", params)
|
|
except RuntimeError as exc:
|
|
return f"Error: {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.
|
|
|
|
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 f"Error: {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 f"Error: {exc}"
|
|
|
|
return json.dumps({"deleted": True, "id": task_id}, 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,
|
|
) -> 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 — either ``"SHOPPING_LIST"`` or ``"TODOS"``.
|
|
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. ``"🛒"``).
|
|
|
|
Returns:
|
|
JSON with the new list object on success, or an error message.
|
|
"""
|
|
if list_type not in ("SHOPPING_LIST", "TODOS"):
|
|
return "Error: list_type must be 'SHOPPING_LIST' or 'TODOS'."
|
|
if len(name) > 200:
|
|
return "Error: 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
|
|
|
|
try:
|
|
data = _authenticated_call("taskcreatelist", params)
|
|
except RuntimeError as exc:
|
|
return f"Error: {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,
|
|
},
|
|
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 f"Error: {exc}"
|
|
|
|
list_obj: dict[str, Any] | None = None
|
|
try:
|
|
with FamilyWallClient() as client:
|
|
client.login(email, password)
|
|
|
|
# Fetch lists and verify the target can be deleted.
|
|
raw = client.call("taskgettasklists", {})
|
|
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 f"Error: 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.
|
|
# taskdeletelist uses 'id' (same pattern as metadelete / taskcategorydelete).
|
|
client.call("taskdeletelist", {"id": list_id})
|
|
client.logout()
|
|
except FamilyWallError as exc:
|
|
return f"Error: Family Wall API error: {exc}"
|
|
except Exception as exc:
|
|
return f"Error: Connection error: {exc}"
|
|
|
|
return json.dumps(
|
|
{"deleted": True, "id": list_id, "name": list_obj.get("name")},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tool: like_post
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@mcp.tool()
|
|
def like_post(post_id: str, like: bool = True) -> str:
|
|
"""Like a wall post/activity with a STAR mood.
|
|
|
|
Note: Unlike (like=False) is not yet supported. The Family Wall API offers
|
|
no discoverable endpoint or parameter to remove a like. Passing like=False
|
|
returns an error without making any API call.
|
|
|
|
Args:
|
|
post_id: Wall post ID from get_activities (e.g. ``wall/23431854_31119189``).
|
|
like: Must be ``True``. ``False`` is reserved for future unlike support.
|
|
|
|
Returns:
|
|
JSON success indicator or an error message.
|
|
"""
|
|
# Unlike is not yet supported: extensive FW_DEBUG=1 testing showed that
|
|
# wallmood with moodType="STAR" is an idempotent SET operation (not a toggle).
|
|
# Tested and ruled out: moodType variations ("NONE", "REMOVE", "DELETE", ""),
|
|
# moodStarShortcut parameter, and alternative endpoints (all return 502).
|
|
# See SPEC.md for full investigation notes.
|
|
if not like:
|
|
return json.dumps(
|
|
{"error": "Unlike is not yet supported. The unlike mechanism is unknown."},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
|
|
# Verified via FW_DEBUG=1:
|
|
# - Parameter 'wall_message_id': post ID as returned by get_activities
|
|
# - Parameter 'moodType': "STAR" (Family Wall's internal like type; "LIKE" is silently
|
|
# mapped to "STAR" server-side — use "STAR" directly)
|
|
# - Response a00.r.r: full wall message object with moodMap showing resulting state
|
|
params: dict[str, Any] = {
|
|
"wall_message_id": post_id,
|
|
"moodType": "STAR",
|
|
}
|
|
|
|
try:
|
|
data = _authenticated_call("wallmood", params)
|
|
except RuntimeError as exc:
|
|
return f"Error: {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}
|
|
|
|
# Surface a warning when the like call apparently had no effect, so the
|
|
# caller can distinguish a successful like from a silent API rejection
|
|
# (e.g. rate limit, unsupported post type, or self-like restriction).
|
|
if not now_liked:
|
|
result["warning"] = (
|
|
"Like may not have been applied. "
|
|
"Possible causes: rate limit, unsupported post type (e.g. FAMILY_CREATED), "
|
|
"or self-like restriction."
|
|
)
|
|
|
|
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
|