obdiag MCP服务化的一种尝试

渠磊 · 2025 年6 月 4 日 19:56

1. 前言

本文并不涉及obdiag的诊断操作，仅作为 obdiag 实现MCP可被大模型进行调用的一种尝试。

2. 前期准备

大模型：cursor自带，也可以使用任何支持MCP的模型调用（如vscode的cline组件、cherry studio等等）

调用节点：已安装obdiag，并完成集群配置的节点

本文所有的指导为在cursor环境下的调用案例，其他环境可联系 obdiag

3. 大模型调用obdiag MCP及其实现

3.1. 启动mcp_server

首先需要在obdiag所在节点上启动我们的obdiag mcp_server.py(demo版本)

from fastmcp import FastMCP, Client
import subprocess

# 创建 FastMCP 实例
mcp = FastMCP("OBDiag MCP Server")

def run_obdiag_command(command: str) -> str:
    """
    运行 obdiag 命令并返回结果
    :param command: 完整的 obdiag 命令
    :return: 指令执行的输出结果
    """
    try:
        print(f"Running command: {command}")
        # 使用 subprocess 执行命令
        result = subprocess.run(command, shell=True, text=True, capture_output=True)
        # 返回标准输出或错误输出
        if result.returncode == 0:
            return result.stdout
        else:
            return f"Error: {result.stderr}"
    except Exception as e:
        return f"Exception occurred: {str(e)}"

# 定义一个工具函数，用于调用 obdiag 指令
@mcp.tool()
async def obdiag_check_run() -> str:
    """
    巡检集群，并返回巡检报告
    :return: 指令执行的输出结果
    """
    print("巡检执行")
    full_command = f"obdiag check run"
    return run_obdiag_command(full_command)
# 启动 MCP 服务
if __name__ == "__main__":
    # 检查是否还存在其他的 MCP 实例
    check_old_mcp = subprocess.run("ps -ef | grep obdiag_mcpserver.py | grep -v grep", shell=True, text=True, capture_output=True)
    if check_old_mcp.returncode == 0:
        print("MCP server is already running. Please stop it before starting a new one.")
        exit(1)
    mcp.run(transport="streamable-http", host="0.0.0.0", port=8000, path="/mcp")