AAP MCP Server

#!/usr/bin/env python3 """ Client utilities and error handling """ import logging from typing import Any, Dict, Optional import httpx logger = logging.getLogger(__name__) class AAPError(Exception): """Base exception for AAP API errors""" pass class AAPAuthenticationError(AAPError): """Authentication failed""" pass class AAPNotFoundError(AAPError): """Resource not found""" pass class AAPValidationError(AAPError): """Request validation failed""" pass def handle_api_error(e: Exception, context: str = "") -> Dict[str, Any]: """ Standardized error handling for API operations Args: e: The exception to handle context: Additional context about what operation failed Returns: Standardized error response dictionary """ error_response = { "success": False, "error_type": type(e).__name__, "context": context } if isinstance(e, httpx.HTTPStatusError): status_code = e.response.status_code error_response.update({ "http_status": status_code, "http_reason": e.response.reason_phrase, "url": str(e.request.url) }) # Try to extract detailed error from response try: error_detail = e.response.json() if isinstance(error_detail, dict): error_response["details"] = error_detail else: error_response["details"] = {"message": str(error_detail)} except: error_response["details"] = {"message": e.response.text[:500]} # Categorize common HTTP errors if status_code == 401: error_response["error"] = "Authentication failed - check your credentials" error_response["suggestion"] = "Verify AAP_TOKEN or AAP_USERNAME/AAP_PASSWORD" elif status_code == 403: error_response["error"] = "Permission denied - insufficient privileges" error_response["suggestion"] = "Check user permissions for this operation" elif status_code == 404: error_response["error"] = "Resource not found" error_response["suggestion"] = "Verify the resource ID or endpoint path" elif status_code == 400: error_response["error"] = "Bad request - invalid parameters" error_response["suggestion"] = "Check request parameters and data format" elif status_code >= 500: error_response["error"] = "Server error - AAP Controller issue" error_response["suggestion"] = "Check AAP Controller status and logs" else: error_response["error"] = f"HTTP {status_code}: {e.response.reason_phrase}" elif isinstance(e, httpx.ConnectError): error_response.update({ "error": "Connection failed - cannot reach AAP Controller", "suggestion": "Check AAP_BASE_URL and network connectivity", "details": {"message": str(e)} }) elif isinstance(e, httpx.TimeoutException): error_response.update({ "error": "Request timeout - AAP Controller not responding", "suggestion": "Try again or check AAP Controller performance", "details": {"message": str(e)} }) else: # Generic error handling error_response.update({ "error": f"Operation failed: {str(e)}", "details": {"message": str(e)} }) # Log the error for debugging logger.error(f"API Error in {context}: {error_response}") return error_response def validate_required_params(params: Dict[str, Any], required: list) -> Optional[Dict[str, Any]]: """ Validate that required parameters are provided Args: params: Dictionary of parameters to validate required: List of required parameter names Returns: Error response if validation fails, None if validation passes """ missing = [param for param in required if params.get(param) is None] if missing: return { "success": False, "error": f"Missing required parameters: {', '.join(missing)}", "error_type": "ValidationError", "missing_parameters": missing, "provided_parameters": list(params.keys()) } return None def safe_api_call(func, context: str = "", **kwargs) -> Dict[str, Any]: """ Safely execute an API call with standardized error handling Args: func: The API function to call context: Context description for error reporting **kwargs: Arguments to pass to the function Returns: API response or standardized error response """ try: result = func(**kwargs) return result except Exception as e: return handle_api_error(e, context)