| """TodoWriteTool — persistent todo list management. |
| |
| Stores todos in ~/.stack-2.9/todos.json |
| |
| Operations: |
| - add : add a new todo item |
| - complete: mark a todo as completed |
| - delete : remove a todo by id |
| - list : list all todos (optionally filtered) |
| |
| Input schema: |
| operation : str — one of add, complete, delete, list |
| task : str — description of the task (required for add) |
| todo_id : str — id of the todo (required for complete/delete) |
| priority : str — low|medium|high|urgent (default: medium, for add) |
| """ |
|
|
| from __future__ import annotations |
|
|
| import json |
| import os |
| import uuid |
| from dataclasses import dataclass |
| from datetime import datetime, timezone |
| from typing import Any |
|
|
| from .base import BaseTool, ToolResult |
| from .registry import get_registry |
|
|
| TOOL_NAME = "TodoWrite" |
| DATA_DIR = os.path.expanduser("~/.stack-2.9") |
| TODOS_FILE = os.path.join(DATA_DIR, "todos.json") |
|
|
|
|
| def _load_todos() -> list[dict[str, Any]]: |
| os.makedirs(DATA_DIR, exist_ok=True) |
| if os.path.exists(TODOS_FILE): |
| try: |
| with open(TODOS_FILE) as f: |
| return json.load(f) |
| except Exception: |
| pass |
| return [] |
|
|
|
|
| def _save_todos(todos: list[dict[str, Any]]) -> None: |
| os.makedirs(DATA_DIR, exist_ok=True) |
| with open(TODOS_FILE, "w") as f: |
| json.dump(todos, f, indent=2, default=str) |
|
|
|
|
| class TodoWriteTool(BaseTool[dict[str, Any], dict[str, Any]]): |
| """Persistent todo list tool supporting add, complete, delete, and list operations.""" |
|
|
| name = TOOL_NAME |
| description = "Manage a persistent session todo list: add, complete, delete, or list items." |
| search_hint = "manage session todo checklist add complete delete" |
|
|
| @property |
| def input_schema(self) -> dict[str, Any]: |
| return { |
| "type": "object", |
| "properties": { |
| "operation": { |
| "type": "string", |
| "enum": ["add", "complete", "delete", "list"], |
| "description": "Operation to perform", |
| }, |
| "task": { |
| "type": "string", |
| "description": "Task description (required for 'add' operation)", |
| }, |
| "todo_id": { |
| "type": "string", |
| "description": "Todo ID (required for 'complete' and 'delete' operations)", |
| }, |
| "priority": { |
| "type": "string", |
| "enum": ["low", "medium", "high", "urgent"], |
| "description": "Priority level for 'add' operation (default: medium)", |
| "default": "medium", |
| }, |
| }, |
| "required": ["operation"], |
| } |
|
|
| def validate_input(self, input_data: dict[str, Any]) -> tuple[bool, str | None]: |
| op = input_data.get("operation") |
| if op == "add" and not input_data.get("task"): |
| return False, "Error: 'task' is required when adding a todo" |
| if op in ("complete", "delete") and not input_data.get("todo_id"): |
| return False, f"Error: 'todo_id' is required for '{op}' operation" |
| return True, None |
|
|
| def execute(self, input_data: dict[str, Any]) -> ToolResult[dict[str, Any]]: |
| op = input_data.get("operation") |
| todos = _load_todos() |
|
|
| if op == "add": |
| return self._add(todos, input_data) |
| elif op == "complete": |
| return self._complete(todos, input_data) |
| elif op == "delete": |
| return self._delete(todos, input_data) |
| elif op == "list": |
| return self._list(todos, input_data) |
| else: |
| return ToolResult(success=False, error=f"Unknown operation: {op}") |
|
|
| def _add(self, todos: list[dict[str, Any]], input_data: dict[str, Any]) -> ToolResult[dict[str, Any]]: |
| todo_id = str(uuid.uuid4())[:8] |
| now = datetime.now(timezone.utc).isoformat() |
| item = { |
| "id": todo_id, |
| "content": input_data["task"], |
| "status": "pending", |
| "priority": input_data.get("priority", "medium"), |
| "created_at": now, |
| "updated_at": now, |
| } |
| todos.append(item) |
| _save_todos(todos) |
| return ToolResult( |
| success=True, |
| data={"id": todo_id, "content": item["content"], "status": "pending", "priority": item["priority"]}, |
| ) |
|
|
| def _complete(self, todos: list[dict[str, Any]], input_data: dict[str, Any]) -> ToolResult[dict[str, Any]]: |
| todo_id = input_data["todo_id"] |
| for t in todos: |
| if t["id"] == todo_id: |
| t["status"] = "completed" |
| t["updated_at"] = datetime.now(timezone.utc).isoformat() |
| _save_todos(todos) |
| return ToolResult(success=True, data={"id": todo_id, "status": "completed"}) |
| return ToolResult(success=False, error=f"Todo #{todo_id} not found") |
|
|
| def _delete(self, todos: list[dict[str, Any]], input_data: dict[str, Any]) -> ToolResult[dict[str, Any]]: |
| todo_id = input_data["todo_id"] |
| original_len = len(todos) |
| todos[:] = [t for t in todos if t["id"] != todo_id] |
| if len(todos) == original_len: |
| return ToolResult(success=False, error=f"Todo #{todo_id} not found") |
| _save_todos(todos) |
| return ToolResult(success=True, data={"id": todo_id, "deleted": True}) |
|
|
| def _list(self, todos: list[dict[str, Any]], input_data: dict[str, Any]) -> ToolResult[dict[str, Any]]: |
| status_filter = input_data.get("status") |
| if status_filter: |
| todos = [t for t in todos if t.get("status") == status_filter] |
| return ToolResult( |
| success=True, |
| data={ |
| "todos": todos, |
| "total": len(todos), |
| "pending": sum(1 for t in _load_todos() if t.get("status") == "pending"), |
| "completed": sum(1 for t in _load_todos() if t.get("status") == "completed"), |
| }, |
| ) |
|
|
| def map_result_to_message(self, result: dict, tool_use_id: str | None = None) -> str: |
| if "error" in result and not result.get("success", True): |
| return result["error"] |
| data = result.get("data", {}) |
| op = data.get("operation", "") |
| if op == "add": |
| return f"Todo #{data['id']} added: {data['content']} [{data['status']}]" |
| elif op == "complete": |
| return f"Todo #{data['id']} marked as completed." |
| elif op == "delete": |
| return f"Todo #{data['id']} deleted." |
| elif op == "list": |
| items = data.get("todos", []) |
| if not items: |
| return "No todos found." |
| lines = [f"{data['total']} todo(s) (pending: {data['pending']}, completed: {data['completed']}):\n"] |
| for t in items: |
| lines.append(f" [{t['status']:9}] #{t['id']} [{t.get('priority','medium'):6}] {t['content']}") |
| return "\n".join(lines) |
| return str(data) |
|
|
|
|
| |
| get_registry().register(TodoWriteTool()) |
|
|
