mcp-familywall/src/mcp_familywall/server.py

"""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:
    """Return the configured Family Wall MCP server instance.

    Returns:
        FastMCP instance with all tools registered.
    """
    return mcp