Webhooks de Imagen y Video

Resumen

Las generaciones de vídeo pueden tardar varios minutos, por lo que mantener una consulta periódica abierta resulta costoso. Configura una generación para que se entregue mediante webhook y ElevenLabs enviará un evento flows_generation a tu ruta cuando la generación alcance el estado completed o failed.

La carga útil del evento es la respuesta final de la ruta GET correspondiente, así que un controlador que ya interpreta la respuesta de consulta periódica no necesita una vía de análisis independiente.

Antes de empezar

La entrega mediante webhook utiliza los webhooks de tu espacio de trabajo suscritos a eventos de generación. Configurarlo requiere dos pasos: crear el webhook y, después, suscribirlo al evento.

{
  "events": ["flows"]
}

Crear y suscribir webhooks requiere el permiso de gestión de webhooks o ser administrador del espacio de trabajo. Un solo evento admite hasta 10 webhooks; a partir de ahí, la solicitud falla con too_many_webhooks.

Se rechaza una generación que solicita entrega mediante webhook cuando no hay ningún webhook suscrito a eventos de generación, así que nunca se genera un resultado sin ningún lugar al que enviarlo.

Solicita la entrega mediante webhook

Añade un objeto webhook a la solicitud de creación. Usa {"type": "all"} para entregar a todos los webhooks suscritos a eventos de generación, lo que mantiene la solicitud estable aunque se añadan o sustituyan webhooks.

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

Para dirigirte a webhooks específicos, configura el campo webhook como una lista de ID. Cada ID debe ser uno de los webhooks del espacio de trabajo suscritos a eventos de generación.

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

La solicitud de creación valida el destino antes de iniciar la generación y devuelve un error cuando la entrega no sería posible:

La entrega mediante webhook funciona bien con generaciones encadenadas: configura webhook en la generación final y toda la cadena se ejecutará en el servidor con un único evento al final. Esto también se aplica cuando la cadena falla a mitad de camino: el error se propaga hasta la generación final, que lo entrega como un evento failed con el motivo dependency_failed.

Carga útil del webhook

Una generación completada entrega la URL de salida y el 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 generación fallida entrega en su lugar la categoría y el mensaje del error:

{
  "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 para decidir qué campos están presentes. Los dos estados finales son los únicos que puede contener un webhook, ya que la entrega solo se produce cuando finaliza una generación.

Gestiona el evento

Un controlador verifica la firma, comprueba el tipo de evento y, después, actúa según data.status. Este ejemplo descarga la salida de una generación completada y registra el motivo de una fallida.

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

Ambos ejemplos descargan dentro de la solicitud por brevedad. Un vídeo grande tarda lo suficiente como para que esto pueda superar el tiempo de espera de entrega, así que en producción pasa el ID de generación a una cola y devuelve 2xx de inmediato. La URL firmada es válida durante aproximadamente una hora, tiempo de sobra para un proceso en segundo plano.

Verifica la firma

El controlador anterior llama a construct_event / constructEvent, que verifica la cabecera ElevenLabs-Signature, valida la marca de tiempo y analiza la carga útil en un solo paso. Verifica siempre antes de confiar en un evento.

Es importante que el listener valide todos los webhooks entrantes. Actualmente, los webhooks admiten autenticación mediante firmas HMAC. Configura la autenticación HMAC de esta forma:

• Almacena de forma segura el secreto compartido generado al crear el webhook
• Verifica la cabecera ElevenLabs-Signature en tu ruta de API mediante el SDK

El SDK de JavaScript incluye constructEvent; el SDK de Python incluye construct_event con rawBody, sig_header y secret (en Python no se llaman payload / signature). Ambos verifican la firma, validan la marca de tiempo y analizan la carga útil 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"}

Comportamiento de entrega

Cada generación entrega exactamente un evento final por cada webhook de destino. La entrega es independiente de la propia generación: un webhook que falla o no es accesible no afecta al resultado, que sigue disponible desde la ruta GET y en la respuesta de lista.

Devuelve un estado 2xx rápidamente desde tu controlador. Los fallos repetidos desactivan automáticamente un webhook y un webhook desactivado hace que se rechacen al crearlas las generaciones posteriores que lo tengan como destino. Diseña el controlador para que sea idempotente y usa el id de generación para eliminar duplicados.

En flujos de trabajo en los que no sea aceptable perder un resultado, trata los webhooks como la vía rápida y concilia periódicamente con flows.image.list o flows.video.list, filtrando por status.

Próximos pasos