HeyGen 视频状态轮询与下载全指南OpenMontage 项目中的异步视频生成实战【免费下载链接】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 以异步方式处理视频生成提交生成任务后视频并不会立即就绪而是经历排队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.SYNCruntime ToolRuntime.API并声明了HEYGEN_API_KEY环境变量作为可用性前提get_status()在缺少该环境变量时返回UNAVAILABLE。检查视频状态的两种途径途径一MCP 工具首选如果已连接 HeyGen MCP 服务器优先使用mcp__heygen__get_video传入videoId参数一次调用即可同时拿到status当前状态video_url成片下载地址thumbnail_url缩略图duration时长秒title标题gif_urlGIF 版本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_KEYyour-api-key-here或.env文件注入所有 HeyGen 请求都依赖该密钥。curl 示例curl -X GET https://api.heygen.com/v2/videos/YOUR_VIDEO_ID \ -H X-Api-Key: $HEYGEN_API_KEYTypeScript 示例fetchinterface 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): PromiseVideoStatusResponse[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 示例requestsimport requests import os def get_video_status(video_id: str) - dict: response requests.get( fhttps://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状态字段是状态机的心脏四个取值分别对应任务的不同生命周期StatusDescriptionpendingVideo is queued for processingprocessingVideo is being generatedcompletedVideo is ready for downloadfailedVideo generation failed只有completed与failed是终态前者携带video_url等可下载产物后者携带failure_code与failure_message供排查。在 OpenMontage 的仓库实现中poll_heygentools/video/_shared.py也遵循同样的状态语义completed时从output.video.video_url或output.video_url提取下载地址failed/error时抛出异常并携带error字段其余状态继续等待。预期生成时间与影响因素视频生成通常需要5–15 分钟高峰期或脚本较长时可能超过 20 分钟。仓库内的poll_heygen默认把单次轮询总超时设定为600 秒10 分钟与官方推荐的“大多数视频 10 分钟内完成”的经验值一致。FactorImpactScript lengthLonger scripts significantly longer processingResolution1080p takes longer than 720pAvatar complexitySome avatars render fasterQueue loadPeak hours may cause 15-20 minute waitsMultiple scenesEach scene adds processing time实践建议原文档 Recommendations超时时间设置为15–20 分钟900,000–1,200,000 ms留足高峰余量台词超过 2 分钟的视频直接按 15 分钟以上预估长视频优先采用异步模式保存video_id稍后再查询而不是让进程长时间挂起等待。仓库侧的另一项佐证来自heygen_video工具的声明式配置tools/video/heygen_video.pyRetryPolicy(max_retries2, backoff_seconds10.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 选择或配额问题。轮询实现从基础版到工程化基础轮询TypeScriptasync function waitForVideo( videoId: string, maxWaitMs 600000, // 10 minutes pollIntervalMs 5000 // 5 seconds ): Promisestring { 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); }带进度回调的轮询TypeScripttype ProgressCallback (status: string, elapsed: number) void; async function waitForVideoWithProgress( videoId: string, onProgress?: ProgressCallback, maxWaitMs 600000, pollIntervalMs 5000 ): Promisestring { 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(fStatus: {status}, Elapsed: {elapsed}s) video_url wait_for_video(video_id, on_progressprogress_callback)仓库源码中的轮询范式指数退避上述示例使用的是固定 5 秒间隔OpenMontage 的poll_heygentools/video/_shared.py则展示了更贴合生产实践的指数退避轮询interval 5.0 while time.time() deadline: response requests.get(url, headersheaders, timeout30) 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(fCompleted but no video_url in output: {data}) if status in {failed, error}: raise RuntimeError(fHeyGen 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 ): Promisevoid { 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, streamTrue, timeout60) response.raise_for_status() with open(output_path, wb) as f: for chunk in response.iter_content(chunk_size8192): f.write(chunk) print(fVideo downloaded to {output_path}) return except Exception as e: last_error e delay initial_delay * (2 ** attempt) # Exponential backoff print(fDownload attempt {attempt 1} failed, retrying in {delay}s...) time.sleep(delay) raise Exception(fFailed 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_videotools/video/_shared.py演示了“轮询到 URL → 请求下载 → 写盘 → 返回 ToolResult”的完整链路video_url poll_heygen(execution_id, api_key, timeout600) output_path Path(inputs.get(output_path, fheygen_video_{execution_id}.mp4)) output_path.parent.mkdir(parentsTrue, exist_okTrue) download requests.get(video_url, timeout120) download.raise_for_status() output_path.write_bytes(download.content)实现细节值得借鉴下载请求单独设置 120 秒超时成片可能很大需要比状态查询更宽裕的预算自动创建输出目录mkdir(parentsTrue, exist_okTrue)返回的ToolResult中携带execution_id、output、format: mp4与artifacts方便上层任务系统记录与追踪对应heygen_video工具声明的side_effects [writes video file to output_path, calls HeyGen API]。完整工作流示例生成 → 轮询 → 下载将前三步串成一个函数即得到可直接复用的端到端流程TypeScriptasync function generateAndDownloadVideo(config: VideoConfig): Promisestring { // 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/executionstools/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 Generationinterface PendingVideo { videoId: string; createdAt: string; script: string; avatarId: string; voiceId: string; } async function startVideoGeneration(config: VideoGenerateRequest): PromisePendingVideo { 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 Laterasync function checkVideoStatus(): Promisevoid { 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, dont 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_typeevent_data其中event_data携带video_id、video_url、callback_id以及幂等处理同一事件可能重复投递需做去重等细节。选型建议单次、交互式生成轮询足够实现简单出错易排查生产流水线、批量生成优先 Webhook配合队列与去重避免无意义的轮询请求也可以两者组合Webhook 为主、轮询兜底覆盖 Webhook 丢失或延迟的极端情况。最佳实践清单使用指数退避exponential backoff——长时间任务逐步拉大轮询间隔参考poll_heygen的 5s → ×1.2 → 封顶 30s 策略降低配额消耗设置合理超时——大多数视频 10 分钟内完成官方建议预留 15–20 分钟900,000–1,200,000 ms安全余量优雅处理失败——完整读取failure_code/failure_message将其作为可操作的反馈写入日志与告警不要吞掉异常考虑 Webhooks——生产环境用推送替代轮询参见 webhooks.md缓存视频 URL——下载链接的有效期有限completed后应尽快下载并本地缓存避免 URL 过期导致 404下载必须带重试——状态为 completed 不代表 URL 立即可用CDN 同步窗口期内请求可能失败务必用带退避的重试逻辑包裹下载completed 无 URL 视为异常——正如poll_heygen所做completed 却取不到video_url时应显式抛错而不是返回空串。与 OpenMontage 仓库的结合方式在本仓库中上述模式已被封装为可被 Agent 直接调用的heygen_video工具工具入口tools/video/heygen_video.py 定义了name heygen_video、provider heygen以及input_schemaprompt必填支持text_to_video/image_to_video两种 operationaspect_ratio可选16:9/9:16/1:1默认veo_3_1供应商变体并声明fallback wan_video及一整套本地/云端备用工具核心实现tools/video/_shared.py 中的generate_heygen_video提交 → 轮询 → 下载与poll_heygen指数退避轮询供应商矩阵HEYGEN_PROVIDERStools/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 的异步视频能力。【免费下载链接】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），仅供参考