项目介绍
使用AI驱动的Ansible及OpenShift自动化与模型上下文协议（MCP）服务器的环境设置
概述
本指南将引导您完成演示中涉及的部分设置，该演示专注于使用Claude Desktop与您的Ansible自动化平台和OpenShift集群环境进行交互。
预备条件
确保已安装以下内容。
必需项
- Ansible自动化平台（AAP）环境
- 带有OpenShift虚拟化的OpenShift集群
- 在您的笔记本电脑上安装Claude Desktop（推荐使用Pro计划以获得最佳效果）
- 笔记本电脑上安装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命令现在可用。
第二步：创建并设置您的项目
# 创建一个新的项目目录
uv init ansible
cd ansible
# 创建虚拟环境并激活它
uv venv
source .venv/bin/activate
# 安装依赖项
uv add "mcp[cli]" httpx
# 创建我们的服务器文件
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
# 认证环境变量
AAP_URL = os.getenv("AAP_URL")
AAP_TOKEN = os.getenv("AAP_TOKEN")
if not AAP_TOKEN:
raise ValueError("AAP_TOKEN是必需的")
# API认证头
HEADERS = {
"Authorization": f"Bearer {AAP_TOKEN}",
"Content-Type": "application/json"
}
# 初始化FastMCP
mcp = FastMCP("ansible")
async def make_request(url: str, method: str = "GET", json: dict = None) -> Any:
"""辅助函数，用于向AAP发送经过身份验证的API请求。"""
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"错误 {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:
"""列出Ansible自动化平台中的所有清单。"""
return await make_request(f"{AAP_URL}/inventories/")
@mcp.tool()
async def get_inventory(inventory_id: str) -> Any:
"""根据ID获取特定清单的详细信息。"""
return await make_request(f"{AAP_URL}/inventories/{inventory_id}/")
@mcp.tool()
async def run_job(template_id: int, extra_vars: dict = {}) -> Any:
"""根据ID运行作业模板，可选地带有额外变量。"""
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:
"""根据ID检查作业的状态。"""
return await make_request(f"{AAP_URL}/jobs/{job_id}/")
@mcp.tool()
async def job_logs(job_id: int) -> str:
"""检索作业的日志。"""
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:
"""在Ansible自动化平台中创建新的项目。"""
payload = {
"name": name,
"description": description,
"organization": organization_id,
"scm_type": source_control_type.lower(), # 默认为Git
"scm_url": source_control_url,
"scm_branch": source_control_branch,
以下省略...
第四步：配置Claude Desktop以使用您的MCP服务器
在我的特定情况下，我想要利用两个MCP服务器：上面提到的Ansible MCP服务器和我在quarkus-mcp-servers Git仓库中找到的Kubernetes MCP服务器。
打开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仪表板，选择访问管理 -> 用户 -> <您的用户> -> 令牌 -> 创建令牌 -> 选择作用域下拉菜单并选择“写入”，然后点击创建令牌。
第五步：重新启动Claude Desktop
如果您已经打开了Claude Desktop，请重新启动它，否则请确保Claude Desktop能够识别MCP服务器。您可以通过确保锤子图标被启动来验证这一点。
注意：锤子旁边的数字会根据可用的MCP工具数量而变化。
单击锤子图标后，可以看到一个工具列表。下面是一个示例。
第六步：测试您的环境
现在一切就绪，看看是否可以与您的Ansible自动化平台和OpenShift集群进行交互。
您可以自由询问如下问题：
- 可用的作业模板有多少？
- 我的OpenShift集群上有多少台虚拟机？
注意：很可能您需要利用Claude Desktop Pro计划才能获得全部功能。
参考资料
Claude Desktop快速入门指南（适用于服务器开发者）
BONUS：添加事件驱动的Ansible MCP服务器
如果您已经设置了事件驱动的Ansible，可以利用下面的事件驱动的Ansible MCP服务器。步骤与上述类似。
- 创建一个
eda.py并存储在您的/absolute/path/to/ansible_mcp中。 - 更新您的
claude_desktop_config.json。 - 重启您的Claude Desktop，并验证锤子图标是否已识别您的新MCP工具。
以下是便于复制粘贴的两个文件。
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"
}
}
}
}
警告：需要提供uv二进制文件的绝对路径。在系统中执行which uv以获取完整路径。
注意：可以从AAP仪表板生成EDA Token。
eda.py MCP服务器
import os
import httpx
from mcp.server.fastmcp import FastMCP
from typing import Optional, Any, Dict
# 认证环境变量
EDA_URL = os.getenv("EDA_URL")
EDA_TOKEN = os.getenv("EDA_TOKEN")
if not EDA_TOKEN:
raise ValueError("EDA_TOKEN是必需的")
# API认证头
HEADERS = {
"Authorization": f"Bearer {EDA_TOKEN}",
"Content-Type": "application/json"
}
# 初始化FastMCP
mcp = FastMCP("eda")
async def make_request(url: str, *, method: str = "GET", params: Optional[Dict] = None, json: Optional[Dict] = None) -> Any:
"""辅助函数，用于向EDA发送经过身份验证的API请求。"""
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"错误 {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:
"""列出事件驱动的Ansible中的所有激活。"""
return await make_request(f"{EDA_URL}/activations/")
@mcp.tool()
async def get_activation(activation_id: int) -> Any:
"""获取特定激活的详细信息。"""
return await make_request(f"{EDA_URL}/activations/{activation_id}/")
@mcp.tool()
async def create_activation(payload: Dict) -> Any:
"""创建新的激活。"""
return await make_request(f"{EDA_URL}/activations/", method="POST", json=payload)
@mcp.tool()
async def disable_activation(activation_id: int) -> Any:
"""禁用激活。"""
return await make_request(f"{EDA_URL}/activations/{activation_id}/disable/", method="POST")
@mcp.tool()
async def enable_activation(activation_id: int) -> Any:
"""启用激活。"""
return await make_request(f"{EDA_URL}/activations/{activation_id}/enable/", method="POST")
@mcp.tool()
async def restart_activation(activation_id: int) -> Any:
"""重启激活。"""
return await make_request(f"{EDA_URL}/activations/{activation_id}/restart/", method="POST")
@mcp.tool()
async def delete_activation(activation_id: int) -> Any:
"""删除激活。"""
return await make_request(f"{EDA_URL}/activations/{activation_id}/", method="DELETE")
@mcp.tool()
async def list_decision_environments() -> Any:
"""列出所有决策环境。"""
return await make_request(f"{EDA_URL}/decision-environments/")
@mcp.tool()
async def create_decision_environment(payload: Dict) -> Any:
"""创建新的决策环境。"""
return await make_request(f"{EDA_URL}/decision-environments/", method="POST", json=payload)
@mcp.tool()
async def list_rulebooks() -> Any:
"""列出EDA中的所有规则集。"""
return await make_request(f"{EDA_URL}/rulebooks/")
@mcp.tool()
async def get_rulebook(rulebook_id: int) -> Any:
"""获取特定规则集的详细信息。"""
return await make_request(f"{EDA_URL}/rulebooks/{rulebook_id}/")
@mcp.tool()
async def list_event_streams() -> Any:
"""列出所有事件流。"""
return await make_request(f"{EDA_URL}/event-streams/")
@mcp.tool()
async def list_rule_audits() -> Any:
"""列出所有规则审计。"""
return await make_request(f"{EDA_URL}/audit-rules/")
@mcp.tool()
async def get_rule_audit(rulebook_id: int) -> Any:
"""获取特定规则的审计。"""
return await make_request(f"{EDA_URL}/audit-rules/{rulebook_id}")
@mcp.tool()
async def get_rule_activation_audit(activation_id: int) -> Any:
"""获取特定规则激活的审计。"""
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")