Image & Video Webhook | ElevenLabs Documentation

操作指南 · 假设你已完成 Image & Video 快速入门。

概览

视频生成可能需要几分钟，持续轮询的成本较高。为生成任务启用 webhook 推送后，生成达到 completed 或 failed 时，ElevenLabs 会向你的端点发送 flows_generation 事件。

事件负载与对应 GET 端点的最终响应相同，因此已能处理轮询响应的处理程序无需额外解析逻辑。

开始前

Webhook 推送使用工作区中订阅了生成事件的 webhook。设置分两步：创建 webhook，然后订阅事件。

创建 webhook

前往 开发者 > Webhooks，创建一个使用可公开访问 HTTPS 回调 URL 的 webhook。请保留返回的签名密钥；验证传入事件时需要它。

订阅生成事件

在 选择要监听的事件 下，勾选 Image & Video API 生成或模板运行完成。同一事件也会推送已完成的模板运行。已创建但未订阅此事件的 webhook 永远不会被调用。

你也可以通过 API 将 flows 事件传给更新工作区 webhook来完成：

{
  "events": ["flows"]
}

创建和订阅 webhook 需要 Webhooks 管理权限或工作区管理员权限。单个事件最多可接收 10 个 webhook；超过此数量，请求会因 too_many_webhooks 而失败。

当没有 webhook 订阅生成事件时，要求 webhook 推送的生成任务会被拒绝，确保不会生成无处可发送的结果。

请求 webhook 推送

在创建请求中添加 webhook 对象。使用 {"type": "all"} 可推送至所有订阅生成事件的 webhook，因此即使新增或替换 webhook，请求也无需变更。

from elevenlabs import VideoGenerationRequest_Veo31FastGenerate001, WebhookTarget_All
generation = elevenlabs.flows.video.create(
    request=VideoGenerationRequest_Veo31FastGenerate001(
        prompt="A corgi rides a tiny surfboard across a sunlit wave at golden hour, cinematic",
        duration_secs=8,
        webhook=WebhookTarget_All(),
    )
)

如需指定 webhook，请将 webhook 字段设为 ID 列表。每个 ID 都必须属于工作区中订阅生成事件的 webhook。

{
  "webhook": {
    "type": "ids",
    "ids": ["Q8mVr2LpXcT4nB6yJdKw"]
  }
}

创建请求会在启动生成前验证目标；若无法推送，会返回错误：

错误状态	原因
`no_webhooks_configured`	请求推送至所有 webhook，但工作区没有 webhook。
`invalid_webhook_id`	列出的 webhook 未订阅生成事件，或已不存在。
`webhook_disabled`	指定的 webhook 已被禁用，无论是手动禁用还是因失败而自动禁用。

Webhook 推送很适合用于链式生成：在最终生成任务中设置 webhook，整条链便会在服务端运行，并在结束时只发送一个事件。即使链在中途失败也是如此——失败会级联到最终生成任务，并以 failed 事件和 dependency_failed 原因推送。

Webhook 负载

已完成的生成任务会推送输出 URL 和 MIME 类型：

{
  "type": "flows_generation",
  "event_timestamp": 1739721600,
  "data": {
    "id": "JWr5N6X9ZTqf8jD2LmQb",
    "status": "completed",
    "content_url": "https://storage.googleapis.com/generations/JWr5N6X9ZTqf8jD2LmQb",
    "content_mime_type": "video/mp4"
  }
}

失败的生成任务则会推送失败类别和消息：

{
  "type": "flows_generation",
  "event_timestamp": 1739721600,
  "data": {
    "id": "JWr5N6X9ZTqf8jD2LmQb",
    "status": "failed",
    "failure_reason": "timeout",
    "error_message": "Timed out while processing. You were not charged for this generation."
  }
}

根据 data.status 判断存在哪些字段。Webhook 只能携带这两种最终状态，因为仅会在生成完成后推送。

content_url 是签名 URL，在事件发送约 1 小时后过期。请及时下载媒体，或重新获取生成任务以取得新的 URL。

处理事件

处理程序会验证签名、检查事件类型，然后根据 data.status 分支。此示例会下载已完成生成任务的输出，并记录失败任务的原因。

# server.py
import os
import requests
from dotenv import load_dotenv
from elevenlabs.client import ElevenLabs
from elevenlabs.errors import BadRequestError
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
load_dotenv()
app = FastAPI()
elevenlabs = ElevenLabs(api_key=os.getenv("ELEVENLABS_API_KEY"))
WEBHOOK_SECRET = os.getenv("WEBHOOK_SECRET")
@app.post("/webhook/flows")
async def receive_generation(request: Request):
    payload = await request.body()
    signature = request.headers.get("elevenlabs-signature")
    try:
        event = elevenlabs.webhooks.construct_event(
            rawBody=payload.decode("utf-8"),
            sig_header=signature,
            secret=WEBHOOK_SECRET,
        )
    except BadRequestError:
        return JSONResponse(content={"error": "Invalid signature"}, status_code=401)
    # construct_event returns a parsed dict, not an object with attributes.
    if event.get("type") != "flows_generation":
        return {"status": "ignored"}
    generation = event["data"]
    if generation["status"] == "completed":
        media = requests.get(generation["content_url"]).content
        with open(f"{generation['id']}.mp4", "wb") as f:
            f.write(media)
    else:
        print(f"Generation {generation['id']} failed: {generation['failure_reason']}")
    return {"status": "received"}

为简洁起见，两个示例都会在请求期间下载。大型视频可能耗时较长，超过推送超时时间，因此生产环境中应将生成任务 ID 交给队列，并立即返回 2xx。签名 URL 的有效期约为 1 小时，足够后台工作器使用。

开发期间如需在本地服务器接收事件，可通过 ngrok 等隧道将其公开，并使用它提供的 HTTPS URL 作为 webhook 回调 URL。

验证签名

上述处理程序调用 construct_event / constructEvent，一步即可验证 ElevenLabs-Signature 标头、验证时间戳并解析负载。始终应在信任事件前进行验证。

监听器必须验证所有传入的 webhook。Webhook 目前支持通过 HMAC 签名进行身份验证。可按以下方式设置 HMAC 身份验证：

安全存储创建 webhook 时生成的共享密钥
使用 SDK 在端点中验证 ElevenLabs-Signature 请求头

JavaScript SDK 提供 constructEvent；Python SDK 提供 construct_event，并使用 rawBody、sig_header 和 secret（在 Python 中，它们不叫 payload / signature）。两者都会验证签名、验证时间戳，并解析 JSON 负载。

Python

JavaScript

使用 FastAPI 的 webhook 处理程序示例：

from dotenv import load_dotenv
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from elevenlabs.client import ElevenLabs
from elevenlabs.errors import BadRequestError
import os
load_dotenv()
app = FastAPI()
elevenlabs = ElevenLabs(
    api_key=os.getenv("ELEVENLABS_API_KEY"),
)
WEBHOOK_SECRET = os.getenv("WEBHOOK_SECRET")
@app.post("/webhook")
async def receive_message(request: Request):
    payload = await request.body()
    signature = request.headers.get("elevenlabs-signature")
    try:
        event = elevenlabs.webhooks.construct_event(
            rawBody=payload.decode("utf-8"),
            sig_header=signature,
            secret=WEBHOOK_SECRET,
        )
    except BadRequestError as e:
        return JSONResponse(content={"error": "Invalid signature"}, status_code=401)
    # construct_event returns a dict (parsed JSON), not an object with attributes
    if event.get("type") == "post_call_transcription":
        print(f"Received transcription: {event.get('data')}")
    return {"status": "received"}

推送行为

每个生成任务会向每个指定的 webhook 精确推送一个最终事件。推送独立于生成任务本身：webhook 失败或无法访问不会影响结果，结果仍可通过 GET 端点和列表响应获取。

处理程序应及时返回 2xx 状态。反复失败会自动禁用 webhook；被禁用的 webhook 会导致后续指定它的生成任务在创建时被拒绝。请将处理程序设计为幂等，并使用生成任务 id 去重。

对于不能接受遗漏结果的工作流，可将 webhook 视为快速路径，并定期通过 flows.image.list 或 flows.video.list 按 status 筛选进行核对。

后续步骤

引用和资源

使用先前的生成任务、上传的资源或内联媒体引导生成。

Webhook 配置

创建、保护和管理工作区的 webhook。

API 参考文档

浏览图像、视频和资源端点。