打通Dify与AI工具生态：将Workflow转为MCP工具的实践
背景
我们公司部署了自己的Dify平台，已有大量构建好的Workflow工作流。这些工作流涵盖了知识库检索、文档分析等核心业务场景，具有很高的复用价值。然而，我司的Dify平台本身并未提供直接将Workflow转换为MCP（Model Context Protocol）的功能，这限制了我们在各种支持MCP的AI工具中使用这些现有资产。
为了解决这个问题，我自己调研开发一个中间件，让Dify的Workflow能够以MCP工具的形式被各种AI应用调用。
什么是MCP？
首先简单介绍下MCP，MCP（Model Context Protocol）是Anthropic推出的开放协议，用于连接AI助手与各种数据源和工具，具体可以看下我之前的文章《一文入门AI圈最近爆火的MCP协议》。通过MCP，我们可以让AI工具访问本地文件、数据库、API等资源，大大扩展其能力边界。目前已有越来越多的AI应用开始支持MCP协议，包括Claude、各种AI开发框架等。
解决方案设计
我的解决思路是开发一个轻量级的HTTP服务，作为Dify Workflow和MCP协议之间的桥梁：
AI工具 ←→ MCP协议 ←→ 我们的服务 ←→ Dify API ←→ Workflow
完整代码实现
"""
Dify Workflow转MCP工具的中间件服务功能说明：
1. 将Dify平台的Workflow转换为符合MCP协议的工具
2. 支持多个Workflow并行服务，每个通过不同路径访问
3. 实现完整的MCP协议：初始化、工具列表、工具调用
4. 异步处理，支持高并发访问架构设计：
AI工具 ←→ MCP协议 ←→ 本服务 ←→ Dify API ←→ Workflow使用方法：
1. 修改get_config函数中的配置信息
2. 替换API_KEY为实际的Dify API密钥
3. 运行服务：python workflow_mcp_server.py
4. 在支持MCP的AI工具中配置服务地址
"""from starlette.applications import Starlette
from starlette.routing import Mount
from starlette.requests import Request
from starlette.responses import JSONResponse, Response
import uvicorn
import asyncio
import logging
import json
import requestsdef get_config(workflowId):"""获取Workflow配置信息参数说明：- name: 工具名称，会显示在AI工具中- prefix: 路径前缀，用于区分不同的workflow- tenantId: Dify租户ID- workflowId: Dify工作流ID- description: 工具描述，帮助AI理解工具用途注意：实际使用时应该从数据库或配置文件动态获取"""return {"name": "资金存管知识库检索","prefix": "04633c4f-8638-43a3-a02e-af23c29f821f","tenantId": "04633c4f-8638-43a3-a02e-af23c29f821f","workflowId": "WKFL-fa2bb193-3c2e-467f-ba39-5fc27ae4798a","description": "调用知识库获取资金存管相关信息"}def make_workflow_request(query, config):"""调用Dify Workflow API参数：- query: 用户输入的查询内容- config: workflow配置信息返回：- Workflow执行结果"""url = "https://dify.example.com/v1//workflow/run"payload = json.dumps({"tenantId": config['tenantId'],"workflowId": config['workflowId'],"inputs": {},"responseMode": "blocking", # 阻塞模式，等待结果返回"userId": "1000123","query": query})headers = {# 注意：这里需要替换为实际的API密钥'Authorization': 'Bearer ${API_KEY}','Content-Type': 'application/json'}try:response = requests.post(url, headers=headers, data=payload)print(f"Dify API响应: {response.text}")return json.loads(response.text)['data']['outputs']except Exception as e:logger.error(f"调用Dify API失败: {e}")return response.textasync def handle_request(scope, receive, send):"""处理HTTP请求的核心函数实现MCP协议的三个主要方法：1. initialize - 初始化握手2. tools/list - 返回可用工具列表3. tools/call - 执行工具调用"""request = Request(scope, receive)path = request.url.pathconfig = get_config(path)try:if scope["method"] == "POST":# 读取请求体body = b""more_body = Truewhile more_body:message = await receive()body += message.get("body", b"")more_body = message.get("more_body", False)try:request_data = json.loads(body.decode())logger.info(f"收到MCP请求: {request_data}")# MCP协议处理 - 初始化if request_data.get("method") == "initialize":"""MCP初始化握手返回协议版本、服务能力和服务器信息"""response_data = {"jsonrpc": "2.0","id": request_data.get("id"),"result": {"protocolVersion": "2024-11-05","capabilities": {"tools": {} # 声明支持工具调用},"serverInfo": {"name": config['name'],"version": "1.0.0"}}}response = JSONResponse(response_data)await response(scope, receive, send)return# MCP协议处理 - 工具列表elif request_data.get("method") == "tools/list":"""返回可用工具列表每个工具包含名称、描述和输入参数schema"""tools = [{"name": config.get('name'),"description": config['description'],"inputSchema": {'properties': {'query': {'description': '用户输入', 'type': 'string'}}, 'required': ['query'], 'type': 'object'}}]response_data = {"jsonrpc": "2.0","id": request_data.get("id"),"result": {"tools": tools}}response = JSONResponse(response_data)await response(scope, receive, send)return# MCP协议处理 - 工具调用elif request_data.get("method") == "tools/call":"""执行工具调用提取参数，调用Dify Workflow，返回结果"""tool_name = request_data["params"]["name"]arguments = request_data["params"].get("arguments", {})# 调用Dify Workflowresult = make_workflow_request(arguments['query'], config)response_data = {"jsonrpc": "2.0","id": request_data.get("id"),"result": {"content": [{"type": "text","text": str(result)}]}}logger.info(f"工具调用结果: {result}")response = JSONResponse(response_data)await response(scope, receive, send)returnexcept json.JSONDecodeError:response = JSONResponse({"error": "Invalid JSON"}, status_code=400)await response(scope, receive, send)return# 默认GET请求响应，用于健康检查response = JSONResponse({"status": "MCP Server Running", "prefix": config.get('prefix', ''),"service": "Dify Workflow to MCP Bridge"})await response(scope, receive, send)returnexcept Exception as e:logger.error(f"处理请求时出错: {e}")response = Response("Internal Server Error", status_code=500)await response(scope, receive, send)# 创建Starlette应用
app = Starlette(routes=[Mount("/", app=handle_request)
])# 配置日志
logging.basicConfig(level=logging.INFO,format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)async def main():"""启动服务器服务配置：- 端口：8081- 主机：0.0.0.0 (允许外部访问)- 访问地址示例：http://127.0.0.1:8081/04633c4f-8638-43a3-a02e-af23c29f821f"""logger.info("正在启动Dify Workflow MCP服务器...")logger.info("服务端口: 8081")logger.info("访问地址: http://127.0.0.1:8081/04633c4f-8638-43a3-a02e-af23c29f821f")config = uvicorn.Config(app=app,host="0.0.0.0",port=8081,log_level="info",access_log=True)server = uvicorn.Server(config)await server.serve()if __name__ == "__main__":try:asyncio.run(main())except KeyboardInterrupt:logger.info("服务器已被用户停止")except Exception as e:logger.error(f"服务器错误: {e}")
使用方法
1. 配置修改
- 在
get_config函数中更新你的Workflow信息 - 替换
make_workflow_request中的${API_KEY}为实际的Dify API密钥 - 根据需要修改Dify API地址
2. 启动服务
python workflow_mcp_server.py
3. 在AI工具中配置
{"mcpServers": {"dify-workflow": {"type": "streamableHttp","url": "http://127.0.0.1:8081/04633c4f-8638-43a3-a02e-af23c29f821f" }}
}
优势与效果
- 资产复用：直接复用现有Dify Workflow，无需重新构建
- 无缝集成：支持所有兼容MCP协议的AI工具
- 灵活扩展：可轻松添加多个Workflow支持
- 统一管理：工作流维护仍在Dify平台进行
总结
在这个项目里，我做了一个转换工具，让公司现有 Dify 平台里的Workflow能够被其他支持 MCP 标准的应用调用。这解决了我们当前 Dify 版本的一个功能限制，让我们之前做好的工作流能继续发挥作用。其实，新版的 Dify 本身已经支持 MCP 标准了，那样集成会更方便。只是我们现在用的平台还没升级到新版本。所以，未来等平台升级了，我这个转换工具可能就用不着了。
不过，做这个项目还是很有意义的：首先，它马上解决了我们眼下的问题，让工作流在新需求到来前就能投入使用。其次，动手做这个工具，让我对 MCP 标准具体是怎么运作的有了更深的了解，这对以后处理类似的技术对接肯定有帮助。对于其他团队，如果也遇到旧版 Dify 需要对接 MCP 应用的情况，这个方案可以作为一个临时的解决办法。另外，如果开发者想研究 MCP 标准的具体实现，我这个工具的代码也可以当作一个参考例子。