> This is a page from the ElevenLabs documentation. For a complete page index, fetch https://elevenlabs.io/docs/llms.txt. For the full documentation in a single file, fetch https://elevenlabs.io/docs/llms-full.txt.

# Image & Video Webhook

> **Note**
>
> **操作指南** · 假设你已完成 [Image & Video 快速入门](/docs/zh/eleven-api/guides/cookbooks/image-and-video)。

## 概览

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

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

## 开始前

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

#### 创建 webhook

前往 [**开发者** > **Webhooks**](https://elevenlabs.io/app/developers/webhooks)，创建一个使用可公开访问 HTTPS 回调 URL 的 webhook。请保留返回的签名密钥；验证传入事件时需要它。

#### 订阅生成事件

在 **选择要监听的事件** 下，勾选 **Image & Video API 生成或模板运行完成**。同一事件也会推送已完成的[模板运行](/docs/zh/eleven-api/guides/how-to/templates/run-a-template)。
已创建但未订阅此事件的 webhook 永远不会被调用。

你也可以通过 API 将 `flows` 事件传给[更新工作区 webhook](/docs/zh/api-reference/webhooks/update)来完成：

```json
{
  "events": ["flows"]
}
```

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

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

## 请求 webhook 推送

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

```python
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(),
    )
)
```

```typescript
const generation = await elevenlabs.flows.video.create({
  modelId: "veo-3.1-fast-generate-001",
  prompt: "A corgi rides a tiny surfboard across a sunlit wave at golden hour, cinematic",
  durationSecs: 8,
  webhook: { type: "all" },
});
```

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

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

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

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

Webhook 推送很适合用于[链式生成](/docs/zh/eleven-api/guides/how-to/image-and-video/references#chain-one-generation-into-the-next)：在最终生成任务中设置 `webhook`，整条链便会在服务端运行，并在结束时只发送一个事件。即使链在中途失败也是如此——失败会级联到最终生成任务，并以 `failed` 事件和 `dependency_failed` 原因推送。

## Webhook 负载

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

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

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

```json
{
  "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 只能携带这两种最终状态，因为仅会在生成完成后推送。

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

## 处理事件

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

```python maxLines=0
# 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"}
```

```typescript maxLines=0
// server.mts
import { ElevenLabsClient } from "@elevenlabs/elevenlabs-js";
import "dotenv/config";
import express from "express";
import { writeFile } from "fs/promises";

const elevenlabs = new ElevenLabsClient();
const app = express();

const WEBHOOK_SECRET = process.env.WEBHOOK_SECRET;

// The raw body is required: verification runs over the exact bytes sent.
app.post("/webhook/flows", express.raw({ type: "application/json" }), async (req, res) => {
  const signature = req.headers["elevenlabs-signature"] as string;

  let event;
  try {
    event = await elevenlabs.webhooks.constructEvent(
      req.body.toString(),
      signature,
      WEBHOOK_SECRET
    );
  } catch {
    return res.status(401).json({ error: "Invalid signature" });
  }

  if (event.type !== "flows_generation") {
    return res.status(200).json({ received: true });
  }

  const generation = event.data;
  if (generation.status === "completed") {
    const response = await fetch(generation.content_url);
    await writeFile(`${generation.id}.mp4`, Buffer.from(await response.arrayBuffer()));
  } else {
    console.error(`Generation ${generation.id} failed: ${generation.failure_reason}`);
  }

  res.status(200).json({ received: true });
});

app.listen(3000);
```

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

> **Tip**
>
> 开发期间如需在本地服务器接收事件，可通过 [ngrok](https://ngrok.com/) 等隧道将其公开，并使用它提供的 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

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

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

#### JavaScript

#### Express

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

```javascript
import { ElevenLabsClient } from '@elevenlabs/elevenlabs-js';
import express from 'express';

const app = express();

const elevenlabs = new ElevenLabsClient();
const WEBHOOK_SECRET = process.env.WEBHOOK_SECRET;

// Use express.text() to preserve raw body for signature verification
app.post('/webhook', express.text({ type: 'application/json' }), async (req, res) => {
  const signature = req.headers['elevenlabs-signature'];
  const payload = req.body; // Raw string body

  let event;
  try {
    event = await elevenlabs.webhooks.constructEvent(payload, signature, WEBHOOK_SECRET);
  } catch (error) {
    return res.status(401).json({ error: 'Invalid signature' });
  }

  // Process the webhook event
  if (event.type === 'post_call_transcription') {
    console.log('Received transcription:', event.data);
  }

  res.status(200).json({ received: true });
});
```

#### Next.js

使用 Next.js API 路由的 webhook 处理程序示例：

**`app/api/webhook/route.ts`**

```typescript app/api/webhook/route.ts
import { NextResponse } from 'next/server';
import type { NextRequest } from 'next/server';
import { ElevenLabsClient } from '@elevenlabs/elevenlabs-js';

const elevenlabs = new ElevenLabsClient();
const WEBHOOK_SECRET = process.env.WEBHOOK_SECRET;

export async function POST(req: NextRequest) {
  const body = await req.text();
  const signature = req.headers.get('elevenlabs-signature');

  let event;
  try {
    event = await elevenlabs.webhooks.constructEvent(body, signature, WEBHOOK_SECRET);
  } catch (error) {
    return NextResponse.json({ error: 'Invalid signature' }, { status: 401 });
  }

  // Process the webhook event
  if (event.type === 'post_call_transcription') {
    console.log('Received transcription:', event.data);
  }

  return NextResponse.json({ received: true }, { status: 200 });
}
```

## 推送行为

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

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

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

## 后续步骤

#### [引用和资源](/docs/zh/eleven-api/guides/how-to/image-and-video/references)

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

#### [Webhook 配置](/docs/zh/eleven-api/resources/webhooks)

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

#### [API 参考文档](/docs/zh/api-reference/flows/image/create)

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