Image & Video वेबहुक्स

ओवरव्यू

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

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

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

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

{
  "events": ["flows"]
}

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

अगर generation events के लिए कोई webhook subscribed नहीं है, तो webhook डिलीवरी का अनुरोध करने वाला जनरेशन रिजेक्ट हो जाता है, ताकि कहीं डिलीवर न हो सकने वाला result कभी जनरेट न हो।

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

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 के उन webhooks में से एक होनी चाहिए जो generation events के लिए subscribed हैं।

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

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

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

Webhook payload

पूरा हुआ जनरेशन 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"
  }
}

विफल जनरेशन इसके बजाय 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 ही ऐसे हैं जिन्हें webhook ले जा सकता है, क्योंकि डिलीवरी केवल जनरेशन पूरा होने पर होती है।

इवेंट संभालें

एक handler signature को verify करता है, इवेंट type जांचता है, फिर data.status पर branch करता है। यह उदाहरण पूरे हुए जनरेशन का output download करता है और विफल जनरेशन का कारण 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 के अंदर download करते हैं। बड़े वीडियो में इतना समय लग सकता है कि यह डिलीवरी timeout से अधिक हो जाए, इसलिए production में generation ID को queue में भेजें और तुरंत 2xx लौटाएं। Signed URL लगभग एक घंटे तक मान्य रहता है, जो background worker के लिए पर्याप्त है।

Signature verify करें

ऊपर दिया handler construct_event / constructEvent को call करता है, जो ElevenLabs-Signature header को verify करता है, timestamp को validate करता है और payload को एक ही step में parse करता है। किसी इवेंट पर भरोसा करने से पहले हमेशा 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"}

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

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

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

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

अगले चरण