Class DeepCodeListTasksTool

nanobot/nanobot/agent/tools/deepcode.py:256–340

List active and recent DeepCode tasks.

class DeepCodeListTasksTool(Tool):
    """List active and recent DeepCode tasks."""

    def __init__(self, api_url: str | None = None):
        self._api_url = api_url or _get_deepcode_url()

    @property
    def name(self) -> str:
        return "deepcode_list_tasks"

    @property
    def description(self) -> str:
        return (
            "List all active and recent DeepCode code generation tasks. "
            "Shows task IDs, status, progress, and results summary."
        )

    @property
    def parameters(self) -> dict[str, Any]:
        return {
            "type": "object",
            "properties": {
                "limit": {
                    "type": "integer",
                    "description": "Maximum number of recent tasks to show. Default: 10",
                    "minimum": 1,
                    "maximum": 50,
                },
            },
        }

    async def execute(self, limit: int = 10, **kwargs: Any) -> str:
        try:
            async with httpx.AsyncClient(timeout=15.0) as client:
                # Fetch active tasks
                active_resp = await client.get(f"{self._api_url}/api/v1/workflows/active")
                active_resp.raise_for_status()
                active_data = active_resp.json()

                # Fetch recent tasks
                recent_resp = await client.get(
                    f"{self._api_url}/api/v1/workflows/recent",
                    params={"limit": limit},
                )
                recent_resp.raise_for_status()
                recent_data = recent_resp.json()

                lines = []

                # Active tasks
                active_tasks = active_data.get("tasks", [])
                if active_tasks:
                    lines.append(f"=== Active Tasks ({len(active_tasks)}) ===")
                    for task in active_tasks:
                        lines.append(
                            f"  [{task.get('status', '?')}] {task.get('task_id', '?')} "
                            f"- {task.get('progress', 0)}% - {task.get('message', '')}"
                        )