1. Dify 工作流里接 MCP Server为什么 Key 会越配越乱如果你正在用 Dify 搭 Agent 或工作流大概率遇到过这种局面天气查询一个 Key、数据库查询一个 Key、内部知识库又一个 Key每个工具节点都要单独填一遍改一次配置要翻五六个地方。更麻烦的是当工具从 HTTP 接口换成 MCP Server(SSE) 之后SSE 长连接动不动就断Dify 的 Python 代码节点报一个Connection reset by peer你根本不知道是网络问题、端口问题还是 Key 没传对。这篇就聚焦一个具体场景在 Dify 工作流里用 Python 代码节点调用自建的 MCP Server(SSE)并且用 TaoToken 统一管理 Key。我会给你一份可以直接抄的config.toml/settings.json骨架、一段 Dify Python 代码节点能跑的 SSE 客户端片段再加一次curl连通性验证动作。目标很明确10 分钟内让 Dify → MCP Server(SSE) 这条链路跑通而不是卡在“Key 填哪儿”和“SSE 为什么又断了”上。先说清楚 MCP 是什么。MCPModel Context Protocol是一套让模型和外部工具对话的协议你可以把它理解成“工具的 USB-C 接口”——以前每个工具要写一套对接代码现在只要工具实现了 MCP任何支持 MCP 的客户端都能即插即用。SSEServer-Sent Events是它的一种传输方式服务端主动往客户端推消息适合需要实时返回的工具调用。适合谁看已经在用 Dify 做 Agent、手里有自建 Python 工具服务、想让多个工具共用一套 Key 管理的人。如果你还没搭过 Dify建议先把 Dify 跑起来再回来这篇不重复讲部署。2. TaoToken 前置统一 Key 到底解决了什么在讲配置之前得先说明白为什么要引入 TaoToken。传统做法是每个 MCP Server 自己管一套鉴权Dify 里每个工具节点填一个 Key。工具一多Key 就散落在各个节点的配置里谁改过、哪个过期了全靠记忆。TaoToken 在这里的角色是统一入口你只需要在 TaoToken 侧维护一份 KeyDify 的 Python 代码节点通过它去访问后端模型或工具服务不用在每个节点里重复填。它的 API 地址是https://taotoken.net/api控制台和文档分别在控制台https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentmcp_sse_difyAPI Keys 管理https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentmcp_sse_dify接入文档https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentmcp_sse_dify注意TaoToken 是统一的 Key 与调用入口不是让你绕过任何合规流程。所有调用仍然走正常鉴权只是把分散的 Key 收敛到一处管理。具体到本篇场景统一 Key 带来两个直接好处。第一Dify 的 Python 代码节点里只需要读一个环境变量不用硬编码多个 Key第二当 MCP Server 需要调用后端模型时走的是同一套凭证SSE 连接建立时的鉴权头也统一了排查问题时只需要看一个地方。如果你后面要做长期编码或 Agent 任务可以了解下 Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentmcp_sse_dify3. 可复制配置config.toml 与 settings.json 骨架这一节给你两份骨架一份给 MCP Server 侧Python 服务读的config.toml一份给 Dify 侧Python 代码节点读的settings.json。两份都只保留必要字段你按自己的环境改。先看 MCP Server 侧的config.toml。假设你的天气 MCP Server 用 fastmcp 写服务启动时需要知道监听地址、端口以及调用后端时用的统一 Key# config.toml —— MCP Server 侧配置 [server] name weather-server transport sse host 0.0.0.0 port 9000 [auth] # 统一 Key 从环境变量读取不写死在文件里 api_key_env TAOTOKEN_API_KEY base_url https://taotoken.net/api [logging] level INFO这里的关键点是api_key_env它指向环境变量名而不是 Key 本身。这样config.toml可以进版本库Key 留在运行环境里。启动服务前先导出export TAOTOKEN_API_KEY你的Key python mcp_weather_fastmcp.py再看 Dify 侧的settings.json。Dify 的 Python 代码节点运行在沙箱里读不到你宿主机的环境变量所以要把连接信息通过节点配置传进去。这份骨架放在代码节点里作为常量读取{ mcp_server: { name: weather-server, sse_url: http://host.docker.internal:9000/sse, timeout_seconds: 30, retry: 3 }, taotoken: { base_url: https://taotoken.net/api, api_key_env: TAOTOKEN_API_KEY } }sse_url里的host.docker.internal是重点。Dify 跑在 Docker 里容器内的localhost指向容器自己不是宿主机。所以访问宿主机上的 MCP Server 必须用host.docker.internal。如果你在 Linux 上跑 Docker这个域名可能不生效需要加--add-hosthost.docker.internal:host-gateway启动参数或者直接用宿主机内网 IP。提示timeout_seconds和retry是给 SSE 连接用的。SSE 是长连接网络抖动容易断设一个合理的重试次数能省很多事。4. Dify Python 代码节点SSE 客户端可复制片段现在进入核心部分。Dify 的 Python 代码节点里你要写一段能连上 MCP Server(SSE) 的客户端。下面这段可以直接复制改一下 URL 和 Key 环境变量名就能用。import json import os import requests # 从节点配置读取实际使用时替换为你的值 SSE_URL http://host.docker.internal:9000/sse TAOTOKEN_BASE https://taotoken.net/api API_KEY os.environ.get(TAOTOKEN_API_KEY, ) def call_mcp_tool(tool_name: str, arguments: dict) - dict: 通过 SSE 调用 MCP Server 的工具。 先建立 SSE 连接拿到 session再发工具调用请求。 headers { Accept: text/event-stream, Authorization: fBearer {API_KEY}, } # 第一步建立 SSE 连接读取服务端推送的 endpoint 事件 session requests.Session() resp session.get(SSE_URL, headersheaders, streamTrue, timeout30) if resp.status_code ! 200: return {error: fSSE 连接失败状态码 {resp.status_code}} endpoint None for line in resp.iter_lines(decode_unicodeTrue): if line and line.startswith(data:): payload line[5:].strip() try: event json.loads(payload) except json.JSONDecodeError: continue if event.get(type) endpoint: endpoint event.get(endpoint) break if not endpoint: return {error: 未从 SSE 拿到 endpoint} # 第二步向 endpoint 发工具调用请求 call_url fhttp://host.docker.internal:9000{endpoint} body { jsonrpc: 2.0, id: 1, method: tools/call, params: {name: tool_name, arguments: arguments}, } call_resp session.post(call_url, jsonbody, headersheaders, timeout30) return call_resp.json() # Dify 代码节点入口 def main(city: str) - dict: result call_mcp_tool(get_weather, {city: city}) return {result: result}这段代码的逻辑分两步先 GET 建立 SSE 连接从服务端推来的事件里拿到endpoint再 POST 到那个 endpoint 发 JSON-RPC 格式的工具调用。这是 MCP SSE 传输的标准流程不是我自己编的。几个容易踩的点。第一streamTrue必须加否则iter_lines拿不到流式数据。第二Accept: text/event-stream头要带上有些服务端会检查。第三Authorization头里的 Key 从环境变量读别硬编码。第四Dify 代码节点的超时时间要设得比timeout_seconds大不然节点先超时了。如果你想让模型自己决定调哪个工具而不是在代码里写死get_weather那更适合用 Dify 的 MCP 工具配置界面而不是代码节点。代码节点适合你已经明确知道要调什么、需要精细控制的场景。5. 验证请求一次 curl 连通性验证配置写完别急着在 Dify 里跑先用curl验证 MCP Server 本身是通的。这一步能帮你把“服务端问题”和“Dify 配置问题”分开。在宿主机上执行curl -N -H Accept: text/event-stream \ -H Authorization: Bearer $TAOTOKEN_API_KEY \ http://localhost:9000/sse-N是关闭缓冲让你实时看到 SSE 推送。正常的话你会看到类似这样的输出event: endpoint data: {type:endpoint,endpoint:/messages?session_idabc123}拿到endpoint之后再发一次工具调用curl -X POST http://localhost:9000/messages?session_idabc123 \ -H Content-Type: application/json \ -H Authorization: Bearer $TAOTOKEN_API_KEY \ -d {jsonrpc:2.0,id:1,method:tools/call,params:{name:get_weather,arguments:{city:beijing}}}如果返回里能看到北京的天气数据说明 MCP Server 和统一 Key 都没问题。接下来把localhost换成host.docker.internal在 Dify 容器里再验一次。这一步过了Dify 代码节点基本就能跑通。注意session_id是动态的每次建立 SSE 连接都会变别把它写死在配置里。6. 本篇常见错排查报错一Connection refused或Failed to establish a new connection九成是地址写错了。Dify 在 Docker 里localhost指向容器自己。改成host.docker.internal。Linux 上如果这个域名不解析检查 Docker 启动参数有没有加host-gateway。报错二SSE 连上了但拿不到 endpoint看服务端日志。fastmcp 的mcp.run(transportsse, ...)默认会推 endpoint 事件如果没推可能是版本问题升级fastmcp到最新。另外确认Accept头带了text/event-stream。报错三401 UnauthorizedKey 没传对。检查三处环境变量有没有导出、Dify 代码节点里os.environ.get的变量名和导出的一致、Authorization头的格式是Bearer key而不是别的。报错四SSE 连接几分钟后自动断这是长连接的正常现象服务端或中间网络会回收空闲连接。在客户端加心跳或重试逻辑settings.json里的retry就是干这个的。如果断得太频繁检查有没有反向代理在超时回收。报错五Dify 代码节点超时Dify 节点默认超时可能比你的 SSE 连接短。把节点的超时调大或者把timeout_seconds调小让客户端先返回。排查顺序建议先curl宿主机 → 再curl容器内 → 最后 Dify 代码节点。一层层往上别一上来就怀疑 Dify。7. 把 Key 收口把链路跑通回到最开始的问题多工具 Key 分散、SSE 连接易断。这篇给的方案是用 TaoToken 把 Key 收口到一处MCP Server 侧通过config.toml读环境变量Dify 侧通过settings.json传连接信息Python 代码节点用标准 SSE 客户端调工具。链路是 Dify → Python 代码节点 → MCP Server(SSE) → 后端服务每一段都有对应的验证动作。如果你在排障或接入阶段卡住了先去 API Keys 页面确认 Key 状态再对照接入文档检查请求格式API Keyshttps://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentmcp_sse_dify接入文档https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentmcp_sse_dify如果你只是想先验证模型能不能正常对话用模型对话页面最快https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentmcp_sse_dify如果你后面要做长期编码或 Agent 任务Coding Plan 更适合https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentmcp_sse_dify最后留一个我实际踩过的坑Dify 代码节点里requests的streamTrue如果忘了加iter_lines会一次性读完然后卡住表现是节点一直转圈不返回。加上之后立刻正常。这种问题看日志看不出来只能靠对 SSE 流程的理解去定位。