Spaces:
Running
Running
File size: 4,506 Bytes
84c328d |
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 133 134 135 136 137 138 139 140 141 142 143 144 145 |
"""MCP tool for completing/uncompleting tasks in the todo list.
[Task]: T042, T043
[From]: specs/004-ai-chatbot/tasks.md
This tool allows the AI agent to mark tasks as complete or incomplete
through natural language conversations.
"""
from typing import Optional, Any
from uuid import UUID
from datetime import datetime
from sqlalchemy import select
from models.task import Task
from core.database import engine
from sqlmodel import Session
# Tool metadata for MCP registration
tool_metadata = {
"name": "complete_task",
"description": """Mark a task as completed or not completed (toggle completion status).
Use this tool when the user wants to:
- Mark a task as complete, done, finished
- Mark a task as incomplete, pending, not done
- Unmark a task as complete (revert to pending)
- Toggle the completion status of a task
Parameters:
- user_id (required): User ID (UUID) who owns the task
- task_id (required): Task ID (UUID) of the task to mark complete/incomplete
- completed (required): True to mark as complete, False to mark as incomplete/pending
Returns: Updated task details with confirmation.
""",
"inputSchema": {
"type": "object",
"properties": {
"user_id": {
"type": "string",
"description": "User ID (UUID) who owns this task"
},
"task_id": {
"type": "string",
"description": "Task ID (UUID) of the task to mark complete/incomplete"
},
"completed": {
"type": "boolean",
"description": "True to mark complete, False to mark incomplete"
}
},
"required": ["user_id", "task_id", "completed"]
}
}
async def complete_task(
user_id: str,
task_id: str,
completed: bool
) -> dict[str, Any]:
"""Mark a task as completed or incomplete.
[From]: specs/004-ai-chatbot/spec.md - US4
Args:
user_id: User ID (UUID string) who owns the task
task_id: Task ID (UUID string) of the task to update
completed: True to mark complete, False to mark incomplete
Returns:
Dictionary with updated task details
Raises:
ValueError: If validation fails or task not found
"""
# Get database session (synchronous)
with Session(engine) as db:
try:
# Fetch the task
stmt = select(Task).where(
Task.id == UUID(task_id),
Task.user_id == UUID(user_id)
)
task = db.scalars(stmt).first()
if not task:
return {
"success": False,
"error": "Task not found",
"message": f"Could not find task with ID {task_id}"
}
# Update completion status
old_status = "completed" if task.completed else "pending"
task.completed = completed
task.updated_at = datetime.utcnow()
# Save to database
db.add(task)
db.commit()
db.refresh(task)
# Build success message
new_status = "completed" if completed else "pending"
action = "marked as complete" if completed else "marked as pending"
message = f"✅ Task '{task.title}' {action}"
return {
"success": True,
"task": {
"id": str(task.id),
"title": task.title,
"description": task.description,
"due_date": task.due_date.isoformat() if task.due_date else None,
"priority": task.priority,
"completed": task.completed,
"created_at": task.created_at.isoformat(),
"updated_at": task.updated_at.isoformat()
},
"message": message,
"old_status": old_status,
"new_status": new_status
}
except ValueError as e:
db.rollback()
raise ValueError(f"Failed to update task completion status: {str(e)}")
# Register tool with MCP server
def register_tool(mcp_server: Any) -> None:
"""Register this tool with the MCP server.
[From]: backend/mcp_server/server.py
Args:
mcp_server: MCP server instance
"""
mcp_server.tool(
name=tool_metadata["name"],
description=tool_metadata["description"]
)(complete_task)
|