Refactor workflow runs list to use MCP tool calls

Author: jtcorbett

src/mcp_agent/cli/cloud/commands/workflows/runs/main.py

@@ -7,18 +7,90 @@
 import yaml
 from rich.table import Table
 
+from mcp_agent.app import MCPApp
 from mcp_agent.cli.core.utils import run_async
-from mcp_agent.cli.mcp_app.api_client import WorkflowExecutionStatus
+from mcp_agent.cli.exceptions import CLIError
+from mcp_agent.config import MCPServerSettings, Settings, LoggerSettings
+from mcp_agent.mcp.gen_client import gen_client
 from ...utils import (
     setup_authenticated_client,
     validate_output_format,
-    handle_server_api_errors,
     resolve_server,
 )
 from mcp_agent.cli.utils.ux import console, print_info
 
 
-@handle_server_api_errors
+async def _list_workflow_runs_async(
+    server_id_or_url: str, limit: Optional[int], status: Optional[str], format: str
+) -> None:
+    """List workflow runs using MCP tool calls to a deployed server."""
+    if server_id_or_url.startswith(("http://", "https://")):
+        server_url = server_id_or_url
+    else:
+        client = setup_authenticated_client()
+        server = resolve_server(client, server_id_or_url)
+
+        if hasattr(server, "appServerInfo") and server.appServerInfo:
+            server_url = server.appServerInfo.serverUrl
+        else:
+            raise CLIError(
+                f"Server '{server_id_or_url}' is not deployed or has no server URL"
+            )
+
+        if not server_url:
+            raise CLIError(f"No server URL found for server '{server_id_or_url}'")
+
+    quiet_settings = Settings(logger=LoggerSettings(level="error"))
+    app = MCPApp(name="workflows_cli", settings=quiet_settings)
+
+    try:
+        async with app.run() as workflow_app:
+            context = workflow_app.context
+
+            sse_url = (
+                f"{server_url}/sse" if not server_url.endswith("/sse") else server_url
+            )
+            context.server_registry.registry["workflow_server"] = MCPServerSettings(
+                name="workflow_server",
+                description=f"Deployed MCP server {server_url}",
+                url=sse_url,
+                transport="sse",
+            )
+
+            async with gen_client(
+                "workflow_server", server_registry=context.server_registry
+            ) as client:
+                result = await client.call_tool("workflows-runs-list", {})
+
+                workflows_data = result.content[0].text if result.content else []
+                if isinstance(workflows_data, str):
+                    workflows_data = json.loads(workflows_data)
+
+                if not workflows_data:
+                    workflows_data = []
+
+                # Apply filtering
+                workflows = workflows_data
+                if status:
+                    status_filter = _get_status_filter(status)
+                    workflows = [w for w in workflows if _matches_status(w, status_filter)]
+
+                if limit:
+                    workflows = workflows[:limit]
+
+                if format == "json":
+                    _print_workflows_json(workflows)
+                elif format == "yaml":
+                    _print_workflows_yaml(workflows)
+                else:
+                    _print_workflows_text(workflows, status, server_id_or_url)
+
+    except Exception as e:
+        raise CLIError(
+            f"Error listing workflow runs for server {server_id_or_url}: {str(e)}"
+        ) from e
+
+
 def list_workflow_runs(
     server_id_or_url: str = typer.Argument(
         ..., help="Server ID, app config ID, or server URL to list workflow runs for"
@@ -46,62 +118,38 @@ def list_workflow_runs(
         mcp-agent cloud workflows runs apcnf_xyz789 --limit 10 --format json
     """
     validate_output_format(format)
-    client = setup_authenticated_client()
-
-    if server_id_or_url.startswith(("http://", "https://")):
-        resolved_server = resolve_server(client, server_id_or_url)
+    run_async(_list_workflow_runs_async(server_id_or_url, limit, status, format))
 
-        if hasattr(resolved_server, "appId"):
-            app_id_or_config_id = resolved_server.appId
-        elif hasattr(resolved_server, "appConfigurationId"):
-            app_id_or_config_id = resolved_server.appConfigurationId
-        else:
-            raise ValueError(
-                f"Could not extract app ID or config ID from server: {server_id_or_url}"
-            )
-    else:
-        app_id_or_config_id = server_id_or_url
-
-    max_results = limit or 100
-
-    status_filter = None
-    if status:
-        status_map = {
-            "running": WorkflowExecutionStatus.WORKFLOW_EXECUTION_STATUS_RUNNING,
-            "failed": WorkflowExecutionStatus.WORKFLOW_EXECUTION_STATUS_FAILED,
-            "timed_out": WorkflowExecutionStatus.WORKFLOW_EXECUTION_STATUS_TIMED_OUT,
-            "timeout": WorkflowExecutionStatus.WORKFLOW_EXECUTION_STATUS_TIMED_OUT,  # alias
-            "canceled": WorkflowExecutionStatus.WORKFLOW_EXECUTION_STATUS_CANCELED,
-            "cancelled": WorkflowExecutionStatus.WORKFLOW_EXECUTION_STATUS_CANCELED,  # alias
-            "terminated": WorkflowExecutionStatus.WORKFLOW_EXECUTION_STATUS_TERMINATED,
-            "completed": WorkflowExecutionStatus.WORKFLOW_EXECUTION_STATUS_COMPLETED,
-            "continued": WorkflowExecutionStatus.WORKFLOW_EXECUTION_STATUS_CONTINUED_AS_NEW,
-            "continued_as_new": WorkflowExecutionStatus.WORKFLOW_EXECUTION_STATUS_CONTINUED_AS_NEW,
-        }
-        status_filter = status_map.get(status.lower())
-        if not status_filter:
-            valid_statuses = "running|failed|timed_out|timeout|canceled|cancelled|terminated|completed|continued|continued_as_new"
-            raise typer.BadParameter(
-                f"Invalid status '{status}'. Valid options: {valid_statuses}"
-            )
 
-    async def list_workflows_async():
-        return await client.list_workflows(
-            app_id_or_config_id=app_id_or_config_id, max_results=max_results
+def _get_status_filter(status: str) -> str:
+    """Convert status string to normalized status."""
+    status_map = {
+        "running": "running",
+        "failed": "failed",
+        "timed_out": "timed_out",
+        "timeout": "timed_out",  # alias
+        "canceled": "canceled",
+        "cancelled": "canceled",  # alias
+        "terminated": "terminated",
+        "completed": "completed",
+        "continued": "continued",
+        "continued_as_new": "continued",
+    }
+    normalized_status = status_map.get(status.lower())
+    if not normalized_status:
+        valid_statuses = "running|failed|timed_out|timeout|canceled|cancelled|terminated|completed|continued|continued_as_new"
+        raise typer.BadParameter(
+            f"Invalid status '{status}'. Valid options: {valid_statuses}"
         )
+    return normalized_status
 
-    response = run_async(list_workflows_async())
-    workflows = response.workflows or []
-
-    if status_filter:
-        workflows = [w for w in workflows if w.execution_status == status_filter]
 
-    if format == "json":
-        _print_workflows_json(workflows)
-    elif format == "yaml":
-        _print_workflows_yaml(workflows)
-    else:
-        _print_workflows_text(workflows, status, server_id_or_url)
+def _matches_status(workflow: dict, status_filter: str) -> bool:
+    """Check if workflow matches the status filter."""
+    workflow_status = workflow.get("execution_status", "")
+    if isinstance(workflow_status, str):
+        return status_filter.lower() in workflow_status.lower()
+    return False
 
 
 def _print_workflows_text(workflows, status_filter, server_id_or_url):
@@ -127,21 +175,48 @@ def _print_workflows_text(workflows, status_filter, server_id_or_url):
     table.add_column("Principal", style="dim", width=15)
 
     for workflow in workflows:
-        status_display = _get_status_display(workflow.execution_status)
-        created_display = (
-            workflow.created_at.strftime("%Y-%m-%d %H:%M:%S")
-            if workflow.created_at
-            else "N/A"
-        )
-        run_id_display = _truncate_string(workflow.run_id or "N/A", 15)
+        # Handle both dict and object formats
+        if isinstance(workflow, dict):
+            workflow_id = workflow.get("workflow_id", "N/A")
+            name = workflow.get("name", "N/A")
+            execution_status = workflow.get("execution_status", "N/A")
+            run_id = workflow.get("run_id", "N/A")
+            created_at = workflow.get("created_at", "N/A")
+            principal_id = workflow.get("principal_id", "N/A")
+        else:
+            workflow_id = getattr(workflow, "workflow_id", "N/A")
+            name = getattr(workflow, "name", "N/A")
+            execution_status = getattr(workflow, "execution_status", "N/A")
+            run_id = getattr(workflow, "run_id", "N/A")
+            created_at = getattr(workflow, "created_at", "N/A")
+            principal_id = getattr(workflow, "principal_id", "N/A")
+        
+        status_display = _get_status_display(execution_status)
+        
+        # Handle created_at formatting
+        if created_at and created_at != "N/A":
+            if hasattr(created_at, "strftime"):
+                created_display = created_at.strftime("%Y-%m-%d %H:%M:%S")
+            else:
+                # Try to parse ISO string
+                try:
+                    from datetime import datetime
+                    dt = datetime.fromisoformat(str(created_at).replace("Z", "+00:00"))
+                    created_display = dt.strftime("%Y-%m-%d %H:%M:%S")
+                except (ValueError, TypeError):
+                    created_display = str(created_at)
+        else:
+            created_display = "N/A"
+        
+        run_id_display = _truncate_string(str(run_id) if run_id else "N/A", 15)
 
         table.add_row(
-            _truncate_string(workflow.workflow_id, 20),
-            _truncate_string(workflow.name, 20),
+            _truncate_string(str(workflow_id) if workflow_id else "N/A", 20),
+            _truncate_string(str(name) if name else "N/A", 20),
             status_display,
             run_id_display,
             created_display,
-            _truncate_string(workflow.principal_id, 15),
+            _truncate_string(str(principal_id) if principal_id else "N/A", 15),
         )
 
     console.print(table)
@@ -163,16 +238,19 @@ def _print_workflows_yaml(workflows):
 
 
 def _workflow_to_dict(workflow):
-    """Convert WorkflowInfo to dictionary."""
+    """Convert workflow dict to standardized dictionary format."""
+    # If it's already a dict, just return it
+    if isinstance(workflow, dict):
+        return workflow
+    
+    # Otherwise convert from object attributes
     return {
-        "workflow_id": workflow.workflow_id,
-        "run_id": workflow.run_id,
-        "name": workflow.name,
-        "created_at": workflow.created_at.isoformat() if workflow.created_at else None,
-        "principal_id": workflow.principal_id,
-        "execution_status": workflow.execution_status.value
-        if workflow.execution_status
-        else None,
+        "workflow_id": getattr(workflow, "workflow_id", None),
+        "run_id": getattr(workflow, "run_id", None),
+        "name": getattr(workflow, "name", None),
+        "created_at": getattr(workflow, "created_at", None).isoformat() if getattr(workflow, "created_at", None) else None,
+        "principal_id": getattr(workflow, "principal_id", None),
+        "execution_status": getattr(workflow, "execution_status", None).value if getattr(workflow, "execution_status", None) else None,
     }
 
 
@@ -184,18 +262,26 @@ def _truncate_string(text: str, max_length: int) -> str:
 
 
 def _get_status_display(status):
-    """Convert WorkflowExecutionStatus to display string with emoji."""
+    """Convert status to display string with emoji."""
     if not status:
         return "❓ Unknown"
-
-    status_map = {
-        WorkflowExecutionStatus.WORKFLOW_EXECUTION_STATUS_RUNNING: "[green]🟢 Running[/green]",
-        WorkflowExecutionStatus.WORKFLOW_EXECUTION_STATUS_COMPLETED: "[blue]✅ Completed[/blue]",
-        WorkflowExecutionStatus.WORKFLOW_EXECUTION_STATUS_FAILED: "[red]❌ Failed[/red]",
-        WorkflowExecutionStatus.WORKFLOW_EXECUTION_STATUS_CANCELED: "[yellow]🟡 Canceled[/yellow]",
-        WorkflowExecutionStatus.WORKFLOW_EXECUTION_STATUS_TERMINATED: "[red]🔴 Terminated[/red]",
-        WorkflowExecutionStatus.WORKFLOW_EXECUTION_STATUS_TIMED_OUT: "[orange]⏰ Timed Out[/orange]",
-        WorkflowExecutionStatus.WORKFLOW_EXECUTION_STATUS_CONTINUED_AS_NEW: "[purple]🔄 Continued[/purple]",
-    }
-
-    return status_map.get(status, "❓ Unknown")
+    
+    # Convert to string for comparison
+    status_str = str(status).lower()
+    
+    if "running" in status_str:
+        return "[green]🟢 Running[/green]"
+    elif "completed" in status_str:
+        return "[blue]✅ Completed[/blue]"
+    elif "failed" in status_str or "error" in status_str:
+        return "[red]❌ Failed[/red]"
+    elif "cancel" in status_str:
+        return "[yellow]🟡 Canceled[/yellow]"
+    elif "terminat" in status_str:
+        return "[red]🔴 Terminated[/red]"
+    elif "timeout" in status_str or "timed_out" in status_str:
+        return "[orange]⏰ Timed Out[/orange]"
+    elif "continued" in status_str:
+        return "[purple]🔄 Continued[/purple]"
+    else:
+        return f"❓ {status}"