programmingwithabdullah/todo-aipowered
0
1"""2MCP Tool: update_task3Purpose: Update an existing task by ID or title4"""5 6from pydantic import Field7from typing import Dict, Any, Optional8from database import get_session_context9 10 11async def update_task(12 user_id: str = Field(description="User's ID from JWT context"),13 task_id: Optional[int] = Field(description="Task ID to update", default=None),14 task_title: Optional[str] = Field(15 description="Task title to search for (if task_id not provided)", default=None16 ),17 title: Optional[str] = Field(18 description="New task title (max 200 chars)", max_length=200, default=None19 ),20 description: Optional[str] = Field(21 description="New task description (max 1000 chars)",22 max_length=1000,23 default=None,24 ),25) -> Dict[str, Any]:26 """27 Update an existing task by ID or title.28 29 Args:30 user_id: User's ID from JWT context31 task_id: Task ID to update (optional if task_title provided)32 task_title: Task title to search for (optional if task_id provided)33 title: New task title (optional, max 200 chars)34 description: New task description (optional, max 1000 chars)35 36 Returns:37 dict: {task_id, title, description, status} or {error, status} or {disambiguation, matches, status}38 """39 try:40 # Validate that at least one identifier is provided41 if task_id is None and not task_title:42 return {43 "error": "Please provide either a task ID or task title to update",44 "status": "error",45 }46 47 # Validate update fields48 if title is not None and len(title) > 200:49 return {"error": "Title must be 200 characters or less", "status": "error"}50 51 if description is not None and len(description) > 1000:52 return {53 "error": "Description must be 1000 characters or less",54 "status": "error",55 }56 57 # Build updates dict58 updates = {}59 if title is not None:60 updates["title"] = title.strip()61 if description is not None:62 updates["description"] = description.strip()63 64 if not updates:65 return {66 "error": "No updates provided. Please specify a new title or description.",67 "status": "error",68 }69 70 with get_session_context() as db:71 from services.task_service import TaskService72 from schemas.task_schemas import TaskUpdate73 from mcp_server.tool_utils import resolve_task74 75 # Resolve the task using shared logic76 task, error_response = resolve_task(user_id, db, task_id, task_title)77 if error_response:78 return error_response79 80 # Perform the update81 task_service = TaskService()82 task_update = TaskUpdate(**updates)83 updated_task = task_service.update_task(user_id, task.id, task_update, db)84 85 return {86 "task_id": updated_task.id,87 "title": updated_task.title,88 "description": updated_task.description,89 "status": "updated",90 }91 92 except Exception as e:93 from sqlalchemy.exc import SQLAlchemyError94 95 if isinstance(e, SQLAlchemyError):96 return {"error": f"Database error: {str(e)}", "status": "error"}97 return {"error": f"Unexpected error: {str(e)}", "status": "error"}98 