HeyGen 视频状态轮询与下载全指南：OpenMontage 项目中的异步视频生成实战

HeyGen 视频状态轮询与下载全指南：OpenMontage 项目中的异步视频生成实战

【免费下载链接】OpenMontageWorld's 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 以异步方式处理视频生成：提交生成任务后，视频并不会立即就绪，而是经历排队（pending）、生成（processing）、完成（completed）或失败（failed）等阶段。本文基于 OpenMontage 仓库中的 HeyGen 技能参考文档（.claude/skills/heygen/references/video-status.md），系统讲解如何通过 MCP 工具或直接 API 查询视频状态、如何设计健壮的轮询与下载逻辑、如何结合仓库源码（tools/video/heygen_video.py、tools/video/_shared.py）落地可复用的生产级调用链。读完本文，你将掌握 HeyGen 视频任务从提交、轮询、失败处理到下载落盘、再到可恢复与 Webhook 替代方案的完整工程模式。

为什么需要轮询：HeyGen 的异步处理模型

HeyGen 的视频生成是典型的异步任务：调用 API 提交生成请求后，服务端立即返回一个video_id（或 execution_id），真正的视频渲染在云端排队执行。因此客户端必须周期性地查询状态接口，直到任务进入completed或failed终态。

在 OpenMontage 仓库中，HeyGen 相关能力封装为heygen_video工具（tools/video/heygen_video.py），其内部通过 tools/video/_shared.py 中的generate_heygen_video与poll_heygen完成“提交 + 轮询 + 下载”的完整闭环。该工具被标记为ToolStability.EXPERIMENTAL、ExecutionMode.SYNC，runtime = ToolRuntime.API，并声明了HEYGEN_API_KEY环境变量作为可用性前提（get_status()在缺少该环境变量时返回UNAVAILABLE）。

检查视频状态的两种途径

途径一：MCP 工具（首选）

如果已连接 HeyGen MCP 服务器，优先使用mcp__heygen__get_video，传入videoId参数，一次调用即可同时拿到：

• status（当前状态）
• video_url（成片下载地址）
• thumbnail_url（缩略图）
• duration（时长，秒）
• title（标题）
• gif_url（GIF 版本）
• captioned_video_url（带字幕版本）
• 以及其他元数据

MCP 工具自动处理认证与请求格式化，省去手工拼接X-Api-Key的繁琐与出错可能。这与仓库 skill 文档（.claude/skills/heygen/SKILL.md）中的工具选择原则一致：有 MCP 工具优先用 MCP 工具，无 MCP 工具时回退到直接 HTTP API。需要说明的是，.claude/skills/heygen/SKILL.md已标记为 DEPRECATED（推荐改用create-video与avatar-video两个聚焦 skill），但其中关于状态查询与轮询的模式完全适用于新 skill 的同类流程。

途径二：直接调用 REST API（无 MCP 时）

状态查询端点为GET https://api.heygen.com/v2/videos/{video_id}，认证方式为请求头X-Api-Key。环境变量配置方式参见 authentication.md 参考文档：通过export HEYGEN_API_KEY="your-api-key-here"或.env文件注入，所有 HeyGen 请求都依赖该密钥。

curl 示例：

curl -X GET "https://api.heygen.com/v2/videos/YOUR_VIDEO_ID" \ -H "X-Api-Key: $HEYGEN_API_KEY"

TypeScript 示例（fetch）：

interface VideoStatusResponse { error: null | string; data: { id: string; status: "pending" | "processing" | "completed" | "failed"; video_url?: string; thumbnail_url?: string; duration?: number; title?: string; created_at?: string; completed_at?: string; gif_url?: string; captioned_video_url?: string; subtitle_url?: string; folder_id?: string; output_language?: string; failure_code?: string; failure_message?: string; }; } async function getVideoStatus(videoId: string): Promise<VideoStatusResponse["data"]> { const response = await fetch( `https://api.heygen.com/v2/videos/${videoId}`, { headers: { "X-Api-Key": process.env.HEYGEN_API_KEY! } } ); const json: VideoStatusResponse = await response.json(); if (json.error) { throw new Error(json.error); } return json.data; }

Python 示例（requests）：

import requests import os def get_video_status(video_id: str) -> dict: response = requests.get( f"https://api.heygen.com/v2/videos/{video_id}", headers={"X-Api-Key": os.environ["HEYGEN_API_KEY"]} ) data = response.json() if data.get("error"): raise Exception(data["error"]) return data["data"]

视频状态类型（Status Types）

状态字段是状态机的心脏，四个取值分别对应任务的不同生命周期：

只有completed与failed是终态：前者携带video_url等可下载产物，后者携带failure_code与failure_message供排查。在 OpenMontage 的仓库实现中，poll_heygen（tools/video/_shared.py）也遵循同样的状态语义：completed时从output.video.video_url（或output.video_url）提取下载地址；failed/error时抛出异常并携带error字段；其余状态继续等待。

预期生成时间与影响因素

视频生成通常需要5–15 分钟，高峰期或脚本较长时可能超过 20 分钟。仓库内的poll_heygen默认把单次轮询总超时设定为600 秒（10 分钟），与官方推荐的“大多数视频 10 分钟内完成”的经验值一致。

实践建议（原文档 Recommendations）：

• 超时时间设置为15–20 分钟（900,000–1,200,000 ms），留足高峰余量；
• 台词超过 2 分钟的视频，直接按 15 分钟以上预估；
• 长视频优先采用异步模式：保存video_id，稍后再查询，而不是让进程长时间挂起等待。

仓库侧的另一项佐证来自heygen_video工具的声明式配置（tools/video/heygen_video.py）：RetryPolicy(max_retries=2, backoff_seconds=10.0, retryable_errors=["rate_limit", "timeout", "server_error"])，即对限流、超时与服务端错误做 2 次重试、10 秒退避，这与“队列负载可能导致等待”的现实场景相呼应。

响应格式解读

完成态（completed）示例

{ "error": null, "data": { "id": "abc123", "status": "completed", "video_url": "https://files.heygen.ai/video/abc123.mp4", "thumbnail_url": "https://files.heygen.ai/thumbnail/abc123.jpg", "duration": 45.2, "title": "My Video", "created_at": "2024-01-15T10:30:00Z", "completed_at": "2024-01-15T10:38:00Z", "gif_url": "https://files.heygen.ai/gif/abc123.gif", "captioned_video_url": null, "subtitle_url": null, "folder_id": null, "output_language": "en" } }

失败态（failed）示例

{ "error": null, "data": { "id": "abc123", "status": "failed", "failure_code": "script_too_long", "failure_message": "Script too long for selected avatar" } }

失败态的关键在于failure_code/failure_message——它们是可操作（actionable）的反馈，应在日志与告警中原样保留，供用户修正脚本长度、Avatar 选择或配额问题。

轮询实现：从基础版到工程化

基础轮询（TypeScript）

async function waitForVideo( videoId: string, maxWaitMs = 600000, // 10 minutes pollIntervalMs = 5000 // 5 seconds ): Promise<string> { const startTime = Date.now(); while (Date.now() - startTime < maxWaitMs) { const status = await getVideoStatus(videoId); switch (status.status) { case "completed": return status.video_url!; case "failed": throw new Error(status.failure_message || "Video generation failed"); case "pending": case "processing": await new Promise((resolve) => setTimeout(resolve, pollIntervalMs)); break; } } throw new Error("Video generation timed out"); }

带进度回调的轮询（TypeScript）

type ProgressCallback = (status: string, elapsed: number) => void; async function waitForVideoWithProgress( videoId: string, onProgress?: ProgressCallback, maxWaitMs = 600000, pollIntervalMs = 5000 ): Promise<string> { const startTime = Date.now(); while (Date.now() - startTime < maxWaitMs) { const elapsed = Date.now() - startTime; const status = await getVideoStatus(videoId); onProgress?.(status.status, elapsed); switch (status.status) { case "completed": return status.video_url!; case "failed": throw new Error(status.failure_message || "Video generation failed"); default: await new Promise((resolve) => setTimeout(resolve, pollIntervalMs)); } } throw new Error("Video generation timed out"); } // Usage const videoUrl = await waitForVideoWithProgress( videoId, (status, elapsed) => { console.log(`Status: ${status}, Elapsed: ${Math.round(elapsed / 1000)}s`); } );

Python 轮询

import time from typing import Optional, Callable def wait_for_video( video_id: str, max_wait_seconds: int = 600, poll_interval: int = 5, on_progress: Optional[Callable[[str, int], None]] = None ) -> str: start_time = time.time() while time.time() - start_time < max_wait_seconds: elapsed = int(time.time() - start_time) status_data = get_video_status(video_id) status = status_data["status"] if on_progress: on_progress(status, elapsed) if status == "completed": return status_data["video_url"] elif status == "failed": raise Exception(status_data.get("failure_message", "Video generation failed")) time.sleep(poll_interval) raise Exception("Video generation timed out") # Usage def progress_callback(status: str, elapsed: int): print(f"Status: {status}, Elapsed: {elapsed}s") video_url = wait_for_video(video_id, on_progress=progress_callback)

仓库源码中的轮询范式：指数退避

上述示例使用的是固定 5 秒间隔；OpenMontage 的poll_heygen（tools/video/_shared.py）则展示了更贴合生产实践的指数退避轮询：

interval = 5.0 while time.time() < deadline: response = requests.get(url, headers=headers, timeout=30) data = response.json().get("data", {}) status = data.get("status", "") if status == "completed": video_url = ( data.get("output", {}).get("video", {}).get("video_url") or data.get("output", {}).get("video_url") ) if video_url: return video_url raise RuntimeError(f"Completed but no video_url in output: {data}") if status in {"failed", "error"}: raise RuntimeError(f"HeyGen generation failed: {data.get('error', 'Unknown')}") time.sleep(min(interval, max(0.0, deadline - time.time()))) interval = min(interval * 1.2, 30.0)

关键设计点：

• 轮询间隔动态增长：从 5 秒起步，每次乘以 1.2，封顶 30 秒——长时间任务下减少请求次数，降低 API 配额消耗与限流概率；
• 总超时兜底：超过deadline（默认 600 秒）抛出TimeoutError，避免无限挂起；
• completed 但无 URL 兜底：即使状态为 completed，若输出中找不到video_url也会抛出明确异常，防止静默返回空值；
• 请求层带 30 秒超时，避免网络抖动时请求悬挂。

下载成片：重试与退避

状态显示completed之后，video_url不一定立即可用——文件可能仍在 CDN 同步。因此下载必须使用重试逻辑（retry with backoff）。

TypeScript（带重试，指数退避）

import fs from "fs"; import path from "path"; async function downloadVideoWithRetry( videoUrl: string, outputPath = "./output/video.mp4", maxRetries = 5, initialDelayMs = 2000 ): Promise<void> { let lastError: Error | null = null; for (let attempt = 0; attempt < maxRetries; attempt++) { try { const response = await fetch(videoUrl); if (!response.ok) { throw new Error(`HTTP ${response.status}: ${response.statusText}`); } const arrayBuffer = await response.arrayBuffer(); fs.writeFileSync(path.resolve(outputPath), Buffer.from(arrayBuffer)); console.log(`Video downloaded to ${outputPath}`); return; } catch (error) { lastError = error as Error; const delay = initialDelayMs * Math.pow(2, attempt); // Exponential backoff console.log(`Download attempt ${attempt + 1} failed, retrying in ${delay}ms...`); await new Promise((resolve) => setTimeout(resolve, delay)); } } throw new Error(`Failed to download after ${maxRetries} attempts: ${lastError?.message}`); }

Python（带重试，指数退避）

import requests import time def download_video_with_retry( video_url: str, output_path: str, max_retries: int = 5, initial_delay: float = 2.0 ) -> None: last_error = None for attempt in range(max_retries): try: response = requests.get(video_url, stream=True, timeout=60) response.raise_for_status() with open(output_path, "wb") as f: for chunk in response.iter_content(chunk_size=8192): f.write(chunk) print(f"Video downloaded to {output_path}") return except Exception as e: last_error = e delay = initial_delay * (2 ** attempt) # Exponential backoff print(f"Download attempt {attempt + 1} failed, retrying in {delay}s...") time.sleep(delay) raise Exception(f"Failed to download after {max_retries} retries: {last_error}")

简单下载（无重试）

适合可人工重试的快速脚本：

async function downloadVideo(videoUrl: string, outputPath = "./output/video.mp4") { const response = await fetch(videoUrl); if (!response.ok) { throw new Error(`Failed to download: ${response.status}`); } const arrayBuffer = await response.arrayBuffer(); fs.writeFileSync(path.resolve(outputPath), Buffer.from(arrayBuffer)); }

仓库源码中的下载与落盘

generate_heygen_video（tools/video/_shared.py）演示了“轮询到 URL → 请求下载 → 写盘 → 返回 ToolResult”的完整链路：

video_url = poll_heygen(execution_id, api_key, timeout=600) output_path = Path(inputs.get("output_path", f"heygen_video_{execution_id}.mp4")) output_path.parent.mkdir(parents=True, exist_ok=True) download = requests.get(video_url, timeout=120) download.raise_for_status() output_path.write_bytes(download.content)

实现细节值得借鉴：

• 下载请求单独设置 120 秒超时（成片可能很大，需要比状态查询更宽裕的预算）；
• 自动创建输出目录（mkdir(parents=True, exist_ok=True)）；
• 返回的ToolResult中携带execution_id、output、format: "mp4"与artifacts，方便上层任务系统记录与追踪（对应heygen_video工具声明的side_effects = ["writes video file to output_path", "calls HeyGen API"]）。

完整工作流示例：生成 → 轮询 → 下载

将前三步串成一个函数，即得到可直接复用的端到端流程（TypeScript）：

async function generateAndDownloadVideo(config: VideoConfig): Promise<string> { // 1. Generate video const generateResponse = await fetch( "https://api.heygen.com/v2/video/generate", { method: "POST", headers: { "X-Api-Key": process.env.HEYGEN_API_KEY!, "Content-Type": "application/json", }, body: JSON.stringify(config), } ); const { data: generateData } = await generateResponse.json(); const videoId = generateData.video_id; console.log(`Video ID: ${videoId}`); // 2. Poll for completion const videoUrl = await waitForVideoWithProgress( videoId, (status, elapsed) => { console.log(`[${Math.round(elapsed / 1000)}s] Status: ${status}`); } ); // 3. Download const outputPath = `./output/${videoId}.mp4`; await downloadVideo(videoUrl, outputPath); return outputPath; }

在 OpenMontage 的工具层，等价流程由generate_heygen_video完成。它先把请求包装为workflow_type: "GenerateVideoNode"提交到POST https://api.heygen.com/v1/workflows/executions（tools/video/_shared.py），从响应中取出execution_id作为轮询句柄，再走poll_heygen→ 下载落盘。若operation为image_to_video，还会先通过upload_image_heygen将本地参考图上传为公开 URL，再注入reference_image_url字段（tools/video/_shared.py）。

可恢复的状态检查：长任务的异步模式

对于超长视频（可能超过 20 分钟），与其让一个进程长期占用，不如保存 video_id、稍后再查。这也是原文档强调的异步模式。

生成后保存状态（Save State After Generation）

interface PendingVideo { videoId: string; createdAt: string; script: string; avatarId: string; voiceId: string; } async function startVideoGeneration(config: VideoGenerateRequest): Promise<PendingVideo> { const videoId = await generateVideo(config); const pending: PendingVideo = { videoId, createdAt: new Date().toISOString(), script: config.video_inputs[0].voice.input_text!, avatarId: config.video_inputs[0].character.avatar_id!, voiceId: config.video_inputs[0].voice.voice_id!, }; // Save to file for later retrieval fs.writeFileSync("pending-video.json", JSON.stringify(pending, null, 2)); console.log(`Video generation started. ID: ${videoId}`); console.log("Check status later with: checkVideoStatus()"); return pending; }

稍后查询状态（Check Status Later）

async function checkVideoStatus(): Promise<void> { if (!fs.existsSync("pending-video.json")) { console.log("No pending video found"); return; } const pending: PendingVideo = JSON.parse( fs.readFileSync("pending-video.json", "utf-8") ); const elapsed = Date.now() - new Date(pending.createdAt).getTime(); console.log(`Checking video ${pending.videoId} (started ${Math.round(elapsed / 60000)} min ago)...`); const status = await getVideoStatus(pending.videoId); switch (status.status) { case "completed": console.log(`Video ready: ${status.video_url}`); console.log(`Duration: ${status.duration}s`); // Clean up pending file fs.unlinkSync("pending-video.json"); // Save result fs.writeFileSync("video-result.json", JSON.stringify({ ...pending, videoUrl: status.video_url, thumbnailUrl: status.thumbnail_url, duration: status.duration, title: status.title, createdAt: status.created_at, completedAt: status.completed_at, }, null, 2)); break; case "failed": console.error(`Video failed: ${status.failure_message}`); fs.unlinkSync("pending-video.json"); break; default: console.log(`Status: ${status.status} - check again in a few minutes`); } }

CLI 友好模式

// generate-video.ts - Start generation and exit async function main() { const pending = await startVideoGeneration(config); console.log(`\nVideo ID saved. Run 'npx tsx check-status.ts' to check progress.`); process.exit(0); // Exit immediately, don't wait } // check-status.ts - Check and optionally wait async function main() { const args = process.argv.slice(2); const shouldWait = args.includes("--wait"); if (shouldWait) { // Poll until complete (with 20 min timeout) const result = await waitForVideo(pending.videoId, apiKey, onProgress, 1200000); console.log(`Done: ${result.video_url}`); } else { // Just check once and report await checkVideoStatus(); } }

该模式与 OpenMontage 中heygen_video工具"先提交、再轮询、最后下载"的执行语义一致，区别在于把进程内的阻塞等待替换为跨进程的持久化状态，更适合批处理（batch）与长任务场景。仓库侧ToolRuntime.API、ExecutionMode.SYNC的声明也从侧面说明：单次工具调用内是同步等待，而面向生产的高吞吐场景则应把状态检查外置化。

替代方案：使用 Webhooks 而非轮询

轮询是简单可靠的兜底，但对生产系统而言，Webhook 推送更高效——不需要维护长连接或周期性请求，HeyGen 在视频完成时主动通知你的服务。

原文档明确指向了 webhooks.md 参考文档，该文档进一步给出了事件类型表（如avatar_video.success表示视频生成完成、avatar_video.fail表示失败、video_translate.success表示翻译完成）、Webhook 注册方式（events数组订阅指定事件类型）、事件负载结构（event_type+event_data，其中event_data携带video_id、video_url、callback_id）以及幂等处理（同一事件可能重复投递，需做去重）等细节。

选型建议：

• 单次、交互式生成：轮询足够，实现简单，出错易排查；
• 生产流水线、批量生成：优先 Webhook，配合队列与去重，避免无意义的轮询请求；
• 也可以两者组合：Webhook 为主、轮询兜底，覆盖 Webhook 丢失或延迟的极端情况。

最佳实践清单

1. 使用指数退避（exponential backoff）——长时间任务逐步拉大轮询间隔（参考poll_heygen的 5s → ×1.2 → 封顶 30s 策略），降低配额消耗；
2. 设置合理超时——大多数视频 10 分钟内完成，官方建议预留 15–20 分钟（900,000–1,200,000 ms）安全余量；
3. 优雅处理失败——完整读取failure_code/failure_message，将其作为可操作的反馈写入日志与告警，不要吞掉异常；
4. 考虑 Webhooks——生产环境用推送替代轮询，参见 webhooks.md；
5. 缓存视频 URL——下载链接的有效期有限，completed后应尽快下载并本地缓存，避免 URL 过期导致 404；
6. 下载必须带重试——状态为 completed 不代表 URL 立即可用，CDN 同步窗口期内请求可能失败，务必用带退避的重试逻辑包裹下载；
7. completed 无 URL 视为异常——正如poll_heygen所做，completed 却取不到video_url时应显式抛错，而不是返回空串。

与 OpenMontage 仓库的结合方式

在本仓库中，上述模式已被封装为可被 Agent 直接调用的heygen_video工具：

• 工具入口：tools/video/heygen_video.py 定义了name = "heygen_video"、provider = "heygen"以及input_schema（prompt必填，支持text_to_video/image_to_video两种 operation，aspect_ratio可选16:9/9:16/1:1，默认veo_3_1供应商变体），并声明fallback = "wan_video"及一整套本地/云端备用工具；
• 核心实现：tools/video/_shared.py 中的generate_heygen_video（提交 → 轮询 → 下载）与poll_heygen（指数退避轮询）；
• 供应商矩阵：HEYGEN_PROVIDERS（tools/video/_shared.py）列出可用的云端模型变体（如veo_3_1、veo3_fast、kling_pro、sora_v2_pro、runway_gen4、seedance_pro、ltx_distilled等），并标注了各自的quality与speed元数据，用于estimate_cost/estimate_runtime的预估；
• 认证前提：HEYGEN_API_KEY环境变量（见 authentication.md），工具在缺少该变量时返回UNAVAILABLE并提供安装指引。

理解video-status.md中的轮询与下载模式，是正确使用这套工具链、乃至自行扩展 HeyGen 集成（批处理、Webhook、定时巡检）的基础。把"提交 → 轮询 → 下载 → 重试"这套状态机逻辑吃透，你就能在任意语言、任意任务框架中稳定地消费 HeyGen 的异步视频能力。

【免费下载链接】OpenMontageWorld's 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