AAP MCP Server

#!/usr/bin/env python3 """ AAP Controller Generic API Access Tool """ from typing import Any, Dict, Optional, Union from fastmcp import FastMCP from pydantic import Field from connectors.aap_connector import get_aap_connector def register_generic_api_tools(mcp: FastMCP): """Register generic API access tools with the MCP server""" @mcp.tool() def generic_aap_api( method: str = Field(description="HTTP method: GET, POST, PATCH, DELETE"), endpoint: str = Field(description="API endpoint path (e.g., 'analytics/job_explorer/', 'jobs/', 'inventories/1/')"), data: Optional[Dict[str, Any]] = Field(None, description="Request body data for POST/PATCH requests"), params: Optional[Dict[str, Any]] = Field(None, description="Query parameters for GET requests"), get_stdout: bool = Field(False, description="Use stdout format for job/adhoc output endpoints") ) -> Dict[str, Any]: """ Generic AAP Controller API access tool. Provides direct access to any AAP Controller API endpoint with proper authentication. Examples: - GET /jobs/: method="GET", endpoint="jobs/" - GET /inventories/2/hosts/: method="GET", endpoint="inventories/2/hosts/" - POST /job_templates/5/launch/: method="POST", endpoint="job_templates/5/launch/", data={"extra_vars": {"var1": "value1"}} - GET /analytics/job_explorer/: method="GET", endpoint="analytics/job_explorer/", params={"period": "month"} - GET /jobs/123/stdout/: method="GET", endpoint="jobs/123/stdout/", get_stdout=true """ try: client = get_aap_connector() method = method.upper() # Clean the endpoint - remove leading/trailing slashes and add trailing slash endpoint = endpoint.strip('/') if not endpoint.endswith('/'): endpoint += '/' if method == "GET": if get_stdout: # Use the special stdout method for job output return client.get_stdout(endpoint, params) else: return client.get(endpoint, params) elif method == "POST": return client.post(endpoint, data) elif method == "PATCH": if not data: return {"error": "data is required for PATCH requests"} return client.patch(endpoint, data) elif method == "DELETE": return client.delete(endpoint) else: return {"error": f"Unsupported HTTP method: {method}"} except Exception as e: return {"error": f"Generic API request failed: {str(e)}"} @mcp.tool() def create_workflow_survey( workflow_id: int = Field(description="Workflow job template ID"), questions: list = Field(description="List of survey questions in proper format"), survey_name: str = Field(default="Workflow Survey", description="Survey name"), survey_description: str = Field(default="Survey for workflow", description="Survey description") ) -> Dict[str, Any]: """ Create a survey for a workflow job template with proper formatting. This tool handles the specific format requirements for AAP survey specs. Example questions format: [ { "question_name": "Environment", "question_description": "Select environment", "required": true, "type": "multiplechoice", "variable": "environment", "default": "dev", "choices": ["dev", "test", "prod"] }, { "question_name": "Application Name", "question_description": "Enter application name", "required": true, "type": "text", "variable": "app_name", "default": "webapp", "min": 1, "max": 50 } ] """ try: client = get_aap_connector() # Convert questions to AAP survey spec format spec = [] for question in questions: survey_question = { "question_name": question["question_name"], "question_description": question.get("question_description", ""), "required": question.get("required", True), "type": question["type"], "variable": question["variable"], "default": question.get("default", ""), "new_question": True } # Add type-specific fields if question["type"] == "multiplechoice": survey_question["choices"] = question.get("choices", []) survey_question["min"] = 0 survey_question["max"] = 1024 elif question["type"] in ["text", "textarea"]: survey_question["choices"] = [] survey_question["min"] = question.get("min", 0) survey_question["max"] = question.get("max", 1024) elif question["type"] == "integer": survey_question["choices"] = [] survey_question["min"] = question.get("min", 0) survey_question["max"] = question.get("max", 9999) elif question["type"] == "float": survey_question["choices"] = [] survey_question["min"] = question.get("min", 0.0) survey_question["max"] = question.get("max", 9999.0) else: survey_question["choices"] = [] survey_question["min"] = 0 survey_question["max"] = 1024 spec.append(survey_question) # Create survey data survey_data = { "name": survey_name, "description": survey_description, "spec": spec } # Make the API call endpoint = f"workflow_job_templates/{workflow_id}/survey_spec/" response = client.post(endpoint, survey_data) return { "success": True, "survey_created": True, "workflow_id": workflow_id, "questions_count": len(spec), "response": response } except Exception as e: # Enhanced error reporting error_msg = str(e) return { "success": False, "error": f"Failed to create workflow survey: {error_msg}", "workflow_id": workflow_id, "debug_info": { "survey_data": survey_data if 'survey_data' in locals() else "Not created", "endpoint": f"workflow_job_templates/{workflow_id}/survey_spec/", "questions_provided": len(questions) if questions else 0 } }