इमेज और वीडियो वेबहुक्स

ओवरव्यू

वीडियो जनरेशन में कई मिनट लग सकते हैं, जिससे पोलिंग कनेक्शन खुला रखना महंगा हो जाता है। किसी generation के लिए webhook डिलीवरी चुनें और जनरेशन के completed या failed होने पर ElevenLabs आपके endpoint पर flows_generation इवेंट भेजता है।

इवेंट payload संबंधित GET endpoint का अंतिम response होता है, इसलिए जो handler पहले से polling response को समझता है, उसे अलग parsing path की जरूरत नहीं होती।

शुरू करने से पहले

Webhook डिलीवरी के लिए आपके workspace में generation events के लिए subscribed webhooks इस्तेमाल होते हैं। इसे सेट अप करने के दो चरण हैं: webhook बनाएं, फिर उसे इवेंट के लिए subscribe करें।

{
  "events": ["flows"]
}

Webhook बनाने और subscribe करने के लिए Webhooks Manage permission या workspace admin होना जरूरी है। एक इवेंट के लिए अधिकतम 10 webhooks स्वीकार किए जाते हैं; इसके बाद request too_many_webhooks के साथ fail हो जाती है।

अगर किसी generation में webhook डिलीवरी मांगी जाती है, लेकिन generation events के लिए कोई webhook subscribed नहीं है, तो उसे reject कर दिया जाता है। इस तरह नतीजा कभी ऐसी जगह के लिए generate नहीं होता जहां उसे भेजा न जा सके।

वेबहुक डिलीवरी का अनुरोध करें

Create request में एक webhook object जोड़ें। generation events के लिए subscribed हर webhook को डिलीवर करने के लिए {"type": "all"} इस्तेमाल करें। इससे webhooks जोड़े या बदले जाने पर भी request स्थिर रहती है।

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

इसके बजाय खास webhooks को target करने के लिए webhook field में IDs की सूची सेट करें। हर ID, workspace के ऐसे webhook की होनी चाहिए जो generation events के लिए subscribed हो।

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

Create request generation शुरू करने से पहले target को validate करती है और जब डिलीवरी संभव न हो, tो error लौटाती है:

Webhook डिलीवरी चेन किए गए generations के साथ अच्छी तरह काम करती है: अंतिम generation पर webhook सेट करें और पूरी chain server-side चलेगी, जिसके अंत में एक ही event मिलेगा। अगर chain बीच में fail हो जाए तब भी यह लागू होता है — failure अंतिम generation तक पहुंचता है, जो इसे dependency_failed कारण वाले failed event के रूप में डिलीवर करता है।

वेबहुक पेलोड

पूरा हुआ generation output URL और MIME type डिलीवर करता है:

{
  "type": "flows_generation",
  "event_timestamp": 1739721600,
  "data": {
    "id": "JWr5N6X9ZTqf8jD2LmQb",
    "status": "completed",
    "content_url": "https://storage.googleapis.com/generations/JWr5N6X9ZTqf8jD2LmQb",
    "content_mime_type": "video/mp4"
  }
}

Failed generation इसके बजाय failure category और message डिलीवर करता है:

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

यह तय करने के लिए कि कौन-से fields मौजूद हैं, data.status पर branch करें। ये दोनों terminal statuses ही वे एकमात्र statuses हैं जो webhook में आ सकते हैं, क्योंकि डिलीवरी केवल generation पूरा होने पर होती है।

इवेंट हैंडल करें

Handler signature को verify करता है, event type जांचता है, फिर data.status पर branch करता है। यह उदाहरण पूरे हुए generation का output डाउनलोड करता है और failed generation का कारण log करता है।

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

संक्षिप्तता के लिए, दोनों उदाहरण request के दौरान ही डाउनलोड करते हैं। बड़े वीडियो में इतना समय लग सकता है कि डिलीवरी timeout पार हो जाए, इसलिए production में generation ID को queue में दें और तुरंत 2xx लौटाएं। Signed URL लगभग एक घंटे तक valid रहता है, जो background worker के लिए पर्याप्त है।

सिग्नेचर सत्यापित करें

ऊपर दिया handler construct_event / constructEvent को कॉल करता है, जो ElevenLabs-Signature header को verify करता है, timestamp को validate करता है और एक ही चरण में payload parse करता है। किसी event पर भरोसा करने से पहले हमेशा verify करें।

लिस्नर के लिए सभी इनकमिंग वेबहुक को वैलिडेट करना ज़रूरी है। वेबहुक फ़िलहाल HMAC सिग्नेचर के ज़रिए ऑथेंटिकेशन सपोर्ट करते हैं। HMAC ऑथेंटिकेशन सेट अप करने के लिए:

• वेबहुक बनाते समय जनरेट हुए शेयर किए गए सीक्रेट को सुरक्षित रूप से स्टोर करें
• SDK का इस्तेमाल करके अपने एंडपॉइंट में ElevenLabs-Signature हेडर को वेरिफ़ाई करें

JavaScript SDK में constructEvent और Python SDK में rawBody, sig_header, और secret के साथ construct_event उपलब्ध है (Python में इनके नाम payload / signature नहीं हैं)। दोनों सिग्नेचर वेरिफ़ाई करते हैं, टाइमस्टैम्प वैलिडेट करते हैं और 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"}

डिलीवरी व्यवहार

हर generation, target किए गए हर webhook पर ठीक एक terminal event डिलीवर करता है। डिलीवरी generation से स्वतंत्र है: जो webhook fail हो या पहुंच से बाहर हो, वह नतीजे को प्रभावित नहीं करता। नतीजा GET endpoint और list response में उपलब्ध रहता है।

अपने handler से तुरंत 2xx status लौटाएं। बार-बार failures होने पर webhook अपने-आप बंद हो जाता है, और बंद webhook को target करने वाले बाद के generations को create समय पर reject कर दिया जाता है। Handler को idempotent बनाएं और duplicate हटाने के लिए generation id इस्तेमाल करें।

जिन workflows में कोई नतीजा छूटना स्वीकार्य नहीं है, उनमें webhooks को fast path मानें और status पर filter करते हुए समय-समय पर flows.image.list या flows.video.list से reconcile करें।

अगले चरण