Allow logger to also take URLs (#451)

### TL;DR

Unified server resolution logic across CLI commands by implementing a centralized `get_app_or_config` method.

### What changed?

- Removed the redundant `resolve_server` and `parse_app_identifier` utility functions
- Replaced all instances of these functions with direct calls to `client.get_app_or_config`
- Updated the log tailing functionality to work with server URLs directly
- Simplified server URL resolution in workflow commands
- Enhanced the `tail_logs` command to accept server URLs as input
- Updated help text and examples to reflect the new capabilities

### How to test?

1. Test the logger tail command with different identifier types:
   ```
   mcp-agent cloud logger tail app_abc123 --follow
   mcp-agent cloud logger tail apcnf_xyz789 --follow
   mcp-agent cloud logger tail https://abc123.mcpcloud.ai --follow
   ```

2. Test server commands with different identifier types:
   ```
   mcp-agent cloud servers describe app_abc123
   mcp-agent cloud servers describe https://abc123.mcpcloud.ai
   ```

3. Test workflow commands with server URLs:
   ```
   mcp-agent cloud workflows list https://abc123.mcpcloud.ai
   ```

<!-- This is an auto-generated comment: release notes by coderabbit.ai -->

## Summary by CodeRabbit

- New Features
  - Logger tail now supports using a direct server URL in addition to identifiers.

- Refactor
  - Unified server resolution across logger, servers, and workflows commands for more consistent behavior.
  - Log fetching and streaming now operate directly on server objects, simplifying CLI flows.

- Documentation
  - Added an example showing how to tail logs using a server URL.

- Chores
  - Removed legacy helper utilities for server parsing/resolution to reduce complexity.

<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Author: jtcorbett

src/mcp_agent/cli/cloud/commands/logger/tail/main.py

@@ -6,7 +6,7 @@
 import signal
 import sys
 from datetime import datetime, timezone
-from typing import Optional, Dict, Any, List
+from typing import Optional, Dict, Any, List, Union
 from urllib.parse import urlparse
 
 import httpx
@@ -19,8 +19,9 @@
 from mcp_agent.cli.auth import load_credentials, UserCredentials
 from mcp_agent.cli.cloud.commands.utils import setup_authenticated_client
 from mcp_agent.cli.core.api_client import UnauthenticatedError
-from mcp_agent.cli.core.utils import parse_app_identifier, resolve_server_url
+from mcp_agent.cli.core.utils import run_async
 from mcp_agent.cli.utils.ux import print_error
+from mcp_agent.cli.mcp_app.api_client import MCPApp, MCPAppConfiguration
 
 console = Console()
 
@@ -29,7 +30,7 @@
 
 def tail_logs(
     app_identifier: str = typer.Argument(
-        help="App ID or app configuration ID to retrieve logs for"
+        help="App ID, app configuration ID, or server URL to retrieve logs for"
     ),
     since: Optional[str] = typer.Option(
         None,
@@ -91,6 +92,9 @@ def tail_logs(
 
         # Follow logs and filter for specific patterns
         mcp-agent cloud logger tail app_abc123 --follow --grep "authentication.*failed"
+
+        # Use server URL instead of app ID
+        mcp-agent cloud logger tail https://abc123.mcpcloud.ai --follow
     """
 
     credentials = load_credentials()
@@ -130,14 +134,14 @@ def tail_logs(
         print_error("--format must be 'text', 'json', or 'yaml'")
         raise typer.Exit(6)
 
-    app_id, config_id = parse_app_identifier(app_identifier)
-
+    client = setup_authenticated_client()
+    server = run_async(client.get_app_or_config(app_identifier))
+    
     try:
         if follow:
             asyncio.run(
                 _stream_logs(
-                    app_id=app_id,
-                    config_id=config_id,
+                    server=server,
                     credentials=credentials,
                     grep_pattern=grep,
                     app_identifier=app_identifier,
@@ -147,16 +151,15 @@ def tail_logs(
         else:
             asyncio.run(
                 _fetch_logs(
-                    app_id=app_id,
-                    config_id=config_id,
-                    credentials=credentials,
+                    server=server,
                     since=since,
                     grep_pattern=grep,
                     limit=limit,
                     order_by=order_by,
                     asc=asc,
                     desc=desc,
                     format=format,
+                    app_identifier=app_identifier,
                 )
             )
 
@@ -168,19 +171,26 @@ def tail_logs(
 
 
 async def _fetch_logs(
-    app_id: Optional[str],
-    config_id: Optional[str],
-    credentials: UserCredentials,
+    server: Union[MCPApp, MCPAppConfiguration],
     since: Optional[str],
     grep_pattern: Optional[str],
     limit: int,
     order_by: Optional[str],
     asc: bool,
     desc: bool,
     format: str,
+    app_identifier: str,
 ) -> None:
     """Fetch logs one-time via HTTP API."""
 
+    # Extract app_id and config_id from the server object
+    if hasattr(server, 'appId'):  # MCPApp
+        app_id = server.appId
+        config_id = None
+    else:  # MCPAppConfiguration
+        app_id = None
+        config_id = server.appConfigurationId
+
     client = setup_authenticated_client()
 
     # Map order_by parameter from CLI to API format
@@ -240,20 +250,23 @@ async def _fetch_logs(
         console.print("[yellow]No logs found matching the criteria[/yellow]")
         return
 
-    _display_logs(filtered_logs, title=f"Logs for {app_id or config_id}", format=format)
+    _display_logs(filtered_logs, title=f"Logs for {app_identifier}", format=format)
 
 
 async def _stream_logs(
-    app_id: Optional[str],
-    config_id: Optional[str],
+    server: Union[MCPApp, MCPAppConfiguration],
     credentials: UserCredentials,
     grep_pattern: Optional[str],
     app_identifier: str,
     format: str,
 ) -> None:
     """Stream logs continuously via SSE."""
 
-    server_url = await resolve_server_url(app_id, config_id, credentials)
+    # Get server URL directly from the server object
+    if not server.appServerInfo or not server.appServerInfo.serverUrl:
+        raise CLIError("Server URL not available - server may not be deployed")
+    
+    server_url = server.appServerInfo.serverUrl
 
     parsed = urlparse(server_url)
     stream_url = f"{parsed.scheme}://{parsed.netloc}/logs"

src/mcp_agent/cli/cloud/commands/servers/delete/main.py

@@ -6,7 +6,6 @@
 from mcp_agent.cli.mcp_app.api_client import MCPApp
 from ...utils import (
     setup_authenticated_client,
-    resolve_server,
     handle_server_api_errors,
     get_server_name,
     get_server_id,
@@ -25,7 +24,7 @@ def delete_server(
 ) -> None:
     """Delete a specific MCP Server."""
     client = setup_authenticated_client()
-    server = resolve_server(client, id_or_url)
+    server = run_async(client.get_app_or_config(id_or_url))
 
     # Determine server type and delete function
     if isinstance(server, MCPApp):

src/mcp_agent/cli/cloud/commands/servers/describe/main.py

@@ -10,10 +10,10 @@
 from ...utils import (
     setup_authenticated_client,
     validate_output_format,
-    resolve_server,
     handle_server_api_errors,
     clean_server_status,
 )
+from mcp_agent.cli.core.utils import run_async
 from mcp_agent.cli.utils.ux import console
 
 
@@ -29,7 +29,7 @@ def describe_server(
     """Describe a specific MCP Server."""
     validate_output_format(format)
     client = setup_authenticated_client()
-    server = resolve_server(client, id_or_url)
+    server = run_async(client.get_app_or_config(id_or_url))
     print_server_description(server, format)
 
 

src/mcp_agent/cli/cloud/commands/utils.py

@@ -6,7 +6,6 @@
 from mcp_agent.cli.auth import load_api_key_credentials
 from mcp_agent.cli.core.api_client import UnauthenticatedError
 from mcp_agent.cli.core.constants import DEFAULT_API_BASE_URL
-from mcp_agent.cli.core.utils import parse_app_identifier, run_async
 from mcp_agent.cli.exceptions import CLIError
 from mcp_agent.cli.mcp_app.api_client import (
     MCPApp,
@@ -48,35 +47,6 @@ def validate_output_format(format: str) -> None:
         )
 
 
-def resolve_server(
-    client: MCPAppClient, id_or_url: str
-) -> Union[MCPApp, MCPAppConfiguration]:
-    """Resolve server from ID.
-
-    Args:
-        client: Authenticated MCP App client
-        id_or_url: Server identifier (app ID or app config ID)
-
-    Returns:
-        Server object (MCPApp or MCPAppConfiguration)
-
-    Raises:
-        CLIError: If server resolution fails
-    """
-    try:
-        app_id, config_id = parse_app_identifier(id_or_url)
-
-        if config_id:
-            return run_async(client.get_app_configuration(app_config_id=config_id))
-        else:
-            return run_async(client.get_app(app_id=app_id))
-
-    except ValueError as e:
-        raise CLIError(str(e)) from e
-    except Exception as e:
-        raise CLIError(f"Failed to resolve server '{id_or_url}': {str(e)}") from e
-
-
 def handle_server_api_errors(func):
     """Decorator to handle common API errors for server commands.
 

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

@@ -12,7 +12,6 @@
 from mcp_agent.mcp.gen_client import gen_client
 from ...utils import (
     setup_authenticated_client,
-    resolve_server,
     handle_server_api_errors,
 )
 
@@ -25,7 +24,7 @@ async def _cancel_workflow_async(
         server_url = server_id_or_url
     else:
         client = setup_authenticated_client()
-        server = resolve_server(client, server_id_or_url)
+        server = run_async(client.get_app_or_config(server_id_or_url))
 
         if hasattr(server, "appServerInfo") and server.appServerInfo:
             server_url = server.appServerInfo.serverUrl

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

@@ -14,7 +14,6 @@
 from mcp_agent.mcp.gen_client import gen_client
 from ...utils import (
     setup_authenticated_client,
-    resolve_server,
     handle_server_api_errors,
 )
 
@@ -27,7 +26,7 @@ async def _describe_workflow_async(
         server_url = server_id_or_url
     else:
         client = setup_authenticated_client()
-        server = resolve_server(client, server_id_or_url)
+        server = run_async(client.get_app_or_config(server_id_or_url))
 
         if hasattr(server, "appServerInfo") and server.appServerInfo:
             server_url = server.appServerInfo.serverUrl

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

@@ -15,7 +15,6 @@
 from mcp_agent.mcp.gen_client import gen_client
 from ...utils import (
     setup_authenticated_client,
-    resolve_server,
     handle_server_api_errors,
     validate_output_format,
 )
@@ -27,7 +26,7 @@ async def _list_workflows_async(server_id_or_url: str, format: str = "text") ->
         server_url = server_id_or_url
     else:
         client = setup_authenticated_client()
-        server = resolve_server(client, server_id_or_url)
+        server = run_async(client.get_app_or_config(server_id_or_url))
 
         if hasattr(server, "appServerInfo") and server.appServerInfo:
             server_url = server.appServerInfo.serverUrl

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

@@ -13,7 +13,6 @@
 from mcp_agent.mcp.gen_client import gen_client
 from ...utils import (
     setup_authenticated_client,
-    resolve_server,
     handle_server_api_errors,
 )
 
@@ -29,7 +28,7 @@ async def _signal_workflow_async(
         server_url = server_id_or_url
     else:
         client = setup_authenticated_client()
-        server = resolve_server(client, server_id_or_url)
+        server = run_async(client.get_app_or_config(server_id_or_url))
 
         if hasattr(server, "appServerInfo") and server.appServerInfo:
             server_url = server.appServerInfo.serverUrl

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

@@ -14,7 +14,6 @@
 from ...utils import (
     setup_authenticated_client,
     validate_output_format,
-    resolve_server,
 )
 from mcp_agent.cli.utils.ux import console, print_info
 
@@ -27,7 +26,7 @@ async def _list_workflow_runs_async(
         server_url = server_id_or_url
     else:
         client = setup_authenticated_client()
-        server = resolve_server(client, server_id_or_url)
+        server = run_async(client.get_app_or_config(server_id_or_url))
 
         if hasattr(server, "appServerInfo") and server.appServerInfo:
             server_url = server.appServerInfo.serverUrl