Webhook di Immagini e Video

Panoramica

La generazione di video può richiedere diversi minuti, rendendo costoso mantenere aperto il polling. Configura una generazione per la consegna tramite webhook e ElevenLabs invierà un evento flows_generation al tuo endpoint quando la generazione raggiunge lo stato completed o failed.

Il payload dell’evento corrisponde alla risposta finale del relativo endpoint GET, quindi un handler che interpreta già la risposta del polling non richiede un percorso di parsing separato.

Prima di iniziare

La consegna tramite webhook usa i webhook del tuo workspace sottoscritti agli eventi di generazione. Per configurarli sono necessari due passaggi: crea il webhook, quindi sottoscrivilo all’evento.

{
  "events": ["flows"]
}

Per creare e sottoscrivere webhook servono l’autorizzazione Gestione webhook o il ruolo di amministratore del workspace. Un singolo evento accetta fino a 10 webhook; oltre questo limite la richiesta non riesce con too_many_webhooks.

Una generazione che richiede la consegna tramite webhook quando nessun webhook è sottoscritto agli eventi di generazione viene rifiutata, quindi un risultato non viene mai generato senza una destinazione a cui consegnarlo.

Richiedi la consegna tramite webhook

Aggiungi un oggetto webhook alla richiesta di creazione. Usa {"type": "all"} per inviare il risultato a ogni webhook sottoscritto agli eventi di generazione, così la richiesta resta invariata quando i webhook vengono aggiunti o sostituiti.

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

Per scegliere webhook specifici, imposta invece il campo webhook su un elenco di ID. Ogni ID deve corrispondere a uno dei webhook del workspace sottoscritti agli eventi di generazione.

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

La richiesta di creazione convalida la destinazione prima di avviare la generazione e restituisce un errore quando la consegna non sarebbe possibile:

La consegna tramite webhook si abbina bene alle generazioni concatenate: imposta webhook sulla generazione finale e l’intera catena viene eseguita lato server con un unico evento alla fine. Questo vale anche quando la catena non riesce a metà: l’errore si propaga alla generazione finale, che lo invia come evento failed con motivo dependency_failed.

Payload del webhook

Una generazione completata invia l’URL dell’output e il tipo 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"
  }
}

Una generazione non riuscita invia invece la categoria e il messaggio dell’errore:

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

Usa data.status per stabilire quali campi sono presenti. I due stati finali sono gli unici che un webhook può contenere, poiché la consegna avviene solo al termine di una generazione.

Gestisci l’evento

Un handler verifica la firma, controlla il tipo di evento, quindi gestisce i vari casi in base a data.status. Questo esempio scarica l’output di una generazione completata e registra il motivo di una generazione non riuscita.

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

Per brevità, entrambi gli esempi scaricano il contenuto durante la richiesta. Un video di grandi dimensioni può richiedere abbastanza tempo da superare il timeout di consegna, quindi in produzione passa l’ID della generazione a una coda e restituisci subito un codice 2xx. L’URL firmato è valido per circa un’ora, un tempo sufficiente per un worker in background.

Verifica la firma

L’handler precedente chiama construct_event / constructEvent, che verifica l’header ElevenLabs-Signature, convalida il timestamp e analizza il payload in un solo passaggio. Verifica sempre la firma prima di considerare attendibile un evento.

È importante che il listener convalidi tutti i webhook in arrivo. I webhook supportano attualmente l’autenticazione tramite firme HMAC. Configura l’autenticazione HMAC:

• Archiviando in modo sicuro il segreto condiviso generato alla creazione del webhook
• Verificando l’header ElevenLabs-Signature nel tuo endpoint tramite l’SDK

L’SDK JavaScript espone constructEvent; l’SDK Python espone construct_event con rawBody, sig_header e secret (in Python non si chiamano payload / signature). Entrambi verificano la firma, convalidano il timestamp e analizzano il payload 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"}

Comportamento di consegna

Ogni generazione invia esattamente un evento finale per ogni webhook selezionato. La consegna è indipendente dalla generazione stessa: un webhook che non riesce o non è raggiungibile non influisce sul risultato, che rimane disponibile dall’endpoint GET e nella risposta dell’elenco.

Restituisci tempestivamente uno stato 2xx dal tuo handler. Errori ripetuti disabilitano automaticamente un webhook e un webhook disabilitato fa sì che le generazioni successive che lo selezionano vengano rifiutate al momento della creazione. Progetta l’handler affinché sia idempotente e usa l’id della generazione per eliminare i duplicati.

Per workflow in cui non puoi accettare la perdita di un risultato, tratta i webhook come percorso rapido e verifica periodicamente con flows.image.list o flows.video.list, filtrando per status.

Passaggi successivi