Spaces:
Running
Running
| """Tool status tool - check status of background tools.""" | |
| import time | |
| import logging | |
| from typing import TYPE_CHECKING, Any, Dict | |
| from reachy_mini_receptionist.tools.core_tools import Tool, ToolDependencies | |
| from reachy_mini_receptionist.tools.tool_constants import SystemTool | |
| if TYPE_CHECKING: | |
| from reachy_mini_receptionist.tools.background_tool_manager import BackgroundToolManager | |
| logger = logging.getLogger(__name__) | |
| class TaskStatus(Tool): | |
| """Check status of background tool tasks.""" | |
| name = "task_status" | |
| description = ( | |
| "Check the status of background tool tasks. " | |
| "Use this when the user asks about running tools or wants to know what's happening in the background." | |
| ) | |
| parameters_schema = { | |
| "type": "object", | |
| "properties": { | |
| "tool_id": { | |
| "type": "string", | |
| "description": "Specific tool ID to check (optional, shows all running tools if omitted)", | |
| }, | |
| }, | |
| "required": [], | |
| } | |
| async def __call__(self, deps: ToolDependencies, **kwargs: Any) -> Dict[str, Any]: | |
| """Get status of background tools.""" | |
| tool_id: str | None = kwargs.get("tool_id") | |
| tool_manager: BackgroundToolManager | None = kwargs.get("tool_manager") | |
| if tool_manager is None: | |
| return {"error": "Tool manager is required."} | |
| logger.info(f"Tool call: tool_status tool_id={tool_id}") | |
| if tool_id: | |
| tool = tool_manager.get_tool(tool_id) | |
| if not tool: | |
| return {"error": f"Tool {tool_id} not found."} | |
| result: Dict[str, Any] = { | |
| "tool_id": tool.tool_id, | |
| "name": tool.tool_name, | |
| "status": tool.status.value, | |
| "started_at": tool.started_at, | |
| } | |
| if tool.completed_at: | |
| result["completed_at"] = tool.completed_at | |
| if tool.progress is not None: | |
| result["progress_percent"] = f"{tool.progress.progress:.0%}" | |
| if tool.progress.message: | |
| result["progress_message"] = tool.progress.message | |
| if tool.result: | |
| result["result"] = tool.result | |
| if tool.error: | |
| result["error"] = tool.error | |
| return result | |
| # Get all running tools | |
| running = tool_manager.get_running_tools() | |
| if not running: | |
| return { | |
| "status": "idle", | |
| "message": "No tools running in the background.", | |
| } | |
| tools_info = [] | |
| for tool in [tool for tool in running if tool.tool_name not in [system_tool.value for system_tool in SystemTool]]: | |
| elapsed = time.monotonic() - tool.started_at | |
| tool_info: Dict[str, Any] = { | |
| "tool_id": tool.tool_id, | |
| "name": tool.tool_name, | |
| "status": tool.status.value, | |
| "elapsed_seconds": round(elapsed, 1), | |
| } | |
| # Add progress if tracking | |
| if tool.progress is not None: | |
| tool_info["progress_percent"] = f"{tool.progress.progress:.0%}" | |
| if tool.progress.message: | |
| tool_info["progress_message"] = tool.progress.message | |
| tools_info.append(tool_info) | |
| return { | |
| "status": "running", | |
| "count": len(tools_info), | |
| "message": f"{len(tools_info)} tool(s) running in the background.", | |
| "tools": tools_info, | |
| } | |