使用Dify+fastmcp 实现mcp服务，内含详细步骤与源码

概要

使用dify+python-fastmcp完成mcp服务器，以实现运维智能客服免知识库进行运维数据获取与分析。

整体架构流程

问题： 配置本地知识库与dify知识库都是不小的工作量，而机器运维场景需要获取实时信息。

基本流程： 使用mcp服务器实现大模型实时根据接口返回数据进行分析，生成分析结果与建议再返回给用户。

场景详情： 本文给出一个运维使用场景，用户可以实时查询某台机器cpu，内存，磁盘使用情况。通过mcp服务器进行远程机器指令调用，将结果返回给agent，最后大模型根据返回结果自行整理语言进行返回

实现效果如下图：

技术细节

1.配置dify

如下图所示，简单在dify中配置mcp服务器配置信息，将dify作为mcp客户端使用。

1.1 添加节点

在dify中创建一个chatflow，然后添加Agent结点。

1.2 Agent节点配置

从这里进入Marketplace。如果发现自己dify没有Agent或者没有MarketPlace的，可能是dify版本过低。这里使用dify1.2版本。

1.3 Agent策略安装与配置

在Marketplace中安装Agent 策略

随后在Agent策略中选择刚刚安装的Agent策略，选择FunctionCalling支持MCP工具

1.4 Agent mcp服务配置

随后配置工具与MCP服务器配置，工具个人测试下来是可选配的。
后续再看看具体是什么功能。因为我把工具无论是删除还是禁用，也不影响mcp服务器的调用。如果有知道的大佬，可以评论区告知一下。

1.5 查询配置

查询这里就和其他的节点配置类似，选择开始的query就行。

2. 基于fastmcp进行mcp服务器搭建

环境：python >= 3.10 已安装fastmcp

2.1 创建server.py函数

使用FastMCP创建一个服务器，使用@mac_server.tool进行工具创建。
填写description进行描述，用来使用dify大模型更容易解析语言去定位使用哪个工具。
RemoteQuickHandler是使用paramiko实现远程调用目标机器类。
main函数中设置transport为sse对应上方dify agent配置中的transport 值
其他几个tool函数，是实现获取内存，cpu，磁盘信息的功能。

from fastmcp import FastMCP
import psutil
from remote_excute import RemoteQuickHandler
# 创建一个MCP服务器实例
mcp_server = FastMCP(name = "My MCP Server",port=8002 , host = "0.0.0.0")#path = "/xxx" 这里可以自行配置path，不填默认sse# 定义一个工具
@mcp_server.tool(description = "这是一个计算两个数加法函数")
def add(a: int, b: int) -> int:return a + b@mcp_server.tool(description = "这是一个获取当前服务器CPU使用率以及前10高CPU进程信息的函数")
def get_cpu_usage() -> dict:"""获取当前服务器CPU总使用率及前10高CPU进程信息"""# 1. 获取整体CPU使用率（间隔0.1秒采样）total_cpu = psutil.cpu_percent(interval=0.1)# 2. 获取所有进程的CPU使用率（需处理权限问题）processes = []for proc in psutil.process_iter(['pid', 'name', 'cpu_percent']):try:# 进程CPU使用率需要主动更新（部分系统需要）proc.cpu_percent()  # 第一次调用初始化cpu_pct = proc.cpu_percent(interval=0.1)  # 第二次调用获取实际值processes.append({"pid": proc.info['pid'],"name": proc.info['name'],"cpu_percent": cpu_pct})except (psutil.NoSuchProcess, psutil.AccessDenied):# 跳过已结束或无权限访问的进程continue# 3. 按CPU使用率降序排序，取前10top_processes = sorted(processes, key=lambda x: x['cpu_percent'], reverse=True)[:10]# 返回结构化数据return {"total_cpu_usage": total_cpu,"top_10_cpu_processes": top_processes}@mcp_server.tool(description = "这是一个查询目标服务器CPU使用率以及前10高CPU进程信息的函数")
def get_host_cpu_usage(host_ip: str) -> dict:"""获取目标服务器C