图像和视频 Webhooks

接收生成结果,无需轮询。

操作指南 · 假设你已完成 图像和视频快速入门。

概览

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

事件载荷是对应 GET 端点的最终响应,因此已能处理轮询响应的处理程序无需单独解析。

开始前

Webhook 投递使用工作区订阅了生成事件的 webhook。设置需要两步:创建 webhook,然后订阅事件。

1

创建 webhook

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

2

订阅生成事件

在 选择要监听的事件 下,勾选 Image & Video API generation completed。已创建但未订阅此事件的 webhook 不会被调用。

也可以通过 API 完成:向 更新工作区 webhook 传递 flows 事件:

{
"events": ["flows"]
}

创建和订阅 webhook 需要 Webhooks Manage 权限或工作区管理员权限。单个事件最多可接受 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 负载。

使用 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 过滤)进行核对。

后续步骤