diff --git a/application/agents/tools/todo_list.py b/application/agents/tools/todo_list.py new file mode 100644 index 00000000..87a3e969 --- /dev/null +++ b/application/agents/tools/todo_list.py @@ -0,0 +1,321 @@ +from datetime import datetime +from typing import Any, Dict, List, Optional +import uuid + +from .base import Tool +from application.core.mongo_db import MongoDB +from application.core.settings import settings + + +class TodoListTool(Tool): + """Todo List + + Manages todo items for users. Supports creating, viewing, updating, and deleting todos. + """ + + def __init__(self, tool_config: Optional[Dict[str, Any]] = None, user_id: Optional[str] = None) -> None: + """Initialize the tool. + + Args: + tool_config: Optional tool configuration. Should include: + - tool_id: Unique identifier for this todo list tool instance (from user_tools._id) + This ensures each user's tool configuration has isolated todos + user_id: The authenticated user's id (should come from decoded_token["sub"]). + """ + self.user_id: Optional[str] = user_id + + # Get tool_id from configuration (passed from user_tools._id in production) + # In production, tool_id is the MongoDB ObjectId string from user_tools collection + if tool_config and "tool_id" in tool_config: + self.tool_id = tool_config["tool_id"] + elif user_id: + # Fallback for backward compatibility or testing + self.tool_id = f"default_{user_id}" + else: + # Last resort fallback (shouldn't happen in normal use) + self.tool_id = str(uuid.uuid4()) + + db = MongoDB.get_client()[settings.MONGO_DB_NAME] + self.collection = db["todos"] + + # ----------------------------- + # Action implementations + # ----------------------------- + def execute_action(self, action_name: str, **kwargs: Any) -> str: + """Execute an action by name. + + Args: + action_name: One of list, create, get, update, complete, delete. + **kwargs: Parameters for the action. + + Returns: + A human-readable string result. + """ + if not self.user_id: + return "Error: TodoListTool requires a valid user_id." + + if action_name == "list": + return self._list() + + if action_name == "create": + return self._create(kwargs.get("title", "")) + + if action_name == "get": + return self._get(kwargs.get("todo_id")) + + if action_name == "update": + return self._update( + kwargs.get("todo_id"), + kwargs.get("title", "") + ) + + if action_name == "complete": + return self._complete(kwargs.get("todo_id")) + + if action_name == "delete": + return self._delete(kwargs.get("todo_id")) + + return f"Unknown action: {action_name}" + + def get_actions_metadata(self) -> List[Dict[str, Any]]: + """Return JSON metadata describing supported actions for tool schemas.""" + return [ + { + "name": "list", + "description": "List all todos for the user.", + "parameters": {"type": "object", "properties": {}}, + }, + { + "name": "create", + "description": "Create a new todo item.", + "parameters": { + "type": "object", + "properties": { + "title": { + "type": "string", + "description": "Title of the todo item." + } + }, + "required": ["title"], + }, + }, + { + "name": "get", + "description": "Get a specific todo by ID.", + "parameters": { + "type": "object", + "properties": { + "todo_id": { + "type": "integer", + "description": "The ID of the todo to retrieve." + } + }, + "required": ["todo_id"], + }, + }, + { + "name": "update", + "description": "Update a todo's title by ID.", + "parameters": { + "type": "object", + "properties": { + "todo_id": { + "type": "integer", + "description": "The ID of the todo to update." + }, + "title": { + "type": "string", + "description": "The new title for the todo." + } + }, + "required": ["todo_id", "title"], + }, + }, + { + "name": "complete", + "description": "Mark a todo as completed.", + "parameters": { + "type": "object", + "properties": { + "todo_id": { + "type": "integer", + "description": "The ID of the todo to mark as completed." + } + }, + "required": ["todo_id"], + }, + }, + { + "name": "delete", + "description": "Delete a specific todo by ID.", + "parameters": { + "type": "object", + "properties": { + "todo_id": { + "type": "integer", + "description": "The ID of the todo to delete." + } + }, + "required": ["todo_id"], + }, + }, + ] + + def get_config_requirements(self) -> Dict[str, Any]: + """Return configuration requirements.""" + return {} + + # ----------------------------- + # Internal helpers + # ----------------------------- + def _coerce_todo_id(self, value: Optional[Any]) -> Optional[int]: + """Convert todo identifiers to sequential integers.""" + if value is None: + return None + + if isinstance(value, int): + return value if value > 0 else None + + if isinstance(value, str): + stripped = value.strip() + if stripped.isdigit(): + numeric_value = int(stripped) + return numeric_value if numeric_value > 0 else None + + return None + + def _get_next_todo_id(self) -> int: + """Get the next sequential todo_id for this user and tool. + + Returns a simple integer (1, 2, 3, ...) scoped to this user/tool. + With 5-10 todos max, scanning is negligible. + """ + # Find all todos for this user/tool and get their IDs + todos = list(self.collection.find( + {"user_id": self.user_id, "tool_id": self.tool_id}, + {"todo_id": 1} + )) + + # Find the maximum todo_id + max_id = 0 + for todo in todos: + todo_id = self._coerce_todo_id(todo.get("todo_id")) + if todo_id is not None: + max_id = max(max_id, todo_id) + + return max_id + 1 + + def _list(self) -> str: + """List all todos for the user.""" + cursor = self.collection.find({"user_id": self.user_id, "tool_id": self.tool_id}) + todos = list(cursor) + + if not todos: + return "No todos found." + + result_lines = ["Todos:"] + for doc in todos: + todo_id = doc.get("todo_id") + title = doc.get("title", "Untitled") + status = doc.get("status", "open") + + line = f"[{todo_id}] {title} ({status})" + result_lines.append(line) + + return "\n".join(result_lines) + + def _create(self, title: str) -> str: + """Create a new todo item.""" + title = (title or "").strip() + if not title: + return "Error: Title is required." + + now = datetime.now() + todo_id = self._get_next_todo_id() + + doc = { + "todo_id": todo_id, + "user_id": self.user_id, + "tool_id": self.tool_id, + "title": title, + "status": "open", + "created_at": now, + "updated_at": now, + } + self.collection.insert_one(doc) + return f"Todo created with ID {todo_id}: {title}" + + def _get(self, todo_id: Optional[Any]) -> str: + """Get a specific todo by ID.""" + parsed_todo_id = self._coerce_todo_id(todo_id) + if parsed_todo_id is None: + return "Error: todo_id must be a positive integer." + + doc = self.collection.find_one({ + "user_id": self.user_id, + "tool_id": self.tool_id, + "todo_id": parsed_todo_id + }) + + if not doc: + return f"Error: Todo with ID {parsed_todo_id} not found." + + title = doc.get("title", "Untitled") + status = doc.get("status", "open") + + result = f"Todo [{parsed_todo_id}]:\nTitle: {title}\nStatus: {status}" + + return result + + def _update(self, todo_id: Optional[Any], title: str) -> str: + """Update a todo's title by ID.""" + parsed_todo_id = self._coerce_todo_id(todo_id) + if parsed_todo_id is None: + return "Error: todo_id must be a positive integer." + + title = (title or "").strip() + if not title: + return "Error: Title is required." + + result = self.collection.update_one( + {"user_id": self.user_id, "tool_id": self.tool_id, "todo_id": parsed_todo_id}, + {"$set": {"title": title, "updated_at": datetime.now()}} + ) + + if result.matched_count == 0: + return f"Error: Todo with ID {parsed_todo_id} not found." + + return f"Todo {parsed_todo_id} updated to: {title}" + + def _complete(self, todo_id: Optional[Any]) -> str: + """Mark a todo as completed.""" + parsed_todo_id = self._coerce_todo_id(todo_id) + if parsed_todo_id is None: + return "Error: todo_id must be a positive integer." + + result = self.collection.update_one( + {"user_id": self.user_id, "tool_id": self.tool_id, "todo_id": parsed_todo_id}, + {"$set": {"status": "completed", "updated_at": datetime.now()}} + ) + + if result.matched_count == 0: + return f"Error: Todo with ID {parsed_todo_id} not found." + + return f"Todo {parsed_todo_id} marked as completed." + + def _delete(self, todo_id: Optional[Any]) -> str: + """Delete a specific todo by ID.""" + parsed_todo_id = self._coerce_todo_id(todo_id) + if parsed_todo_id is None: + return "Error: todo_id must be a positive integer." + + result = self.collection.delete_one({ + "user_id": self.user_id, + "tool_id": self.tool_id, + "todo_id": parsed_todo_id + }) + + if result.deleted_count == 0: + return f"Error: Todo with ID {parsed_todo_id} not found." + + return f"Todo {parsed_todo_id} deleted." diff --git a/application/agents/tools/tool_manager.py b/application/agents/tools/tool_manager.py index fb45b987..855f1b53 100644 --- a/application/agents/tools/tool_manager.py +++ b/application/agents/tools/tool_manager.py @@ -28,7 +28,7 @@ class ToolManager: module = importlib.import_module(f"application.agents.tools.{tool_name}") for member_name, obj in inspect.getmembers(module, inspect.isclass): if issubclass(obj, Tool) and obj is not Tool: - if tool_name in {"mcp_tool", "notes", "memory"} and user_id: + if tool_name in {"mcp_tool", "notes", "memory", "todo_list"} and user_id: return obj(tool_config, user_id) else: return obj(tool_config) @@ -36,7 +36,7 @@ class ToolManager: def execute_action(self, tool_name, action_name, user_id=None, **kwargs): if tool_name not in self.tools: raise ValueError(f"Tool '{tool_name}' not loaded") - if tool_name in {"mcp_tool", "memory"} and user_id: + if tool_name in {"mcp_tool", "memory", "todo_list"} and user_id: tool_config = self.config.get(tool_name, {}) tool = self.load_tool(tool_name, tool_config, user_id) return tool.execute_action(action_name, **kwargs) diff --git a/frontend/public/toolIcons/tool_todo_list.svg b/frontend/public/toolIcons/tool_todo_list.svg new file mode 100644 index 00000000..3eb8ef24 --- /dev/null +++ b/frontend/public/toolIcons/tool_todo_list.svg @@ -0,0 +1 @@ + diff --git a/tests/test_todo_tool.py b/tests/test_todo_tool.py new file mode 100644 index 00000000..a193e8ee --- /dev/null +++ b/tests/test_todo_tool.py @@ -0,0 +1,156 @@ +import pytest +from application.agents.tools.todo_list import TodoListTool +from application.core.settings import settings + + +class FakeCursor(list): + def sort(self, key, direction): + reverse = direction == -1 + sorted_list = sorted(self, key=lambda d: d.get(key, 0), reverse=reverse) + return FakeCursor(sorted_list) + + def limit(self, count): + return FakeCursor(self[:count]) + + def __iter__(self): + return self + + def __next__(self): + if not self: + raise StopIteration + return self.pop(0) + + +class FakeCollection: + def __init__(self): + self.docs = {} + + def create_index(self, *args, **kwargs): + pass + + def insert_one(self, doc): + key = (doc["user_id"], doc["tool_id"], int(doc["todo_id"])) + self.docs[key] = doc + return type("res", (), {"inserted_id": key}) + + def find_one(self, query): + key = (query.get("user_id"), query.get("tool_id"), int(query.get("todo_id"))) + return self.docs.get(key) + + def find(self, query): + user_id = query.get("user_id") + tool_id = query.get("tool_id") + filtered = [ + doc for (uid, tid, _), doc in self.docs.items() + if uid == user_id and tid == tool_id + ] + return FakeCursor(filtered) + + def update_one(self, query, update, upsert=False): + key = (query.get("user_id"), query.get("tool_id"), int(query.get("todo_id"))) + if key in self.docs: + self.docs[key].update(update.get("$set", {})) + return type("res", (), {"matched_count": 1}) + elif upsert: + new_doc = {**query, **update.get("$set", {})} + self.docs[key] = new_doc + return type("res", (), {"matched_count": 1}) + else: + return type("res", (), {"matched_count": 0}) + + def delete_one(self, query): + key = (query.get("user_id"), query.get("tool_id"), int(query.get("todo_id"))) + if key in self.docs: + del self.docs[key] + return type("res", (), {"deleted_count": 1}) + return type("res", (), {"deleted_count": 0}) + + +@pytest.fixture +def todo_tool(monkeypatch) -> TodoListTool: + """Provides a TodoListTool with a fake MongoDB backend.""" + fake_collection = FakeCollection() + fake_client = {settings.MONGO_DB_NAME: {"todos": fake_collection}} + monkeypatch.setattr("application.core.mongo_db.MongoDB.get_client", lambda: fake_client) + return TodoListTool({"tool_id": "test_tool"}, user_id="test_user") + + +def test_create_and_get(todo_tool: TodoListTool): + res = todo_tool.execute_action("todo_create", title="Write tests", description="Write pytest cases") + assert res["status_code"] == 201 + todo_id = res["todo_id"] + + get_res = todo_tool.execute_action("todo_get", todo_id=todo_id) + assert get_res["status_code"] == 200 + assert get_res["todo"]["title"] == "Write tests" + assert get_res["todo"]["description"] == "Write pytest cases" + + +def test_get_all_todos(todo_tool: TodoListTool): + todo_tool.execute_action("todo_create", title="Task 1") + todo_tool.execute_action("todo_create", title="Task 2") + + list_res = todo_tool.execute_action("todo_list") + assert list_res["status_code"] == 200 + titles = [todo["title"] for todo in list_res["todos"]] + assert "Task 1" in titles + assert "Task 2" in titles + + +def test_update_todo(todo_tool: TodoListTool): + create_res = todo_tool.execute_action("todo_create", title="Initial Title") + todo_id = create_res["todo_id"] + + update_res = todo_tool.execute_action("todo_update", todo_id=todo_id, updates={"title": "Updated Title", "status": "done"}) + assert update_res["status_code"] == 200 + + get_res = todo_tool.execute_action("todo_get", todo_id=todo_id) + assert get_res["todo"]["title"] == "Updated Title" + assert get_res["todo"]["status"] == "done" + + +def test_delete_todo(todo_tool: TodoListTool): + create_res = todo_tool.execute_action("todo_create", title="To Delete") + todo_id = create_res["todo_id"] + + delete_res = todo_tool.execute_action("todo_delete", todo_id=todo_id) + assert delete_res["status_code"] == 200 + + get_res = todo_tool.execute_action("todo_get", todo_id=todo_id) + assert get_res["status_code"] == 404 + + +def test_isolation_and_default_tool_id(monkeypatch): + """Ensure todos are isolated by tool_id and user_id.""" + fake_collection = FakeCollection() + fake_client = {settings.MONGO_DB_NAME: {"todos": fake_collection}} + monkeypatch.setattr("application.core.mongo_db.MongoDB.get_client", lambda: fake_client) + + # Same user, different tool_id + tool1 = TodoListTool({"tool_id": "tool_1"}, user_id="u1") + tool2 = TodoListTool({"tool_id": "tool_2"}, user_id="u1") + + r1_create = tool1.execute_action("todo_create", title="from tool 1") + r2_create = tool2.execute_action("todo_create", title="from tool 2") + + r1 = tool1.execute_action("todo_get", todo_id=r1_create["todo_id"]) + r2 = tool2.execute_action("todo_get", todo_id=r2_create["todo_id"]) + + assert r1["status_code"] == 200 + assert r1["todo"]["title"] == "from tool 1" + + assert r2["status_code"] == 200 + assert r2["todo"]["title"] == "from tool 2" + + # Same user, no tool_id → should default to same value + t3 = TodoListTool({}, user_id="default_user") + t4 = TodoListTool({}, user_id="default_user") + + assert t3.tool_id == "default_default_user" + assert t4.tool_id == "default_default_user" + + create_res = t3.execute_action("todo_create", title="shared default") + r = t4.execute_action("todo_get", todo_id=create_res["todo_id"]) + + assert r["status_code"] == 200 + assert r["todo"]["title"] == "shared default"