1. 为什么我要把 Dify 工作流塞进 MCP 里
Dify 工作流本身已经很好用了，可视化编排、节点调试、API 发布一条龙。但用久了会发现一个尴尬的地方：每次想让 AI 助手帮我跑一个工作流，都得手动打开 Dify 页面、点运行、复制输入、粘贴结果。如果一天要跑几十次，这种重复劳动就很折磨人。
MCP（Model Context Protocol）解决的正是这个问题。它本质上是一个标准接口，让大模型能直接调用外部工具。你可以把 Dify 工作流包装成一个 MCP 工具，然后在 Claude Desktop、ChatWise、Cherry Studio 这类支持 MCP 的客户端里，直接用自然语言触发工作流执行。比如你说一句"帮我查一下 gitbook 是什么"，背后就是 MCP Server 收到请求、转发给 Dify 工作流的 chat-messages 接口、拿到流式响应、再把结果回传给客户端。
这套方案适合谁？我总结了三类人：一是已经在用 Dify 做业务编排、想进一步降低操作成本的开发者；二是想学 MCP Server 开发、但苦于找不到真实场景练手的人；三是手里有多个 AI 工具、希望用统一 Key 和统一通道管理调用链的团队。本文会从零走一遍完整路径：搭 Dify 工作流、写 MCP Server、配置客户端、用 TaoToken 统一 Key 完成鉴权，最后做一次端到端联调验证。
需要提前说明的是，Dify 可以本地部署，也可以用官方云服务，接口地址改一下就行。我这边用的是本地部署版本，访问地址是http://127.0.0.1/v1/chat-messages。MCP Server 用 Python 写，依赖mcp库里的FastMCP。整个链路里最容易被忽略的是鉴权环节——Dify 自己的 API Key 和 MCP 客户端调用大模型用的 Key 是两回事，后面我会用 TaoToken 把这两层统一起来，避免 Key 散落在各个配置文件里。
2. TaoToken 前置准备：统一 Key 与 API 通道
在写代码之前，先把鉴权这层理清楚。很多教程会跳过这一步，直接让你把 Dify 的 API Key 硬编码在脚本里，结果就是：脚本里一个 Key、客户端配置里一个 Key、Dify 后台又一个 Key，时间一长自己都记不清哪个是哪个。更麻烦的是，如果 MCP Server 里还要调用大模型做二次处理，那就又多一层 Key。
我的做法是用 TaoToken 作为统一的 API 通道。它提供兼容 OpenAI 风格的接口，Base URL 是https://taotoken.net/api，你只需要在控制台生成一个 Key，就能同时用于模型对话和后续的 Coding Plan 场景。这样 MCP Server 里调用大模型、客户端里配置模型、Dify 工作流里如果需要外部模型节点，都可以指向同一个通道，Key 只维护一份。
具体操作路径是这样的：先打开官网https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=注册账号，然后进控制台https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_content=console&utm_campaign=rewrite创建 API Key。创建完之后，在 API Keys 页面https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api-keys&utm_campaign=rewrite可以看到完整的 Key 列表和用量统计。如果你只是想先验证模型能不能通，可以直接用模型对话页面https://taotoken.net/chat?utm_source=taotoken_aicg_blog_end&utm_content=model_chat&utm_campaign=rewrite发一条消息试试。
这里有个细节要注意：TaoToken 的 Key 和 Dify 自己的 API Key 是两个独立的东西。Dify 的 Key 用于访问 Dify 工作流的 chat-messages 接口，TaoToken 的 Key 用于访问大模型接口。在 MCP Server 里，我会把 Dify 的 Key 作为参数传入，而 TaoToken 的 Key 则通过环境变量注入，这样脚本本身不存任何敏感信息。
如果你后续要做长期编码或者 Agent 类任务，可以关注一下 Coding Plan 页面https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding_plan&utm_campaign=rewrite，它针对高频调用场景做了额度优化。接入文档在https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite，里面有完整的接口说明和示例。Claude Code 相关的接入说明在https://taotoken.net/claude-code-anthropic?utm_source=taotoken_aicg_blog_end&utm_content=claudecode&utm_campaign=rewrite，如果你用 Claude Code 做开发，可以参考那份配置。
3. 可复制配置：MCP Server 与 Dify 工作流对接
这一节是全文的核心，我会给出完整的 MCP Server 代码、Dify 工作流的关键节点参数，以及客户端的配置文件片段。你照着复制就能跑起来。
3.1 环境准备与项目初始化
先确认本地有 Python 3.10 以上版本，因为mcp库依赖这个版本。然后安装 uv，在 PowerShell 里执行：
powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"装完之后把 uv 的 bin 目录加到系统 Path 里。Win + R 输入sysdm.cpl，进"高级"→"环境变量"，在用户变量里找到 Path，新建一条C:\Users\{你的用户名}\.local\bin。确认后重开 PowerShell，执行uv --version能看到版本号就说明装好了。
接着创建项目目录：
uv init mcptool cd mcptool uv venv --python=python3.10 .venv\Scripts\activate uv add mcp httpx requests这几条命令做完，你会得到一个干净的虚拟环境，里面装好了mcp、httpx、requests三个库。mcp是核心，httpx和requests用于发 HTTP 请求。
3.2 Dify 工作流的关键节点配置
Dify 工作流的搭建不是本文重点，我简单说一下我用的那个基础工作流：用户输入 → 调用本地 SearXNG 查询 → 大模型整合 → 输出。你在 Dify 里搭好之后，进"访问 API"页面，能看到接口地址和 API Key。
关键参数有三个：接口地址是http://127.0.0.1/v1/chat-messages，请求方式是 POST，鉴权用Authorization: Bearer {api_key}。请求体里response_mode建议选streaming，因为 Dify 对阻塞式输出有超时限制，工作流节点一多就容易触发超时错误。流式输出虽然处理起来麻烦一点，但稳定性高很多。
3.3 MCP Server 完整代码
在项目目录下新建mcptool.py，把下面这段代码完整复制进去：
import requests import json import sys from typing import Any, Optional, Dict from requests.exceptions import RequestException, Timeout, ConnectionError from mcp.server.fastmcp import FastMCP mcp = FastMCP("mcptool") @mcp.tool() async def send_dify_chat_request( query: str, inputs: Optional[Dict[str, Any]] = None, user_id: str = "user123", api_key: Optional[str] = None, api_url: Optional[str] = None, response_mode: str = "streaming", timeout: int = 30 ) -> Dict[str, Any]: """ 向Dify API发送聊天请求并处理响应 Args: query: 用户的提问内容 inputs: 允许传入App定义的各变量值, 默认为空字典 user_id: 用户标识，用于定义终端用户身份，默认为"user123" api_key: Dify API密钥，如不提供则使用默认值 api_url: Dify API URL，如不提供则使用默认值 response_mode: 响应模式，可选"streaming"或"blocking"，默认为"streaming" timeout: 请求超时时间（秒），默认为30 Returns: Dict: 包含AI回答、使用情况等信息的字典 """ if api_key is None: api_key = "你的Dify API密钥" if api_url is None: api_url = "http://127.0.0.1/v1/chat-messages" print(f"向Dify API发送请求: 查询={query}, 用户={user_id}") headers = { "Authorization": f"Bearer {api_key}", "Content-Type": "application/json" } if inputs is None: inputs = {} data = { "inputs": inputs, "query": query, "response_mode": response_mode, "user": user_id } result = { "success": False, "answer": "", "usage_info": None, "error": None, "workflow_info": {} } try: print(f"发送请求到 {api_url}") response = requests.post(api_url, headers=headers, json=data, stream=True, timeout=timeout) if response.status_code == 200: full_answer = "" usage_info = None workflow_info = {} error_occurred = False try: for line in response.iter_lines(): if line: line = line.decode('utf-8') if line.startswith('data: '): data_str = line[6:] try: data_json = json.loads(data_str) event_type = data_json.get("event") if event_type == "message": full_answer += data_json.get("answer", "") elif event_type == "message_end": if "metadata" in data_json: usage_info = data_json.get("metadata", {}).get("usage") elif event_type == "error": error_occurred = True error_message = data_json.get("error", "未知错误") print(f"API错误: {error_message}") result["error"] = f"API错误: {error_message}" break elif event_type == "workflow_started": workflow_info["id"] = data_json.get("workflow_run_id") print(f"工作流开始: {workflow_info['id']}") elif event_type == "workflow_finished": status = data_json.get("data", {}).get("status") print(f"工作流完成: {status}") workflow_info["status"] = status if status == "failed": error_msg = data_json.get("data", {}).get("error", "未知错误") result["error"] = f"工作流执行失败: {error_msg}" elif event_type in ["node_started", "node_finished"]: node_id = data_json.get("data", {}).get("node_id") node_status = data_json.get("data", {}).get("status", "running") if event_type == "node_finished" and node_status == "failed": error_msg = data_json.get("data", {}).get("error", "未知错误") result["error"] = f"节点 {node_id} 执行失败: {error_msg}" except json.JSONDecodeError as e: result["error"] = f"JSON解析错误: {e}" except Exception as e: result["error"] = f"读取响应流时出错: {e}" if not error_occurred: print("AI回答:", full_answer) if usage_info: print(f"总令牌: {usage_info.get('total_tokens', 'N/A')}") result["success"] = not error_occurred result["answer"] = full_answer result["usage_info"] = usage_info result["workflow_info"] = workflow_info else: error_message = "未知错误" try: error_data = response.json() error_message = error_data.get("error", {}).get("message", str(error_data)) except (json.JSONDecodeError, ValueError, KeyError): error_message = response.text result["error"] = f"请求失败，状态码: {response.status_code}, 错误信息: {error_message}" if response.status_code == 401: result["error_type"] = "authentication_error" elif response.status_code == 403: result["error_type"] = "permission_error" elif response.status_code == 404: result["error_type"] = "not_found_error" elif response.status_code == 429: result["error_type"] = "rate_limit_error" elif 500 <= response.status_code < 600: result["error_type"] = "server_error" except Timeout: result["error"] = "请求超时: 服务器没有在预期的时间内响应" result["error_type"] = "timeout_error" except ConnectionError: result["error"] = "连接错误: 无法连接到服务器，请检查网络连接和服务器状态" result["error_type"] = "connection_error" except RequestException as e: result["error"] = f"请求异常: {e}" result["error_type"] = "request_error" except Exception as e: result["error"] = f"发生未预期的错误: {e}" result["error_type"] = "unexpected_error" return result if __name__ == "__main__": mcp.run(transport='stdio')这段代码的核心逻辑是：用FastMCP("mcptool")定义服务框架，用@mcp.tool()装饰器把send_dify_chat_request注册为工具。函数内部用requests.post发流式请求，逐行解析 SSE 事件，把message事件里的answer拼起来，最后返回一个结构化字典。mcp.run(transport='stdio')启动服务，通过标准输入输出和客户端通信。
3.4 客户端配置片段
如果你用 Claude Desktop，打开claude_desktop_config.json，加入下面这段：
{ "mcpServers": { "mcptool": { "command": "uv", "args": [ "--directory", "D:\\Desktop\\Claude-Files\\mcptool", "run", "mcptool.py" ] } } }注意--directory后面的路径要改成你自己存放脚本的位置。保存后重启 Claude Desktop，在工具列表里就能看到send_dify_chat_request这个工具。
如果你用 ChatWise 或 Cherry Studio，配置方式类似，只是命令格式略有不同。ChatWise 里填uv --directory D:\Desktop\Claude-Files\mcptool run mcptool.py就行。Cherry Studio 的 MCP 配置稍微绕一点，但把同样的命令填进去也能用。
4. 验证请求：一次端到端联调
配置写完之后，必须做一次完整的联调验证，否则你永远不知道是 MCP Server 没起来、还是 Dify 工作流没通、还是 Key 配错了。
第一步，先在终端里单独跑一下 MCP Server，确认脚本本身不报错：
cd D:\Desktop\Claude-Files\mcptool .venv\Scripts\activate python mcptool.py如果没有任何输出就卡住了，这是正常的，因为mcp.run(transport='stdio')在等客户端连接。按 Ctrl+C 退出。
第二步，在 Claude Desktop 里发一条消息，比如"帮我用 mcptool 查一下 gitbook 是什么"。Claude 会识别到需要调用send_dify_chat_request工具，弹出授权提示，你点允许。然后观察终端输出，应该能看到类似这样的日志：
向Dify API发送请求: 查询=gitbook是什么, 用户=user123 发送请求到 http://127.0.0.1/v1/chat-messages 工作流开始: xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx 工作流完成: succeeded AI回答: GitBook 是一个基于 Node.js 的文档协作平台... 总令牌: 1234第三步，回到 Claude Desktop，你应该能看到工具返回的结果被整合进了对话里。如果一切正常，说明整条链路是通的：Claude → MCP Server → Dify 工作流 → 返回结果 → Claude 展示。
这里有个验证技巧：先在 Dify 的"访问 API"页面用 curl 单独测一下工作流能不能通。命令是：
curl -X POST 'http://127.0.0.1/v1/chat-messages' \ -H 'Authorization: Bearer {你的Dify API Key}' \ -H 'Content-Type: application/json' \ -d '{"inputs": {}, "query": "gitbook是什么", "response_mode": "streaming", "user": "test"}'如果 curl 能返回流式数据，说明 Dify 侧没问题，问题就在 MCP Server 或客户端配置上。如果 curl 就报错，那先解决 Dify 的问题。
5. 本篇常见错排查
这一节我整理了实际踩过的坑，按报错信息分类，你对照着排查。
401 认证失败：最常见的原因是 Dify API Key 填错了，或者 Key 前面多了空格。检查mcptool.py里api_key的值，确认和 Dify"访问 API"页面里的一致。如果你用的是 TaoToken 的 Key 去调 Dify 接口，那肯定 401，因为这是两个不同的鉴权体系。Dify 的 Key 只用于 Dify 接口，TaoToken 的 Key 用于模型接口。
local proxy failed / connection refused：这个报错通常出现在客户端启动 MCP Server 的时候。原因是uv命令找不到，或者--directory路径写错了。先在 PowerShell 里手动执行一遍uv --directory 你的路径 run mcptool.py，看能不能跑起来。如果提示uv不是内部命令，说明 Path 没配好，回去检查环境变量。
reading choices 相关报错：这个一般出现在 MCP Server 内部调用大模型接口的时候。如果你在脚本里加了调用大模型的逻辑，检查 Base URL 是不是https://taotoken.net/api，Model ID 是不是填对了。TaoToken 的接口兼容 OpenAI 格式，但 Model ID 要用它支持的名称，具体可以在模型对话页面测试。
OAuth 相关报错：如果你用的是 Claude Code 或者某些需要 OAuth 的客户端，可能会遇到 token 过期的问题。这种情况重新走一遍授权流程就行。Claude Code 的接入说明在https://taotoken.net/claude-code-anthropic?utm_source=taotoken_aicg_blog_end&utm_content=claudecode&utm_campaign=rewrite，里面有完整的配置步骤。
工作流执行失败但 HTTP 200：这种情况说明请求发出去了，Dify 也收到了，但工作流内部某个节点报错了。看终端日志里node_finished事件的error字段，能定位到具体是哪个节点失败。常见原因是 SearXNG 没启动、或者大模型节点的 API Key 过期了。
流式响应解析出错：如果日志里出现JSON解析错误，大概率是 Dify 返回的数据格式和预期不一致。检查response_mode是不是streaming，以及 Dify 版本是否支持 SSE 格式。有些老版本 Dify 的流式格式略有不同，需要调整解析逻辑。
排查的时候记住一个原则：先隔离，再定位。先用 curl 测 Dify，再用终端测 MCP Server，最后才测客户端。每层都通了，整条链路才通。
6. 把统一 Key 用起来：接入文档与后续动作
走到这里，你已经有了一个能跑的 MCP Server，能通过自然语言操纵 Dify 工作流。但如果你想让这套东西真正用在日常开发里，还有两件事值得做。
第一件是把 Key 管理统一起来。现在mcptool.py里还硬编码着 Dify 的 API Key，这不安全。更好的做法是用环境变量注入，或者干脆把 Dify 的调用也走 TaoToken 的通道（如果你的 Dify 工作流里需要调用外部模型）。TaoToken 的接入文档在https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite，里面有环境变量配置的示例。API Keys 管理页面在https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api-keys&utm_campaign=rewrite，你可以在那里创建多个 Key，按项目隔离。
第二件是扩展工具集。现在只有一个send_dify_chat_request工具，你可以继续加：比如list_workflows用来列出所有工作流、get_workflow_status用来查执行状态、cancel_workflow用来中断执行。每加一个工具，就在mcptool.py里加一个@mcp.tool()装饰的函数，然后在客户端重启一下就能用。
如果你后续要做更复杂的 Agent 场景，比如让 AI 自动编排多个工作流、根据结果决定下一步调哪个工具，那可以考虑用 Coding Plan 的额度方案，它在高频调用下更划算。具体在https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding_plan&utm_campaign=rewrite可以看详情。
最后提醒一句：MCP Server 的调试日志默认打到 stdout，但 stdio 传输模式下 stdout 是给协议通信用 的，所以你的print语句可能会干扰通信。如果发现客户端连不上但脚本单独跑没问题，先把所有print改成写文件或者用sys.stderr。这个坑我踩过，排查了半天才发现是日志输出把协议数据冲掉了。
