TaskFlow / src /tools /update_task.py
BilalCode's picture
taskflow todo app
310260a
Raw
History Blame
5.27 kB
"""
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:
# Validate that at least one field is provided
if title is None and description is None:
return {
"status": "error",
"error": "No fields provided to update"
}
# Validate title if provided
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"
}
# Validate description length if provided
if description is not None and len(description) > 2000:
return {
"status": "error",
"error": "Description exceeds 2000 characters"
}
# Query task with user_id scoping for security
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"
}
# Update fields
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 structured result
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:
# Log error and return structured error response
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"]
}
}
}