AI导哈

Ansible与OpenShift自动化MCP服务器

通过MCP服务器实现Ansible Automation Platform和OpenShift集群的AI驱动自动化交互。

  • by rlopez133
  • Python
  • 本地部署
  • 20 星标
  • 更新于
在 GitHub 查看 rlopez133/mcp

使用说明

使用模型上下文协议（MCP）服务器设置AI驱动的Ansible与OpenShift自动化环境

概述

本指南将引导您完成演示中MCP服务器和Claude桌面部分的设置，重点介绍如何使用Claude桌面与您的Ansible自动化平台和OpenShift集群环境进行交互。

先决条件

确保您已安装以下内容。

必需的

  • 一个Ansible自动化平台（AAP）环境
  • 一个带有OpenShift虚拟化的OpenShift集群
  • 克劳德桌面版 安装在您的笔记本电脑上（为获得最佳效果，需使用专业版套餐）
  • 您的笔记本电脑上已安装 Python 3.10 或更高版本
  • 确保您已通过您的OpenShift集群进行身份验证（例如，导出kubeconfig）

第一步：设置您的笔记本电脑环境

安装 uv 并设置好你的Python项目和环境。

curl -LsSf https://astral.sh/uv/install.sh | sh

安装 jbang 这将在使用 Kubernetes MCP 服务器时被用到。（需要全局安装 jbang，建议使用 Homebrew 安装方式。如果本地安装（使用 curl 方式），Claude 将无法访问它。）。

重启你的终端以确保 uv 并且 jbang 命令现已可用。

第二步：创建并设置您的项目

# 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

第三步：构建您的Ansible自动化控制器MCP服务器

这是我用来与我的自动化控制器进行交互的MCP服务器。请随意复制/粘贴这段内容到你的（系统/代码中） ansible.py 文件。

注：若要连接到自签名SSL，请将异步客户端编辑为使用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")

步骤4：配置Claude Desktop以使用您的MCP服务器

在我特定的情况下，我想利用两个MCP服务器：一个是上面提到的Ansible MCP服务器，另一个是在我这里找到的Kubernetes MCP服务器 Quarkus-MCP-服务器（或译为：Quarkus MCP 服务器） Git 仓库

打开 claude_desktop_config.json 在MacOS上，它位于

~/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"
      ]
    }
  }
}

保存文件。

警告：您的绝对路径 uv 需要二进制文件。请执行 which uv 在您的系统上操作以获取完整路径。

注：如果您需要创建AAP_TOKEN，请前往AAP仪表板，选择访问管理 -> 用户 -> <您的用户名> -> 令牌 -> 创建令牌 -> 在范围下拉菜单中选择“写入”并点击创建令牌。

步骤5：重新启动Claude桌面版

如果你已经打开了Claude Desktop，请重新启动它；否则，请确保Claude Desktop正在连接到MCP服务器。你可以通过确认锤子图标是否启动来验证这一点。

注：锤子旁边的数字会根据可用的MCP工具数量而变化。

一旦你点击锤子图标，就会看到一系列工具列表。以下是示例。

步骤6：测试您的环境

现在，所有设置都已完成，看看你是否能够与Ansible自动化平台和OpenShift集群进行交互。

请随时向它提问，比如：

  • 有多少个作业模板可用？
  • 我的OpenShift集群上有多少台虚拟机？

注：为了获得全部功能，您很可能需要使用Claude桌面专业版计划。

参考文献

服务器开发者版Claude Desktop快速入门指南

奖金福利：添加事件驱动的Ansible MCP服务器

如果您已经设置了事件驱动Ansible，那么您可以利用下面的事件驱动Ansible MCP服务器。操作指南与上述类似。

  • 创建一个 eda.py 并将其存储在您的 /absolute/path/to/ansible_mcp
  • 更新您的 claude_desktop_config.json
  • 重启您的Claude桌面程序，并验证锤子是否已加载您的新MCP工具

以下是两个文件的列表，便于复制/粘贴。

claude_desktop_config.json 翻译为中文是：“claude 桌面配置文件.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"
        }
    }
  }
}

警告：您的绝对路径 uv 需要二进制文件。请执行 which uv 在您的系统上获取完整路径。

注：可以从AAP仪表板生成EDA代币。

eda.py MCP 服务器

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")

相关 MCP