Spaces:
Sleeping
Sleeping
File size: 4,496 Bytes
4b4f221 | 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 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 | """
MCP Tool: complete_task
This tool allows the AI agent to mark a task as completed.
"""
from typing import Dict, Any
from ..server import mcp_server
from sqlmodel import Session, select
from ...models.task import Task
@mcp_server.register_tool("complete_task")
async def complete_task(user_id: str, task_id: str) -> Dict[str, Any]:
"""
Mark a task as completed for the specified user.
Args:
user_id: The ID of the user who owns the task
task_id: The ID of the task to mark as completed
Returns:
Dictionary containing the updated task information
"""
try:
# Import database session here to avoid circular imports
from sqlmodel import Session
from src.database.connection import engine
from src.services.task_service import TaskService
from src.models.task import TaskUpdate
# Validate inputs
if not user_id or not task_id:
return {
"success": False,
"error": "Both user_id and task_id are required"
}
# Create database session
with Session(engine) as db_session:
task_service = TaskService()
# Update the task to mark as completed
task_update = TaskUpdate(completed=True)
updated_task = task_service.update_task(
task_id=task_id,
user_id=user_id,
task_update=task_update,
db=db_session
)
# Convert to dict for response
task_dict = {
"id": updated_task.id,
"user_id": updated_task.user_id,
"title": updated_task.title,
"description": updated_task.description,
"completed": updated_task.completed,
"priority": updated_task.priority.value if hasattr(updated_task.priority, 'value') else updated_task.priority,
"created_at": updated_task.created_at.isoformat() if hasattr(updated_task.created_at, 'isoformat') else str(updated_task.created_at),
"updated_at": updated_task.updated_at.isoformat() if hasattr(updated_task.updated_at, 'isoformat') else str(updated_task.updated_at)
}
return {
"success": True,
"task": task_dict
}
except Exception as e:
return {
"success": False,
"error": f"Failed to complete task: {str(e)}"
}
# For integration with the existing Phase II task system
def complete_task_with_db_session(user_id: str, task_id: str, db_session: Session = None) -> Dict[str, Any]:
"""
Mark a task as completed for the specified user using database session.
Args:
user_id: The ID of the user who owns the task
task_id: The ID of the task to mark as completed
db_session: Database session to use for the operation
Returns:
Dictionary containing the updated task information
"""
try:
# Validate inputs
if not user_id or not task_id:
return {
"success": False,
"error": "Both user_id and task_id are required"
}
# Query the database for the specific task belonging to the user
statement = select(Task).where(Task.user_id == user_id).where(Task.id == task_id)
result = db_session.exec(statement)
task = result.first()
if not task:
return {
"success": False,
"error": "Task not found or does not belong to user"
}
# Update the task to mark as completed
task.completed = True
db_session.add(task)
db_session.commit()
db_session.refresh(task)
# Convert to dictionary for response
task_dict = {
"id": task.id,
"user_id": task.user_id,
"title": task.title,
"description": task.description,
"completed": task.completed,
"priority": task.priority.value if hasattr(task.priority, 'value') else task.priority,
"created_at": getattr(task, 'created_at', None),
"updated_at": getattr(task, 'updated_at', None)
}
return {
"success": True,
"task": task_dict
}
except Exception as e:
return {
"success": False,
"error": f"Failed to complete task: {str(e)}"
} |