Webhooks de Imagem e Vídeo

Visão geral

A geração de vídeos pode levar vários minutos, o que torna caro manter consultas periódicas abertas. Ative a entrega por webhook para uma geração, e a ElevenLabs enviará um evento flows_generation ao seu endpoint quando a geração chegar a completed ou failed.

A carga útil do evento é a resposta final do endpoint GET correspondente, portanto, um manipulador que já entende a resposta da consulta periódica não precisa de um caminho de análise separado.

Antes de começar

A entrega por webhook usa os webhooks que seu espaço de trabalho assinou para eventos de geração. A configuração tem duas etapas: criar o webhook e depois assiná-lo para o evento.

{
  "events": ["flows"]
}

Criar e assinar webhooks exige a permissão Gerenciar Webhooks ou ser administrador do espaço de trabalho. Um único evento aceita até 10 webhooks; acima disso, a solicitação falha com too_many_webhooks.

Uma geração que solicita entrega por webhook quando nenhum webhook está assinado para eventos de geração é rejeitada, para que um resultado nunca seja gerado sem um destino de entrega.

Solicitar entrega por webhook

Adicione um objeto webhook à solicitação de criação. Use {"type": "all"} para entregar a todos os webhooks assinados para eventos de geração, o que mantém a solicitação estável conforme webhooks são adicionados ou substituídos.

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 direcionar a webhooks específicos, defina o campo webhook como uma lista de IDs. Cada ID deve ser de um dos webhooks do espaço de trabalho assinados para eventos de geração.

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

A solicitação de criação valida o destino antes de iniciar a geração e retorna um erro quando a entrega não for possível:

A entrega por webhook funciona bem com gerações encadeadas: defina webhook na geração final, e toda a cadeia será executada no servidor com um único evento ao final. Isso também vale quando a cadeia falha no meio do processo — a falha é propagada para a geração final, que a entrega como um evento failed com o motivo dependency_failed.

Carga útil do webhook

Uma geração concluída entrega a URL de saída e o 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"
  }
}

Uma geração com falha entrega a categoria e a mensagem de falha:

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

Use data.status para decidir quais campos estão presentes. Os dois status finais são os únicos que um webhook pode conter, já que a entrega ocorre apenas quando uma geração termina.

Processar o evento

Um manipulador verifica a assinatura, confere o tipo de evento e, então, ramifica com base em data.status. Este exemplo baixa a saída de uma geração concluída e registra o motivo de uma geração que falhou.

# 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 os exemplos fazem o download dentro da solicitação por brevidade. Um vídeo grande pode levar tempo suficiente para exceder o tempo limite de entrega, então, em produção, envie o ID da geração a uma fila e retorne 2xx imediatamente. A URL assinada é válida por cerca de uma hora, o que é suficiente para um worker em segundo plano.

Verificar a assinatura

O manipulador acima chama construct_event / constructEvent, que verifica o cabeçalho ElevenLabs-Signature, valida o registro de data e hora e analisa a carga útil em uma única etapa. Sempre verifique antes de confiar em um evento.

É importante que o listener valide todos os webhooks recebidos. Atualmente, os webhooks oferecem suporte à autenticação por assinaturas HMAC. Configure a autenticação HMAC:

• Armazenando com segurança o segredo compartilhado gerado na criação do webhook
• Verificando o cabeçalho ElevenLabs-Signature no seu endpoint usando o SDK

O SDK JavaScript disponibiliza constructEvent; o SDK Python disponibiliza construct_event com rawBody, sig_header e secret (em Python, eles não se chamam payload / signature). Ambos verificam a assinatura, validam o carimbo de data e hora e analisam o 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 de entrega

Cada geração entrega exatamente um evento final por webhook direcionado. A entrega é independente da própria geração: um webhook que falha ou está inacessível não afeta o resultado, que continua disponível no endpoint GET e na resposta da lista.

Retorne prontamente um status 2xx do seu manipulador. Falhas repetidas desativam um webhook automaticamente, e um webhook desativado faz com que gerações subsequentes que o direcionam sejam rejeitadas no momento da criação. Projete o manipulador para ser idempotente e use o id da geração para eliminar duplicatas.

Para fluxos de trabalho em que não é aceitável perder um resultado, trate os webhooks como o caminho rápido e reconcilie periodicamente com flows.image.list ou flows.video.list, filtrando por status.

Próximas etapas