OpenMontage 中的 HeyGen Webhooks 事件通知接入指南从轮询到推送到生产级视频生成回调【免费下载链接】OpenMontageWorlds first open-source, agentic video production system. 12 production pipelines, 100 tools, 700 agent skill and production-knowledge files. Turn your AI coding assistant into a full video production studio.项目地址: https://gitcode.com/GitHub_Trending/op/OpenMontage导读HeyGen 的 API 生成数字人视频avatar video属于耗时异步任务Webhooks 机制让服务端在视频生成完成、失败或翻译结束等时刻主动把结果推送到你的应用替代低效的轮询。本指南以 .agents/skills/heygen/references/webhooks.md 为核心系统讲解事件类型、端点搭建、URL 注册、事件负载结构、callback_id 关联、签名校验与失败重试等完整链路并结合 OpenMontage 仓库中真实调用 HeyGen 的 Python 实现与轮询代码帮助读者在自己的生产环境中构建一套可靠、安全的 HeyGen 异步回调系统。Webhooks 为什么必要异步任务的推送式通知HeyGen 数字人视频生成是典型的长时间异步流程通常 5–15 分钟峰值或长脚本可超 20 分钟。同步等待显然不可行而最常见的替代方案是轮询polling客户端按固定间隔反复查询任务状态直到拿到结果。Webhooks 的做法则完全相反——由 HeyGen 在你的任务状态发生变化时向你的服务器主动推送 HTTP POST 通知。相比之下它能推送通知的事件包括视频生成完成Video generation completes视频生成失败Video generation fails翻译完成Translation completes数字人 Avatar 训练完成Avatar training completes其他异步操作的结束Other async operations finish推送模式避免了对状态的反复查询既降低 API 调用量与成本也让任务何时结束这一信息以近乎实时的方式触达服务端天然适合将完成事件继续驱动下游流水线如下载成片、写库、通知用户、触发剪辑。需要注意的是OpenMontage 仓库当前的工具实现仍以轮询为主tools/video/_shared.py 中的poll_heygen()与 tools/video/heygen_video.py 提供了同步等成片的参考实现而 Webhooks 是文档推荐的生产系统升级方向两者的取舍详见下文对比章节。搭建一个符合要求的 Webhook 端点你的 Webhook 端点需要满足三个基本约束接受 POST 请求Accept POST requests快速返回 200 状态Return 200 status quickly异步处理事件Handle events asynchronously第 2、3 条是核心设计原则推送方HeyGen只关心你收到了而不是你处理完了。因此正确模式是先立刻应答 200再把事件体交给后台任务去处理避免慢业务阻塞 HTTP 响应而触发上游重试。Express.js 示例下面是一个完整的 TypeScript/Express 端点使用express.json()解析请求体先应答再异步处理import express from express; import crypto from crypto; const app express(); app.use(express.json()); // Webhook endpoint app.post(/webhook/heygen, async (req, res) { // Acknowledge receipt immediately res.status(200).send(OK); // Process event asynchronously processWebhookEvent(req.body).catch(console.error); }); async function processWebhookEvent(event: HeyGenWebhookEvent) { console.log(Received event: ${event.event_type}); switch (event.event_type) { case avatar_video.success: await handleVideoSuccess(event); break; case avatar_video.fail: await handleVideoFailure(event); break; case video_translate.success: await handleTranslationSuccess(event); break; default: console.log(Unknown event type: ${event.event_type}); } } app.listen(3000, () { console.log(Webhook server running on port 3000); });Python Flask 示例Python 侧同样遵循先应答、再处理的范式示例中用threading.Thread把事件处理放到独立线程from flask import Flask, request, jsonify import threading app Flask(__name__) app.route(/webhook/heygen, methods[POST]) def heygen_webhook(): event request.json # Acknowledge immediately response jsonify({status: received}) # Process asynchronously thread threading.Thread( targetprocess_webhook_event, args(event,) ) thread.start() return response, 200 def process_webhook_event(event): event_type event.get(event_type) print(fReceived event: {event_type}) if event_type avatar_video.success: handle_video_success(event) elif event_type avatar_video.fail: handle_video_failure(event) elif event_type video_translate.success: handle_translation_success(event) if __name__ __main__: app.run(port3000)在线程模型之外更健壮的做法是接入任务队列如 Celery、BullMQ、SQS把事件体入队由 worker 消费——这正是文档最佳实践中Queue processing一条的落地方式。完整 Webhook 事件类型清单订阅事件时event_type决定了你的应用能收到哪些回调Event TypeDescriptionavatar_video.successVideo generation completedavatar_video.failVideo generation failedvideo_translate.successTranslation completedvideo_translate.failTranslation failedinstant_avatar.successInstant avatar createdinstant_avatar.failInstant avatar creation failed从类型命名可以推断HeyGen 的 Webhook 事件横跨视频生成avatar_video、**视频翻译video_translate与即时数字人创建instant_avatar**三大业务成功与失败成对出现方便你在订阅时精确取舍。事件负载Payload结构详解所有事件统一采用{ event_type, event_data }双层结构event_type标识事件种类event_data携带业务数据。下面以视频生成事件为例拆解。视频成功事件TypeScript 接口定义interface VideoSuccessEvent { event_type: avatar_video.success; event_data: { video_id: string; video_url: string; thumbnail_url: string; duration: number; callback_id?: string; }; }对应的 JSON 实例{ event_type: avatar_video.success, event_data: { video_id: abc123, video_url: https://files.heygen.ai/video/abc123.mp4, thumbnail_url: https://files.heygen.ai/thumbnail/abc123.jpg, duration: 45.2, callback_id: your_custom_id } }视频失败事件interface VideoFailureEvent { event_type: avatar_video.fail; event_data: { video_id: string; error: string; callback_id?: string; }; }{ event_type: avatar_video.fail, event_data: { video_id: abc123, error: Script too long for selected avatar, callback_id: your_custom_id } }值得注意callback_id是可选字段它承载着你系统内的业务标识如订单号、任务号是跨系统追踪的关键其详细用法见下文Callback ID一节。失败负载中的error字段通常携带可读的错误原因例如示例中的 Script too long for selected avatar直接可用于日志与告警。注册 Webhook URL搭建好端点后需要通过 HeyGen 控制台Dashboard或 API 把你的 URL 订阅事件集合告诉 HeyGen。请求字段FieldTypeReqDescriptionurlstring✓Your webhook endpoint URLeventsarray✓Event types to subscribe tosecretstringShared secret for signature verificationurl必须是你公网可达的 HTTPS 端点events是上文事件类型表中你要监听的事件数组secret为可选字段一旦设置HeyGen 将用它对你的回调做签名供你端到端验签。通过 API 注册curlcurl -X POST https://api.heygen.com/v1/webhook/endpoint.add \ -H X-Api-Key: $HEYGEN_API_KEY \ -H Content-Type: application/json \ -d { url: https://your-domain.com/webhook/heygen, events: [avatar_video.success, avatar_video.fail] }这里的鉴权头X-Api-Key: $HEYGEN_API_KEY与 HeyGen 所有 API 请求保持一致API Key 通过环境变量注入——这与仓库技能文档 authentication.md 及工具实现一致见 heygen_video.py 中的os.environ.get(HEYGEN_API_KEY)。通过 API 注册TypeScriptinterface WebhookConfig { url: string; // Required events: string[]; // Required secret?: string; } async function registerWebhook(config: WebhookConfig): Promisevoid { const response await fetch(https://api.heygen.com/v1/webhook/endpoint.add, { method: POST, headers: { X-Api-Key: process.env.HEYGEN_API_KEY!, Content-Type: application/json, }, body: JSON.stringify(config), }); const json await response.json(); if (json.error) { throw new Error(json.error); } }注册通常是一次性配置建议纳入基础设施代码IaC管理如需变更订阅事件集合在 Dashboard 或调用同族管理端点更新即可。用 Callback ID 把回调与业务请求关联起来回调是无状态的 HTTP POST你的服务端无法仅凭事件内容知道这条视频对应哪个内部订单/工单。解决方案就是callback_id在发起视频生成时带上你自己的业务标识HeyGen 会原样把它写进后续所有相关事件的event_data.callback_id中。生成视频时带上 Callback IDconst videoConfig { video_inputs: [...], callback_id: order_12345, // Your custom identifier };在 Webhook 处理器中反查业务记录async function handleVideoSuccess(event: VideoSuccessEvent) { const { video_id, video_url, callback_id } event.event_data; if (callback_id) { // Look up your original request const order await getOrderByCallbackId(callback_id); await updateOrderWithVideo(order.id, video_url); } }这样即使系统重启、事件乱序到达服务端也能通过callback_id幂等地把成片 URL 回写到正确的业务记录上。Webhook 安全签名校验与来源验证公网端点天然暴露在恶意请求之下任何指向/webhook/heygen的伪造 POST 都可能污染你的业务数据。安全防护分两层验签与校验事件合法性。校验 Webhook 签名如果你在注册时配置了secretHeyGen 会对负载做 HMAC-SHA256 签名并放入请求头。验签实现如下import crypto from crypto; function verifyWebhookSignature( payload: string, signature: string, secret: string ): boolean { const expectedSignature crypto .createHmac(sha256, secret) .update(payload) .digest(hex); return crypto.timingSafeEqual( Buffer.from(signature), Buffer.from(expectedSignature) ); } // In your webhook handler app.post(/webhook/heygen, (req, res) { const signature req.headers[x-heygen-signature] as string; const payload JSON.stringify(req.body); if (!verifyWebhookSignature(payload, signature, WEBHOOK_SECRET)) { return res.status(401).send(Invalid signature); } // Process event... });两个工程细节值得强调请求头假定为x-heygen-signature真实请求头名称以 HeyGen 官方文档为准实现时务必核对比较签名使用crypto.timingSafeEqual而非以避免时序侧信道攻击——注意timingSafeEqual要求两个 Buffer 等长生产代码需先对长度做防御性判断。校验事件来源与合法性即便没有secret也应做基本的字段与事件类型校验拒绝明显伪造的请求function isValidHeygenEvent(event: any): boolean { // Check required fields if (!event.event_type || !event.event_data) { return false; } // Check event type is known const validEventTypes [ avatar_video.success, avatar_video.fail, video_translate.success, video_translate.fail, ]; return validEventTypes.includes(event.event_type); }以上两种手段结合 HTTPS 强制加密传输构成端点的基本安全基线。更完整的鉴权方式X-Api-Key请求头、错误码、限流参见 authentication.md。处理失败与重试可靠的异步消费Webhook 是尽力而为的推送你的一次性处理逻辑可能因数据库抖动、下游服务不可用等原因失败。因此消费方必须自带重试语义。文档给出带指数退避exponential backoff的通用处理器模板async function processWebhookEvent(event: HeyGenWebhookEvent) { const maxRetries 3; for (let attempt 1; attempt maxRetries; attempt) { try { await handleEvent(event); return; } catch (error) { console.error(Attempt ${attempt} failed:, error); if (attempt maxRetries) { // Exponential backoff await new Promise((r) setTimeout(r, Math.pow(2, attempt) * 1000)); } } } // Store failed event for manual review await storeFailedEvent(event); }设计要点最多重试 3 次每次失败按2^attempt秒退避2s → 4s → 8s退避采用幂级增长避免集中重试对下游形成流量尖峰3 次仍失败后不要静默丢弃把事件落盘/入库storeFailedEvent供人工复查或延迟补偿任务再次触发。若你的业务把 Webhook 事件当作关键链路如成片入库即对外可交付还应配合死信队列与监控告警保证最终一定有人处理。Webhook 与轮询的取舍文档给出了两者直观的对照这里完整继承并结合仓库现状做补充AspectWebhookPollingLatencyImmediateDepends on intervalEfficiencyHigh (push)Low (repeated requests)ComplexityRequires endpointSimpler to implementReliabilityNeeds retry handlingGuaranteed deliveryCostLower API usageHigher API usage从仓库现实看OpenMontage 目前接入 HeyGen 采用的就是轮询方案。tools/video/_shared.py 的poll_heygen()展示了同步轮询的生产级细节def poll_heygen(execution_id: str, api_key: str, timeout: int 600) - str: ... interval 5.0 while time.time() deadline: ... if status completed: video_url ( data.get(output, {}).get(video, {}).get(video_url) or data.get(output, {}).get(video_url) ) ... if status in {failed, error}: raise RuntimeError(...) time.sleep(min(interval, max(0.0, deadline - time.time()))) interval min(interval * 1.2, 30.0)这段实现与技能文档 video-status.md 高度吻合起始间隔 5 秒、超时上限 600 秒、间隔按 1.2 倍递增并封顶 30 秒即指数退避并显式处理completed / failed / error三种终态。它验证了轮询实现更简单、可保证交付的判断——同步工具模式天然适合单次调用的 CLI 场景。选型建议在长驻服务或流水线系统中当视频完成需要驱动后续步骤、且你不想让进程空转等待时优先考虑 Webhookwebhooks.md一次性的脚本式调用则用轮询即可。视频下载相关注意事项URL 时效、下载重试可参考 video-status.md。本地测试 Webhook生产端点接入前必须先在本地验证事件处理逻辑。Webhook 的回调目标需要公网可达本地开发通常借助内网穿透工具。用 ngrok 暴露本地服务# Start ngrok tunnel ngrok http 3000 # Use ngrok URL as webhook endpoint # https://abc123.ngrok.io/webhook/heygen启动 ngrok 后把打印出的公网地址如https://abc123.ngrok.io拼上你的路由/webhook/heygen即为可注册的url。本地模拟推送事件你也可以绕过 ngrok直接把构造好的事件 POST 到http://localhost:3000实现纯本地的快速回归// Test webhook locally async function simulateWebhook(event: HeyGenWebhookEvent) { const response await fetch(http://localhost:3000/webhook/heygen, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify(event), }); console.log(Response: ${response.status}); } // Simulate success event await simulateWebhook({ event_type: avatar_video.success, event_data: { video_id: test_123, video_url: https://example.com/test.mp4, thumbnail_url: https://example.com/test.jpg, duration: 30, callback_id: test_callback, }, });建议把成功事件、失败事件、非法签名、未知 event_type、畸形 JSON五类用例固化为一套本地测试脚本覆盖处理器的主要分支模拟成功事件尤其适合验证先 200 应答、再异步消费的时序行为。生产级 Webhook 端点最佳实践清单综合文档与上文分析一个可用于生产的 HeyGen Webhook 端点应满足Respond quickly—— 5 秒内返回 200事件一律异步处理Handle duplicates—— 同一事件可能被多次推送处理逻辑需幂等善用video_id/callback_id做去重键Implement retries—— 为临时性处理失败预留指数退避重试Log everything—— 落盘原始事件负载便于事后排查与回放Use callback IDs—— 用callback_id贯穿请求、回调、业务记录全链路Secure endpoints—— 校验 HMAC 签名、校验事件类型、强制 HTTPSMonitor health—— 持续观测回调成功率、处理延迟与重试率Queue processing—— 重量级处理一律进入任务队列由 worker 消费。在 OpenMontage 中的定位与延伸阅读在 OpenMontage 的 Agent 技能体系中Webhook 能力是被多套技能共享的基础设施知识除了本文依据的 .agents/skills/heygen/references/webhooks.mdavatar-videoreferences/webhooks.md与create-videoreferences/webhooks.md技能也维护了各自的 Webhook 参考文件分别面向精确数字人/场景控制v2 API与提示词驱动的视频生成Video Agent API两条工作流。阅读时请注意版本前提旧的heygen技能已在 SKILL.md 中标记为Deprecated新增集成应优先参考create-video/avatar-video两个新技能。本指南中 curl 示例基于文档提供的https://api.heygen.com/v1/webhook/endpoint.add端点与avatar_video.*事件命名实际接入时请以 HeyGen 当前 API 版本及官方事件名称为准。相关上下文可继续阅读轮询式状态获取与成片下载video-status.md鉴权与请求头规范authentication.md仓库内的同步轮询参考实现tools/video/_shared.py、tools/video/heygen_video.py提示词与数字人/声音选择prompt-optimizer.md、avatars.md、voices.md【免费下载链接】OpenMontageWorlds first open-source, agentic video production system. 12 production pipelines, 100 tools, 700 agent skill and production-knowledge files. Turn your AI coding assistant into a full video production studio.项目地址: https://gitcode.com/GitHub_Trending/op/OpenMontage创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考