Completed new latest CLI (#2426)

skyvern/cli/tasks.py

@@ -1,13 +1,62 @@
 """Task-related CLI helpers."""
 
+from __future__ import annotations
+
+import json
+import os
+from typing import Optional
+
 import typer
+from dotenv import load_dotenv
+from rich.panel import Panel
+
+from skyvern.client import Skyvern
+from skyvern.config import settings
 
 from .console import console
 
-tasks_app = typer.Typer()
+tasks_app = typer.Typer(help="Manage Skyvern tasks and operations.")
 
 
-@tasks_app.command()
-def placeholder() -> None:
-    """Placeholder command for task management."""
-    console.print("Task operations are not yet implemented.")
+@tasks_app.callback()
+def tasks_callback(
+    ctx: typer.Context,
+    api_key: Optional[str] = typer.Option(
+        None,
+        "--api-key",
+        help="Skyvern API key",
+        envvar="SKYVERN_API_KEY",
+    ),
+) -> None:
+    """Store API key in Typer context."""
+    ctx.obj = {"api_key": api_key}
+
+
+def _get_client(api_key: Optional[str] = None) -> Skyvern:
+    """Instantiate a Skyvern SDK client using environment variables."""
+    load_dotenv()
+    load_dotenv(".env")
+    key = api_key or os.getenv("SKYVERN_API_KEY") or settings.SKYVERN_API_KEY
+    return Skyvern(base_url=settings.SKYVERN_BASE_URL, api_key=key)
+
+
+def _list_workflow_tasks(client: Skyvern, run_id: str) -> list[dict]:
+    """Return tasks for the given workflow run."""
+    resp = client.agent._client_wrapper.httpx_client.request(
+        "api/v1/tasks",
+        method="GET",
+        params={"workflow_run_id": run_id, "page_size": 100, "page": 1},
+    )
+    resp.raise_for_status()
+    return resp.json()
+
+
+@tasks_app.command("list")
+def list_tasks(
+    ctx: typer.Context,
+    workflow_run_id: str = typer.Option(..., "--workflow-run-id", "-r", help="Workflow run ID"),
+) -> None:
+    """List tasks for a workflow run."""
+    client = _get_client(ctx.obj.get("api_key") if ctx.obj else None)
+    tasks = _list_workflow_tasks(client, workflow_run_id)
+    console.print(Panel(json.dumps(tasks, indent=2), border_style="cyan"))