| """ |
| MCP tool definitions for Todo task management. |
| |
| Per @specs/001-chatbot-mcp/contracts/mcp-tools.json |
| All tools enforce user isolation by requiring user_id in every request. |
| """ |
| from sqlmodel import Session, select |
| from typing import Optional, TYPE_CHECKING |
| from datetime import datetime |
| from uuid import UUID |
|
|
| from models.task import TaskTable |
| from config import engine |
|
|
| if TYPE_CHECKING: |
| from mcp.server import SimpleMCPRegistry |
|
|
|
|
| def register_task_tools(mcp_server) -> None: |
| """ |
| Register all task management tools with the MCP server. |
| |
| Args: |
| mcp_server: The SimpleMCPRegistry instance |
| |
| This function is called during MCP server initialization to register |
| all 5 task tools: add_task, list_tasks, complete_task, delete_task, update_task. |
| Per @specs/001-chatbot-mcp/plan.md, all task operations go through MCP. |
| """ |
|
|
| @mcp_server.tool() |
| async def add_task( |
| user_id: str, |
| title: str, |
| description: Optional[str] = None |
| ) -> dict: |
| """ |
| Create a new todo task for a user. |
| |
| Use this when the user requests to create, add, or make a new task. |
| |
| Args: |
| user_id: ID of the user who will own the task (UUID string) |
| title: Task title (short, descriptive name) |
| description: Optional detailed description |
| |
| Returns: |
| Dict with task_id, title, description, created_at, success, error |
| """ |
| try: |
| with Session(engine) as session: |
| task = TaskTable( |
| user_id=UUID(user_id), |
| title=title, |
| description=description, |
| completed=False, |
| created_at=datetime.utcnow() |
| ) |
| session.add(task) |
| session.commit() |
| session.refresh(task) |
|
|
| return { |
| "task_id": str(task.id), |
| "title": task.title, |
| "description": task.description, |
| "created_at": task.created_at.isoformat(), |
| "success": True, |
| "error": None |
| } |
| except Exception as e: |
| return { |
| "task_id": None, |
| "title": None, |
| "description": None, |
| "created_at": None, |
| "success": False, |
| "error": str(e) |
| } |
|
|
| @mcp_server.tool() |
| async def list_tasks( |
| user_id: str, |
| include_completed: bool = True |
| ) -> dict: |
| """ |
| List all tasks for a user. |
| |
| Use this when the user asks to see, show, or display their tasks. |
| |
| Args: |
| user_id: ID of the user whose tasks to list (UUID string) |
| include_completed: Whether to include completed tasks (default: true) |
| |
| Returns: |
| Dict with tasks list, count, success, error |
| """ |
| try: |
| with Session(engine) as session: |
| statement = select(TaskTable).where(TaskTable.user_id == UUID(user_id)) |
|
|
| if not include_completed: |
| statement = statement.where(TaskTable.completed == False) |
|
|
| tasks = session.exec(statement).all() |
|
|
| return { |
| "tasks": [ |
| { |
| "id": str(task.id), |
| "title": task.title, |
| "description": task.description, |
| "completed": task.completed, |
| "created_at": task.created_at.isoformat(), |
| "completed_at": task.completed_at.isoformat() if task.completed_at else None |
| } |
| for task in tasks |
| ], |
| "count": len(tasks), |
| "success": True, |
| "error": None |
| } |
| except Exception as e: |
| return { |
| "tasks": [], |
| "count": 0, |
| "success": False, |
| "error": str(e) |
| } |
|
|
| @mcp_server.tool() |
| async def complete_task( |
| user_id: str, |
| task_id: str |
| ) -> dict: |
| """ |
| Mark a task as completed. |
| |
| Use this when the user asks to complete, finish, or check off a task. |
| |
| Args: |
| user_id: ID of the user who owns the task (UUID string) |
| task_id: ID of the task to mark as complete (UUID string) |
| |
| Returns: |
| Dict with task_id, title, completed, completed_at, success, error |
| """ |
| try: |
| with Session(engine) as session: |
| task = session.query(TaskTable).filter( |
| TaskTable.id == UUID(task_id), |
| TaskTable.user_id == UUID(user_id) |
| ).first() |
|
|
| if not task: |
| return { |
| "task_id": task_id, |
| "title": None, |
| "completed": False, |
| "completed_at": None, |
| "success": False, |
| "error": "Task not found or access denied" |
| } |
|
|
| task.completed = True |
| task.completed_at = datetime.utcnow() |
| session.add(task) |
| session.commit() |
| session.refresh(task) |
|
|
| return { |
| "task_id": str(task.id), |
| "title": task.title, |
| "completed": task.completed, |
| "completed_at": task.completed_at.isoformat(), |
| "success": True, |
| "error": None |
| } |
| except Exception as e: |
| return { |
| "task_id": task_id, |
| "title": None, |
| "completed": False, |
| "completed_at": None, |
| "success": False, |
| "error": str(e) |
| } |
|
|
| @mcp_server.tool() |
| async def delete_task( |
| user_id: str, |
| task_id: str |
| ) -> dict: |
| """ |
| Delete a task permanently. |
| |
| CRITICAL: The task_id parameter must be a valid UUID from the user's existing tasks. |
| If the user provides a task number or title, you MUST map it to the correct task_id. |
| Example: If user says "delete task 1", find the task with index 0 in their task list and use its ID. |
| Always confirm which task you're deleting by showing the task title before proceeding. |
| |
| Use this when the user asks to delete, remove, or get rid of a task. |
| |
| Args: |
| user_id: ID of the user who owns the task (UUID string) |
| task_id: ID of the task to delete (UUID string) - must match an existing task ID for this user |
| |
| Returns: |
| Dict with task_id, title, deleted, success, error |
| """ |
| try: |
| with Session(engine) as session: |
| task = session.query(TaskTable).filter( |
| TaskTable.id == UUID(task_id), |
| TaskTable.user_id == UUID(user_id) |
| ).first() |
|
|
| if not task: |
| return { |
| "task_id": task_id, |
| "title": None, |
| "deleted": False, |
| "success": False, |
| "error": "Task not found or access denied" |
| } |
|
|
| title = task.title |
| session.delete(task) |
| session.commit() |
|
|
| return { |
| "task_id": task_id, |
| "title": title, |
| "deleted": True, |
| "success": True, |
| "error": None |
| } |
| except Exception as e: |
| return { |
| "task_id": task_id, |
| "title": None, |
| "deleted": False, |
| "success": False, |
| "error": str(e) |
| } |
|
|
| @mcp_server.tool() |
| async def update_task( |
| user_id: str, |
| task_id: str, |
| title: Optional[str] = None, |
| description: Optional[str] = None |
| ) -> dict: |
| """ |
| Update a task's title or description. |
| |
| CRITICAL: The task_id parameter must be a valid UUID from the user's existing tasks. |
| If the user provides a task number or title, you MUST map it to the correct task_id. |
| Example: If user says "update task 1 to buy milk", find the task with index 0 in their task list and use its ID. |
| |
| Use this when the user asks to change, modify, or edit a task. |
| |
| Args: |
| user_id: ID of the user who owns the task (UUID string) |
| task_id: ID of the task to update (UUID string) - must match an existing task ID for this user |
| title: New task title (optional) |
| description: New task description (optional, empty string to clear) |
| |
| Returns: |
| Dict with task_id, title, description, updated_at, success, error |
| """ |
| try: |
| with Session(engine) as session: |
| task = session.query(TaskTable).filter( |
| TaskTable.id == UUID(task_id), |
| TaskTable.user_id == UUID(user_id) |
| ).first() |
|
|
| if not task: |
| return { |
| "task_id": task_id, |
| "title": None, |
| "description": None, |
| "updated_at": None, |
| "success": False, |
| "error": "Task not found or access denied" |
| } |
|
|
| if title is not None: |
| task.title = title |
| if description is not None: |
| task.description = description if description else None |
|
|
| session.add(task) |
| session.commit() |
| session.refresh(task) |
|
|
| return { |
| "task_id": str(task.id), |
| "title": task.title, |
| "description": task.description, |
| "updated_at": task.created_at.isoformat(), |
| "success": True, |
| "error": None |
| } |
| except Exception as e: |
| return { |
| "task_id": task_id, |
| "title": None, |
| "description": None, |
| "updated_at": None, |
| "success": False, |
| "error": str(e) |
| } |
|
|
