Webhooki Image & Video

Przegląd

Generowanie filmów może trwać kilka minut, więc ciągłe odpytywanie jest kosztowne. Włącz dostarczanie przez webhook dla generowania, a ElevenLabs wyśle zdarzenie flows_generation na twój endpoint, gdy generowanie uzyska status completed lub failed.

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

Zanim zaczniesz

Dostarczanie przez webhook korzysta z webhooków w twoim workspace subskrybowanych do zdarzeń 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. Jedno zdarzenie obsługuje maksymalnie 10 webhooków; powyżej tego limitu żądanie kończy się błędem too_many_webhooks.

Generowanie żądające dostarczenia przez webhook, gdy żaden webhook nie subskrybuje zdarzeń generowania, zostanie odrzucone, więc wynik nigdy nie powstanie bez miejsca, do którego można go dostarczyć.

Poproś o dostarczenie przez webhook

Dodaj obiekt webhook do żądania utworzenia. Użyj {"type": "all"}, aby dostarczać do każdego webhooka subskrybowanego do zdarzeń generowania — żądanie pozostanie stabilne, gdy webhooki będą 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 wskazać konkretne webhooki, ustaw pole webhook jako listę identyfikatorów. Każdy identyfikator musi należeć do webhooka workspace subskrybowanego do zdarzeń 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 sprawdza się przy połączonych generowaniach: ustaw webhook w końcowym generowaniu, a cały łańcuch uruchomi się po stronie serwera i wyśle jedno zdarzenie na końcu. Dotyczy to także błędu w trakcie łańcucha — błąd przechodzi do końcowego generowania, które dostarcza go jako zdarzenie failed z przyczyną dependency_failed.

Ładunek webhooka

Ukończone generowanie dostarcza URL wyniku 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 za to 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."
  }
}

Sprawdzaj data.status, aby ustalić, które pola są dostępne. Te dwa statusy końcowe są jedynymi, które może zawierać webhook, ponieważ dostarczenie następuje dopiero po zakończeniu generowania.

Obsłuż zdarzenie

Handler weryfikuje podpis, sprawdza typ zdarzenia, a potem analizuje data.status. Ten przykład pobiera wynik ukończonego generowania i loguje 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ą dane w ramach żądania. Duży film może być pobierany 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 przez około godzinę, co wystarcza workerowi działającemu 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 na każdy wskazany webhook. Dostarczanie nie zależy od samego generowania: webhook, który zawiedzie lub jest niedostępny, nie wpływa na wynik, który nadal jest 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 sprawia, że kolejne generowania go wskazujące są odrzucane podczas tworzenia. Zaprojektuj handler tak, by był idempotentny, i używaj id generowania do usuwania duplikatów.

W workflow, w których pominięty wynik jest niedopuszczalny, traktuj webhooki jako szybką ścieżkę i okresowo uzgadniaj wyniki za pomocą flows.image.list lub flows.video.list, filtrując według status.

Kolejne kroki