Class DeepCodeStatusTool

nanobot/nanobot/agent/tools/deepcode.py:177–253

Check the status and progress of a DeepCode task.

class DeepCodeStatusTool(Tool):
    """Check the status and progress of a DeepCode task."""

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

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

    @property
    def description(self) -> str:
        return (
            "Check the status and progress of a DeepCode code generation task. "
            "Provide the task_id returned by deepcode_paper2code or deepcode_chat2code. "
            "Returns current status, progress percentage, and result when complete."
        )

    @property
    def parameters(self) -> dict[str, Any]:
        return {
            "type": "object",
            "properties": {
                "task_id": {
                    "type": "string",
                    "description": "The task ID to check status for",
                },
            },
            "required": ["task_id"],
        }

    async def execute(self, task_id: str, **kwargs: Any) -> str:
        try:
            async with httpx.AsyncClient(timeout=15.0) as client:
                response = await client.get(f"{self._api_url}/api/v1/workflows/status/{task_id}")
                response.raise_for_status()
                data = response.json()

                status = data.get("status", "unknown")
                progress = data.get("progress", 0)
                message = data.get("message", "")
                result = data.get("result")
                error = data.get("error")

                lines = [
                    f"Task ID: {task_id}",
                    f"Status: {status}",
                    f"Progress: {progress}%",
                ]

                if message:
                    lines.append(f"Message: {message}")

                if status == "completed" and result:
                    lines.append(f"\nResult:\n{result}")
                elif status == "error" and error:
                    lines.append(f"\nError: {error}")
                elif status == "waiting_for_input":