TaskFlow / src /tools /update_task.py
BilalCode's picture
taskflow todo app
310260a
Raw
History Blame Contribute Delete
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"]
}
}
}