Ansible Automation Platform

🚀 FastMCP 工具服务器与 AAP 和 EDA 的集成设置说明

本项目聚焦于将 FastMCP 工具服务器与 AAP 和 EDA 进行集成，通过一系列步骤和代码实现，能让你轻松完成两者的集成，实现对 AAP 和 EDA 服务的便捷操作。

🚀 快速开始

✨ 必需条件

• 安装 Python 3.8 或更高版本。
• 安装 fastmcp 库，可使用以下命令：

pip install fastmcp

📦 安装指南

1️⃣ 创建 AAP 工具服务器（eda.py）

🌐 环境变量配置

在 eda.py 文件中设置以下环境变量：

AAP_TOKEN=<你的 AAP 访问令牌>
AAP_URL=<你的 AAP 服务 URL，例如：https://example.com/api/controller/v2>

💻 核心功能代码

import os
import httpx
from mcp.server.fastmcp import FastMCP
from typing import Any, Dict

# 加载环境变量
AAP_TOKEN = os.getenv("AAP_TOKEN")
AAP_URL = os.getenv("AAP_URL")

if not AAP_TOKEN:
    raise ValueError("AAP_TOKEN 环境变量未设置")

# 设置请求头
HEADERS = {
    "Authorization": f"Bearer {AAP_TOKEN}",
    "Content-Type": "application/json"
}

# 初始化 FastMCP 服务
mcp = FastMCP("eda")

async def make_request(url: str, method: str = "GET", json: Dict = None) -> Any:
    """辅助函数：用于向 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, 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:
    """列出所有激活项"""
    return await make_request(f"{AAP_URL}/activations/")

@mcp.tool()
async def get_activation(activation_id: int) -> Any:
    """获取特定激活项的详细信息"""
    return await make_request(f"{AAP_URL}/activations/{activation_id}/")

@mcp.tool()
async def create_activation(payload: Dict) -> Any:
    """创建新的激活项"""
    return await make_request(f"{AAP_URL}/activations/", method="POST", json=payload)

@mcp.tool()
async def disable_activation(activation_id: int) -> Any:
    """禁用特定激活项"""
    return await make_request(f"{AAP_URL}/activations/{activation_id}/disable/", method="POST")

@mcp.tool()
async def enable_activation(activation_id: int) -> Any:
    """启用特定激活项"""
    return await make_request(f"{AAP_URL}/activations/{activation_id}/enable/", method="POST")

@mcp.tool()
async def restart_activation(activation_id: int) -> Any:
    """重启特定激活项"""
    return await make_request(f"{AAP_URL}/activations/{activation_id}/restart/", method="POST")

@mcp.tool()
async def delete_activation(activation_id: int) -> Any:
    """删除特定激活项"""
    return await make_request(f"{AAP_URL}/activations/{activation_id}/", method="DELETE")

@mcp.tool()
async def list_rules() -> Any:
    """列出所有规则"""
    return await make_request(f"{AAP_URL}/rules/")

@mcp.tool()
async def get_rule(rule_id: int) -> Any:
    """获取特定规则的详细信息"""
    return await make_request(f"{AAP_URL}/rules/{rule_id}/")

# 启动 FastMCP 服务
if __name__ == "__main__":
    mcp.start()

2️⃣ 创建 EDA 工具服务器（activation.py）

🌐 环境变量配置

在 activation.py 文件中设置以下环境变量：

ACTIVATION_TOKEN=<你的 EDA 访问令牌>
ACTIVATION_URL=<你的 EDA 服务 URL，例如：https://example.com/eda/api>

💻 核心功能代码

import os
import httpx
from mcp.server.fastmcp import FastMCP
from typing import Any, Dict

# 加载环境变量
ACTIVATION_TOKEN = os.getenv("ACTIVATION_TOKEN")
ACTIVATION_URL = os.getenv("ACTIVATION_URL")

if not ACTIVATION_TOKEN:
    raise ValueError("ACTIVATION_TOKEN 环境变量未设置")

# 设置请求头
HEADERS = {
    "Authorization": f"Bearer {ACTIVATION_TOKEN}",
    "Content-Type": "application/json"
}

# 初始化 FastMCP 服务
mcp = FastMCP("activation")

async def make_request(url: str, method: str = "GET", json: Dict = None) -> Any:
    """辅助函数：用于向 EDA 发送认证请求"""
    async with httpx.AsyncClient() as client:
        response = await client.request(method, url, headers=HEADERS, 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_rulesets() -> Any:
    """列出所有规则集"""
    return await make_request(f"{ACTIVATION_URL}/rulesets/")

@mcp.tool()
async def get_rule_set(rule_set_id: int) -> Any:
    """获取特定规则集的详细信息"""
    return await make_request(f"{ACTIVATION_URL}/rulesets/{rule_set_id}/")

# 启动 FastMCP 服务
if __name__ == "__main__":
    mcp.start()

3️⃣ 运行工具服务器

▶️ 启动 AAP 工具服务器

python eda.py

▶️ 启动 EDA 工具服务器

python activation.py

💻 使用示例

基础用法

调用 AAP 工具服务器的 list_activations 方法

curl -X GET "http://localhost:8000/list_activations"

调用 EDA 工具服务器的 list_rulesets 方法

curl -X GET "http://localhost:8001/list_rulesets"

⚠️ 注意事项

⚠️ 重要提示

• 确保环境变量正确设置，否则工具服务器无法正常运行。
• 网络问题可能导致请求超时，请检查网络连接。
• 如果需要处理大量数据，建议优化代码性能。

通过以上步骤，您可以轻松集成 FastMCP 工具服务器与 AAP 和 EDA 服务。