refactor logger tail http requests to api_client

Author: jtcorbett

examples/basic/mcp_basic_agent/requirements.txt

@@ -1,5 +1,5 @@
 # Core framework dependency
-mcp-agent @ file://../../../  # Link to the local mcp-agent project root
+mcp-agent==0.1.16
 
 # Additional dependencies specific to this example
 anthropic

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

@@ -17,7 +17,8 @@
 
 from mcp_agent.cli.exceptions import CLIError
 from mcp_agent.cli.auth import load_credentials, UserCredentials
-from mcp_agent.cli.core.constants import DEFAULT_API_BASE_URL
+from mcp_agent.cli.cloud.commands.servers.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
 
 console = Console()
@@ -179,36 +180,22 @@ async def _fetch_logs(
 ) -> None:
     """Fetch logs one-time via HTTP API."""
 
-    api_base = DEFAULT_API_BASE_URL
-    headers = {
-        "Authorization": f"Bearer {credentials.api_key}",
-        "Content-Type": "application/json",
-    }
-    
-    payload = {}
-    
-    if app_id:
-        payload["app_id"] = app_id
-    elif config_id:
-        payload["app_configuration_id"] = config_id
-    else:
-        raise CLIError("Unable to determine app or configuration ID from provided identifier")
-    
-    if since:
-        payload["since"] = since
-    if limit:
-        payload["limit"] = limit
+    client = setup_authenticated_client()
 
+    # Map order_by parameter from CLI to API format
+    order_by_param = None
     if order_by:
         if order_by == "timestamp":
-            payload["orderBy"] = "LOG_ORDER_BY_TIMESTAMP"
+            order_by_param = "LOG_ORDER_BY_TIMESTAMP"
         elif order_by == "severity":
-            payload["orderBy"] = "LOG_ORDER_BY_LEVEL"
+            order_by_param = "LOG_ORDER_BY_LEVEL"
 
+    # Map order parameter from CLI to API format  
+    order_param = None
     if asc:
-        payload["order"] = "LOG_ORDER_ASC"
+        order_param = "LOG_ORDER_ASC"
     elif desc:
-        payload["order"] = "LOG_ORDER_DESC"
+        order_param = "LOG_ORDER_DESC"
 
     with Progress(
         SpinnerColumn(),
@@ -219,23 +206,26 @@ async def _fetch_logs(
         progress.add_task("Fetching logs...", total=None)
 
         try:
-            async with httpx.AsyncClient(timeout=30.0) as client:
-                response = await client.post(
-                    f"{api_base}/mcp_app/get_app_logs",
-                    json=payload,
-                    headers=headers,
-                )
-                
-                if response.status_code == 401:
-                    raise CLIError("Authentication failed. Try running 'mcp-agent login'")
-                elif response.status_code == 404:
-                    raise CLIError("App or configuration not found")
-                elif response.status_code != 200:
-                    raise CLIError(f"API request failed: {response.status_code} {response.text}")
-                
-                data = response.json()
-                log_entries = data.get("logEntries", [])
-                
+            response = await client.get_app_logs(
+                app_id=app_id,
+                app_configuration_id=config_id,
+                since=since,
+                limit=limit,
+                order_by=order_by_param,
+                order=order_param,
+            )
+            # Convert LogEntry models to dictionaries for compatibility with display functions
+            log_entries = [entry.model_dump() for entry in response.log_entries_list]
+            
+        except UnauthenticatedError:
+            raise CLIError("Authentication failed. Try running 'mcp-agent login'")
+        except httpx.HTTPStatusError as e:
+            if e.response.status_code == 404:
+                raise CLIError("App or configuration not found")
+            elif e.response.status_code == 401:
+                raise CLIError("Authentication failed. Try running 'mcp-agent login'")
+            else:
+                raise CLIError(f"API request failed: {e.response.status_code} {e.response.text}")
         except httpx.RequestError as e:
             raise CLIError(f"Failed to connect to API: {e}")
 

src/mcp_agent/cli/mcp_app/api_client.py

@@ -104,6 +104,28 @@ def is_valid_server_url_format(server_url: str) -> bool:
     return parsed.scheme in {"http", "https"} and bool(parsed.netloc)
 
 
+class LogEntry(BaseModel):
+    """Represents a single log entry."""
+    timestamp: Optional[str] = None
+    level: Optional[str] = None
+    message: Optional[str] = None
+    # Allow additional fields that might be present
+    
+    class Config:
+        extra = "allow"
+
+
+class GetAppLogsResponse(BaseModel):
+    """Response from get_app_logs API endpoint."""
+    logEntries: Optional[List[LogEntry]] = []
+    log_entries: Optional[List[LogEntry]] = []
+    
+    @property
+    def log_entries_list(self) -> List[LogEntry]:
+        """Get log entries regardless of field name format."""
+        return self.logEntries or self.log_entries or []
+
+
 class MCPAppClient(APIClient):
     """Client for interacting with the MCP App API service over HTTP."""
 
@@ -586,3 +608,62 @@ async def can_delete_app_configuration(self, app_config_id: str) -> bool:
             resource_name=f"MCP_APP_CONFIG:{app_config_id}",
             action="MANAGE:MCP_APP_CONFIG",
         )
+
+    async def get_app_logs(
+        self,
+        app_id: Optional[str] = None,
+        app_configuration_id: Optional[str] = None,
+        since: Optional[str] = None,
+        limit: Optional[int] = None,
+        order_by: Optional[str] = None,
+        order: Optional[str] = None,
+    ) -> GetAppLogsResponse:
+        """Get logs for an MCP App or App Configuration via the API.
+
+        Args:
+            app_id: The UUID of the app to get logs for (mutually exclusive with app_configuration_id)
+            app_configuration_id: The UUID of the app configuration to get logs for (mutually exclusive with app_id)
+            since: Time filter for logs (e.g., "1h", "24h", "7d")
+            limit: Maximum number of log entries to return
+            order_by: Field to order by ("LOG_ORDER_BY_TIMESTAMP" or "LOG_ORDER_BY_LEVEL")
+            order: Log ordering direction ("LOG_ORDER_ASC" or "LOG_ORDER_DESC")
+
+        Returns:
+            GetAppLogsResponse: The retrieved log entries
+
+        Raises:
+            ValueError: If neither or both app_id and app_configuration_id are provided, or if IDs are invalid
+            httpx.HTTPStatusError: If the API returns an error (e.g., 404, 403)
+            httpx.HTTPError: If the request fails
+        """
+        # Validate inputs
+        if not app_id and not app_configuration_id:
+            raise ValueError("Either app_id or app_configuration_id must be provided")
+        if app_id and app_configuration_id:
+            raise ValueError("Only one of app_id or app_configuration_id can be provided")
+        
+        if app_id and not is_valid_app_id_format(app_id):
+            raise ValueError(f"Invalid app ID format: {app_id}")
+        if app_configuration_id and not is_valid_app_config_id_format(app_configuration_id):
+            raise ValueError(f"Invalid app configuration ID format: {app_configuration_id}")
+
+        # Prepare request payload
+        payload = {}
+        if app_id:
+            payload["app_id"] = app_id
+        if app_configuration_id:
+            payload["app_configuration_id"] = app_configuration_id
+        if since:
+            payload["since"] = since
+        if limit:
+            payload["limit"] = limit
+        if order_by:
+            payload["order_by"] = order_by
+        if order:
+            payload["order"] = order
+
+        response = await self.post("/mcp_app/get_app_logs", payload)
+
+        # Parse the response
+        data = response.json()
+        return GetAppLogsResponse(**data)