| """ |
| Update Task MCP Tool |
| |
| This tool enables the AI agent to update existing task details. |
| Supports updating title and/or description. |
| """ |
|
|
| from typing import Dict, Any, Optional |
| from sqlmodel import Session, select |
| from datetime import datetime |
| from ..models.task import Task |
| from .mcp_server import MCPContext |
|
|
|
|
| async def update_task_internal( |
| ctx: MCPContext, |
| task_id: int, |
| title: Optional[str] = None, |
| description: Optional[str] = None |
| ) -> Dict[str, Any]: |
| """ |
| Internal MCP tool for updating a task's details. |
| |
| This function is called by the AgentService with user_id pre-bound. |
| It updates the task's title and/or description. |
| |
| Args: |
| ctx: MCP context containing db_engine and user_id |
| task_id: ID of the task to update |
| title: New task title (optional, 1-200 chars) |
| description: New task description (optional, max 2000 chars) |
| |
| Returns: |
| Dict containing: |
| - task: Updated task details |
| - status: "success" or "error" |
| - error: Error message (only if status is "error") |
| |
| Error Cases: |
| - Task not found: Returns error "Task not found" |
| - Task belongs to different user: Returns error "Task not found" (security) |
| - Empty title: Returns error "Title cannot be empty" |
| - Title too long: Returns error "Title exceeds 200 characters" |
| - No fields to update: Returns error "No fields provided to update" |
| - Database error: Returns error with message |
| """ |
| try: |
| |
| if title is None and description is None: |
| return { |
| "status": "error", |
| "error": "No fields provided to update" |
| } |
|
|
| |
| if title is not None: |
| title = title.strip() |
| if not title: |
| return { |
| "status": "error", |
| "error": "Title cannot be empty" |
| } |
| if len(title) > 200: |
| return { |
| "status": "error", |
| "error": "Title exceeds 200 characters" |
| } |
|
|
| |
| if description is not None and len(description) > 2000: |
| return { |
| "status": "error", |
| "error": "Description exceeds 2000 characters" |
| } |
|
|
| |
| with ctx.get_session() as session: |
| statement = select(Task).where( |
| Task.id == task_id, |
| Task.user_id == ctx.user_id |
| ) |
| task = session.exec(statement).first() |
|
|
| if not task: |
| return { |
| "status": "error", |
| "error": "Task not found" |
| } |
|
|
| |
| if title is not None: |
| task.title = title |
| if description is not None: |
| task.description = description |
|
|
| task.updated_at = datetime.utcnow() |
|
|
| session.add(task) |
| session.commit() |
| session.refresh(task) |
|
|
| |
| return { |
| "status": "success", |
| "task": { |
| "id": task.id, |
| "title": task.title, |
| "description": task.description, |
| "completed": task.completed, |
| "created_at": task.created_at.isoformat(), |
| "updated_at": task.updated_at.isoformat() |
| } |
| } |
|
|
| except Exception as e: |
| |
| print(f"Error updating task: {str(e)}") |
| return { |
| "status": "error", |
| "error": f"Database error: {str(e)}" |
| } |
|
|
|
|
| def get_tool_definition() -> Dict[str, Any]: |
| """ |
| Get OpenAI function calling definition for update_task tool. |
| |
| Returns: |
| Tool definition in OpenAI function calling format |
| """ |
| return { |
| "type": "function", |
| "function": { |
| "name": "update_task", |
| "description": ( |
| "Update an existing task's title or description. " |
| "Use this when the user wants to modify, change, or edit a task. " |
| "Examples: 'change X to Y', 'update the task', 'edit task 3', " |
| "'rename X to Y', 'add details to the task'." |
| ), |
| "parameters": { |
| "type": "object", |
| "properties": { |
| "task_id": { |
| "type": "integer", |
| "description": "The ID of the task to update" |
| }, |
| "title": { |
| "type": "string", |
| "description": "New task title (optional, 1-200 characters)" |
| }, |
| "description": { |
| "type": "string", |
| "description": "New task description (optional, max 2000 characters)" |
| } |
| }, |
| "required": ["task_id"] |
| } |
| } |
| } |
|
|