File size: 3,166 Bytes
310260a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 | """
Delete Task MCP Tool
This tool enables the AI agent to permanently delete tasks.
Enforces user_id scoping for security.
"""
from typing import Dict, Any
from sqlmodel import Session, select
from ..models.task import Task
from .mcp_server import MCPContext
async def delete_task_internal(
ctx: MCPContext,
task_id: int
) -> Dict[str, Any]:
"""
Internal MCP tool for deleting a task.
This function is called by the AgentService with user_id pre-bound.
It permanently removes the task from the database.
Args:
ctx: MCP context containing db_engine and user_id
task_id: ID of the task to delete
Returns:
Dict containing:
- status: "success" or "error"
- message: Confirmation message
- deleted_task_id: ID of the deleted task
- 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)
- Database error: Returns error with message
"""
try:
# 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"
}
# Store task title for confirmation message
task_title = task.title
# Delete task
session.delete(task)
session.commit()
# Return structured result
return {
"status": "success",
"message": f"Task '{task_title}' deleted successfully",
"deleted_task_id": task_id
}
except Exception as e:
# Log error and return structured error response
print(f"Error deleting 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 delete_task tool.
Returns:
Tool definition in OpenAI function calling format
"""
return {
"type": "function",
"function": {
"name": "delete_task",
"description": (
"Permanently delete a task. "
"Use this when the user wants to remove, delete, or cancel a task. "
"Examples: 'delete X', 'remove the task', 'cancel task 3', "
"'get rid of X', 'delete my first task'."
),
"parameters": {
"type": "object",
"properties": {
"task_id": {
"type": "integer",
"description": "The ID of the task to delete"
}
},
"required": ["task_id"]
}
}
}
|