Webhooks provide instant notifications when your image generation tasks complete, fail, or timeout. 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/photo' , methods = [ 'POST' ])
def handle_webhook ():
data = request.json
if data[ 'status' ] == 'completed' :
print ( f "Image ready: { data[ 'image_url' ] } " )
else :
print ( f "Generation failed: { data[ 'message' ] } " )
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/o1-photo/generate \
-H "Authorization: Bearer YOUR_API_KEY" \
-H "Content-Type: application/json" \
-d '{
"prompt": "A futuristic cyberpunk city",
"callback_url": "https://your-app.com/webhook/photo"
}'
Handle Notifications
Your endpoint will receive instant notifications when tasks complete.
Payload Structure
Sent when image generation completes successfully. {
"generation_id" : "photo-550e8400-e29b-41d4-a716" ,
"status" : "completed" ,
"image_url" : "https://cdn.example.com/photos/photo-550e8400.png" ,
"model" : "kling-o1-photo" ,
"message" : null
}
Always "completed" for successful generations
UUID of the completed task
Direct download URL for the generated image
Always null for successful generations
Reliability Features
Retry Policy: Up to 5 attempts with 30-second intervalsBackoff Strategy: Linear backoff between retriesFailure 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/photo' , methods = [ 'POST' ])
def kling_photo_webhook ():
try :
data = request.json
generation_id = data.get( 'generation_id' )
if data.get( 'status' ) == 'completed' :
image_url = data.get( 'image_url' )
logging.info( f "Image ready for { generation_id } : { image_url } " )
# Process successful generation
process_image(generation_id, image_url)
else :
error_code = data.get( 'message' )
logging.error( f "Generation failed for { generation_id } : { error_code } " )
# Handle failure
handle_failure(generation_id, error_code)
return jsonify({ 'status' : 'received' }), 200
except Exception as e:
logging.error( f "Webhook error: { str (e) } " )
return jsonify({ 'error' : 'Internal error' }), 500
def process_image ( generation_id , image_url ):
# Your image processing logic here
pass
def handle_failure ( generation_id , error_code ):
# Your error handling logic here
pass
Important: Always return a 200 status code quickly to acknowledge receipt. Perform heavy processing asynchronously to avoid timeouts.