AAP MCP Server

#!/usr/bin/env python3 """ AAP Controller MCP Tools Package This package contains modular tools for interacting with the Ansible Automation Platform Controller API through MCP. """ from functools import wraps from typing import Any, Dict, List, Optional, Callable from connectors import handle_api_error, validate_required_params def validate_action_params(action_requirements: Dict[str, List[str]]): """ Decorator to validate required parameters for specific actions Args: action_requirements: Dict mapping action names to lists of required parameters Example: @validate_action_params({ "create": ["template_data"], "update": ["template_id", "template_data"], "delete": ["template_id"] }) def my_tool_function(action, template_id=None, template_data=None, **kwargs): # Parameters already validated based on action pass """ def decorator(func: Callable) -> Callable: @wraps(func) def wrapper(*args, **kwargs): # Extract action from kwargs or first positional arg action = kwargs.get('action') or (args[0] if args else None) if not action: return { "success": False, "error": "Action parameter is required", "error_type": "ValidationError" } # Check if action has specific requirements if action in action_requirements: required_params = action_requirements[action] validation_error = validate_required_params(kwargs, required_params) if validation_error: return validation_error # Call the original function try: return func(*args, **kwargs) except Exception as e: return handle_api_error(e, f"{func.__name__}({action})") return wrapper return decorator def create_action_handler(action_map: Dict[str, Callable]): """ Create an optimized action handler that maps actions to functions Args: action_map: Dict mapping action names to handler functions Returns: Function that handles action routing """ def handle_action(action: str, **kwargs) -> Dict[str, Any]: if action not in action_map: return { "success": False, "error": f"Unknown action: {action}", "error_type": "ValidationError", "available_actions": list(action_map.keys()) } handler = action_map[action] try: return handler(**kwargs) except Exception as e: return handle_api_error(e, f"action_{action}") return handle_action def batch_api_calls(api_calls: List[Callable], context: str = "batch_operation") -> Dict[str, Any]: """ Execute multiple API calls in sequence with error handling Args: api_calls: List of callable functions that make API calls context: Context for error reporting Returns: Results from all API calls with error handling """ results = [] errors = [] for i, call in enumerate(api_calls): try: result = call() results.append({"index": i, "success": True, "result": result}) except Exception as e: error_info = handle_api_error(e, f"{context}_call_{i}") results.append({"index": i, "success": False, "error": error_info}) errors.append(error_info) return { "success": len(errors) == 0, "total_calls": len(api_calls), "successful_calls": len(api_calls) - len(errors), "failed_calls": len(errors), "results": results, "errors": errors } def paginate_results(api_func: Callable, params: Optional[Dict] = None, page_size: int = 100) -> Dict[str, Any]: """ Helper to paginate through API results efficiently Args: api_func: Function that makes the API call params: Parameters for the API call page_size: Number of items per page Returns: All paginated results """ if params is None: params = {} all_results = [] page = 1 params = params.copy() params["page_size"] = page_size try: while True: params["page"] = page response = api_func(params=params) if not response or "results" not in response: break results = response["results"] all_results.extend(results) # Check if there are more pages if not response.get("next") or len(results) < page_size: break page += 1 return { "success": True, "count": len(all_results), "results": all_results } except Exception as e: return handle_api_error(e, "paginate_results") # Common action requirement patterns for reuse COMMON_ACTION_REQUIREMENTS = { "list_actions": { "list": [], "list_all": [] }, "crud_actions": { "create": ["data"], "update": ["id", "data"], "delete": ["id"], "get": ["id"] }, "template_actions": { "list_job_templates": [], "create_job_template": ["template_data"], "update_job_template": ["template_id", "template_data"], "delete_job_template": ["template_id"], "launch_job_template": ["template_id"] }, "job_actions": { "list_jobs": [], "launch": ["job_template_id"], "cancel": ["job_id"], "relaunch": ["job_id"], "get_status": ["job_id"], "get_logs": ["job_id"] } }