Spaces:
Running
Running
File size: 2,747 Bytes
676582c |
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 |
"""
Delete Task MCP Tool
MCP tool for deleting tasks via natural language.
Supports task identification by ID or title.
Implements user context injection for security.
"""
import logging
from typing import Union
from sqlmodel import Session, select
from ...models.task import Task
from ...core.database import get_session
from ..tool_registry import ToolExecutionResult
logger = logging.getLogger(__name__)
async def delete_task(
task_identifier: Union[int, str],
user_id: int # Injected by backend, never from LLM
) -> ToolExecutionResult:
"""
Delete a task permanently.
SECURITY: user_id is injected by the backend via MCPToolRegistry.
The LLM cannot specify or modify the user_id.
Args:
task_identifier: Task ID (integer) or task title (string)
user_id: User ID (injected by backend for security)
Returns:
ToolExecutionResult with success status and confirmation message
"""
try:
# Query task from database
db: Session = next(get_session())
try:
# Build query based on identifier type
if isinstance(task_identifier, int):
# Search by ID
statement = select(Task).where(
Task.id == task_identifier,
Task.user_id == user_id
)
identifier_type = "ID"
else:
# Search by title (exact match)
statement = select(Task).where(
Task.title == task_identifier,
Task.user_id == user_id
)
identifier_type = "title"
task = db.exec(statement).first()
# Check if task exists
if not task:
logger.warning(f"Task not found for deletion: {identifier_type}={task_identifier}, user_id={user_id}")
return ToolExecutionResult(
success=False,
error=f"Task not found with {identifier_type}: {task_identifier}"
)
# Store task title for confirmation message
task_title = task.title
# Delete task
db.delete(task)
db.commit()
logger.info(f"Task deleted successfully: id={task.id}, user_id={user_id}, title={task_title}")
return ToolExecutionResult(
success=True,
message=f"Task '{task_title}' has been deleted successfully."
)
finally:
db.close()
except Exception as e:
logger.error(f"Error deleting task: {str(e)}")
return ToolExecutionResult(
success=False,
error=f"Failed to delete task: {str(e)}"
)
|