设置说明：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"
}
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}/")
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"
}
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}/")
if __name__ == "__main__":
mcp.start()
3. 运行工具服务器
启动 AAP 工具服务器
python eda.py
启动 EDA 工具服务器
python activation.py
4. 示例用法
调用 AAP 工具服务器的 list_activations 方法
curl -X GET "http://localhost:8000/list_activations"
调用 EDA 工具服务器的 list_rulesets 方法
curl -X GET "http://localhost:8001/list_rulesets"
5. 注意事项
- 确保环境变量正确设置，否则工具服务器无法正常运行。
- 网络问题可能导致请求超时，请检查网络连接。
- 如果需要处理大量数据，建议优化代码性能。
通过以上步骤，您可以轻松集成 FastMCP 工具服务器与 AAP 和 EDA 服务。
