Ansible Automation Platform

🚀 FastMCPツールサーバとAAPおよびEDAの統合設定説明

FastMCPツールサーバを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サービスに簡単に統合できます。