打通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 requests

def 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.text

async def handle_request(scope, receive, send):
    """
    处理HTTP请求的核心函数
    
    实现MCP协议的三个主要方法：
    1. initialize - 初始化握手
    2. tools/list - 返回可用工具列表
    3. tools/call - 执行工具调用
    """
    request = Request(scope, receive)
    path = request.url.path
    config = get_config(path)
    
    try:
        if scope["method"] == "POST":
            # 读取请求体
            body = b""
            more_body = True
            while 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 Workflow
                    result = 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)
                    return
                    
            except 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)
        return
        
    except 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" 
    }
  }
}

优势与效果

1. 资产复用：直接复用现有Dify Workflow，无需重新构建
2. 无缝集成：支持所有兼容MCP协议的AI工具
3. 灵活扩展：可轻松添加多个Workflow支持
4. 统一管理：工作流维护仍在Dify平台进行

总结

  在这个项目里，我做了一个转换工具，让公司现有 Dify 平台里的Workflow能够被其他支持 MCP 标准的应用调用。这解决了我们当前 Dify 版本的一个功能限制，让我们之前做好的工作流能继续发挥作用。其实，新版的 Dify 本身已经支持 MCP 标准了，那样集成会更方便。只是我们现在用的平台还没升级到新版本。所以，未来等平台升级了，我这个转换工具可能就用不着了。

  不过，做这个项目还是很有意义的：首先，它马上解决了我们眼下的问题，让工作流在新需求到来前就能投入使用。其次，动手做这个工具，让我对 MCP 标准具体是怎么运作的有了更深的了解，这对以后处理类似的技术对接肯定有帮助。对于其他团队，如果也遇到旧版 Dify 需要对接 MCP 应用的情况，这个方案可以作为一个临时的解决办法。另外，如果开发者想研究 MCP 标准的具体实现，我这个工具的代码也可以当作一个参考例子。