Ansible Automation Platform

🚀 FastMCP Tool Server Integration with AAP and EDA

This project focuses on integrating the FastMCP tool server with AAP and EDA services, enabling seamless interaction and efficient data processing between different systems.

🚀 Quick Start

✅ Prerequisites

• Install Python 3.8 or a higher version.
• Install the fastmcp library:

pip install fastmcp

📦 Installation

Step 1: Create the AAP Tool Server (eda.py)

Environment Variable Configuration

Set the following environment variables in the eda.py file:

AAP_TOKEN=<Your AAP access token>
AAP_URL=<Your AAP service URL, e.g., https://example.com/api/controller/v2>

Core Functionality Code

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

# Load environment variables
AAP_TOKEN = os.getenv("AAP_TOKEN")
AAP_URL = os.getenv("AAP_URL")

if not AAP_TOKEN:
    raise ValueError("AAP_TOKEN environment variable is not set")

# Set request headers
HEADERS = {
    "Authorization": f"Bearer {AAP_TOKEN}",
    "Content-Type": "application/json"
}

# Initialize the FastMCP service
mcp = FastMCP("eda")

async def make_request(url: str, method: str = "GET", json: Dict = None) -> Any:
    """Helper function: used to send authenticated 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, 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"""
    return await make_request(f"{AAP_URL}/activations/")

@mcp.tool()
async def get_activation(activation_id: int) -> Any:
    """Get detailed information for a specific activation"""
    return await make_request(f"{AAP_URL}/activations/{activation_id}/")

@mcp.tool()
async def create_activation(payload: Dict) -> Any:
    """Create a new activation"""
    return await make_request(f"{AAP_URL}/activations/", method="POST", json=payload)

@mcp.tool()
async def disable_activation(activation_id: int) -> Any:
    """Disable a specific activation"""
    return await make_request(f"{AAP_URL}/activations/{activation_id}/disable/", method="POST")

@mcp.tool()
async def enable_activation(activation_id: int) -> Any:
    """Enable a specific activation"""
    return await make_request(f"{AAP_URL}/activations/{activation_id}/enable/", method="POST")

@mcp.tool()
async def restart_activation(activation_id: int) -> Any:
    """Restart a specific activation"""
    return await make_request(f"{AAP_URL}/activations/{activation_id}/restart/", method="POST")

@mcp.tool()
async def delete_activation(activation_id: int) -> Any:
    """Delete a specific activation"""
    return await make_request(f"{AAP_URL}/activations/{activation_id}/", method="DELETE")

@mcp.tool()
async def list_rules() -> Any:
    """List all rules"""
    return await make_request(f"{AAP_URL}/rules/")

@mcp.tool()
async def get_rule(rule_id: int) -> Any:
    """Get detailed information for a specific rule"""
    return await make_request(f"{AAP_URL}/rules/{rule_id}/")

# Start the FastMCP service
if __name__ == "__main__":
    mcp.start()

Step 2: Create the EDA Tool Server (activation.py)

Environment Variable Configuration

Set the following environment variables in the activation.py file:

ACTIVATION_TOKEN=<Your EDA access token>
ACTIVATION_URL=<Your EDA service URL, e.g., https://example.com/eda/api>

Core Functionality Code

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

# Load environment variables
ACTIVATION_TOKEN = os.getenv("ACTIVATION_TOKEN")
ACTIVATION_URL = os.getenv("ACTIVATION_URL")

if not ACTIVATION_TOKEN:
    raise ValueError("ACTIVATION_TOKEN environment variable is not set")

# Set request headers
HEADERS = {
    "Authorization": f"Bearer {ACTIVATION_TOKEN}",
    "Content-Type": "application/json"
}

# Initialize the FastMCP service
mcp = FastMCP("activation")

async def make_request(url: str, method: str = "GET", json: Dict = None) -> Any:
    """Helper function: used to send authenticated requests to 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"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_rulesets() -> Any:
    """List all rule sets"""
    return await make_request(f"{ACTIVATION_URL}/rulesets/")

@mcp.tool()
async def get_rule_set(rule_set_id: int) -> Any:
    """Get detailed information for a specific rule set"""
    return await make_request(f"{ACTIVATION_URL}/rulesets/{rule_set_id}/")

# Start the FastMCP service
if __name__ == "__main__":
    mcp.start()

Step 3: Run the Tool Servers

Start the AAP Tool Server

python eda.py

Start the EDA Tool Server

python activation.py

💻 Usage Examples

Call the list_activations Method of the AAP Tool Server

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

Call the list_rulesets Method of the EDA Tool Server

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

⚠️ Important Note

• Ensure that the environment variables are set correctly; otherwise, the tool servers will not run properly.
• Network issues may cause request timeouts. Please check your network connection.
• If you need to process large amounts of data, it is recommended to optimize the code performance.

By following the above steps, you can easily integrate the FastMCP tool server with AAP and EDA services.