diff --git a/pyproject.toml b/pyproject.toml index d2811b5..91a2a8a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "mcp-familywall" -version = "0.1.5" +version = "0.2.0" description = "MCP server for Family Wall — read your family's lists and tasks via Claude" readme = "README.md" requires-python = ">=3.12" diff --git a/src/mcp_familywall/__init__.py b/src/mcp_familywall/__init__.py index 1276d02..d3ec452 100644 --- a/src/mcp_familywall/__init__.py +++ b/src/mcp_familywall/__init__.py @@ -1 +1 @@ -__version__ = "0.1.5" +__version__ = "0.2.0" diff --git a/src/mcp_familywall/modules/__init__.py b/src/mcp_familywall/modules/__init__.py new file mode 100644 index 0000000..6e956de --- /dev/null +++ b/src/mcp_familywall/modules/__init__.py @@ -0,0 +1 @@ +# Package marker — tools are registered in server.py. diff --git a/src/mcp_familywall/modules/lists.py b/src/mcp_familywall/modules/lists.py new file mode 100644 index 0000000..c9e3ce8 --- /dev/null +++ b/src/mcp_familywall/modules/lists.py @@ -0,0 +1,23 @@ +"""List name translation helpers.""" + +from __future__ import annotations + +# Mapping of Family Wall system list identifiers to German display names. +# Extend as new system names are discovered. +SYSTEM_NAMES: dict[str, str] = { + "SYS-CAT-SHOPPINGLIST": "Einkaufsliste", +} + + +def translate_name(name: str) -> str: + """Translate system list names to German display names. + + Unknown names are returned unchanged. + + Args: + name: Raw list name from the Family Wall API. + + Returns: + Human-readable German name, or the original name if no mapping exists. + """ + return SYSTEM_NAMES.get(name, name) diff --git a/src/mcp_familywall/server.py b/src/mcp_familywall/server.py index 2d5d4e0..5f05924 100644 --- a/src/mcp_familywall/server.py +++ b/src/mcp_familywall/server.py @@ -1,17 +1,233 @@ -"""MCP server for Family Wall. - -Tools (get_circles, get_lists, get_tasks) are implemented in Gruppe 2. -""" +"""MCP server for Family Wall — read-only tools for circles, lists and tasks.""" 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 the list of task categories from an accgetallfamily response. + + SPEC says a00.r.r[], but a02.r.r[] has also been observed. + Both paths are tried defensively; the first non-empty result wins. + + Args: + data: Raw response body from accgetallfamily. + + Returns: + List of raw list-category dicts (may be empty). + """ + for key in ("a00", "a02"): + try: + items = data[key]["r"]["r"] + if isinstance(items, list) and items: + logger.debug("Lists found under %s.r.r (%d items)", key, len(items)) + return items # type: ignore[return-value] + except (KeyError, TypeError): + continue + return [] + + +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 (Kreise) the account belongs to. + + Each circle has an id and a name. + Because the response structure of the famlistfamily endpoint has not yet + been verified against a live API call, the raw response is returned as + JSON for the first verification pass. + + Returns: + JSON string — either a list of {id, name} objects once the structure + is confirmed, or the raw API response for verification. + """ + 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("famlistfamily") + client.logout() + except FamilyWallError as exc: + return f"Error: {exc}" + except Exception as exc: + return f"Connection error: {exc}" + + # Response structure not yet verified — return raw JSON for inspection. + # TODO: once confirmed, extract and return [{id, name}, ...] list. + return json.dumps(data, ensure_ascii=False, indent=2) + + +# --------------------------------------------------------------------------- +# Tool: get_lists +# --------------------------------------------------------------------------- + + +@mcp.tool() +def get_lists(scope: str | None = None) -> str: + """Return all task lists, optionally filtered by circle name. + + Args: + scope: Optional circle name to filter by. When None, all lists from + all circles are returned. + + Returns: + JSON string — list of objects with keys: + id, name, type, open, total. + """ + try: + data = _accgetallfamily() + except RuntimeError as exc: + return f"Error: {exc}" + + raw_lists = _extract_lists(data) + if not raw_lists: + # Return raw response for debugging if no lists found at expected path + return json.dumps( + {"warning": "No lists found at expected path", "raw": data}, + ensure_ascii=False, + indent=2, + ) + + result = [] + for item in raw_lists: + # TODO: apply scope filtering once the circle field is identified + # in the response (field name not yet verified via live API call). + result.append( + { + "id": item.get("metaId"), + "name": translate_name(item.get("name", "")), + "type": item.get("taskListType"), + "open": item.get("remainingTaskNumber"), + "total": item.get("totalTaskNumber"), + } + ) + + 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 specific list. + + Args: + list_id: The metaId of the list (from get_lists). + only_open: When True (default), only incomplete tasks are returned. + + Returns: + JSON string — list of objects with keys: + id, text, description, 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, + } + ) + + return json.dumps(result, ensure_ascii=False, indent=2) + + +# --------------------------------------------------------------------------- +# Factory +# --------------------------------------------------------------------------- + def create_server() -> FastMCP: - """Create and return the Family Wall MCP server. + """Return the configured Family Wall MCP server instance. Returns: - Configured FastMCP instance (no tools registered yet — see Gruppe 2). + FastMCP instance with all tools registered. """ - return FastMCP("familywall") + return mcp