feat: add Task CRUD tools – create, update, toggle, delete (v0.4.0)

Implements four new MCP write tools via taskcreate2, taskupdate2,
taskmark, and metadelete endpoints. Confirmation prompts noted in
docstrings for destructive/mutating operations. Body parameters
documented in SPEC.md as pending verification via FW_DEBUG=1.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-04-15 15:06:38 +02:00
parent 332b01718e
commit 5aff3ac9bf
4 changed files with 249 additions and 5 deletions
+171 -1
View File
@@ -1,4 +1,4 @@
"""MCP server for Family Wall — read-only tools for circles, lists and tasks."""
"""MCP server for Family Wall — tools for circles, lists, tasks (read + write)."""
from __future__ import annotations
@@ -299,6 +299,176 @@ def get_activities(limit: int = 20):
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) -> 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.
description: Optional longer description.
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
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) -> str:
"""Update the text and/or description of an existing task.
IMPORTANT: Ask the user for confirmation before calling this tool.
At least one of *text* or *description* 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).
Returns:
JSON success indicator or an error message.
"""
if text is None and description is None:
return "Error: At least one of 'text' or 'description' 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
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] = {
"metaId": task_id,
"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", {"metaId": task_id})
except RuntimeError as exc:
return f"Error: {exc}"
return json.dumps({"deleted": True, "id": task_id}, ensure_ascii=False, indent=2)
# ---------------------------------------------------------------------------
# Factory
# ---------------------------------------------------------------------------