OpenMontage 中的 HeyGen Webhook 集成指南：从事件订阅、签名校验到重试设计

在 HeyGen 数字人视频生产链路中，异步任务的结果获取主要有两种方式：轮询（polling）与 Webhook 推送。本文基于 OpenMontage 仓库中 avatar-video 技能的参考文档 webhooks.md，系统讲解 Webhook 端点的搭建、事件类型与 Payload 结构、注册流程、callback_id 追踪机制、HMAC 签名校验与失败重试策略。读完后你将能够在生产系统中用“事件推送 + 异步处理 + 幂等去重”替代长轮询，完整掌握 HeyGen 异步操作的落地工程方案。

Webhook 在 HeyGen 异步工作流中的定位

HeyGen 的视频生成是异步操作——提交 POST /v2/video/generate 后需要数分钟才能得到结果。默认工作流（见 SKILL.md）的第 5 步是“Poll for completion — GET /v2/videos/{video_id} until status is completed”，即反复查询状态直到完成，对应参考文档 video-status.md 中给出的 pending / processing / completed / failed 四种状态与轮询实现。

Webhook 的价值在于：由 HeyGen 在事件发生时主动把通知推送到你的服务器，覆盖以下场景：

• 视频生成完成
• 视频生成失败
• 翻译（translation）完成
• 数字人（avatar）训练完成
• 其他异步操作结束

从仓库源码结构看，OpenMontage 目前的 HeyGen 集成走的是同步轮询路线：heygen_video.py 中的 HeyGenVideo 工具声明 execution_mode = ExecutionMode.SYNC，其底层调用 _shared.py 的 poll_heygen(execution_id, api_key, timeout=600)——轮询 status 字段，遇到 completed 时从 output.video.video_url 提取下载地址，遇到 failed / error 时抛错，默认超时 600 秒。也就是说，仓库内现有实现适合 Agent 单次任务内的“等结果”场景；而当你搭建的是批量、多任务、无人值守的生产系统时，本文介绍的 Webhook 推送模式就是更合理的选择——video-status.md 也明确建议“For production systems, webhooks are more efficient than polling”。

搭建 Webhook 端点：快速 ACK + 异步处理

端点设计遵循三条铁律：

1. 接受 POST 请求；
2. 快速返回 200 状态码（不要在 HTTP 响应里做重活）；
3. 事件处理放到异步任务中执行。

Express.js 实现

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 实现

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(
        target=process_webhook_event,
        args=(event,)
    )
    thread.start()

    return response, 200

def process_webhook_event(event):
    event_type = event.get("event_type")
    print(f"Received 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(port=3000)

两种实现的共同点是：先返回 200，再启动独立的任务处理事件——这是防止 HeyGen 因超时而丢弃通知的关键。

事件类型与 Payload 结构

事件类型一览

Event Type	Description
avatar_video.success	Video generation completed
avatar_video.fail	Video generation failed
video_translate.success	Translation completed
video_translate.fail	Translation failed
instant_avatar.success	Instant avatar created
instant_avatar.fail	Instant avatar creation failed

视频成功事件

类型定义：

interface VideoSuccessEvent {
  event_type: "avatar_video.success";
  event_data: {
    video_id: string;
    video_url: string;
    thumbnail_url: string;
    duration: number;
    callback_id?: string;
  };
}

实际 Payload 示例：

{
  "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"
  }
}

对比 video-status.md 中 GET /v2/videos/{video_id} 返回的 completed 状态（包含 video_url、thumbnail_url、duration、gif_url、captioned_video_url 等字段），成功 Webhook 事件是同一份结果的“推送版”——拿到 video_url 后即可按下载流程落盘（注意：该文档提示 video_url 有时效性，应及时下载并缓存本地路径）。

视频失败事件

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"
  }
}

失败事件中的 error 信息（如 "Script too long for selected avatar"）可直接映射到用户可读的失败原因，对应轮询模式下 failure_code: "script_too_long" / failure_message 的语义，便于在两种模式间做统一错误处理。

注册 Webhook 端点

通过 HeyGen 控制台（dashboard）或 API 配置 Webhook URL。注册请求的字段如下：

Field	Type	Req	Description
url	string	✓	Your webhook endpoint URL
events	array	✓	Event types to subscribe to
secret	string		Shared secret for signature verification

通过 API 注册

curl -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"]
  }'

TypeScript 注册封装

interface WebhookConfig {
  url: string;                                 // Required
  events: string[];                            // Required
  secret?: string;
}

async function registerWebhook(config: WebhookConfig): Promise<void> {
  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);
  }
}

鉴权方式与 avatar-video 技能全链路一致：X-Api-Key 请求头 + HEYGEN_API_KEY 环境变量（SKILL.md 的 Authentication 一节亦同）。

用 callback_id 追踪任务归属

callback_id 是把“发起请求的业务实体”与“最终收到的 Webhook 事件”关联起来的纽带，适合订单系统、批量任务等场景。

发起视频生成时携带

const 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 时只能靠 video_id 反查本地记录表，而 callback_id 是自定义字符串（如 order_12345），可以把 HeyGen 任务直接挂到业务主键上，省去中间映射。

Webhook 安全：签名校验与来源验证

校验 HMAC 签名

如果 HeyGen 提供了签名验证能力（注册时配置 secret），应校验 x-heygen-signature 请求头，避免被伪造请求触发副作用：

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...
});

注意两点：签名计算基于原始 payload 字符串；比较使用 timingSafeEqual 防止时序侧信道攻击。

校验事件结构合法性

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);
}

签名校验回答“请求是否来自 HeyGen”，结构校验回答“内容是否是一个合法事件”，两层都通过后才进入业务处理。

失败处理：重试与死信存储

处理逻辑自身也可能失败（数据库抖动、下游服务不可用）。推荐模式是指数退避重试 + 失败事件落库人工处理：

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 次、指数退避、最终落库”的结构与仓库中工具层的重试设计思路一致，例如 heygen_video.py 中 RetryPolicy(max_retries=2, backoff_seconds=10.0, retryable_errors=["rate_limit", "timeout", "server_error"])——只不过 Webhook 场景下重试的是本地事件处理，而非 API 调用本身。

Webhook 与轮询的选型对比

Aspect	Webhook	Polling
Latency	Immediate	Depends on interval
Efficiency	High (push)	Low (repeated requests)
Complexity	Requires endpoint	Simpler to implement
Reliability	Needs retry handling	Guaranteed delivery
Cost	Lower API usage	Higher API usage

选型建议：

• Agent 单次任务 / 脚本化调用：轮询更简单——无需暴露公网端点、无需部署服务，OpenMontage 仓库现有实现（poll_heygen，默认 600 秒超时）即此类场景；
• 生产系统 / 批量生成 / 多任务并行：Webhook 更优——推送即时、API 调用量低，但需要部署可公网访问（或经隧道暴露）的端点，并自行处理重复事件与重试。

两者并非互斥：常见做法是 Webhook 为主、轮询兜底（长时间未收到事件时主动查一次状态）。

本地测试：ngrok 隧道与事件模拟

用 ngrok 暴露本地服务

# Start ngrok tunnel
ngrok http 3000

# Use ngrok URL as webhook endpoint
# https://abc123.ngrok.io/webhook/heygen

把 ngrok 提供的 URL 填入注册请求的 url 字段即可在本地收事件。

本地模拟事件投递

// 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",
  },
});

用模拟脚本逐一遍历上表中的六种事件类型（含 fail 分支），可以在不消耗 HeyGen 额度的情况下完整验证端点的 ACK、去重与路由逻辑。

最佳实践清单

1. 快速响应——5 秒内返回 200，重活全部异步化；
2. 处理重复事件——同一事件可能被多次投递，处理逻辑必须幂等（建议以 event_type + video_id + callback_id 作为去重键）；
3. 实现重试——对临时性处理失败做指数退避重试；
4. 全量记录日志——保存每个 Webhook payload，便于事后排查；
5. 使用 callback ID——用业务标识贯穿请求与回调；
6. 加固端点——校验 HMAC 签名、强制 HTTPS；
7. 监控健康度——跟踪 Webhook 成功率并告警；
8. 队列化处理——重处理（下载视频、转码、入库）交给任务队列，避免阻塞 ACK。

小结

webhooks.md 为 HeyGen 的异步视频生产提供了完整的“推送侧”工程方案：端点快速 ACK + 异步处理、六类事件与两类核心 Payload、endpoint.add 注册接口、callback_id 业务追踪、HMAC 签名与来源校验、指数退避重试。结合仓库现状（tools/video/heygen_video.py 与 tools/video/_shared.py 中的 poll_heygen 同步轮询实现）可以推断，OpenMontage 当前以轮询服务 Agent 会话内的即时取件需求；当你把 avatar-video 技能（数字人、多场景、透明 WebM、批量规格生成）扩展到长期运行的生产系统时，本文的 Webhook 架构就是自然的演进方向。