Image & Video Webhook

概要

ビデオ生成には数分かかることがあり、ポーリング接続を維持するコストは高くなります。生成でWebhook配信を有効にすると、生成がcompletedまたはfailedになると、ElevenLabsがエンドポイントにflows_generationイベントを送信します。

イベントペイロードは対応するGETエンドポイントの最終レスポンスです。そのため、すでにポーリングレスポンスを処理できるハンドラーであれば、別のパース処理は必要ありません。

始める前に

Webhook配信では、ワークスペースで生成イベントを購読しているWebhookを使用します。設定は、Webhookの作成とイベントの購読という2つの手順で行います。

{
  "events": ["flows"]
}

Webhookの作成と購読には、Webhooks Manage権限またはワークスペース管理者権限が必要です。1つのイベントにつき最大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"]
  }
}

作成リクエストでは、生成開始前に対象を検証します。配信できない場合はエラーを返します。

Webhook配信は、連鎖生成と相性がよい機能です。最後の生成にwebhookを設定すると、チェーン全体がサーバーサイドで実行され、最後に1つのイベントだけが送信されます。チェーンの途中で失敗した場合も同様です。失敗は最後の生成まで連鎖し、dependency_failed理由を持つ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が配信されるのは生成完了時だけなので、Webhookに含まれる終端ステータスはこの2つだけです。

イベントを処理する

ハンドラーでは、署名を検証し、イベントタイプを確認してから、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時間有効で、バックグラウンドワーカーには十分な時間です。

署名を検証する

上記のハンドラーではconstruct_event／constructEventを呼び出します。これにより、ElevenLabs-Signatureヘッダーの検証、タイムスタンプの検証、ペイロードのパースを一度に行います。イベントを信頼する前に必ず検証してください。

受信側は、すべての受信Webhookを検証することが重要です。Webhookは現在、HMAC署名による認証に対応しています。HMAC認証は次の手順で設定します。

• Webhookの作成時に生成された共有シークレットを安全に保存する
• SDKを使用してエンドポイントでElevenLabs-Signatureヘッダーを検証する

JavaScript SDKではconstructEventを、Python SDKでは**rawBody、sig_header、secret**を指定するconstruct_eventを利用できます（Pythonではpayload／signatureという名前ではありません）。どちらも署名の検証、タイムスタンプの検証、JSONペイロードの解析を行います。

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ごとに終端イベントが1つだけ配信されます。配信は生成そのものから独立しています。Webhookが失敗または到達不能になっても結果には影響せず、結果はGETエンドポイントとリストレスポンスで引き続き利用できます。

ハンドラーからは速やかに2xxステータスを返してください。失敗が繰り返されるとWebhookは自動的に無効化され、無効化されたWebhookを対象にする後続の生成は作成時に拒否されます。ハンドラーは冪等に設計し、生成idを使って重複を排除してください。

結果の見逃しが許容できないワークフローでは、Webhookを高速パスとして扱い、statusでフィルタリングしたflows.image.listまたはflows.video.listと定期的に照合してください。

次のステップ