Skip to main content
Webhooks provide instant notifications when your video generation tasks complete or fail. No more polling!

​
Setup Guide

1

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/sora', 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.get('error_message')}")
        
    return jsonify({'status': 'received'}), 200
2

Add Webhook URL

Include your webhook URL when creating a generation task.
curl -X POST https://platform.runblob.io/v1/sora/generate \
  -H "Authorization: Bearer YOUR_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "prompt": "A sunset over mountains",
    "orientation": "9:16",
    "duration": 10,
    "callback_url": "https://your-app.com/webhook/sora"
  }'
3

Handle Notifications

Your endpoint will receive instant notifications when tasks complete or fail.

​
Payload Structure

Sent when video generation completes successfully.
{
  "generation_id": "550e8400-e29b-41d4-a716-446655440000",
  "status": "completed",
  "prompt": "A beautiful sunset over the ocean",
  "video_url": "https://cdn.runblob.io/videos/2025/11/550e8400.mp4",
  "created_at": "2025-11-24T10:00:00.000000",
  "completed_at": "2025-11-24T10:05:30.000000"
}
​
generation_id
string
UUID of the completed task
​
status
string
Always "completed" for successful generations
​
prompt
string
Original text prompt
​
video_url
string
Direct download URL for the generated video
​
created_at
string
ISO 8601 time of task creation
​
completed_at
string
ISO 8601 time of completion

​
Reliability Features

Retry Policy: Up to 5 attempts with exponential backoffBackoff Strategy: 30s, 60s, 120s, 240s, 480sFailure 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 HTTPSValidation: Always validate the generation_id in your webhook handlerIP Allowlist: Consider restricting access to RunBlob’s IP ranges

​
Example Implementations

from flask import Flask, request, jsonify
import logging

app = Flask(__name__)
logging.basicConfig(level=logging.INFO)

@app.route('/webhook/sora', methods=['POST'])
def sora_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:
            error = data.get('error_message')
            logging.error(f"Generation failed for {generation_id}: {error}")
            # Handle failure
            handle_failure(generation_id, error)
        
        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, error_message):
    # Your error handling logic here
    pass

if __name__ == '__main__':
    app.run(port=3000)
Important: Always return a 200 status code quickly to acknowledge receipt. Perform heavy processing asynchronously to avoid timeouts.

​
Testing Webhooks

​
Local Development

Use tools like ngrok to expose your local server: 
# Start ngrok
ngrok http 3000

# Use the provided HTTPS URL in your callback_url
# Example: https://abc123.ngrok.io/webhook/sora

​
Webhook Verification

Always verify the webhook data: 
def verify_webhook(data):
    required_fields = ['generation_id', 'status', 'prompt']
    
    # Check required fields
    if not all(field in data for field in required_fields):
        return False
    
    # Verify generation_id format (UUID)
    try:
        uuid.UUID(data['generation_id'])
    except ValueError:
        return False
    
    # Verify status value
    if data['status'] not in ['completed', 'failed']:
        return False
    
    return True
For production environments, consider implementing additional security measures like webhook signatures or API key validation.