feat: Gruppe 1 – Projektgerüst, Auth, CLI (v0.1.0)
- pyproject.toml: hatchling build, mcp-familywall entry-point, deps - src/mcp_familywall/__init__.py: version 0.1.0 - src/mcp_familywall/config.py: YAML config loader (schema_version: 1) - src/mcp_familywall/auth.py: keyring credential resolution (FW_EMAIL/FW_PASSWORD → keyring) - src/mcp_familywall/fw_client.py: httpx client, login/logout/call, FW_DEBUG logging - src/mcp_familywall/cli.py: click CLI with setup / check / serve commands - .gitignore, README.md, CLAUDE.md, SPEC.md Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,184 @@
|
||||
"""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,
|
||||
"type": "email",
|
||||
}
|
||||
_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"))
|
||||
msg = f"Login failed: {error_data}"
|
||||
raise FamilyWallError(msg, response_data=body)
|
||||
|
||||
if "r" not in body:
|
||||
raise FamilyWallError("Login failed: unexpected response format", response_data=body)
|
||||
|
||||
# Extract JSESSIONID from the Set-Cookie header
|
||||
session_id = response.cookies.get("JSESSIONID")
|
||||
if not session_id:
|
||||
raise FamilyWallError("Login succeeded but no JSESSIONID cookie returned.")
|
||||
|
||||
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)
|
||||
|
||||
return body
|
||||
Reference in New Issue
Block a user