Webhook Callbacks

Receive automatic notifications when your content generation tasks complete or fail.

How It Works

1. Include a callbackUrl when creating a task
2. We'll send a POST request to your URL when the task completes or fails
3. Your endpoint should respond with a 200 status code

Webhook Payload

Text-to-Video Success

{
  "taskId": "task_abc123",
  "status": "success",
  "statusCode": 200,
  "resultUrl": "https://cdn.heo365.com/sora2/video.mp4",
  "createdAt": "2024-10-24T10:00:00Z",
  "completedAt": "2024-10-24T10:03:00Z",
  "requestParams": {
    "projectName": "sora2",
    "modelName": "sora2_text_to_video",
    "prompt": "A beautiful sunset over the ocean",
    "imageUrl": "",
    "aspectRatio": "9:16",
    "duration": 10,
    "callbackUrl": "https://your-domain.com/webhook"
  }
}

Image-to-Video Success

{
  "taskId": "task_abc123",
  "status": "success",
  "statusCode": 200,
  "resultUrl": "https://cdn.heo365.com/sora2/video.mp4",
  "createdAt": "2024-10-24T10:00:00Z",
  "completedAt": "2024-10-24T10:03:00Z",
  "requestParams": {
    "projectName": "sora2",
    "modelName": "sora2_image_to_video",
    "prompt": "Animate this image with gentle camera movement",
    "imageUrl": "https://example.com/image.jpg",
    "aspectRatio": "9:16",
    "duration": 10,
    "callbackUrl": "https://your-domain.com/webhook"
  }
}

Failed Task

{
  "taskId": "task_abc123",
  "status": "failed",
  "statusCode": 200,
  "errorMessage": "Failed to generate, system error please try again",
  "resultUrl": null,
  "createdAt": "2024-10-24T10:00:00Z",
  "completedAt": "2024-10-24T10:05:00Z",
  "requestParams": {
    "projectName": "sora2",
    "modelName": "sora2_text_to_video",
    "prompt": "Invalid prompt content",
    "imageUrl": "",
    "aspectRatio": "9:16",
    "duration": 10,
    "callbackUrl": "https://your-domain.com/webhook"
  }
}

Payload Fields

Field	Type	Description
taskId	string	Unique identifier for the task
status	string	Task status: success or failed
statusCode	number	HTTP status code (always 200 for webhook)
errorMessage	string	Error description (only when status is failed)
resultUrl	string | null	CDN URL of the generated video, null if failed
createdAt	string	ISO 8601 timestamp when task was created
completedAt	string	ISO 8601 timestamp when task completed
requestParams	object	Original parameters used to create the task
requestParams.projectName	string	Project name (sora2)
requestParams.modelName	string	Model used for generation
requestParams.prompt	string	Video description prompt
requestParams.imageUrl	string	Image URL for image-to-video, empty string for text-to-video
requestParams.aspectRatio	string	Video aspect ratio (9:16 or 16:9, default: 9:16)
requestParams.duration	number	Video duration in seconds
requestParams.remixTargetId	string	Previous task ID if this is a remix
requestParams.callbackUrl	string	Webhook URL

Payload Structure

The webhook payload structure matches the Query Task API response format (without the success wrapper), ensuring consistency across the API.

Requirements

Webhook Requirements

• Your webhook URL must be publicly accessible
• Must use HTTPS (HTTP is not supported)
• Should respond within 10 seconds
• Should return a 200 status code

Implementation Examples

Python (Flask)

from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/webhook', methods=['POST'])
def webhook():
    data = request.json
    
    task_id = data.get('taskId')
    status = data.get('status')
    request_params = data.get('requestParams', {})
    
    print(f'Task {task_id} is {status}')
    print('Original request:', request_params)
    
    if status == 'success':
        result_url = data.get('resultUrl')
        print(f'Video URL: {result_url}')
        # Process the completed video
        # e.g., save to database, notify user, etc.
    elif status == 'failed':
        error_message = data.get('errorMessage')
        print(f'Task failed: {error_message}')
        # Handle failure
        # e.g., log error, notify user, etc.
    
    return jsonify({'status': 'ok'}), 200

if __name__ == '__main__':
    app.run(port=5000)

Python (FastAPI)

from fastapi import FastAPI, Request
from pydantic import BaseModel
from typing import Optional, Dict, Any

app = FastAPI()

class WebhookPayload(BaseModel):
    taskId: str
    status: str
    resultUrl: Optional[str] = None
    errorMessage: Optional[str] = None
    createdAt: str
    completedAt: str
    requestParams: Dict[str, Any]

@app.post('/webhook')
async def webhook(payload: WebhookPayload):
    print(f'Task {payload.taskId} is {payload.status}')
    
    if payload.status == 'success':
        print(f'Video URL: {payload.resultUrl}')
        # Process the completed video
    elif payload.status == 'failed':
        print(f'Task failed: {payload.errorMessage}')
        # Handle failure
    
    return {'status': 'ok'}

if __name__ == '__main__':
    import uvicorn
    uvicorn.run(app, host='0.0.0.0', port=5000)

Node.js (Express)

const express = require('express');
const app = express();

app.use(express.json());

app.post('/webhook', (req, res) => {
  const data = req.body;
  
  const taskId = data.taskId;
  const status = data.status;
  const requestParams = data.requestParams || {};
  
  console.log(`Task ${taskId} is ${status}`);
  console.log('Original request:', requestParams);
  
  if (status === 'success') {
    const resultUrl = data.resultUrl;
    console.log(`Video URL: ${resultUrl}`);
    // Process the completed video
    // e.g., save to database, notify user, etc.
  } else if (status === 'failed') {
    const errorMessage = data.errorMessage;
    console.log(`Task failed: ${errorMessage}`);
    // Handle failure
    // e.g., log error, notify user, etc.
  }
  
  res.status(200).json({ status: 'ok' });
});

app.listen(5000, () => {
  console.log('Webhook server listening on port 5000');
});

Next.js API Route

// pages/api/webhook.ts
import type { NextApiRequest, NextApiResponse } from 'next';

interface WebhookPayload {
  taskId: string;
  status: 'success' | 'failed';
  resultUrl?: string;
  errorMessage?: string;
  createdAt: string;
  completedAt: string;
  requestParams: Record<string, any>;
}

export default async function handler(
  req: NextApiRequest,
  res: NextApiResponse
) {
  if (req.method !== 'POST') {
    return res.status(405).json({ error: 'Method not allowed' });
  }

  const data: WebhookPayload = req.body;

  console.log(`Task ${data.taskId} is ${data.status}`);

  if (data.status === 'success') {
    console.log(`Video URL: ${data.resultUrl}`);
    // Process the completed video
    // e.g., save to database, notify user via email/websocket
  } else if (data.status === 'failed') {
    console.log(`Task failed: ${data.errorMessage}`);
    // Handle failure
    // e.g., log error, notify user
  }

  res.status(200).json({ status: 'ok' });
}

Security Best Practices

Security Recommendations

1. Verify Task ID: Check the task ID against your database to ensure the request is legitimate
2. Use HTTPS: Always use HTTPS for your webhook endpoint
3. Validate Payload: Validate the webhook payload structure before processing
4. Implement Retry Logic: Handle temporary failures gracefully
5. Log Requests: Log all webhook requests for debugging and auditing

Enhanced Security Example

from flask import Flask, request, jsonify, abort

app = Flask(__name__)

# Store your task IDs in a database
VALID_TASK_IDS = set()  # Replace with database lookup

@app.route('/webhook', methods=['POST'])
def webhook():
    data = request.json
    
    # 1. Validate payload structure
    required_fields = ['taskId', 'status']
    if not all(field in data for field in required_fields):
        abort(400, 'Invalid payload structure')
    
    # 2. Verify task ID
    task_id = data.get('taskId')
    if task_id not in VALID_TASK_IDS:
        abort(403, 'Invalid task ID')
    
    # 3. Process webhook
    status = data.get('status')
    
    if status == 'success':
        result_url = data.get('resultUrl')
        # Update database
        # Notify user
        print(f'Task {task_id} completed: {result_url}')
    elif status == 'failed':
        error_message = data.get('errorMessage')
        # Update database
        # Notify user
        print(f'Task {task_id} failed: {error_message}')
    
    # 4. Remove from pending tasks
    VALID_TASK_IDS.discard(task_id)
    
    return jsonify({'status': 'ok'}), 200

Testing Webhooks

Using ngrok for Local Testing

# Install ngrok
brew install ngrok  # macOS
# or download from https://ngrok.com/

# Start your local server
python webhook_server.py

# In another terminal, expose your local server
ngrok http 5000

# Use the ngrok URL as your callbackUrl
# Example: https://abc123.ngrok.io/webhook

Test Webhook Endpoint

# Test your webhook endpoint with curl
curl -X POST https://your-domain.com/webhook \
  -H "Content-Type: application/json" \
  -d '{
    "taskId": "test_123",
    "status": "success",
    "resultUrl": "https://cdn.heo365.com/sora2/test.mp4",
    "createdAt": "2024-10-24T10:00:00Z",
    "completedAt": "2024-10-24T10:03:00Z",
    "requestParams": {
      "projectName": "sora2",
      "modelName": "sora2_text_to_video",
      "prompt": "Test prompt"
    }
  }'

Retry Policy

Webhook Retry Policy

If your webhook endpoint fails to respond or returns a non-200 status code:

• We will retry up to 3 times
• With exponential backoff: 10s, 30s, 90s
• After 3 failed attempts, we will stop retrying

Troubleshooting

Common Issues

Issue	Solution
Webhook not received	Check if your URL is publicly accessible and uses HTTPS
Timeout errors	Ensure your endpoint responds within 10 seconds
4xx/5xx errors	Check your server logs for errors in your webhook handler
Duplicate webhooks	Implement idempotency by checking task ID

Debugging Tips

1. Check Server Logs: Review your server logs for incoming requests
2. Test Locally: Use ngrok to test webhooks on your local machine
3. Validate Response: Ensure your endpoint returns a 200 status code
4. Monitor Performance: Track webhook processing time
5. Handle Errors: Implement proper error handling and logging

Complete Example

Here's a complete example with database integration and user notification:

from flask import Flask, request, jsonify, abort
from sqlalchemy import create_engine, Column, String, Integer, DateTime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from datetime import datetime
import smtplib
from email.mime.text import MIMEText

app = Flask(__name__)

# Database setup
engine = create_engine('sqlite:///tasks.db')
Base = declarative_base()
Session = sessionmaker(bind=engine)

class Task(Base):
    __tablename__ = 'tasks'
    
    id = Column(String, primary_key=True)
    user_email = Column(String)
    status = Column(String)
    result_url = Column(String, nullable=True)
    created_at = Column(DateTime)
    completed_at = Column(DateTime, nullable=True)

Base.metadata.create_all(engine)

def send_email(to_email, subject, body):
    """Send email notification to user"""
    msg = MIMEText(body)
    msg['Subject'] = subject
    msg['From'] = 'noreply@firebirdgen.com'
    msg['To'] = to_email
    
    # Configure your SMTP server
    # smtp = smtplib.SMTP('smtp.gmail.com', 587)
    # smtp.send_message(msg)
    print(f'Email sent to {to_email}: {subject}')

@app.route('/webhook', methods=['POST'])
def webhook():
    data = request.json
    
    # Validate payload
    if not data or 'taskId' not in data:
        abort(400, 'Invalid payload')
    
    task_id = data.get('taskId')
    status = data.get('status')
    
    # Get task from database
    session = Session()
    task = session.query(Task).filter_by(id=task_id).first()
    
    if not task:
        session.close()
        abort(404, 'Task not found')
    
    # Update task status
    task.status = status
    task.completed_at = datetime.utcnow()
    
    if status == 'success':
        task.result_url = data.get('resultUrl')
        
        # Send success email
        send_email(
            task.user_email,
            'Your video is ready!',
            f'Your video has been generated successfully.\n\n'
            f'Download: {task.result_url}'
        )
    elif status == 'failed':
        error_message = data.get('errorMessage', 'Unknown error')
        
        # Send failure email
        send_email(
            task.user_email,
            'Video generation failed',
            f'Unfortunately, your video generation failed.\n\n'
            f'Error: {error_message}'
        )
    
    session.commit()
    session.close()
    
    return jsonify({'status': 'ok'}), 200

if __name__ == '__main__':
    app.run(port=5000)

Next Steps

• Create Task - Learn how to create tasks with webhook callbacks
• Query Task - Alternative to webhooks for checking task status