Spaces:
Running
Running
File size: 5,223 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 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 |
"""MCP tool for marking all tasks as complete or incomplete.
[Task]: T044, T045
[From]: specs/004-ai-chatbot/tasks.md
This tool allows the AI agent to mark all tasks with a completion status
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_all_tasks",
"description": """Mark all tasks as completed or not completed.
Use this tool when the user wants to:
- Mark all tasks as complete, done, or finished
- Mark all tasks as incomplete or pending
- Complete every task in their list
Parameters:
- user_id (required): User ID (UUID) who owns the tasks
- completed (required): True to mark all complete, False to mark all incomplete
- status_filter (optional): Only affect tasks with this status ('pending' or 'completed')
Returns: Summary with count of tasks updated.
""",
"inputSchema": {
"type": "object",
"properties": {
"user_id": {
"type": "string",
"description": "User ID (UUID) who owns these tasks"
},
"completed": {
"type": "boolean",
"description": "True to mark all tasks complete, False to mark all incomplete"
},
"status_filter": {
"type": "string",
"enum": ["pending", "completed"],
"description": "Optional: Only affect tasks with this status. If not provided, affects all tasks."
}
},
"required": ["user_id", "completed"]
}
}
async def complete_all_tasks(
user_id: str,
completed: bool,
status_filter: Optional[str] = None
) -> dict[str, Any]:
"""Mark all tasks as completed or incomplete.
[From]: specs/004-ai-chatbot/spec.md - US4
Args:
user_id: User ID (UUID string) who owns the tasks
completed: True to mark all complete, False to mark all incomplete
status_filter: Optional filter to only affect tasks with current status
Returns:
Dictionary with count of tasks updated and confirmation message
Raises:
ValueError: If validation fails
"""
# Get database session (synchronous)
with Session(engine) as db:
try:
# Build query based on filter
stmt = select(Task).where(Task.user_id == UUID(user_id))
# Apply status filter if provided
if status_filter == "pending":
stmt = stmt.where(Task.completed == False)
elif status_filter == "completed":
stmt = stmt.where(Task.completed == True)
# Fetch matching tasks
tasks = list(db.scalars(stmt).all())
if not tasks:
return {
"success": False,
"error": "No tasks found",
"message": f"Could not find any tasks{' matching the filter' if status_filter else ''}"
}
# Count tasks before update
task_count = len(tasks)
already_correct = sum(1 for t in tasks if t.completed == completed)
# If all tasks already have the desired status
if already_correct == task_count:
status_word = "completed" if completed else "pending"
return {
"success": True,
"updated_count": 0,
"skipped_count": task_count,
"message": f"All {task_count} task(s) are already {status_word}."
}
# Update completion status for all tasks
updated_count = 0
for task in tasks:
if task.completed != completed:
task.completed = completed
task.updated_at = datetime.utcnow()
db.add(task)
updated_count += 1
# Save to database
db.commit()
# Build success message
action = "completed" if completed else "marked as pending"
if status_filter:
filter_msg = f" {status_filter} tasks"
else:
filter_msg = ""
message = f"✅ {updated_count} task{'' if updated_count == 1 else 's'}{filter_msg} marked as {action}"
return {
"success": True,
"updated_count": updated_count,
"skipped_count": already_correct,
"total_count": task_count,
"message": message
}
except ValueError as e:
db.rollback()
raise ValueError(f"Failed to update tasks: {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_all_tasks)
|