Ansible & OpenShift Automation
Fornece ferramentas para interagir com a API da Ansible Automation Platform para tarefas de automação.
Documentação
Configuração do Ambiente para a Automação Ansible & OpenShift Automation com IA e Servidores de Protocolos de Contexto de Modelo (MCP)
Visão Geral
Este guia irá orientá-lo na configuração das partes de Servidores MCP + Claude Desktop da demonstração que focou em usar o Claude Desktop para interagir com seus ambientes de Ansible Automation Platform e OpenShift Cluster.
Pré-requisitos
Certifique-se de ter o seguinte instalado.
Necessário
- Um ambiente Ansible Automation Platform (AAP)
- Um cluster OpenShift com OpenShift Virtualization
- Claude Desktop instalado no seu laptop (Plano Pro necessário para melhores resultados)
- Python 3.10 ou superior instalado no seu laptop
- Certifique-se de estar autenticado com seu cluster OpenShift (por exemplo, exportando o kubeconfig)
Passo Um: Configure seu ambiente de laptop
Instale uv e configure seu projeto e ambiente Python.
curl -LsSf https://astral.sh/uv/install.sh | sh
Instale o jbang que será usado ao utilizar o Kubernetes MCP Server. (O jbang precisa ser instalado globalmente, recomenda-se usar o padrão de instalação do homebrew. Se você instalá-lo localmente (o padrão curl), o Claude não conseguirá acessá-lo).
Reinicie seu terminal para garantir que os comandos uv e jbang estejam agora disponíveis.
Passo Dois: Crie e configure seu projeto
# 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
Passo 3: Construindo seu servidor MCP do Ansible Automation Controller
Este é o servidor MCP que usei para interagir com meu controlador de automação. Sinta-se à vontade para copiar/colar isso no seu arquivo ansible.py.
Nota: para conectar a SSL autoassinado, edite o cliente assíncrono para ser 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")
Passo 4: Configurando o Claude Desktop para usar seus servidores MCP
No meu caso específico, quero aproveitar dois servidores MCP: o servidor MCP do Ansible acima e o servidor MCP do Kubernetes que encontrei no repositório Git quarkus-mcp-servers
Abra o claude_desktop_config.json, que no MacOS está localizado em
~/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"
]
}
}
}
Salve o arquivo.
AVISO: O caminho absoluto para o binário uv é necessário. Execute um which uv no seu sistema para obter o caminho completo.
NOTA: Se você precisar criar o AAP_TOKEN, vá ao Painel do AAP, selecione Gerenciamento de Acesso -> Usuários -> <your_user> -> Tokens -> Criar token -> Selecione o menu suspenso Escopo e selecione 'Write' e clique em Criar token.
Passo 5: Reinicie o Claude Desktop
Se você já tinha o Claude Desktop aberto, reinicie-o; caso contrário, certifique-se de que o Claude Desktop está detectando os servidores MCP. Você pode verificar isso garantindo que o ícone de martelo esteja ativado.
NOTA: O número ao lado do martelo variará de acordo com a quantidade de ferramentas MCP disponíveis.
Ao clicar no ícone de martelo, você pode ver uma lista de ferramentas. Abaixo está um exemplo.
Passo 6: Teste seu ambiente
Agora com tudo configurado, veja se você consegue interagir com sua plataforma Ansible Automation Platform e cluster OpenShift.
Sinta-se à vontade para fazer perguntas como:
- Quantos modelos de trabalho (Job Templates) estão disponíveis?
- Quantas VMs estão no meu cluster OpenShift?
NOTA: É muito provável que você precise aproveitar o Plano Pro do Claude Desktop para obter a funcionalidade completa.
Referências
Início Rápido do Claude Desktop para Desenvolvedores de Servidores
BÔNUS: Adicionando o servidor MCP do Event Driven Ansible
Se você configurou o Event Driven Ansible, pode aproveitar o servidor MCP do Event Driven Ansible abaixo. As instruções são semelhantes às acima.
- Crie um
eda.pye armazene-o no seu/absolute/path/to/ansible_mcp - Atualize seu
claude_desktop_config.json - Reinicie seu Claude Desktop e verifique se o martelo detectou suas novas ferramentas MCP
Os dois arquivos estão listados abaixo para fácil cópia/cola.
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"
}
}
}
}
AVISO: O caminho absoluto para o binário uv é necessário. Execute um which uv no seu sistema para obter o caminho completo.
NOTA: Um token EDA pode ser gerado a partir do Painel do AAP.
eda.py MCP Server
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")