画像＆ビデオのWebhook

概要

ビデオ生成には数分かかる場合があり、その間ポーリング接続を維持するとコストがかかります。生成でWebhook配信を有効にすると、生成がcompletedまたはfailedに達した時点で、ElevenLabsからエンドポイントにflows_generationイベントが送信されます。

イベントペイロードは対応するGETエンドポイントの最終レスポンスです。そのため、すでにポーリングレスポンスを処理できるハンドラーなら、別の解析処理は不要です。

始める前に

Webhook配信では、ワークスペースで生成イベントに登録済みのWebhookを使用します。設定は2ステップです。まずWebhookを作成し、次にイベントを登録します。

{
  "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で配信されるのはこの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を高速経路として扱い、定期的にflows.image.listまたはflows.video.listをstatusでフィルタリングして照合してください。

次のステップ