mcp-familywall/src/mcp_familywall/fw_client.py

"""Family Wall HTTP client.

Session strategy: Login → API call → Logout per MCP tool call. No session caching.

Debug logging is enabled when the environment variable FW_DEBUG=1 is set.
Passwords are always masked in debug output.
"""

from __future__ import annotations

import logging
import os
import sys
from typing import Any

import httpx

logger = logging.getLogger(__name__)

BASE_URL = "https://api.familywall.com/api"

_debug = os.environ.get("FW_DEBUG") == "1"


def _debug_log(label: str, data: Any) -> None:
    """Write debug info to stderr when FW_DEBUG=1."""
    if _debug:
        print(f"[FW_DEBUG] {label}: {data}", file=sys.stderr)


def _mask_password(data: dict[str, Any]) -> dict[str, Any]:
    """Return a copy of data with the password field masked."""
    masked = dict(data)
    if "password" in masked:
        masked["password"] = "***"
    return masked


class FamilyWallError(Exception):
    """Raised when the Family Wall API returns an error response."""

    def __init__(self, message: str, response_data: Any = None) -> None:
        super().__init__(message)
        self.response_data = response_data


class FamilyWallClient:
    """Synchronous HTTP client for the Family Wall API.

    Usage::

        client = FamilyWallClient()
        client.login(email, password)
        data = client.call("accgetallfamily", {"a01call": "taskcategorysync"})
        client.logout()

    Or use as a context manager::

        with FamilyWallClient() as client:
            client.login(email, password)
            data = client.call("famlistfamily")
    """

    def __init__(self) -> None:
        self._http: httpx.Client = httpx.Client(timeout=30)
        self._session_id: str | None = None

    def __enter__(self) -> FamilyWallClient:
        return self

    def __exit__(self, *args: object) -> None:
        self._http.close()

    def login(self, email: str, password: str) -> None:
        """Authenticate with Family Wall and store the session ID.

        Args:
            email: Family Wall account email.
            password: Family Wall account password.

        Raises:
            FamilyWallError: If authentication fails.
            httpx.HTTPError: On network-level errors.
        """
        endpoint = f"{BASE_URL}/log2in"
        payload = {
            "identifier": email,
            "password": password,
        }
        _debug_log("LOGIN request", _mask_password(payload))

        response = self._http.post(
            endpoint,
            data=payload,
            headers={"Content-Type": "application/x-www-form-urlencoded"},
        )
        response.raise_for_status()
        body: dict[str, Any] = response.json()
        _debug_log("LOGIN response", body)

        if "ex" in body or "un" in body:
            error_data = body.get("ex", body.get("un"))
            _debug_log("LOGIN ERROR", f"status={response.status_code} body={body}")
            msg = f"Login failed: {error_data}"
            raise FamilyWallError(msg, response_data=body)

        # Response structure: {"a00": {"r": {"r": {"tokenCsrf": "...", ...}}}}
        try:
            session_id: str = body["a00"]["r"]["r"]["tokenCsrf"]
        except (KeyError, TypeError) as exc:
            _debug_log("LOGIN ERROR", f"status={response.status_code} body={body}")
            raise FamilyWallError(
                "Login failed: unexpected response format", response_data=body
            ) from exc

        if not session_id:
            raise FamilyWallError("Login succeeded but tokenCsrf is empty.")

        self._session_id = session_id
        logger.debug("Logged in; session acquired")

        # Configure the HTTP client to send the session cookie and CSRF token on all future calls
        self._http.cookies.set("JSESSIONID", session_id)
        self._http.headers.update({"Tokencsrf": session_id})

    def logout(self) -> None:
        """Invalidate the current session server-side.

        Silently ignores errors — the session will eventually expire anyway.
        """
        if not self._session_id:
            return

        endpoint = f"{BASE_URL}/log2out"
        _debug_log("LOGOUT request", {})
        try:
            response = self._http.post(
                endpoint,
                data={},
                headers={"Content-Type": "application/x-www-form-urlencoded"},
            )
            _debug_log("LOGOUT response", response.text)
        except httpx.HTTPError as exc:
            logger.debug("Logout request failed (session may already be expired): %s", exc)
        finally:
            self._session_id = None
            self._http.cookies.clear()
            if "Tokencsrf" in self._http.headers:
                del self._http.headers["Tokencsrf"]

    def call(self, endpoint: str, params: dict[str, Any] | None = None) -> dict[str, Any]:
        """Call a Family Wall API endpoint.

        Must be called after a successful :meth:`login`.

        Args:
            endpoint: API endpoint name (e.g. ``"famlistfamily"``).
            params: Optional form parameters to send.

        Returns:
            Parsed JSON response body.

        Raises:
            FamilyWallError: If not logged in, or if the API returns an error.
            httpx.HTTPError: On network-level errors.
        """
        if not self._session_id:
            raise FamilyWallError("Not logged in. Call login() first.")

        url = f"{BASE_URL}/{endpoint}"
        payload = params or {}
        _debug_log(f"CALL {endpoint} request", payload)

        response = self._http.post(
            url,
            data=payload,
            headers={"Content-Type": "application/x-www-form-urlencoded"},
        )
        response.raise_for_status()
        body: dict[str, Any] = response.json()
        _debug_log(f"CALL {endpoint} response", body)

        if "ex" in body or "un" in body:
            error_data = body.get("ex", body.get("un"))
            msg = f"API error from {endpoint!r}: {error_data}"
            raise FamilyWallError(msg, response_data=body)

        # Some endpoints return per-call errors nested under a00.un.un or a00.ex.ex
        # instead of top-level — detect and surface them.
        a00 = body.get("a00", {})
        if isinstance(a00, dict):
            if "un" in a00:
                nested = a00["un"]
                msg = f"API error from {endpoint!r}: {nested}"
                raise FamilyWallError(msg, response_data=body)
            if "ex" in a00:
                nested = a00["ex"]
                msg = f"API error from {endpoint!r}: {nested}"
                raise FamilyWallError(msg, response_data=body)

        return body