Ansible & OpenShift Automation
Proporciona herramientas para interactuar con la API de Ansible Automation Platform para tareas de automatización.
Documentación
Configuración del Entorno para la Automatización de Ansible y OpenShift con IA mediante Servidores de Protocolos de Contexto de Modelo (MCP)
Resumen
Esta guía te llevará a través de la configuración de los Servidores MCP y las partes de Claude Desktop de la demostración que se centra en usar Claude Desktop para interactuar con tus entornos de Ansible Automation Platform y clústeres de OpenShift.
Requisitos previos
Asegúrate de tener instalado lo siguiente.
Requerido
- Un entorno de Ansible Automation Platform (AAP)
- Un clúster de OpenShift con OpenShift Virtualization
- Claude Desktop instalado en tu laptop (se requiere el Plan Pro para obtener los mejores resultados)
- Python 3.10 o superior instalado en tu laptop
- Asegúrate de estar autenticado con tu clúster de OpenShift (por ejemplo, exportando kubeconfig)
Paso Uno: Configura tu entorno de laptop
Instala uv y configura tu proyecto y entorno de Python.
curl -LsSf https://astral.sh/uv/install.sh | sh
Instala jbang que se utilizará al usar el Servidor MCP de Kubernetes. (jbang debe instalarse globalmente; se recomienda usar el patrón de instalación de homebrew. Si lo instalas localmente (el patrón de curl), Claude no podrá acceder a él).
Reinicia tu terminal para asegurarte de que los comandos uv y jbang estén ahora disponibles.
Paso Dos: Crea y configura tu proyecto
# Create a new directory for our project
uv init ansible
cd ansible
# Create virtual environment and activate it
uv venv
source .venv/bin/activate
# Install dependencies
uv add "mcp[cli]" httpx
# Create our server file
touch ansible.py
Paso 3: Construye tu Servidor MCP de Ansible Automation Controller
Este es el Servidor MCP que utilicé para interactuar con mi controlador de automatización. Siéntete libre de copiar/pegar esto en tu archivo ansible.py.
Nota: para conectarte a SSL autofirmado, edita el cliente asíncrono para que sea https.AsyncClient(verify=False)
import os
import httpx
from mcp.server.fastmcp import FastMCP
from typing import Any
# Environment variables for authentication
AAP_URL = os.getenv("AAP_URL")
AAP_TOKEN = os.getenv("AAP_TOKEN")
if not AAP_TOKEN:
raise ValueError("AAP_TOKEN is required")
# Headers for API authentication
HEADERS = {
"Authorization": f"Bearer {AAP_TOKEN}",
"Content-Type": "application/json"
}
# Initialize FastMCP
mcp = FastMCP("ansible")
async def make_request(url: str, method: str = "GET", json: dict = None) -> Any:
"""Helper function to make authenticated API requests to AAP."""
async with httpx.AsyncClient() as client:
response = await client.request(method, url, headers=HEADERS, json=json)
if response.status_code not in [200, 201]:
return f"Error {response.status_code}: {response.text}"
return response.json() if "application/json" in response.headers.get("Content-Type", "") else response.text
@mcp.tool()
async def list_inventories() -> Any:
"""List all inventories in Ansible Automation Platform."""
return await make_request(f"{AAP_URL}/inventories/")
@mcp.tool()
async def get_inventory(inventory_id: str) -> Any:
"""Get details of a specific inventory by ID."""
return await make_request(f"{AAP_URL}/inventories/{inventory_id}/")
@mcp.tool()
async def run_job(template_id: int, extra_vars: dict = {}) -> Any:
"""Run a job template by ID, optionally with extra_vars."""
return await make_request(f"{AAP_URL}/job_templates/{template_id}/launch/", method="POST", json={"extra_vars": extra_vars})
@mcp.tool()
async def job_status(job_id: int) -> Any:
"""Check the status of a job by ID."""
return await make_request(f"{AAP_URL}/jobs/{job_id}/")
@mcp.tool()
async def job_logs(job_id: int) -> str:
"""Retrieve logs for a job."""
return await make_request(f"{AAP_URL}/jobs/{job_id}/stdout/")
@mcp.tool()
async def create_project(
name: str,
organization_id: int,
source_control_url: str,
source_control_type: str = "git",
description: str = "",
execution_environment_id: int = None,
content_signature_validation_credential_id: int = None,
source_control_branch: str = "",
source_control_refspec: str = "",
source_control_credential_id: int = None,
clean: bool = False,
update_revision_on_launch: bool = False,
delete: bool = False,
allow_branch_override: bool = False,
track_submodules: bool = False,
) -> Any:
"""Create a new project in Ansible Automation Platform."""
payload = {
"name": name,
"description": description,
"organization": organization_id,
"scm_type": source_control_type.lower(), # Git is default
"scm_url": source_control_url,
"scm_branch": source_control_branch,
"scm_refspec": source_control_refspec,
"scm_clean": clean,
"scm_delete_on_update": delete,
"scm_update_on_launch": update_revision_on_launch,
"allow_override": allow_branch_override,
"scm_track_submodules": track_submodules,
}
if execution_environment_id:
payload["execution_environment"] = execution_environment_id
if content_signature_validation_credential_id:
payload["signature_validation_credential"] = content_signature_validation_credential_id
if source_control_credential_id:
payload["credential"] = source_control_credential_id
return await make_request(f"{AAP_URL}/projects/", method="POST", json=payload)
@mcp.tool()
async def create_job_template(
name: str,
project_id: int,
playbook: str,
inventory_id: int,
job_type: str = "run",
description: str = "",
credential_id: int = None,
execution_environment_id: int = None,
labels: list[str] = None,
forks: int = 0,
limit: str = "",
verbosity: int = 0,
timeout: int = 0,
job_tags: list[str] = None,
skip_tags: list[str] = None,
extra_vars: dict = None,
privilege_escalation: bool = False,
concurrent_jobs: bool = False,
provisioning_callback: bool = False,
enable_webhook: bool = False,
prevent_instance_group_fallback: bool = False,
) -> Any:
"""Create a new job template in Ansible Automation Platform."""
payload = {
"name": name,
"description": description,
"job_type": job_type,
"project": project_id,
"playbook": playbook,
"inventory": inventory_id,
"forks": forks,
"limit": limit,
"verbosity": verbosity,
"timeout": timeout,
"ask_variables_on_launch": bool(extra_vars),
"ask_tags_on_launch": bool(job_tags),
"ask_skip_tags_on_launch": bool(skip_tags),
"ask_credential_on_launch": credential_id is None,
"ask_execution_environment_on_launch": execution_environment_id is None,
"ask_labels_on_launch": labels is None,
"ask_inventory_on_launch": False, # Inventory is required, so not prompting
"ask_job_type_on_launch": False, # Job type is required, so not prompting
"become_enabled": privilege_escalation,
"allow_simultaneous": concurrent_jobs,
"scm_branch": "",
"webhook_service": "github" if enable_webhook else "",
"prevent_instance_group_fallback": prevent_instance_group_fallback,
}
if credential_id:
payload["credential"] = credential_id
if execution_environment_id:
payload["execution_environment"] = execution_environment_id
if labels:
payload["labels"] = labels
if job_tags:
payload["job_tags"] = job_tags
if skip_tags:
payload["skip_tags"] = skip_tags
if extra_vars:
payload["extra_vars"] = extra_vars
return await make_request(f"{AAP_URL}/job_templates/", method="POST", json=payload)
@mcp.tool()
async def list_inventory_sources() -> Any:
"""List all inventory sources in Ansible Automation Platform."""
return await make_request(f"{AAP_URL}/inventory_sources/")
@mcp.tool()
async def get_inventory_source(inventory_source_id: int) -> Any:
"""Get details of a specific inventory source."""
return await make_request(f"{AAP_URL}/inventory_sources/{inventory_source_id}/")
@mcp.tool()
async def create_inventory_source(
name: str,
inventory_id: int,
source: str,
credential_id: int,
source_vars: dict = None,
update_on_launch: bool = True,
timeout: int = 0,
) -> Any:
"""Create a dynamic inventory source. Claude will ask for the source type and credential before proceeding."""
valid_sources = [
"file", "constructed", "scm", "ec2", "gce", "azure_rm", "vmware", "satellite6", "openstack",
"rhv", "controller", "insights", "terraform", "openshift_virtualization"
]
if source not in valid_sources:
return f"Error: Invalid source type '{source}'. Please select from: {', '.join(valid_sources)}"
if not credential_id:
return "Error: Credential is required to create an inventory source."
payload = {
"name": name,
"inventory": inventory_id,
"source": source,
"credential": credential_id,
"source_vars": source_vars,
"update_on_launch": update_on_launch,
"timeout": timeout,
}
return await make_request(f"{AAP_URL}/inventory_sources/", method="POST", json=payload)
@mcp.tool()
async def update_inventory_source(inventory_source_id: int, update_data: dict) -> Any:
"""Update an existing inventory source."""
return await make_request(f"{AAP_URL}/inventory_sources/{inventory_source_id}/", method="PATCH", json=update_data)
@mcp.tool()
async def delete_inventory_source(inventory_source_id: int) -> Any:
"""Delete an inventory source."""
return await make_request(f"{AAP_URL}/inventory_sources/{inventory_source_id}/", method="DELETE")
@mcp.tool()
async def sync_inventory_source(inventory_source_id: int) -> Any:
"""Manually trigger a sync for an inventory source."""
return await make_request(f"{AAP_URL}/inventory_sources/{inventory_source_id}/update/", method="POST")
@mcp.tool()
async def create_inventory(
name: str,
organization_id: int,
description: str = "",
kind: str = "",
host_filter: str = "",
variables: dict = None,
prevent_instance_group_fallback: bool = False,
) -> Any:
"""Create an inventory in Ansible Automation Platform."""
payload = {
"name": name,
"organization": organization_id,
"description": description,
"kind": kind,
"host_filter": host_filter,
"variables": variables,
"prevent_instance_group_fallback": prevent_instance_group_fallback,
}
return await make_request(f"{AAP_URL}/inventories/", method="POST", json=payload)
@mcp.tool()
async def delete_inventory(inventory_id: int) -> Any:
"""Delete an inventory from Ansible Automation Platform."""
return await make_request(f"{AAP_URL}/inventories/{inventory_id}/", method="DELETE")
@mcp.tool()
async def list_job_templates() -> Any:
"""List all job templates available in Ansible Automation Platform."""
return await make_request(f"{AAP_URL}/job_templates/")
@mcp.tool()
async def get_job_template(template_id: int) -> Any:
"""Retrieve details of a specific job template."""
return await make_request(f"{AAP_URL}/job_templates/{template_id}/")
@mcp.tool()
async def list_jobs() -> Any:
"""List all jobs available in Ansible Automation Platform."""
return await make_request(f"{AAP_URL}/jobs/")
@mcp.tool()
async def list_recent_jobs(hours: int = 24) -> Any:
"""List all jobs executed in the last specified hours (default 24 hours)."""
from datetime import datetime, timedelta
time_filter = (datetime.utcnow() - timedelta(hours=hours)).isoformat() + "Z"
return await make_request(f"{AAP_URL}/jobs/?created__gte={time_filter}")
if __name__ == "__main__":
mcp.run(transport="stdio")
Paso 4: Configura Claude Desktop para usar tus Servidores MCP
En mi caso particular, quiero aprovechar dos Servidores MCP: el Servidor MCP de Ansible mencionado anteriormente y el Servidor MCP de Kubernetes que encontré dentro del repositorio de Git quarkus-mcp-servers
Abre el claude_desktop_config.json, que en MacOS se encuentra en
~/Library/Application\ Support/Claude/claude_desktop_config.json
{
"mcpServers": {
"ansible": {
"command": "/absolute/path/to/uv",
"args": [
"--directory",
"/absolute/path/to/ansible_mcp",
"run",
"ansible.py"
],
"env": {
"AAP_TOKEN": "<aap-token>",
"AAP_URL": "https://<aap-url>/api/controller/v2"
}
},
"kubernetes": {
"command": "jbang",
"args": [
"--quiet",
"https://github.com/quarkiverse/quarkus-mcp-servers/blob/main/kubernetes/src/main/java/io/quarkiverse/mcp/servers/kubernetes/MCPServerKubernetes.java"
]
}
}
}
Guarda el archivo.
ADVERTENCIA: Se requiere la ruta absoluta a tu binario uv. Haz un which uv en tu sistema para obtener la ruta completa.
NOTA: Si necesitas crear el AAP_TOKEN, ve al Panel de AAP, selecciona Administración de Acceso -> Usuarios -> <your_user> -> Tokens -> Crear token -> Selecciona el menú desplegable de Alcance y selecciona 'Escribir' y haz clic en Crear token.
Paso 5: Relanza Claude Desktop
Si ya tenías Claude Desktop abierto, relánzalo; de lo contrario, asegúrate de que Claude Desktop esté detectando los servidores MCP. Puedes verificar esto asegurándote de que el ícono de martillo esté lanzado.
NOTA: El número junto al martillo variará según la cantidad de herramientas MCP disponibles.
Una vez que hagas clic en el ícono del martillo, puedes ver una lista de herramientas. A continuación se muestra un ejemplo.
Paso 6: Prueba tu entorno
Ahora con todo configurado, mira si puedes interactuar con tu Ansible Automation Platform y tu clúster de OpenShift.
Siéntete libre de hacerle preguntas como:
- ¿Cuántas Plantillas de Trabajo están disponibles?
- ¿Cuántas máquinas virtuales hay en mi clúster de OpenShift?
NOTA: Es muy probable que necesites aprovechar el Plan Pro de Claude Desktop para obtener la funcionalidad completa.
Referencias
Inicio rápido de Claude Desktop para desarrolladores de servidores
BONUS: Agregar el Servidor MCP de Event Driven Ansible
Si has configurado Event Driven Ansible, puedes aprovechar el Servidor MCP de Event Driven Ansible que se muestra a continuación. Las instrucciones son similares a las anteriores.
- Crea un
eda.pyy guárdalo en tu/absolute/path/to/ansible_mcp - Actualiza tu
claude_desktop_config.json - Reinicia tu Claude Desktop y verifica que el martillo haya detectado tus nuevas herramientas MCP
Los dos archivos se enumeran a continuación para facilitar copiar/pegar.
claude_desktop_config.json
{
"mcpServers": {
"ansible": {
"command": "/absolute/path/to/uv",
"args": [
"--directory",
"/absolute/path/to/ansible_mcp",
"run",
"ansible.py"
],
"env": {
"AAP_TOKEN": "<aap-token>",
"AAP_URL": "https://<aap-url>/api/controller/v2"
}
},
"kubernetes": {
"command": "jbang",
"args": [
"--quiet",
"https://github.com/quarkiverse/quarkus-mcp-servers/blob/main/kubernetes/src/main/java/io/quarkiverse/mcp/servers/kubernetes/MCPServerKubernetes.java"
]
},
"eda": {
"command": "/absolute/path/to/uv",
"args": [
"--directory",
"/absolute/path/to/ansible_mcp",
"run",
"eda.py"
],
"env": {
"EDA_TOKEN": "<eda-token-can-be-same-as-aap-token>",
"EDA_URL": "https://<aap-url>/api/eda/v1"
}
}
}
}
ADVERTENCIA: Se requiere la ruta absoluta a tu binario uv. Haz un which uv en tu sistema para obtener la ruta completa.
NOTA: Se puede generar un Token EDA desde el Panel de AAP.
Servidor MCP eda.py
import os
import httpx
from mcp.server.fastmcp import FastMCP
from typing import Optional, Any, Dict
# Environment variables for authentication
EDA_URL = os.getenv("EDA_URL")
EDA_TOKEN = os.getenv("EDA_TOKEN")
if not EDA_TOKEN:
raise ValueError("EDA_TOKEN is required")
# Headers for API authentication
HEADERS = {
"Authorization": f"Bearer {EDA_TOKEN}",
"Content-Type": "application/json"
}
# Initialize FastMCP
mcp = FastMCP("eda")
async def make_request(url: str, *, method: str = "GET", params: Optional[Dict] = None, json: Optional[Dict] = None) -> Any:
"""Helper function to make authenticated API requests to EDA."""
async with httpx.AsyncClient() as client:
#logging.info(f"make_request.url = {url}")
#logging.info(f"make_request.method = {method}")
#logging.info(f"make_request.params = {params}")
#logging.info(f"make_request.json = {json}")
response = await client.request(method, url, headers=HEADERS, params=params, json=json)
if response.status_code not in [200, 201, 204]:
return f"Error {response.status_code}: {response.text}"
return response.json() if "application/json" in response.headers.get("Content-Type", "") else response.text
@mcp.tool()
async def list_activations() -> Any:
"""List all activations in Event-Driven Ansible."""
return await make_request(f"{EDA_URL}/activations/")
@mcp.tool()
async def get_activation(activation_id: int) -> Any:
"""Get details of a specific activation."""
return await make_request(f"{EDA_URL}/activations/{activation_id}/")
@mcp.tool()
async def create_activation(payload: Dict) -> Any:
"""Create a new activation."""
return await make_request(f"{EDA_URL}/activations/", method="POST", json=payload)
@mcp.tool()
async def disable_activation(activation_id: int) -> Any:
"""Disable an activation."""
return await make_request(f"{EDA_URL}/activations/{activation_id}/disable/", method="POST")
@mcp.tool()
async def enable_activation(activation_id: int) -> Any:
"""Enable an activation."""
return await make_request(f"{EDA_URL}/activations/{activation_id}/enable/", method="POST")
@mcp.tool()
async def restart_activation(activation_id: int) -> Any:
"""Restart an activation."""
return await make_request(f"{EDA_URL}/activations/{activation_id}/restart/", method="POST")
@mcp.tool()
async def delete_activation(activation_id: int) -> Any:
"""Delete an activation."""
return await make_request(f"{EDA_URL}/activations/{activation_id}/", method="DELETE")
@mcp.tool()
async def list_decision_environments() -> Any:
"""List all decision environments."""
return await make_request(f"{EDA_URL}/decision-environments/")
@mcp.tool()
async def create_decision_environment(payload: Dict) -> Any:
"""Create a new decision environment."""
return await make_request(f"{EDA_URL}/decision-environments/", method="POST", json=payload)
@mcp.tool()
async def list_rulebooks() -> Any:
"""List all rulebooks in EDA."""
return await make_request(f"{EDA_URL}/rulebooks/")
@mcp.tool()
async def get_rulebook(rulebook_id: int) -> Any:
"""Retrieve details of a specific rulebook."""
return await make_request(f"{EDA_URL}/rulebooks/{rulebook_id}/")
@mcp.tool()
async def list_event_streams() -> Any:
"""List all event streams."""
return await make_request(f"{EDA_URL}/event-streams/")
@mcp.tool()
async def list_rule_audits() -> Any:
"""List all rule audits"""
return await make_request(f"{EDA_URL}/audit-rules/")
@mcp.tool()
async def get_rule_audit(rulebook_id: int) -> Any:
"""Get the audit of a specific rule"""
return await make_request(f"{EDA_URL}/audit-rules/{rulebook_id}")
@mcp.tool()
async def get_rule_activation_audit(activation_id: int) -> Any:
"""Get the audit of a specific rule activation"""
params = {"activation_instance_id": str(activation_id)}
return await make_request(f"{EDA_URL}/audit-rules/", params=params)
if __name__ == "__main__":
mcp.run(transport="stdio")