1. 先搞清楚WorkBuddy、MCP 和「腾讯混元生图」这三者到底在什么层面打交道很多人一看到“给 WorkBuddy 接入自定义 MCP 连接器”第一反应是这又是个套壳包装的 AI 工具链其实不是。WorkBuddy 的底层定位是一个面向开发者工作流的智能体运行时环境——它不自己造模型也不硬编码能力而是通过一套标准化的协议把外部服务的能力“插”进来。这个协议就是 MCPModel Context Protocol。MCP 不是 API也不是 SDK更不是某种硬件接口协议网上有人混淆成“MCP 是硬件协议”这是典型误解。它本质是一套轻量级、面向语义交互的 JSON-RPC 扩展规范核心目标就一个让 AI 智能体在执行任务时能像调用本地函数一样安全、可追溯、带上下文地调用外部服务。比如你让 WorkBuddy “生成一张春天樱花盛开的海报”它内部不会自己画图而是通过 MCP 协议把“生成图像”这个意图结构化为一个generate_image方法调用连同 prompt、尺寸、风格等参数发给后端某个图像服务。而这个后端服务就是我们今天要接入的「腾讯混元生图」云托管实例。关键点来了腾讯混元生图本身提供的是 HTTP RESTful 接口支持同步返回或 SSE 流式响应。但 WorkBuddy 的 MCP 客户端只认一种通信方式基于 WebSocket 的 MCP Server。它不直接消费原始的 HTTP 接口而是要求你部署一个中间层——一个实现了 MCP Server 规范的网关服务。这个网关干两件事一是监听 WorkBuddy 发来的 MCP 请求比如call_tool二是把请求翻译成对腾讯混元生图 API 的调用并把结果按 MCP 格式封装回传。所以“接入自定义 MCP 连接器”的真实含义是你得亲手写一个 MCP Server它既是 WorkBuddy 的“听令者”又是腾讯混元生图的“代笔人”。这个 Server 不是配置出来的是代码写出来的它不依赖任何现成 SDK因为目前官方并没有发布“腾讯混元 MCP”的一体化适配包。你得自己搭桥。我第一次试的时候直接把混元的 API 文档 URL 粘贴进 WorkBuddy 的 MCP 配置框填上wss://api.xiaozhi.me/mcp/?token...这种地址结果 WorkBuddy 报错Connection refused。后来才明白那个wss://地址根本不是混元的而是某个第三方 MCP 中继服务的地址和腾讯混元毫无关系。这种混淆在社区里非常普遍也是新手最容易卡住的第一道墙。提示MCP Server 必须是 WebSocket 服务wss://或ws://且必须实现listTools、callTool等标准方法。它和你调用的后端 API如混元是解耦的。你可以用 Python 写也可以用 Java、Node.js甚至 Rust只要它能跑通 MCP 协议就行。2. 为什么选 SSE 云托管——不是为了“流式”而是为了“可控性”与“合规性”标题里特意强调了「SSE 云托管」这绝不是为了赶时髦。很多教程一上来就教你怎么用 WebSocket 实现实时推送但在这里SSEServer-Sent Events才是更务实、更稳定的选择。原因有三层层层递进第一层是技术匹配度。腾讯混元生图的官方 API明确支持两种响应模式sync同步阻塞和sse服务端事件流。sync模式简单粗暴但有个致命缺陷当图片生成耗时较长比如复杂 prompt 要 8~12 秒WorkBuddy 的 MCP 客户端会因超时而断开连接报出stream disconnected before completion: idle timeout waiting for sse这个错误。这不是你的代码问题是 WorkBuddy 内置的默认超时策略通常为 5 秒和混元实际响应时间不匹配导致的。而 SSE 天然支持长连接、分段推送混元在生成过程中会持续发送event: progress、event: result等事件客户端可以逐帧接收、实时渲染进度条最后拿到完整 Base64 图片。这完美绕开了超时陷阱。第二层是架构清晰度。SSE 是单向服务端→客户端的 HTTP 协议比 WebSocket 更轻量、更易调试。你不需要维护复杂的连接状态、心跳保活、重连逻辑。一个简单的 HTTP POST 请求发过去然后打开一个 EventSource 连接等着收数据整个流程就像读取一个不断增长的日志文件。我在做 PoC概念验证时先用curl -N命令手动测试混元的 SSE 接口确认能稳定收到data: {progress: 50%}和data: {result: base64...}之后再把这段逻辑封装进 MCP Server心里就有底了。如果一开始就啃 WebSocket光是解决connection reset by peer这类网络抖动问题就能耗掉你两天。第三层是部署合规性。所谓“云托管”指的是把你的 MCP Server 部署在腾讯云函数 SCFServerless Cloud Function或轻量应用服务器 Lighthouse 上而不是本地开发机。SCF 对 HTTP 请求有严格的冷启动和执行时间限制最长 300 秒但它对 SSE 的支持非常友好。你只需要在 SCF 函数里开启一个 HTTP 服务监听/mcp路径当 WorkBuddy 的 WebSocket 连接上来后你的函数就启动一个后台协程去调用混元的 SSE 接口并把收到的每一个data:事件转换成 MCP 的tool_call_result消息通过 WebSocket 回推给 WorkBuddy。整个过程SCF 只负责“转发”不承担模型推理压力成本极低且天然符合企业级安全审计要求——所有流量都走 HTTPS密钥管理由云平台统一处理。注意不要被wss://api.xiaozhi.me/mcp/这类地址误导。那是一个公开的 MCP 中继 Demo它的后端可能是任意模型且 token 有效期极短。生产环境必须用自己的云托管实例绑定自己的腾讯云 API 密钥SecretId/SecretKey并做好 token 鉴权。3. 从零手写 MCP ServerPython FastAPI httpx 的最小可行方案现在进入实操环节。我们放弃一切花哨框架用最精简的技术栈写出一个能跑通的 MCP Server。选型理由很实在Python 生态对 HTTP/SSE 支持最成熟FastAPI 自带异步 HTTP 服务和 WebSocket 支持httpx 是目前 Python 里对 SSE 流式解析最友好的 HTTP 客户端。整个代码不到 200 行但覆盖了所有核心路径。3.1 初始化项目与依赖新建一个目录workbuddy-mcp-hunyuan执行pip install fastapi uvicorn httpx python-dotenv创建main.py这是整个服务的入口。我们不追求高并发先确保单连接、单请求能闭环。3.2 定义 MCP 工具描述mcp.json 的灵魂MCP 协议要求 Server 在启动时必须能响应listTools请求返回一个 JSON 数组描述它能提供哪些工具。这个数组就是mcp.json文件的实质内容。很多人以为mcp.json是一个静态配置文件其实它是 Server 动态生成的契约。我们的mcp.json就一条工具{ name: hunyuan_generate_image, description: 使用腾讯混元大模型生成指定描述的图像, input_schema: { type: object, properties: { prompt: { type: string, description: 图像生成的中文或英文提示词越详细越好 }, size: { type: string, enum: [1024x1024, 768x1024, 1024x768], default: 1024x1024, description: 输出图像尺寸 } }, required: [prompt] } }注意input_schema的写法它直接决定了 WorkBuddy 在 UI 上如何生成表单。enum字段会让 WorkBuddy 给你一个下拉菜单而不是一个开放文本框极大降低用户输错的风险。3.3 核心逻辑WebSocket 连接 SSE 转发这是最关键的 80 行代码。FastAPI 的 WebSocket 路由如下app.websocket(/mcp) async def mcp_websocket(websocket: WebSocket): await websocket.accept() try: while True: # 1. 接收 WorkBuddy 发来的 MCP 请求 data await websocket.receive_json() if data.get(method) listTools: # 返回工具列表 await websocket.send_json({ jsonrpc: 2.0, id: data.get(id), result: [HUNYUAN_TOOL_SCHEMA] }) elif data.get(method) callTool: # 2. 解析调用参数 params data.get(params, {}) prompt params.get(prompt, ) size params.get(size, 1024x1024) # 3. 构造混元 SSE 请求 headers { Authorization: fBearer {os.getenv(HUNYUAN_API_KEY)}, Content-Type: application/json } payload { model: hunyuan-vision, prompt: prompt, size: size, stream: True } # 4. 异步发起 SSE 请求并实时转发 async with httpx.AsyncClient() as client: async with client.stream(POST, HUNYUAN_SSE_URL, jsonpayload, headersheaders) as response: if response.status_code ! 200: await websocket.send_json({ jsonrpc: 2.0, id: data.get(id), error: {code: -32000, message: f混元API调用失败: {response.status_code}} }) break # 5. 逐行解析 SSE 数据流 async for line in response.aiter_lines(): if line.strip() : continue if line.startswith(data:): try: # 去掉 data: 前缀解析 JSON json_str line[6:].strip() event_data json.loads(json_str) # 6. 将混元事件映射为 MCP 结果 if result in event_data and url in event_data[result]: # 最终结果返回图片 URL await websocket.send_json({ jsonrpc: 2.0, id: data.get(id), result: {image_url: event_data[result][url]} }) elif progress in event_data: # 进度更新作为 MCP 的 partial result await websocket.send_json({ jsonrpc: 2.0, id: data.get(id), result: {progress: event_data[progress]} }) except json.JSONDecodeError: pass # 忽略非 JSON 行如注释 except WebSocketDisconnect: print(WorkBuddy 客户端断开连接) except Exception as e: print(f处理请求时出错: {e})这段代码的精妙之处在于第 5 步和第 6 步httpx.AsyncClient.stream()能真正实现异步流式读取aiter_lines()保证了你不会等到整个 SSE 响应结束才开始处理。每收到一行data: {...}就立刻解析、映射、回推。WorkBuddy 的 UI 就能实时显示“生成中30%”、“生成中75%”最后弹出图片。这种体验是同步 API 永远无法提供的。3.4 启动与调试让 WorkBuddy 真正“看见”你的 Server启动服务uvicorn main:app --host 0.0.0.0 --port 8000 --reload此时你的 MCP Server 监听在http://localhost:8000/mcp。但 WorkBuddy 要求的是wss://所以你需要一个反向代理。最简单的方法是用ngrokngrok http 8000它会给你一个类似https://abc123.ngrok.io的公网地址。把这个地址加上/mcp路径填入 WorkBuddy 的 MCP 设置里格式为wss://abc123.ngrok.io/mcp注意是wssngrok 会自动升级。提示本地调试时WorkBuddy 的日志面板CtrlShiftI 打开 DevTools → Console是你的最佳朋友。每一次连接、每一次listTools请求、每一次callTool都会打印出来。如果看到Failed to connect to MCP server90% 的概率是你的 ngrok 地址没填对或者 FastAPI 服务根本没起来。先curl http://localhost:8000/mcp看是否返回 405 Method Not Allowed说明服务起来了再查 ngrok 日志。4. WorkBuddy 端的配置与技能编排让“生图”成为一条自然指令Server 端写好了只是完成了 50%。剩下 50%是让 WorkBuddy 理解、信任并流畅地使用它。这一步很多人草草跳过结果发现 WorkBuddy 死活不调用你的工具或者调用了但返回乱码。4.1 MCP 连接器配置的三个生死细节在 WorkBuddy 的设置 → MCP → 添加新连接器中填写以下信息Name:腾讯混元生图 (SSE)URL:wss://abc123.ngrok.io/mcp你的 ngrok 地址Authentication: 选择None因为我们没加鉴权或Token如果你在 Server 端加了 Bearer Token 校验这三个字段任何一个填错都会导致连接失败。特别注意 URL必须以wss://开头不能是http://路径必须是/mcp不能是/api/mcp或其他末尾不能有多余的/。我曾因为多打了一个斜杠调试了 40 分钟。4.2 创建 Skill把工具变成一句自然语言WorkBuddy 的核心价值是让 AI 用自然语言驱动工具。所以你不能只停在“能调用”这一步必须创建一个 Skill。点击 WorkBuddy 主界面右上角的 New Skill填入Name:生成图片Description:根据我的文字描述生成一张高质量的图片Trigger:When I say 生成一张... or 画一幅... or 给我看看...这里用正则表达式生成一张.*|画一幅.*|给我看看.*Action:Call tool: hunyuan_generate_imageParameters Mapping: 这里最关键把用户说的话自动提取成工具参数。prompt:{{trigger.text}}把整句话作为 promptsize:1024x1024固定值或设为{{user.input.size}}让用户选择保存后你就可以对 WorkBuddy 说“生成一张穿着宇航服的橘猫在月球表面跳跃的图片”它会自动识别触发词提取整句话作为prompt调用你的 MCP Server最终把混元生成的图片 URL 返回给你。4.3 调试 Skill 的黄金法则看tool_call日志Skill 不生效别急着改代码。打开 WorkBuddy 的Activity Log活动日志找到你刚说的那句话。如果 Skill 被正确触发你会看到一条tool_call记录里面包含完整的params。如果params里prompt是空的说明你的 Parameter Mapping 写错了如果根本没这条记录说明 Trigger 的正则没匹配上或者 Skill 没启用。这是最直接、最不可辩驳的证据链。我踩过的一个深坑是我把 Trigger 写成了生成一张.*图片结果用户说“生成一张橘猫”就没触发。后来改成生成一张.*覆盖所有可能性问题立刻解决。经验是Trigger 正则宁宽勿窄参数提取逻辑放在Parameters Mapping里做精细化处理。5. 生产环境加固从本地 Demo 到企业级可用的五道防线一个能跑通的 Demo 和一个能上线的生产服务中间隔着五道墙。这五道墙不是可选项而是必选项。我在线上环境部署时每一道都亲自验证过。5.1 防线一密钥安全——绝不硬编码必须用环境变量 云密钥管理HUNYUAN_API_KEY这种敏感信息绝对不能写死在main.py里也不能放在.env文件里提交到 Git。正确做法是本地开发用.env文件HUNYUAN_API_KEYyour_real_key_here云托管SCF在函数配置里添加环境变量HUNYUAN_API_KEY值从腾讯云 KMS密钥管理系统中获取。SCF 函数启动时会自动解密并注入到环境变量中。这样你的代码里永远只写os.getenv(HUNYUAN_API_KEY)密钥本身从未出现在任何代码或配置文件中。5.2 防线二连接池与超时——避免混元 API 拒绝服务混元 API 有 QPS 限制。如果你的 MCP Server 没有连接池每个请求都新建 TCP 连接很容易触发限流。在httpx.AsyncClient初始化时必须配置limits httpx.Limits(max_connections20, max_keepalive_connections10) timeout httpx.Timeout(30.0, connect10.0, read30.0) async with httpx.AsyncClient(limitslimits, timeouttimeout) as client:connect10.0确保 DNS 解析和 TCP 握手不超过 10 秒read30.0给混元留足生成时间max_connections20控制并发数防止雪崩。5.3 防线三错误重试与降级——当混元挂了WorkBuddy 不能卡死SSE 流可能因网络中断而断开。你的代码里必须有重试逻辑for attempt in range(3): try: async with client.stream(...) as response: # ... 处理流 ... break # 成功则跳出循环 except (httpx.ConnectTimeout, httpx.ReadTimeout): if attempt 2: await websocket.send_json({... error: 重试三次后仍失败 ...}) break await asyncio.sleep(1) # 指数退避更高级的降级是当混元不可用时自动 fallback 到一个本地缓存的“占位图”URL或者返回一段友好的提示“图像生成服务暂时繁忙请稍后再试”。5.4 防线四日志与监控——没有日志的线上服务等于盲人开车在main.py里每一处关键节点都要打日志INFO: “收到 listTools 请求”INFO: “收到 callTool 请求prompt: [前20字]...”WARNING: “混元 API 返回 429已触发限流”ERROR: “SSE 流解析 JSON 失败原始行: {line}”这些日志全部通过logging模块输出并在 SCF 控制台里配置日志投递到 CLS云日志服务。当用户反馈“图片没出来”你不用问东问西直接去 CLS 查关键词hunyuan_generate_image5 秒内定位到是网络问题、密钥问题还是混元侧问题。5.5 防线五版本兼容——MCP 协议在快速迭代你的 Server 必须跟上MCP 协议不是一成不变的。WorkBuddy 新版本可能会增加cancelTool方法或者修改tool_call_result的字段名。你的 Server 必须有版本号并在listTools返回中声明{ name: hunyuan_generate_image, version: 1.2.0, description: ..., ... }同时在代码里用if data.get(jsonrpc) 2.0做基础校验对未知 method 返回标准的Method not found错误。这样即使未来协议升级你的 Server 也不会崩溃只会优雅地告知 WorkBuddy “这个功能我还不支持”。最后分享一个血泪教训我在一次 WorkBuddy 更新后发现所有 MCP 连接都失败了。查日志发现新版本在callTool请求里多加了一个context字段而我的旧版 Server 解析params时直接params.get(prompt)忽略了context的存在导致整个params字典被当成None。修复方法很简单params data.get(params, {})永远给get加默认值。这个小技巧救了我三次。6. 进阶思考不止于“生图”MCP 连接器的真正价值在于“工作流编织”当你把腾讯混元生图成功接入 WorkBuddy 后一个更宏大的图景就展开了。MCP 的终极魅力从来不是“调用一个 API”而是“把多个异构服务像乐高积木一样拼成一个自动化工作流”。举个真实案例我们团队要做一个“周报自动生成”Skill。流程是WorkBuddy 先调用github_list_issues一个查询 GitHub Issues 的 MCP 工具把返回的 Issue 列表喂给qwen_summarize另一个调用通义千问的 MCP 工具做摘要把摘要结果再喂给hunyuan_generate_image生成一张代表本周工作重点的抽象图最后把文字摘要和图片 URL一起发给notion_create_pageNotion API 的 MCP 工具自动创建一页周报。这个流程里WorkBuddy 是大脑四个 MCP Server 是四肢。它们之间不共享内存不直连数据库只通过标准的 MCP 消息传递。任何一个环节出问题比如混元挂了WorkBuddy 会收到明确的 error然后可以选择重试、跳过或者通知管理员。这种松耦合、高内聚的架构才是现代 AI 工作流的基石。所以不要满足于“手把手教你接入”。真正的高手会把这篇教程当作起点开始思考我的业务里还有哪些重复、机械、需要跨系统操作的任务它们背后是否都藏着一个等待被 MCP 化的 API当你能把 CRM、ERP、BI 系统的 API都变成一个个callTool那么 WorkBuddy 就不再是一个聊天窗口而是一个真正意义上的“数字员工操作系统”。我在上周刚刚把公司内部的 Jira 查询、飞书消息推送、以及腾讯混元生图串成了一个“Bug 修复可视化报告”工作流。当工程师提了一个新 BugWorkBuddy 自动抓取 Jira 详情生成修复方案摘要画出修复前后对比图并把整套材料推送到飞书群。整个过程无需人工干预。那一刻我意识到MCP 不是技术而是一种新的生产力范式——它把“写脚本”的门槛降到了“说人话”的程度。这条路才刚刚开始。
WorkBuddy接入腾讯混元生图的MCP Server实战指南
1. 先搞清楚WorkBuddy、MCP 和「腾讯混元生图」这三者到底在什么层面打交道很多人一看到“给 WorkBuddy 接入自定义 MCP 连接器”第一反应是这又是个套壳包装的 AI 工具链其实不是。WorkBuddy 的底层定位是一个面向开发者工作流的智能体运行时环境——它不自己造模型也不硬编码能力而是通过一套标准化的协议把外部服务的能力“插”进来。这个协议就是 MCPModel Context Protocol。MCP 不是 API也不是 SDK更不是某种硬件接口协议网上有人混淆成“MCP 是硬件协议”这是典型误解。它本质是一套轻量级、面向语义交互的 JSON-RPC 扩展规范核心目标就一个让 AI 智能体在执行任务时能像调用本地函数一样安全、可追溯、带上下文地调用外部服务。比如你让 WorkBuddy “生成一张春天樱花盛开的海报”它内部不会自己画图而是通过 MCP 协议把“生成图像”这个意图结构化为一个generate_image方法调用连同 prompt、尺寸、风格等参数发给后端某个图像服务。而这个后端服务就是我们今天要接入的「腾讯混元生图」云托管实例。关键点来了腾讯混元生图本身提供的是 HTTP RESTful 接口支持同步返回或 SSE 流式响应。但 WorkBuddy 的 MCP 客户端只认一种通信方式基于 WebSocket 的 MCP Server。它不直接消费原始的 HTTP 接口而是要求你部署一个中间层——一个实现了 MCP Server 规范的网关服务。这个网关干两件事一是监听 WorkBuddy 发来的 MCP 请求比如call_tool二是把请求翻译成对腾讯混元生图 API 的调用并把结果按 MCP 格式封装回传。所以“接入自定义 MCP 连接器”的真实含义是你得亲手写一个 MCP Server它既是 WorkBuddy 的“听令者”又是腾讯混元生图的“代笔人”。这个 Server 不是配置出来的是代码写出来的它不依赖任何现成 SDK因为目前官方并没有发布“腾讯混元 MCP”的一体化适配包。你得自己搭桥。我第一次试的时候直接把混元的 API 文档 URL 粘贴进 WorkBuddy 的 MCP 配置框填上wss://api.xiaozhi.me/mcp/?token...这种地址结果 WorkBuddy 报错Connection refused。后来才明白那个wss://地址根本不是混元的而是某个第三方 MCP 中继服务的地址和腾讯混元毫无关系。这种混淆在社区里非常普遍也是新手最容易卡住的第一道墙。提示MCP Server 必须是 WebSocket 服务wss://或ws://且必须实现listTools、callTool等标准方法。它和你调用的后端 API如混元是解耦的。你可以用 Python 写也可以用 Java、Node.js甚至 Rust只要它能跑通 MCP 协议就行。2. 为什么选 SSE 云托管——不是为了“流式”而是为了“可控性”与“合规性”标题里特意强调了「SSE 云托管」这绝不是为了赶时髦。很多教程一上来就教你怎么用 WebSocket 实现实时推送但在这里SSEServer-Sent Events才是更务实、更稳定的选择。原因有三层层层递进第一层是技术匹配度。腾讯混元生图的官方 API明确支持两种响应模式sync同步阻塞和sse服务端事件流。sync模式简单粗暴但有个致命缺陷当图片生成耗时较长比如复杂 prompt 要 8~12 秒WorkBuddy 的 MCP 客户端会因超时而断开连接报出stream disconnected before completion: idle timeout waiting for sse这个错误。这不是你的代码问题是 WorkBuddy 内置的默认超时策略通常为 5 秒和混元实际响应时间不匹配导致的。而 SSE 天然支持长连接、分段推送混元在生成过程中会持续发送event: progress、event: result等事件客户端可以逐帧接收、实时渲染进度条最后拿到完整 Base64 图片。这完美绕开了超时陷阱。第二层是架构清晰度。SSE 是单向服务端→客户端的 HTTP 协议比 WebSocket 更轻量、更易调试。你不需要维护复杂的连接状态、心跳保活、重连逻辑。一个简单的 HTTP POST 请求发过去然后打开一个 EventSource 连接等着收数据整个流程就像读取一个不断增长的日志文件。我在做 PoC概念验证时先用curl -N命令手动测试混元的 SSE 接口确认能稳定收到data: {progress: 50%}和data: {result: base64...}之后再把这段逻辑封装进 MCP Server心里就有底了。如果一开始就啃 WebSocket光是解决connection reset by peer这类网络抖动问题就能耗掉你两天。第三层是部署合规性。所谓“云托管”指的是把你的 MCP Server 部署在腾讯云函数 SCFServerless Cloud Function或轻量应用服务器 Lighthouse 上而不是本地开发机。SCF 对 HTTP 请求有严格的冷启动和执行时间限制最长 300 秒但它对 SSE 的支持非常友好。你只需要在 SCF 函数里开启一个 HTTP 服务监听/mcp路径当 WorkBuddy 的 WebSocket 连接上来后你的函数就启动一个后台协程去调用混元的 SSE 接口并把收到的每一个data:事件转换成 MCP 的tool_call_result消息通过 WebSocket 回推给 WorkBuddy。整个过程SCF 只负责“转发”不承担模型推理压力成本极低且天然符合企业级安全审计要求——所有流量都走 HTTPS密钥管理由云平台统一处理。注意不要被wss://api.xiaozhi.me/mcp/这类地址误导。那是一个公开的 MCP 中继 Demo它的后端可能是任意模型且 token 有效期极短。生产环境必须用自己的云托管实例绑定自己的腾讯云 API 密钥SecretId/SecretKey并做好 token 鉴权。3. 从零手写 MCP ServerPython FastAPI httpx 的最小可行方案现在进入实操环节。我们放弃一切花哨框架用最精简的技术栈写出一个能跑通的 MCP Server。选型理由很实在Python 生态对 HTTP/SSE 支持最成熟FastAPI 自带异步 HTTP 服务和 WebSocket 支持httpx 是目前 Python 里对 SSE 流式解析最友好的 HTTP 客户端。整个代码不到 200 行但覆盖了所有核心路径。3.1 初始化项目与依赖新建一个目录workbuddy-mcp-hunyuan执行pip install fastapi uvicorn httpx python-dotenv创建main.py这是整个服务的入口。我们不追求高并发先确保单连接、单请求能闭环。3.2 定义 MCP 工具描述mcp.json 的灵魂MCP 协议要求 Server 在启动时必须能响应listTools请求返回一个 JSON 数组描述它能提供哪些工具。这个数组就是mcp.json文件的实质内容。很多人以为mcp.json是一个静态配置文件其实它是 Server 动态生成的契约。我们的mcp.json就一条工具{ name: hunyuan_generate_image, description: 使用腾讯混元大模型生成指定描述的图像, input_schema: { type: object, properties: { prompt: { type: string, description: 图像生成的中文或英文提示词越详细越好 }, size: { type: string, enum: [1024x1024, 768x1024, 1024x768], default: 1024x1024, description: 输出图像尺寸 } }, required: [prompt] } }注意input_schema的写法它直接决定了 WorkBuddy 在 UI 上如何生成表单。enum字段会让 WorkBuddy 给你一个下拉菜单而不是一个开放文本框极大降低用户输错的风险。3.3 核心逻辑WebSocket 连接 SSE 转发这是最关键的 80 行代码。FastAPI 的 WebSocket 路由如下app.websocket(/mcp) async def mcp_websocket(websocket: WebSocket): await websocket.accept() try: while True: # 1. 接收 WorkBuddy 发来的 MCP 请求 data await websocket.receive_json() if data.get(method) listTools: # 返回工具列表 await websocket.send_json({ jsonrpc: 2.0, id: data.get(id), result: [HUNYUAN_TOOL_SCHEMA] }) elif data.get(method) callTool: # 2. 解析调用参数 params data.get(params, {}) prompt params.get(prompt, ) size params.get(size, 1024x1024) # 3. 构造混元 SSE 请求 headers { Authorization: fBearer {os.getenv(HUNYUAN_API_KEY)}, Content-Type: application/json } payload { model: hunyuan-vision, prompt: prompt, size: size, stream: True } # 4. 异步发起 SSE 请求并实时转发 async with httpx.AsyncClient() as client: async with client.stream(POST, HUNYUAN_SSE_URL, jsonpayload, headersheaders) as response: if response.status_code ! 200: await websocket.send_json({ jsonrpc: 2.0, id: data.get(id), error: {code: -32000, message: f混元API调用失败: {response.status_code}} }) break # 5. 逐行解析 SSE 数据流 async for line in response.aiter_lines(): if line.strip() : continue if line.startswith(data:): try: # 去掉 data: 前缀解析 JSON json_str line[6:].strip() event_data json.loads(json_str) # 6. 将混元事件映射为 MCP 结果 if result in event_data and url in event_data[result]: # 最终结果返回图片 URL await websocket.send_json({ jsonrpc: 2.0, id: data.get(id), result: {image_url: event_data[result][url]} }) elif progress in event_data: # 进度更新作为 MCP 的 partial result await websocket.send_json({ jsonrpc: 2.0, id: data.get(id), result: {progress: event_data[progress]} }) except json.JSONDecodeError: pass # 忽略非 JSON 行如注释 except WebSocketDisconnect: print(WorkBuddy 客户端断开连接) except Exception as e: print(f处理请求时出错: {e})这段代码的精妙之处在于第 5 步和第 6 步httpx.AsyncClient.stream()能真正实现异步流式读取aiter_lines()保证了你不会等到整个 SSE 响应结束才开始处理。每收到一行data: {...}就立刻解析、映射、回推。WorkBuddy 的 UI 就能实时显示“生成中30%”、“生成中75%”最后弹出图片。这种体验是同步 API 永远无法提供的。3.4 启动与调试让 WorkBuddy 真正“看见”你的 Server启动服务uvicorn main:app --host 0.0.0.0 --port 8000 --reload此时你的 MCP Server 监听在http://localhost:8000/mcp。但 WorkBuddy 要求的是wss://所以你需要一个反向代理。最简单的方法是用ngrokngrok http 8000它会给你一个类似https://abc123.ngrok.io的公网地址。把这个地址加上/mcp路径填入 WorkBuddy 的 MCP 设置里格式为wss://abc123.ngrok.io/mcp注意是wssngrok 会自动升级。提示本地调试时WorkBuddy 的日志面板CtrlShiftI 打开 DevTools → Console是你的最佳朋友。每一次连接、每一次listTools请求、每一次callTool都会打印出来。如果看到Failed to connect to MCP server90% 的概率是你的 ngrok 地址没填对或者 FastAPI 服务根本没起来。先curl http://localhost:8000/mcp看是否返回 405 Method Not Allowed说明服务起来了再查 ngrok 日志。4. WorkBuddy 端的配置与技能编排让“生图”成为一条自然指令Server 端写好了只是完成了 50%。剩下 50%是让 WorkBuddy 理解、信任并流畅地使用它。这一步很多人草草跳过结果发现 WorkBuddy 死活不调用你的工具或者调用了但返回乱码。4.1 MCP 连接器配置的三个生死细节在 WorkBuddy 的设置 → MCP → 添加新连接器中填写以下信息Name:腾讯混元生图 (SSE)URL:wss://abc123.ngrok.io/mcp你的 ngrok 地址Authentication: 选择None因为我们没加鉴权或Token如果你在 Server 端加了 Bearer Token 校验这三个字段任何一个填错都会导致连接失败。特别注意 URL必须以wss://开头不能是http://路径必须是/mcp不能是/api/mcp或其他末尾不能有多余的/。我曾因为多打了一个斜杠调试了 40 分钟。4.2 创建 Skill把工具变成一句自然语言WorkBuddy 的核心价值是让 AI 用自然语言驱动工具。所以你不能只停在“能调用”这一步必须创建一个 Skill。点击 WorkBuddy 主界面右上角的 New Skill填入Name:生成图片Description:根据我的文字描述生成一张高质量的图片Trigger:When I say 生成一张... or 画一幅... or 给我看看...这里用正则表达式生成一张.*|画一幅.*|给我看看.*Action:Call tool: hunyuan_generate_imageParameters Mapping: 这里最关键把用户说的话自动提取成工具参数。prompt:{{trigger.text}}把整句话作为 promptsize:1024x1024固定值或设为{{user.input.size}}让用户选择保存后你就可以对 WorkBuddy 说“生成一张穿着宇航服的橘猫在月球表面跳跃的图片”它会自动识别触发词提取整句话作为prompt调用你的 MCP Server最终把混元生成的图片 URL 返回给你。4.3 调试 Skill 的黄金法则看tool_call日志Skill 不生效别急着改代码。打开 WorkBuddy 的Activity Log活动日志找到你刚说的那句话。如果 Skill 被正确触发你会看到一条tool_call记录里面包含完整的params。如果params里prompt是空的说明你的 Parameter Mapping 写错了如果根本没这条记录说明 Trigger 的正则没匹配上或者 Skill 没启用。这是最直接、最不可辩驳的证据链。我踩过的一个深坑是我把 Trigger 写成了生成一张.*图片结果用户说“生成一张橘猫”就没触发。后来改成生成一张.*覆盖所有可能性问题立刻解决。经验是Trigger 正则宁宽勿窄参数提取逻辑放在Parameters Mapping里做精细化处理。5. 生产环境加固从本地 Demo 到企业级可用的五道防线一个能跑通的 Demo 和一个能上线的生产服务中间隔着五道墙。这五道墙不是可选项而是必选项。我在线上环境部署时每一道都亲自验证过。5.1 防线一密钥安全——绝不硬编码必须用环境变量 云密钥管理HUNYUAN_API_KEY这种敏感信息绝对不能写死在main.py里也不能放在.env文件里提交到 Git。正确做法是本地开发用.env文件HUNYUAN_API_KEYyour_real_key_here云托管SCF在函数配置里添加环境变量HUNYUAN_API_KEY值从腾讯云 KMS密钥管理系统中获取。SCF 函数启动时会自动解密并注入到环境变量中。这样你的代码里永远只写os.getenv(HUNYUAN_API_KEY)密钥本身从未出现在任何代码或配置文件中。5.2 防线二连接池与超时——避免混元 API 拒绝服务混元 API 有 QPS 限制。如果你的 MCP Server 没有连接池每个请求都新建 TCP 连接很容易触发限流。在httpx.AsyncClient初始化时必须配置limits httpx.Limits(max_connections20, max_keepalive_connections10) timeout httpx.Timeout(30.0, connect10.0, read30.0) async with httpx.AsyncClient(limitslimits, timeouttimeout) as client:connect10.0确保 DNS 解析和 TCP 握手不超过 10 秒read30.0给混元留足生成时间max_connections20控制并发数防止雪崩。5.3 防线三错误重试与降级——当混元挂了WorkBuddy 不能卡死SSE 流可能因网络中断而断开。你的代码里必须有重试逻辑for attempt in range(3): try: async with client.stream(...) as response: # ... 处理流 ... break # 成功则跳出循环 except (httpx.ConnectTimeout, httpx.ReadTimeout): if attempt 2: await websocket.send_json({... error: 重试三次后仍失败 ...}) break await asyncio.sleep(1) # 指数退避更高级的降级是当混元不可用时自动 fallback 到一个本地缓存的“占位图”URL或者返回一段友好的提示“图像生成服务暂时繁忙请稍后再试”。5.4 防线四日志与监控——没有日志的线上服务等于盲人开车在main.py里每一处关键节点都要打日志INFO: “收到 listTools 请求”INFO: “收到 callTool 请求prompt: [前20字]...”WARNING: “混元 API 返回 429已触发限流”ERROR: “SSE 流解析 JSON 失败原始行: {line}”这些日志全部通过logging模块输出并在 SCF 控制台里配置日志投递到 CLS云日志服务。当用户反馈“图片没出来”你不用问东问西直接去 CLS 查关键词hunyuan_generate_image5 秒内定位到是网络问题、密钥问题还是混元侧问题。5.5 防线五版本兼容——MCP 协议在快速迭代你的 Server 必须跟上MCP 协议不是一成不变的。WorkBuddy 新版本可能会增加cancelTool方法或者修改tool_call_result的字段名。你的 Server 必须有版本号并在listTools返回中声明{ name: hunyuan_generate_image, version: 1.2.0, description: ..., ... }同时在代码里用if data.get(jsonrpc) 2.0做基础校验对未知 method 返回标准的Method not found错误。这样即使未来协议升级你的 Server 也不会崩溃只会优雅地告知 WorkBuddy “这个功能我还不支持”。最后分享一个血泪教训我在一次 WorkBuddy 更新后发现所有 MCP 连接都失败了。查日志发现新版本在callTool请求里多加了一个context字段而我的旧版 Server 解析params时直接params.get(prompt)忽略了context的存在导致整个params字典被当成None。修复方法很简单params data.get(params, {})永远给get加默认值。这个小技巧救了我三次。6. 进阶思考不止于“生图”MCP 连接器的真正价值在于“工作流编织”当你把腾讯混元生图成功接入 WorkBuddy 后一个更宏大的图景就展开了。MCP 的终极魅力从来不是“调用一个 API”而是“把多个异构服务像乐高积木一样拼成一个自动化工作流”。举个真实案例我们团队要做一个“周报自动生成”Skill。流程是WorkBuddy 先调用github_list_issues一个查询 GitHub Issues 的 MCP 工具把返回的 Issue 列表喂给qwen_summarize另一个调用通义千问的 MCP 工具做摘要把摘要结果再喂给hunyuan_generate_image生成一张代表本周工作重点的抽象图最后把文字摘要和图片 URL一起发给notion_create_pageNotion API 的 MCP 工具自动创建一页周报。这个流程里WorkBuddy 是大脑四个 MCP Server 是四肢。它们之间不共享内存不直连数据库只通过标准的 MCP 消息传递。任何一个环节出问题比如混元挂了WorkBuddy 会收到明确的 error然后可以选择重试、跳过或者通知管理员。这种松耦合、高内聚的架构才是现代 AI 工作流的基石。所以不要满足于“手把手教你接入”。真正的高手会把这篇教程当作起点开始思考我的业务里还有哪些重复、机械、需要跨系统操作的任务它们背后是否都藏着一个等待被 MCP 化的 API当你能把 CRM、ERP、BI 系统的 API都变成一个个callTool那么 WorkBuddy 就不再是一个聊天窗口而是一个真正意义上的“数字员工操作系统”。我在上周刚刚把公司内部的 Jira 查询、飞书消息推送、以及腾讯混元生图串成了一个“Bug 修复可视化报告”工作流。当工程师提了一个新 BugWorkBuddy 自动抓取 Jira 详情生成修复方案摘要画出修复前后对比图并把整套材料推送到飞书群。整个过程无需人工干预。那一刻我意识到MCP 不是技术而是一种新的生产力范式——它把“写脚本”的门槛降到了“说人话”的程度。这条路才刚刚开始。
