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 服務。