Spaces:
Sleeping
Sleeping
File size: 3,554 Bytes
f2b5c2a | 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 108 109 110 111 112 113 | """
Add Task MCP Tool.
This tool allows the AI to create new tasks for users through natural language.
"""
import logging
from typing import Dict, Any
from sqlmodel import Session
from src.mcp.tools.base import MCPToolBase
from src.models.task import Task
from src.validation.security_guard import security_guard
logger = logging.getLogger(__name__)
class AddTaskTool(MCPToolBase):
"""MCP tool for adding tasks."""
@property
def name(self) -> str:
return "add_task"
@property
def description(self) -> str:
return "Create a new task for the authenticated user. Extracts task title and optional description from natural language input."
@property
def parameters(self) -> Dict[str, Any]:
return {
"user_id": {
"description": "ID of the authenticated user (automatically provided by system)",
"type": "integer",
"required": True
},
"title": {
"description": "Task title extracted from user's message",
"type": "string",
"required": True
},
"description": {
"description": "Optional task description or additional details",
"type": "string",
"required": False
}
}
async def execute(self, user_id: int, title: str, description: str = "", db: Session = None, **kwargs) -> Dict[str, Any]:
"""
Execute add_task tool.
Args:
user_id: Authenticated user ID
title: Task title
description: Optional task description
db: Database session
Returns:
Dictionary with success status and task details
"""
try:
# Validate user_id
security_guard.validate_user_id(user_id)
# Validate title
if not title or not title.strip():
logger.error("Task title is empty")
return self.create_error_result(
"Task title cannot be empty",
"INVALID_TITLE"
)
# Create task in database (wrapped in transaction)
task = Task(
user_id=user_id,
title=title.strip(),
description=description.strip() if description else "",
completed=False
)
db.add(task)
db.commit()
db.refresh(task)
logger.info(f"Task created: ID={task.id}, User={user_id}, Title='{title[:50]}'")
# Return structured success response
return self.create_success_result(
message=f"Task '{title}' created successfully",
data={
"task_id": task.id,
"title": task.title,
"description": task.description,
"completed": task.completed,
"created_at": task.created_at.isoformat()
}
)
except ValueError as e:
logger.error(f"Validation error in add_task: {str(e)}")
return self.create_error_result(str(e), "VALIDATION_ERROR")
except Exception as e:
logger.error(f"Error creating task: {str(e)}")
db.rollback()
return self.create_error_result(
"Failed to create task. Please try again.",
"DATABASE_ERROR"
)
# Tool instance
add_task_tool = AddTaskTool()
|