1. 从零搭建 MCP 服务操纵 Dify 工作流:为什么值得折腾
如果你已经在本地跑着 Dify,手里也攒了好几个工作流,那你大概率遇到过这种别扭:工作流本身挺好用,但每次都得打开 Dify 的调试页面,手动填参数、点运行、复制结果,再贴回聊天窗口。想让 Claude、ChatWise 这类客户端直接调用它,又发现没有一个统一的入口。MCP(Model Context Protocol)就是来解决这个问题的——它相当于给大模型装了一个标准插座,任何实现了 MCP 协议的服务都能被客户端识别成工具,模型自己决定什么时候调用、传什么参数。
这篇要做的,是用 uv 管理 Python 环境、用 FastMCP 写一个 MCP Server,把 Dify 的工作流包装成一个可被模型调用的工具。同时用 TaoToken 统一管理 Key,避免 Dify 的 API Key、模型 Key、客户端 Key 散落在四五个配置文件里。适合谁看:本地部署过 Dify、会一点 Python、想让 AI 客户端直接触发工作流的开发者。整套链路跑通后,你在 Claude 里说一句“帮我查一下 gitbook 是什么”,模型会自动调用你写的 MCP 工具,工具再去请求 Dify 工作流,最后把整合后的答案返回。
我试过把 Dify 的 Key 硬编码在脚本里,换台机器就得改代码,后来统一收到 TaoToken 才消停。下面按“环境准备 → 写 Server → 配置客户端 → 验证 → 排障”的顺序走一遍,每一步都能复制。
2. uv + FastMCP 环境准备与 Dify 工作流 API 对接
先说清楚依赖关系。MCP 的 Python SDK 要求 Python 3.10 及以上,低于这个版本mcp库装不上。uv 是 Astral 出的包管理器,比 pip 快很多,而且能直接管理虚拟环境和 Python 版本,特别适合这种“一个工具一个环境”的场景。Dify 这边,本地部署和官方云服务都行,区别只是 API 地址不同:本地一般是http://127.0.0.1/v1/chat-messages,云端换成你自己的域名。
2.1 安装 uv 并配置环境变量
Windows 下打开 PowerShell,执行官方安装脚本:
powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"装完之后 uv 的可执行文件在C:\Users\{你的用户名}\.local\bin,这个路径默认不在 PATH 里,得手动加。按Win + R输入sysdm.cpl回车,进“高级”→“环境变量”,在“用户变量”里找到 Path,双击,新建一条C:\Users\{你的用户名}\.local\bin。确定后关掉所有终端,重新开一个 PowerShell,验证:
uv --version能打印出版本号就说明 PATH 生效了。如果提示uv 不是内部或外部命令,八成是没重开终端,或者用户名写错了。
2.2 创建项目与虚拟环境
找个你放项目的目录,右键“在终端中打开”,执行:
uv init mcptool cd mcptool uv venv --python=3.10 .venv\Scripts\activateuv venv --python=3.10会拉一个 3.10 的解释器建虚拟环境,激活后命令行前面会出现(mcptool)前缀。接着装依赖:
uv add mcp httpx requests这里mcp是协议库,httpx和requests用来发 HTTP 请求。装完可以uv pip list看一眼,确认mcp版本在 1.x 以上。
2.3 在 Dify 里拿到工作流 API 信息
打开 Dify,进入你要暴露的工作流,点右上角“访问 API”。页面里有两个关键信息:一个是 API Key(形如app-xxxxxxxx),一个是请求地址。本地部署的地址通常是http://127.0.0.1/v1/chat-messages,注意/v1这个前缀别漏。请求体是 JSON,核心字段有四个:inputs(工作流定义的变量)、query(用户输入)、response_mode(streaming或blocking)、user(终端用户标识)。
这里有个坑:Dify 的blocking模式会等工作流全部跑完才返回,如果工作流里有搜索、大模型整合这类耗时节点,很容易撞上平台超时。所以下面代码统一用streaming,但在客户端侧把流读完再返回,兼顾稳定性和完整性。
3. 可复制的 FastMCP Server 配置与 Dify 调用参数
这一节是核心,直接给能跑的代码和配置。先看 MCP Server 的完整脚本,再讲客户端怎么接。
3.1 编写 mcptool.py
在项目根目录新建mcptool.py,内容如下。关键点:用FastMCP("mcptool")初始化服务,用@mcp.tool()把函数注册成工具,函数签名和 docstring 会被客户端读成工具说明。
import json from typing import Any, Optional, Dict import requests 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 工作流发送请求并返回整合后的答案。 Args: query: 用户提问内容 inputs: 工作流定义的变量字典,默认空 user_id: 终端用户标识 api_key: Dify API Key,不传则用默认值 api_url: Dify API 地址,不传则用默认值 response_mode: streaming 或 blocking timeout: 超时秒数 """ if api_key is None: api_key = "app-你的Dify密钥" if api_url is None: api_url = "http://127.0.0.1/v1/chat-messages" 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: 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 for line in response.iter_lines(): if not line: continue line = line.decode("utf-8") if not line.startswith("data: "): continue try: payload = json.loads(line[6:]) except json.JSONDecodeError: continue event = payload.get("event") if event == "message": full_answer += payload.get("answer", "") elif event == "message_end": usage_info = payload.get("metadata", {}).get("usage") elif event == "workflow_started": workflow_info["id"] = payload.get("workflow_run_id") elif event == "workflow_finished": status = payload.get("data", {}).get("status") workflow_info["status"] = status if status == "failed": error_occurred = True result["error"] = payload.get("data", {}).get("error") elif event == "error": error_occurred = True result["error"] = payload.get("error", "未知错误") result["success"] = not error_occurred result["answer"] = full_answer result["usage_info"] = usage_info result["workflow_info"] = workflow_info else: result["error"] = f"HTTP {response.status_code}: {response.text}" except Timeout: result["error"] = "请求超时" except ConnectionError: result["error"] = "连接失败,检查 Dify 是否在运行" except RequestException as e: result["error"] = f"请求异常: {e}" return result if __name__ == "__main__": mcp.run(transport="stdio")mcp.run(transport="stdio")是必须的,它让服务通过标准输入输出和客户端通信,这是本地 MCP 最常用的传输方式。
3.2 用 TaoToken 统一 Key 接入
上面代码里api_key是硬编码的,实际用起来不优雅。更稳的做法是把 Dify 的 Key、以及后续可能接的模型 Key 都收到 TaoToken 统一管理。TaoToken 的 API 地址是https://taotoken.net/api,你可以在它的控制台里创建 Key,然后让 MCP Server 从环境变量读取,而不是写死在代码里。
改法很简单,把api_key = "app-你的Dify密钥"换成:
import os if api_key is None: api_key = os.environ.get("DIFY_API_KEY", "")然后在启动客户端前设置环境变量。这样换机器、换工作流都不用动代码。如果你还想让 MCP 工具调用模型做二次加工,模型侧也走 TaoToken 的 Key,一个 Key 管住所有下游,省得在四五个平台之间来回切换。
3.3 客户端配置:Claude Desktop 与 ChatWise
Claude Desktop 的配置文件在claude_desktop_config.json,加入:
{ "mcpServers": { "mcptool": { "command": "uv", "args": [ "--directory", "D:\\Desktop\\Claude-Files\\mcptool", "run", "mcptool.py" ], "env": { "DIFY_API_KEY": "app-你的Dify密钥" } } } }注意--directory后面换成你自己的项目路径,Windows 下反斜杠要转义成\\。ChatWise 这类客户端配置逻辑一样,只是把命令写成一行:
uv --directory D:\Desktop\Claude-Files\mcptool run mcptool.pyCherry Studio 也能用,但它的 MCP 加载偶尔不如前两者顺滑,配置项填对之后多试几次即可。
4. 端到端验证:确认 Dify 工作流被 MCP 工具正常触发
配置写完,重启客户端。以 Claude Desktop 为例,重启后在输入框旁边能看到工具图标,点开应该列出send_dify_chat_request,说明 MCP Server 被成功加载。如果没看到,先看客户端日志,通常是路径写错或 uv 没在 PATH 里。
验证动作分两步。第一步,直接在对话里问一个需要工作流处理的问题,比如“帮我查一下 gitbook 是什么,用我的工作流”。模型会判断需要调用工具,弹出授权提示,确认后它会把query传给你的函数。第二步,观察返回。正常情况下你会看到模型先显示“正在使用 mcptool”,然后给出整合后的答案。同时你的 PowerShell 里如果手动跑过uv run mcptool.py,会看到向 Dify API 发送请求这类调试输出。
如果想脱离客户端单独测,可以在项目目录里跑一个临时脚本:
import asyncio from mcptool import send_dify_chat_request async def main(): r = await send_dify_chat_request(query="gitbook 是什么") print(r["success"], r["answer"][:200]) asyncio.run(main())返回True加上一段答案,就说明 Dify 工作流这条链路是通的。这一步能帮你把“客户端问题”和“服务端问题”分开定位。
5. 常见报错排查:401、local proxy failed 与 reading choices
实际搭的时候,报错基本集中在几个地方,对照着看能省不少时间。
401 认证失败:最常见。要么是 Dify 的 API Key 填错,要么是环境变量没传进去。先在 PowerShell 里echo $env:DIFY_API_KEY确认值存在,再检查 Dify 后台那个 Key 有没有被删。注意 Dify 的 Key 是app-开头,别和模型供应商的 Key 搞混。
local proxy failed / 连接失败:说明 MCP Server 起来了,但请求 Dify 的地址不通。本地部署的话确认 Dify 容器在跑,http://127.0.0.1/v1/chat-messages这个地址在浏览器里能访问到(会返回 405 之类,但至少不是拒绝连接)。如果 Dify 跑在 Docker 里而 MCP 跑在宿主机,地址可能得换成宿主机的局域网 IP。
reading choices / JSON 解析错误:这类报错通常出现在流式响应处理里。Dify 返回的每一行是data: {...},如果某行不是标准 JSON,json.loads会抛异常。上面的代码里已经用try/except包住了,遇到解析不了的行直接跳过。如果你自己改代码时去掉了这个保护,就会看到reading 'choices'之类的报错——那是模型侧在解析空响应。
OAuth / 授权弹窗不出现:有些客户端对 MCP 工具的授权是静默的,第一次调用可能不弹窗直接执行。如果一直没反应,检查客户端版本,老版本对 stdio 传输支持不完整。
工具列表为空:MCP Server 启动失败但客户端没报错。手动在终端跑uv run mcptool.py,如果卡住不动是正常的(它在等 stdio 输入),如果直接报错就是依赖或语法问题。另外确认mcp.run(transport="stdio")在__main__里被调用。
排查顺序建议:先手动跑通 Python 脚本 → 再确认客户端能列出工具 → 最后测端到端调用。每一步单独验证,比一上来就全链路调试高效得多。
6. 把 MCP 工具接进日常:Key 统一与后续扩展
链路跑通之后,真正省事的地方在于扩展。你现在有了一个 MCP Server 的骨架,想加新工具,无非是再写一个@mcp.tool()函数,把 Dify 的另一个工作流包进来,或者接别的 HTTP 服务。Key 这块,建议全部走 TaoToken 统一管理:Dify 的 Key、模型 Key、其他第三方服务的 Key 都从环境变量注入,代码里不出现明文。这样你换工作流、换模型、换机器,改的只是环境变量,不用碰代码。
如果你打算长期跑编码类或 Agent 类任务,可以考虑用 TaoToken 的 Coding Plan,把模型调用和工具调用都收在一条链路上,省得每个环节单独配鉴权。需要看具体接入方式的话,模型对话入口和接入文档里都有说明。工具本身不复杂,难的是把散落的 Key 和配置理顺,理顺之后,剩下的就是往里加工作流了。