Webhooki Image & Video

Przegląd

Generowanie wideo może trwać kilka minut, więc utrzymywanie otwartego odpytywania jest kosztowne. Włącz dostarczanie przez webhook dla generowania, a ElevenLabs wyśle do twojego endpointu zdarzenie flows_generation, gdy generowanie osiągnie status completed lub failed.

Ładunek zdarzenia to końcowa odpowiedź z odpowiadającego endpointu GET, więc handler, który obsługuje już odpowiedź z odpytywania, nie potrzebuje osobnej ścieżki parsowania.

Zanim zaczniesz

Dostarczanie przez webhook używa webhooków w twoim workspace, które subskrybują zdarzenia generowania. Konfiguracja wymaga dwóch kroków: utwórz webhook, a potem zasubskrybuj go do zdarzenia.

{
  "events": ["flows"]
}

Tworzenie i subskrybowanie webhooków wymaga uprawnienia Webhooks Manage lub roli administratora workspace. Pojedyncze zdarzenie może przyjąć maksymalnie 10 webhooków; po przekroczeniu tego limitu żądanie kończy się błędem too_many_webhooks.

Generowanie, które żąda dostarczenia przez webhook, gdy żaden webhook nie subskrybuje zdarzeń generowania, zostaje odrzucone, więc wynik nigdy nie jest generowany bez miejsca docelowego.

Zażądaj dostarczenia przez webhook

Dodaj obiekt webhook do żądania utworzenia. Użyj {"type": "all"}, aby dostarczać do każdego webhooka subskrybującego zdarzenia generowania — dzięki temu żądanie pozostaje stabilne, gdy webhooki są dodawane lub zastępowane.

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

Aby kierować dostarczanie do konkretnych webhooków, ustaw pole webhook jako listę identyfikatorów. Każdy identyfikator musi należeć do webhooka workspace subskrybującego zdarzenia generowania.

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

Żądanie utworzenia sprawdza cel przed rozpoczęciem generowania i zwraca błąd, gdy dostarczenie nie byłoby możliwe:

Dostarczanie przez webhook dobrze współgra z łańcuchami generowania: ustaw webhook dla końcowego generowania, a cały łańcuch wykona się po stronie serwera z jednym zdarzeniem na końcu. Dotyczy to także sytuacji, gdy łańcuch nie powiedzie się w trakcie — błąd przechodzi do końcowego generowania, które dostarcza go jako zdarzenie failed z przyczyną dependency_failed.

Ładunek webhooka

Zakończone generowanie dostarcza adres URL wyjścia i typ 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"
  }
}

Nieudane generowanie dostarcza zamiast tego kategorię błędu i komunikat:

{
  "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."
  }
}

Sprawdź data.status, aby ustalić, które pola są dostępne. Dwa statusy końcowe to jedyne, które może zawierać webhook, ponieważ dostarczenie następuje tylko po zakończeniu generowania.

Obsłuż zdarzenie

Handler weryfikuje podpis, sprawdza typ zdarzenia, a potem rozgałęzia się według data.status. Ten przykład pobiera wynik zakończonego generowania i zapisuje w logach przyczynę nieudanego.

# 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"}

Dla zwięzłości oba przykłady pobierają plik w trakcie żądania. Duże wideo może pobierać się na tyle długo, że przekroczy limit czasu dostarczenia, więc na produkcji przekaż identyfikator generowania do kolejki i od razu zwróć 2xx. Podpisany URL jest ważny około godziny, więc to wystarczy dla workera działającego w tle.

Zweryfikuj podpis

Powyższy handler wywołuje construct_event / constructEvent, co w jednym kroku weryfikuje nagłówek ElevenLabs-Signature, sprawdza znacznik czasu i parsuje ładunek. Zawsze weryfikuj zdarzenie, zanim mu zaufasz.

Ważne, by odbiornik weryfikował wszystkie przychodzące webhooki. Webhooki obecnie obsługują uwierzytelnianie za pomocą podpisów HMAC. Aby skonfigurować uwierzytelnianie HMAC:

• Bezpiecznie przechowuj współdzielony sekret wygenerowany podczas tworzenia webhooka
• Zweryfikuj nagłówek ElevenLabs-Signature w swoim endpointzie za pomocą SDK

SDK JavaScript udostępnia constructEvent, a SDK Python construct_event z parametrami rawBody, sig_header i secret (w Pythonie nie nazywają się one payload / signature). Oba weryfikują podpis, sprawdzają znacznik czasu i parsują dane 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"}

Zachowanie dostarczania

Każde generowanie dostarcza dokładnie jedno zdarzenie końcowe do każdego docelowego webhooka. Dostarczanie jest niezależne od samego generowania: webhook, który zawiedzie lub jest niedostępny, nie wpływa na wynik, który pozostaje dostępny z endpointu GET i w odpowiedzi listy.

Szybko zwracaj status 2xx z handlera. Powtarzające się błędy automatycznie wyłączają webhook, a wyłączony webhook powoduje odrzucenie kolejnych generowań, które go wskazują, już przy tworzeniu. Zaprojektuj handler tak, by był idempotentny, i używaj id generowania do usuwania duplikatów.

W workflow, w których pominięcie wyniku jest niedopuszczalne, traktuj webhooki jako szybką ścieżkę i okresowo uzgadniaj wyniki za pomocą flows.image.list lub flows.video.list, filtrując po status.

Kolejne kroki