Spaces:
Sleeping
Sleeping
File size: 4,368 Bytes
1397957 | 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 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 | from typing import Dict, Any, List, Optional
from pydantic import BaseModel
from .tool import BaseTool, ToolContext, ToolResult
from ..core.storage import Storage
class TodoItem(BaseModel):
id: str
content: str
status: str = "pending" # pending, in_progress, completed, cancelled
priority: str = "medium" # high, medium, low
class TodoTool(BaseTool):
@property
def id(self) -> str:
return "todo"
@property
def description(self) -> str:
return (
"Manage a todo list for tracking tasks. Use this to create, update, "
"and track progress on multi-step tasks. Supports pending, in_progress, "
"completed, and cancelled statuses."
)
@property
def parameters(self) -> Dict[str, Any]:
return {
"type": "object",
"properties": {
"action": {
"type": "string",
"enum": ["read", "write"],
"description": "Action to perform: 'read' to get todos, 'write' to update todos"
},
"todos": {
"type": "array",
"description": "List of todos (required for 'write' action)",
"items": {
"type": "object",
"properties": {
"id": {"type": "string"},
"content": {"type": "string"},
"status": {
"type": "string",
"enum": ["pending", "in_progress", "completed", "cancelled"]
},
"priority": {
"type": "string",
"enum": ["high", "medium", "low"]
}
},
"required": ["id", "content", "status", "priority"]
}
}
},
"required": ["action"]
}
async def execute(self, args: Dict[str, Any], ctx: ToolContext) -> ToolResult:
action = args["action"]
session_id = ctx.session_id
if action == "read":
return await self._read_todos(session_id)
elif action == "write":
todos_data = args.get("todos", [])
return await self._write_todos(session_id, todos_data)
else:
return ToolResult(
title="Todo Error",
output=f"Unknown action: {action}",
metadata={"error": "invalid_action"}
)
async def _read_todos(self, session_id: str) -> ToolResult:
todos = await Storage.read(["todo", session_id])
if not todos:
return ToolResult(
title="Todo List",
output="No todos found for this session.",
metadata={"count": 0}
)
items = [TodoItem(**t) for t in todos]
lines = self._format_todos(items)
return ToolResult(
title="Todo List",
output="\n".join(lines),
metadata={"count": len(items)}
)
async def _write_todos(self, session_id: str, todos_data: List[Dict]) -> ToolResult:
items = [TodoItem(**t) for t in todos_data]
await Storage.write(["todo", session_id], [t.model_dump() for t in items])
lines = self._format_todos(items)
return ToolResult(
title="Todo List Updated",
output="\n".join(lines),
metadata={"count": len(items)}
)
def _format_todos(self, items: List[TodoItem]) -> List[str]:
status_icons = {
"pending": "[ ]",
"in_progress": "[~]",
"completed": "[x]",
"cancelled": "[-]"
}
priority_icons = {
"high": "!!!",
"medium": "!!",
"low": "!"
}
lines = []
for item in items:
icon = status_icons.get(item.status, "[ ]")
priority = priority_icons.get(item.priority, "")
lines.append(f"{icon} {priority} {item.content} (id: {item.id})")
return lines if lines else ["No todos."]
|