Dify+fastmcp 实战：从零搭建可复用的 MCP 服务（附完整源码与配置骨架）

1. 为什么我要把运维查询做成 MCP 服务Dify 编排 Agent 时最让人头疼的不是提示词而是「实时数据从哪来」。我一开始也走过弯路把机器指标先同步到本地知识库再让 Dify 检索。结果知识库更新慢、数据滞后运维同事问「这台机器现在 CPU 多少」Agent 只能给出十分钟前的快照完全没法用。后来我把思路换成 MCPModel Context Protocol让 Dify 作为 MCP 客户端通过 fastmcp 暴露一个本地服务Agent 需要数据时直接调用工具函数函数内部再去远程执行命令、返回结构化结果。这样知识库可以完全不建Agent 拿到的是实时值还能自己组织语言给出分析建议。这篇就按「Dify fastmcp 搭建可复用 MCP 服务」的完整链路来写包含可复制的服务端骨架、Dify 工具配置片段、settings.json / config.toml 示例以及启动验证和调用排查。适合需要把内部工具接入 AI 工作流的开发者Python 3.10 以上即可跟做。2. 前置准备TaoToken 与运行环境在动手写代码前先把两件事定下来模型调用通道和本地依赖。模型侧我用 TaoToken 统一管理它的 API 地址是https://taotoken.net/api兼容 OpenAI 风格的调用方式Dify 里配置模型供应商时直接填这个 base_url 就行。如果你还没建 Key可以先去控制台生成一个后面 Dify 的 Agent 节点会用到。模型对话入口https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodel_chatutm_campaignrewriteAPI Keys 管理https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite接入文档https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite本地环境需要python3 --version # 确认 3.10 pip install fastmcp psutil paramikofastmcp负责把普通 Python 函数注册成 MCP 工具psutil采集本机指标paramiko用于远程执行命令。装完后可以先python3 -c import fastmcp; print(fastmcp.__version__)确认版本。注意远程执行账号密码不要硬编码在源码里示例里我会用环境变量占位你落地时务必替换成自己的密钥管理方式。3. 可复制的 fastmcp 服务端骨架3.1 server.py 主结构核心是用FastMCP创建实例再用mcp_server.tool装饰器注册工具。description 写得越清楚Dify 里的大模型越容易判断该调哪个工具。from fastmcp import FastMCP import psutil import os from remote_execute import RemoteQuickHandler mcp_server FastMCP( nameMy MCP Server, host0.0.0.0, port8002, ) mcp_server.tool(description计算两个整数相加用于验证 MCP 通道是否连通) def add(a: int, b: int) - int: return a b mcp_server.tool(description获取当前服务器 CPU 总使用率及前 10 高 CPU 进程) def get_cpu_usage() - dict: total_cpu psutil.cpu_percent(interval0.1) processes [] for proc in psutil.process_iter([pid, name, cpu_percent]): try: proc.cpu_percent() cpu_pct proc.cpu_percent(interval0.1) processes.append({ pid: proc.info[pid], name: proc.info[name], cpu_percent: cpu_pct, }) except (psutil.NoSuchProcess, psutil.AccessDenied): continue top_processes sorted(processes, keylambda x: x[cpu_percent], reverseTrue)[:10] return {total_cpu_usage: total_cpu, top_10_cpu_processes: top_processes} if __name__ __main__: mcp_server.run(transportsse)transportsse是关键Dify 的 Agent 节点目前对接 MCP 走的就是 SSE 通道两边必须一致。3.2 远程查询工具远程机器指标查询封装成独立工具入参是host_ip内部用 paramiko 执行命令并解析。mcp_server.tool(description查询目标服务器 CPU 使用率及前 10 高 CPU 进程) def get_host_cpu_usage(host_ip: str) - dict: handle RemoteQuickHandler( iphost_ip, usernameos.getenv(REMOTE_USER), passwordos.getenv(REMOTE_PASS), ) cmd_total vmstat 1 2 | tail -1 | awk {print 100 - $15} cmd_proc ps -eo user,comm,pcpu --sort-pcpu | head -n 11 status, result_total handle.command_with_result(cmd_total) status, result_proc handle.command_with_result(cmd_proc) if result_total[FAIL]: return {error: result_total[FAIL][0][msg]} if result_proc[FAIL]: return {error: result_proc[FAIL][0][msg]} cpu_usage float(result_total[SUCCESS][0][msg].strip()) lines result_proc[SUCCESS][0][msg].strip().split(\n) top_processes [] for line in lines[1:11]: parts line.strip().split() if len(parts) 3: top_processes.append({ user: parts[0], command: .join(parts[1:-1]), cpu_usage: parts[-1], }) return {cpu_usage: cpu_usage, top_processes: top_processes}内存和磁盘工具结构类似内存用free -b解析字节再转 GB磁盘用df -h配合单位换算函数。这里有个坑df -h输出单位可能是 G/M/T直接当 GB 用会算错必须写一个_convert_to_gb做归一化。3.3 remote_execute.py 封装import paramiko class RemoteQuickHandler: def __init__(self, ip, username, password, port22): self.ip ip self.username username self.password password self.port port def command_with_result(self, command: str) - tuple: client paramiko.SSHClient() client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) try: client.connect(self.ip, portself.port, usernameself.username, passwordself.password, timeout10) stdin, stdout, stderr client.exec_command(command) out stdout.read().decode(utf-8, errorsignore) err stderr.read().decode(utf-8, errorsignore) if err.strip(): return FAIL, {FAIL: [{msg: err.strip()}], SUCCESS: []} return OK, {FAIL: [], SUCCESS: [{msg: out}]} except Exception as e: return FAIL, {FAIL: [{msg: str(e)}], SUCCESS: []} finally: client.close()返回结构统一成{FAIL: [...], SUCCESS: [...]}上层工具函数只判断 FAIL 是否为空逻辑会清爽很多。4. Dify 侧配置与 settings.json / config.toml 示例4.1 Dify Agent 节点配置在 Dify 里新建一个 Chatflow添加 Agent 节点。如果找不到 Agent 或 Marketplace多半是版本太低建议 1.2 及以上。进入 Marketplace 安装 Agent 策略策略选择 FunctionCalling它支持 MCP 工具调用。然后在 Agent 节点的 MCP 服务配置里填入{ mcpServers: { ops-mcp: { transport: sse, url: http://127.0.0.1:8002/sse, timeout: 30 } } }如果你用的是 Dify 的 config.toml 方式管理可以写成[[mcp_servers]] name ops-mcp transport sse url http://127.0.0.1:8002/sse timeout 30查询节点选择开始节点的query变量这样用户输入会直接传给 Agent。工具列表在测试阶段可以留空实测下来删掉或禁用工具并不影响 MCP 服务调用具体作用可以后续再研究。4.2 模型供应商配置在 Dify 的模型供应商里新增 OpenAI 兼容类型base_url 填https://taotoken.net/apiAPI Key 填你在控制台生成的 Key。保存后测试连通性能返回模型列表就说明通道没问题。5. 启动验证与调用排查5.1 启动服务python3 server.py看到类似Uvicorn running on http://0.0.0.0:8002就说明 SSE 服务起来了。先用 curl 验证通道curl -N http://127.0.0.1:8002/sse正常会持续输出 SSE 事件流包含 endpoint 信息。如果卡住无输出检查端口是否被占用、防火墙是否放行。5.2 Dify 内测试在 Dify 点击发布更新进入预览聊天页输入「查一下 192.168.1.10 的 CPU 使用率」。Agent 会先调用get_host_cpu_usage拿到结构化数据后组织语言返回。如果返回的是原始 JSON 而不是自然语言说明 Agent 策略没选 FunctionCalling或者模型不支持工具调用。5.3 常见报错对照现象可能原因处理Dify 提示 MCP 连接失败url 路径不对确认是/sse而非根路径工具调用返回空description 太模糊补充工具用途和参数说明远程执行超时目标机不可达或账号错先用 ssh 手动验证连通性内存数值异常字节未转 GB检查free -b解析逻辑磁盘单位混乱df -h 单位未归一补_convert_to_gb函数6. 把 MCP 服务接进长期工作流一次跑通只是起点。如果你打算把这类 MCP 服务长期挂在 Agent 工作流里建议用 Coding Plan 管理调用额度避免临时 Key 过期导致线上中断Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite控制台https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite我自己的做法是把 server.py 拆成工具层和传输层工具层只关心业务逻辑传输层用 fastmcp 的 SSE 暴露。这样以后要加新工具只写一个带mcp_server.tool的函数就行Dify 侧不用改配置。另外远程执行账号一定走环境变量或密钥服务别图省事写死在代码里这是踩过坑之后最想提醒你的一点。