"""Check the status of background tool tasks or list all tasks."""
from __future__ import annotations
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from tool_context import ToolContext
TOOL_ALLOW_REPEAT = True
TOOL_NAME = "check_task"
TOOL_DESCRIPTION = (
"Check the result of a background tool task by its task_id, "
"or list all tracked tasks. Tools that take longer than 10 "
"seconds return a task_id instead of their result -- use this "
"tool to retrieve the actual result once it completes."
)
TOOL_PARAMETERS = {
"type": "object",
"properties": {
"task_id": {
"type": "string",
"description": (
"The task_id returned by a backgrounded tool call. "
"Required when action is 'status' (the default)."
),
},
"action": {
"type": "string",
"enum": ["status", "list"],
"description": (
"'status' (default) to check a specific task, "
"'list' to show all tracked tasks."
),
},
},
}
[docs]
async def run(
task_id: str | None = None,
action: str = "status",
ctx: ToolContext | None = None,
) -> str:
"""Execute this tool and return the result.
Args:
task_id (str | None): Background task identifier.
action (str): The action value.
ctx (ToolContext | None): Tool execution context providing access to bot internals.
Returns:
str: Result string.
"""
if ctx is None or ctx.task_manager is None:
return "Error: Task manager not available."
if action == "list":
uid = getattr(ctx, "user_id", "") or ""
return await ctx.task_manager.list_tasks(user_id=uid)
if not task_id:
return "Error: task_id is required for status checks."
uid = getattr(ctx, "user_id", "") or ""
return await ctx.task_manager.get_result(task_id, user_id=uid)