Spaces:
Running
Running
File size: 2,704 Bytes
81aa0b5 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 | """Todo list tool β Claude Code-style task tracking.
Todos are persisted per-session in memory. Each todo has:
- id (string)
- content (string)
- status: pending | in_progress | completed
- priority: high | medium | low
"""
from __future__ import annotations
import threading
from typing import Any
# βββ Per-session state ββββββββββββββββββββββββββββββββββββββββββββββββββ
_sessions: dict[str, list[dict[str, Any]]] = {}
_lock = threading.Lock()
def _get_session_id(session_id: str | None) -> str:
return session_id or "default"
def todo_read(session_id: str | None = None) -> dict[str, Any]:
"""Read the current todo list for a session."""
sid = _get_session_id(session_id)
with _lock:
todos = list(_sessions.get(sid, []))
return {"success": True, "todos": todos}
def todo_write(
todos: list[dict[str, Any]],
session_id: str | None = None,
) -> dict[str, Any]:
"""Replace the entire todo list for a session.
Each todo: {id, content, status, priority}
"""
sid = _get_session_id(session_id)
# Validate and normalize
normalized: list[dict[str, Any]] = []
for t in todos:
if not isinstance(t, dict):
continue
normalized.append({
"id": str(t.get("id", "")),
"content": str(t.get("content", "")),
"status": t.get("status", "pending") if t.get("status") in {"pending", "in_progress", "completed"} else "pending",
"priority": t.get("priority", "medium") if t.get("priority") in {"high", "medium", "low"} else "medium",
})
with _lock:
_sessions[sid] = normalized
return {"success": True, "todos": normalized, "count": len(normalized)}
def todo_update(
todo_id: str,
status: str | None = None,
content: str | None = None,
session_id: str | None = None,
) -> dict[str, Any]:
"""Update a single todo by id."""
sid = _get_session_id(session_id)
with _lock:
todos = _sessions.get(sid, [])
for t in todos:
if t["id"] == todo_id:
if status in {"pending", "in_progress", "completed"}:
t["status"] = status
if content is not None:
t["content"] = content
return {"success": True, "todo": t}
return {"success": False, "error": f"Todo not found: {todo_id}"}
def todo_clear(session_id: str | None = None) -> dict[str, Any]:
"""Clear all todos for a session."""
sid = _get_session_id(session_id)
with _lock:
_sessions.pop(sid, None)
return {"success": True}
|