106 lines
3.4 KiB
Python
106 lines
3.4 KiB
Python
"""TaskUpdateTool — update an existing task's status, dependencies, or fields.
|
|
|
|
Supports status transitions (pending -> in_progress -> completed),
|
|
dependency tracking via addBlockedBy, and field updates.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, ClassVar
|
|
|
|
from ..base import Tool, ToolParameter, ToolResult
|
|
from .task_storage import TaskStorage
|
|
|
|
VALID_STATUSES = {"pending", "in_progress", "completed", "deleted"}
|
|
|
|
|
|
class TaskUpdateTool(Tool):
|
|
"""Update an existing task's status, dependencies, or metadata."""
|
|
|
|
name: ClassVar[str] = "task_update"
|
|
description: ClassVar[str] = (
|
|
"Update a task's status, blocked-by dependencies, or other fields. "
|
|
"Use status 'deleted' to remove a task permanently."
|
|
)
|
|
parameters: ClassVar[list[ToolParameter]] = [
|
|
ToolParameter(
|
|
name="task_id",
|
|
type="string",
|
|
description="The task ID to update (e.g. 'task-1')",
|
|
),
|
|
ToolParameter(
|
|
name="status",
|
|
type="string",
|
|
description="New status for the task",
|
|
required=False,
|
|
enum=["pending", "in_progress", "completed", "deleted"],
|
|
),
|
|
ToolParameter(
|
|
name="addBlockedBy",
|
|
type="array",
|
|
description="Task IDs that block this task",
|
|
required=False,
|
|
items={"type": "string"},
|
|
),
|
|
ToolParameter(
|
|
name="subject",
|
|
type="string",
|
|
description="New subject for the task",
|
|
required=False,
|
|
),
|
|
ToolParameter(
|
|
name="description",
|
|
type="string",
|
|
description="New description for the task",
|
|
required=False,
|
|
),
|
|
ToolParameter(
|
|
name="activeForm",
|
|
type="string",
|
|
description="Present continuous form shown while in progress",
|
|
required=False,
|
|
),
|
|
ToolParameter(
|
|
name="owner",
|
|
type="string",
|
|
description="Agent or user who owns this task",
|
|
required=False,
|
|
),
|
|
ToolParameter(
|
|
name="conversation_id",
|
|
type="string",
|
|
description="Conversation scope for task storage",
|
|
),
|
|
]
|
|
|
|
async def execute(self, **kwargs: Any) -> ToolResult:
|
|
conversation_id: str = kwargs["conversation_id"]
|
|
task_id: str = kwargs["task_id"]
|
|
|
|
storage = TaskStorage(conversation_id)
|
|
|
|
# Handle deletion
|
|
status = kwargs.get("status")
|
|
if status == "deleted":
|
|
deleted = storage.delete(task_id)
|
|
if not deleted:
|
|
return ToolResult.fail(f"Task not found: {task_id}")
|
|
return ToolResult.success({"deleted": task_id})
|
|
|
|
# Build updates dict from provided optional fields
|
|
updates: dict[str, Any] = {}
|
|
for field in ("status", "subject", "description", "activeForm", "owner"):
|
|
if field in kwargs and kwargs[field] is not None:
|
|
updates[field] = kwargs[field]
|
|
|
|
if "addBlockedBy" in kwargs and kwargs["addBlockedBy"] is not None:
|
|
updates["addBlockedBy"] = kwargs["addBlockedBy"]
|
|
|
|
if not updates:
|
|
return ToolResult.fail("No updates provided")
|
|
|
|
task = storage.update(task_id, updates)
|
|
if task is None:
|
|
return ToolResult.fail(f"Task not found: {task_id}")
|
|
|
|
return ToolResult.success(task)
|