Webhooks provide instant notifications when your video generation tasks complete or fail. No more polling!
Setup Guide
Create Webhook Endpoint
Set up an HTTPS endpoint that can receive POST requests. from flask import Flask, request, jsonify
app = Flask( __name__ )
@app.route ( '/webhook' , methods = [ 'POST' ])
def handle_webhook ():
data = request.json
if data[ 'status' ] == 'completed' :
print ( f "Video ready: { data[ 'video_url' ] } " )
else :
print ( f "Generation failed: { data[ 'generation_id' ] } " )
return jsonify({ 'status' : 'received' }), 200
Add Webhook URL
Include your webhook URL when creating a generation task. curl -X POST https://platform.runblob.io/v1/kling/generate \
-H "Authorization: Bearer YOUR_API_KEY" \
-H "Content-Type: application/json" \
-d '{
"prompt": "A sunset over mountains",
"callback_url": "https://your-app.com/webhook"
}'
Handle Notifications
Your endpoint will receive instant notifications when tasks complete.
Payload Structure
Sent when video generation completes successfully. {
"generation_id" : "550e8400-e29b-41d4-a716-446655440000" ,
"status" : "completed" ,
"video_url" : "https://cdn.runblob.io/videos/550e8400.mp4" ,
"completed_at" : "2025-11-22T10:32:15Z"
}
UUID of the completed task
Always "completed" for successful generations
Direct download URL for the generated video
ISO 8601 time of completion
Reliability Features
Retry Policy: Up to 5 attempts with exponential backoffBackoff Strategy: 30 seconds, 60 seconds, 120 seconds, 240 seconds, 480 secondsFailure Handling: After 5 failed attempts, the webhook is marked as failed
Request Timeout: 30 seconds per webhook requestConnection Timeout: 10 seconds to establish connectionBest Practice: Respond quickly with a 200 status code
HTTPS Required: All webhook URLs must use HTTPSIP Allowlist: Consider restricting access to RunBlob’s IP rangesValidation: Always validate the generation_id in your webhook handler
Example Implementations
from flask import Flask, request, jsonify
import logging
app = Flask( __name__ )
logging.basicConfig( level = logging. INFO )
@app.route ( '/webhook' , methods = [ 'POST' ])
def kling_webhook ():
try :
data = request.json
generation_id = data.get( 'generation_id' )
if data.get( 'status' ) == 'completed' :
video_url = data.get( 'video_url' )
logging.info( f "Video ready for { generation_id } : { video_url } " )
# Process successful generation
process_video(generation_id, video_url)
else :
logging.error( f "Generation failed for { generation_id } " )
# Handle failure
handle_failure(generation_id)
return jsonify({ 'status' : 'received' }), 200
except Exception as e:
logging.error( f "Webhook error: { str (e) } " )
return jsonify({ 'error' : 'Internal error' }), 500
def process_video ( generation_id , video_url ):
# Your video processing logic here
pass
def handle_failure ( generation_id ):
# Your error handling logic here
pass
Important: Always return a 200 status code quickly to acknowledge receipt. Perform heavy processing asynchronously to avoid timeouts.