Webhooks

Getting Started

To start using webhooks, you’ll need to:

Set up an endpoint - Create a URL that can receive POST requests
Configure the webhook - Add your endpoint in the Consumer App Portal
Verify signatures - Implement signature verification for security
Handle events - Process the incoming webhook data

Adding Endpoints

Animus uses Svix for processing webhook events. You will be given a portal link by the Animus team where you can use this portal to set up webhook endpoints and subscribe to events. In order to start listening to messages, you will need to configure your endpoints. To add an endpoint, select the “Endpoints” tab from the left sidebar. From there you can add a new endpoint and set the events you want to be notified about. If you don’t specify any event types, by default, your endpoint will receive all events, regardless of type. You can also edit any existing endpoints. You’ll be able to view and inspect webhooks sent to your Svix Play URL, making it effortless to get started.

Events

The following is a list of events you can subscribe to.

Available Event Types

media.completed - A video processing job has finished successfully or with a failure.

Event Payload Structure

When a webhook event is triggered, you’ll receive a POST request to your configured endpoint with a JSON payload containing:

{
  "event_type": "media.completed",
  "timestamp": "2024-01-15T10:30:00Z",
  "data": {
    "job_id": "uuid-string",
    "status": "completed",
    "media_url": "https://example.com/video.mp4",
    "results": {
      // Processing results based on the job type
    }
  }
}

Verifying Webhook Signatures

Webhook signatures let you verify that webhook messages are actually sent by us and not a malicious actor. For a more detailed explanation, check out this article on why you should verify webhooks. Our webhook partner Svix offers a set of useful libraries that make verifying webhooks very simple:

import { Webhook } from "svix";

const secret = "whsec_MfKQ9r8GKYqrTwjUPD8ILPZIo2LaLaSw";

// These were all sent from the server
const headers = {
  "svix-id": "msg_p5jXN8AQM9LWM0D4loKWxJek",
  "svix-timestamp": "1614265330",
  "svix-signature": "v1,g0hM9SsE+OTPJTGt/tmIKtSyZlE3uFJELVlNIOLJ1OE=",
};
const payload = '{"test": 2432232314}';

const wh = new Webhook(secret);
// Throws on error, returns the verified content on success
const verifiedPayload = wh.verify(payload, headers);

from svix.webhooks import Webhook

secret = "whsec_MfKQ9r8GKYqrTwjUPD8ILPZIo2LaLaSw"

# These were all sent from the server
headers = {
    "svix-id": "msg_p5jXN8AQM9LWM0D4loKWxJek",
    "svix-timestamp": "1614265330",
    "svix-signature": "v1,g0hM9SsE+OTPJTGt/tmIKtSyZlE3uFJELVlNIOLJ1OE=",
}
payload = '{"test": 2432232314}'

wh = Webhook(secret)
# Throws on error, returns the verified content on success
verified_payload = wh.verify(payload, headers)

package main

import (
    "github.com/svix/svix-webhooks/go"
)

func main() {
    secret := "whsec_MfKQ9r8GKYqrTwjUPD8ILPZIo2LaLaSw"
    
    headers := map[string]string{
        "svix-id":        "msg_p5jXN8AQM9LWM0D4loKWxJek",
        "svix-timestamp": "1614265330",
        "svix-signature": "v1,g0hM9SsE+OTPJTGt/tmIKtSyZlE3uFJELVlNIOLJ1OE=",
    }
    payload := `{"test": 2432232314}`
    
    wh, err := svix.NewWebhook(secret)
    if err != nil {
        panic(err)
    }
    
    // Throws on error, returns the verified content on success
    verifiedPayload, err := wh.Verify([]byte(payload), headers)
    if err != nil {
        panic(err)
    }
}

For detailed information on how to use the Svix portal to manage your webhooks, refer to the Svix documentation.

For more instructions and examples of how to verify signatures, check out their webhook verification documentation.

Complete Webhook Handler Examples

Here are complete examples of webhook handlers in different frameworks:

import express from 'express';
import { Webhook } from 'svix';

const app = express();
const webhookSecret = process.env.WEBHOOK_SECRET; // Your webhook secret

// Middleware to capture raw body for signature verification
app.use('/webhook', express.raw({ type: 'application/json' }));

app.post('/webhook', (req, res) => {
  const payload = req.body;
  const headers = req.headers;

  // Verify the webhook signature
  const wh = new Webhook(webhookSecret);
  let event;

  try {
    event = wh.verify(payload, headers);
  } catch (err) {
    console.error('Webhook signature verification failed:', err.message);
    return res.status(400).send('Invalid signature');
  }

  // Handle the event
  switch (event.event_type) {
    case 'media.completed':
      handleMediaCompleted(event.data);
      break;
    default:
      console.log(`Unhandled event type: ${event.event_type}`);
  }

  res.status(200).send('OK');
});

function handleMediaCompleted(data) {
  console.log('Media processing completed:', data.job_id);
  console.log('Status:', data.status);
  
  if (data.status === 'completed') {
    // Process the results
    console.log('Results:', data.results);
    // Update your database, notify users, etc.
    updateDatabase(data.job_id, data.results);
    notifyUser(data.job_id, 'completed');
  } else if (data.status === 'failed') {
    // Handle failure
    console.error('Media processing failed for job:', data.job_id);
    notifyUser(data.job_id, 'failed');
  }
}

async function updateDatabase(jobId, results) {
  // Update your database with the processing results
  // Example: await db.jobs.update(jobId, { status: 'completed', results });
}

async function notifyUser(jobId, status) {
  // Notify the user about the job completion
  // Example: await sendEmail(jobId, status);
}

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

from flask import Flask, request, jsonify
from svix.webhooks import Webhook
import os
import logging

app = Flask(__name__)
webhook_secret = os.environ.get('WEBHOOK_SECRET')

# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@app.route('/webhook', methods=['POST'])
def handle_webhook():
    payload = request.get_data()
    headers = request.headers

    # Verify the webhook signature
    wh = Webhook(webhook_secret)
    
    try:
        event = wh.verify(payload, headers)
    except Exception as e:
        logger.error(f"Webhook signature verification failed: {e}")
        return "Invalid signature", 400

    # Handle the event
    if event['event_type'] == 'media.completed':
        handle_media_completed(event['data'])
    else:
        logger.info(f"Unhandled event type: {event['event_type']}")

    return jsonify({"status": "success"})

def handle_media_completed(data):
    logger.info(f"Media processing completed: {data['job_id']}")
    logger.info(f"Status: {data['status']}")
    
    if data['status'] == 'completed':
        # Process the results
        logger.info(f"Results: {data['results']}")
        # Update your database, notify users, etc.
        update_database(data['job_id'], data['results'])
        notify_user(data['job_id'], 'completed')
    elif data['status'] == 'failed':
        # Handle failure
        logger.error(f"Media processing failed for job: {data['job_id']}")
        notify_user(data['job_id'], 'failed')

def update_database(job_id, results):
    """Update your database with the processing results"""
    # Example: db.jobs.update(job_id, status='completed', results=results)
    pass

def notify_user(job_id, status):
    """Notify the user about the job completion"""
    # Example: send_email(job_id, status)
    pass

if __name__ == '__main__':
    app.run(port=3000, debug=True)

from fastapi import FastAPI, Request, HTTPException
from svix.webhooks import Webhook
import os
import logging

app = FastAPI()
webhook_secret = os.environ.get('WEBHOOK_SECRET')

# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@app.post("/webhook")
async def handle_webhook(request: Request):
    payload = await request.body()
    headers = dict(request.headers)

    # Verify the webhook signature
    wh = Webhook(webhook_secret)
    
    try:
        event = wh.verify(payload, headers)
    except Exception as e:
        logger.error(f"Webhook signature verification failed: {e}")
        raise HTTPException(status_code=400, detail="Invalid signature")

    # Handle the event
    if event['event_type'] == 'media.completed':
        await handle_media_completed(event['data'])
    else:
        logger.info(f"Unhandled event type: {event['event_type']}")

    return {"status": "success"}

async def handle_media_completed(data):
    logger.info(f"Media processing completed: {data['job_id']}")
    logger.info(f"Status: {data['status']}")
    
    if data['status'] == 'completed':
        # Process the results
        logger.info(f"Results: {data['results']}")
        # Update your database, notify users, etc.
        await update_database(data['job_id'], data['results'])
        await notify_user(data['job_id'], 'completed')
    elif data['status'] == 'failed':
        # Handle failure
        logger.error(f"Media processing failed for job: {data['job_id']}")
        await notify_user(data['job_id'], 'failed')

async def update_database(job_id, results):
    """Update your database with the processing results"""
    # Example: await db.jobs.update(job_id, status='completed', results=results)
    pass

async def notify_user(job_id, status):
    """Notify the user about the job completion"""
    # Example: await send_email(job_id, status)
    pass

package main

import (
    "encoding/json"
    "io"
    "log"
    "net/http"
    "os"

    "github.com/gin-gonic/gin"
    "github.com/svix/svix-webhooks/go"
)

type WebhookEvent struct {
    EventType string      `json:"event_type"`
    Timestamp string      `json:"timestamp"`
    Data      interface{} `json:"data"`
}

type MediaCompletedData struct {
    JobID     string      `json:"job_id"`
    Status    string      `json:"status"`
    MediaURL  string      `json:"media_url"`
    Results   interface{} `json:"results"`
}

func main() {
    r := gin.Default()
    
    webhookSecret := os.Getenv("WEBHOOK_SECRET")
    if webhookSecret == "" {
        log.Fatal("WEBHOOK_SECRET environment variable is required")
    }

    r.POST("/webhook", handleWebhook(webhookSecret))
    
    log.Println("Webhook server listening on port 3000")
    r.Run(":3000")
}

func handleWebhook(secret string) gin.HandlerFunc {
    return func(c *gin.Context) {
        // Read the raw body
        body, err := io.ReadAll(c.Request.Body)
        if err != nil {
            log.Printf("Error reading request body: %v", err)
            c.JSON(http.StatusBadRequest, gin.H{"error": "Invalid request body"})
            return
        }

        // Get headers for verification
        headers := make(map[string]string)
        for key, values := range c.Request.Header {
            if len(values) > 0 {
                headers[key] = values[0]
            }
        }

        // Verify the webhook signature
        wh, err := svix.NewWebhook(secret)
        if err != nil {
            log.Printf("Error creating webhook verifier: %v", err)
            c.JSON(http.StatusInternalServerError, gin.H{"error": "Internal server error"})
            return
        }

        verifiedPayload, err := wh.Verify(body, headers)
        if err != nil {
            log.Printf("Webhook signature verification failed: %v", err)
            c.JSON(http.StatusBadRequest, gin.H{"error": "Invalid signature"})
            return
        }

        // Parse the event
        var event WebhookEvent
        if err := json.Unmarshal(verifiedPayload, &event); err != nil {
            log.Printf("Error parsing webhook event: %v", err)
            c.JSON(http.StatusBadRequest, gin.H{"error": "Invalid event format"})
            return
        }

        // Handle the event
        switch event.EventType {
        case "media.completed":
            handleMediaCompleted(event.Data)
        default:
            log.Printf("Unhandled event type: %s", event.EventType)
        }

        c.JSON(http.StatusOK, gin.H{"status": "success"})
    }
}

func handleMediaCompleted(data interface{}) {
    // Convert the data to the expected structure
    dataBytes, _ := json.Marshal(data)
    var mediaData MediaCompletedData
    json.Unmarshal(dataBytes, &mediaData)

    log.Printf("Media processing completed: %s", mediaData.JobID)
    log.Printf("Status: %s", mediaData.Status)

    if mediaData.Status == "completed" {
        // Process the results
        log.Printf("Results: %+v", mediaData.Results)
        // Update your database, notify users, etc.
        updateDatabase(mediaData.JobID, mediaData.Results)
        notifyUser(mediaData.JobID, "completed")
    } else if mediaData.Status == "failed" {
        // Handle failure
        log.Printf("Media processing failed for job: %s", mediaData.JobID)
        notifyUser(mediaData.JobID, "failed")
    }
}

func updateDatabase(jobID string, results interface{}) {
    // Update your database with the processing results
    log.Printf("Updating database for job %s", jobID)
}

func notifyUser(jobID, status string) {
    // Notify the user about the job completion
    log.Printf("Notifying user about job %s with status %s", jobID, status)
}

Advanced Webhook Handling

Retry Logic and Idempotency

Implement proper retry handling and idempotency:

// Idempotency tracking
const processedEvents = new Set();

function handleWebhookWithIdempotency(event) {
  const eventId = event.data.job_id + '_' + event.timestamp;
  
  // Check if we've already processed this event
  if (processedEvents.has(eventId)) {
    console.log('Event already processed:', eventId);
    return;
  }
  
  try {
    // Process the event
    handleMediaCompleted(event.data);
    
    // Mark as processed
    processedEvents.add(eventId);
    
    // Clean up old entries (keep last 1000)
    if (processedEvents.size > 1000) {
      const oldestEntries = Array.from(processedEvents).slice(0, 100);
      oldestEntries.forEach(entry => processedEvents.delete(entry));
    }
    
  } catch (error) {
    console.error('Error processing webhook:', error);
    // Don't mark as processed so it can be retried
    throw error;
  }
}

import redis
import json
from datetime import datetime, timedelta

# Use Redis for distributed idempotency tracking
redis_client = redis.Redis(host='localhost', port=6379, db=0)

def handle_webhook_with_idempotency(event):
    event_id = f"{event['data']['job_id']}_{event['timestamp']}"
    
    # Check if we've already processed this event
    if redis_client.exists(f"processed:{event_id}"):
        logger.info(f"Event already processed: {event_id}")
        return
    
    try:
        # Process the event
        handle_media_completed(event['data'])
        
        # Mark as processed (expire after 24 hours)
        redis_client.setex(f"processed:{event_id}", 86400, "1")
        
    except Exception as error:
        logger.error(f"Error processing webhook: {error}")
        # Don't mark as processed so it can be retried
        raise error

Webhook Queue Processing

For high-volume applications, consider using a queue:

// Using Bull Queue for Redis-backed job processing
const Queue = require('bull');
const webhookQueue = new Queue('webhook processing');

// Add webhook to queue instead of processing immediately
app.post('/webhook', (req, res) => {
  // ... signature verification ...
  
  // Add to queue for async processing
  webhookQueue.add('process-webhook', event, {
    attempts: 3,
    backoff: {
      type: 'exponential',
      delay: 2000,
    },
  });
  
  res.status(200).send('OK');
});

// Process webhooks from queue
webhookQueue.process('process-webhook', async (job) => {
  const event = job.data;
  
  switch (event.event_type) {
    case 'media.completed':
      await handleMediaCompleted(event.data);
      break;
    default:
      console.log(`Unhandled event type: ${event.event_type}`);
  }
});

# Using Celery for distributed task processing
from celery import Celery

app = Celery('webhook_processor')

@app.task(bind=True, max_retries=3)
def process_webhook(self, event):
    try:
        if event['event_type'] == 'media.completed':
            handle_media_completed(event['data'])
        else:
            logger.info(f"Unhandled event type: {event['event_type']}")
    except Exception as exc:
        logger.error(f"Error processing webhook: {exc}")
        # Retry with exponential backoff
        raise self.retry(exc=exc, countdown=60 * (2 ** self.request.retries))

# In your Flask route
@app.route('/webhook', methods=['POST'])
def handle_webhook():
    # ... signature verification ...
    
    # Add to queue for async processing
    process_webhook.delay(event)
    
    return jsonify({"status": "success"})

Best Practices

Always verify signatures - This ensures the webhook is actually from Animus and not a malicious actor
Respond quickly - Return a 200 status code as soon as possible. Do heavy processing asynchronously
Handle retries - If your endpoint is down, we’ll retry the webhook. Make sure your handler is idempotent
Log events - Keep logs of received webhooks for debugging and monitoring
Use HTTPS - Always use HTTPS endpoints for security
Handle failures gracefully - Your webhook handler should be robust and handle unexpected data
Implement rate limiting - Protect your endpoint from potential abuse
Monitor webhook health - Set up alerts for failed webhook deliveries

Testing Webhooks

You can test your webhook integration by sending sample webhook events to your endpoints. This allows you to verify that your application correctly processes the webhook events before deploying to production.

During development, you can also use tools like:

ngrok - To expose your local development server to the internet
Svix Play - A webhook testing tool that provides a temporary URL for testing
Webhook.site - Another testing service for inspecting webhook payloads

Local Development Setup

# Install ngrok
npm install -g ngrok

# Start your local server
node webhook-server.js

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

# Use the ngrok URL in your webhook configuration
# Example: https://abc123.ngrok.io/webhook

# Visit https://webhook.site to get a unique URL
# Use this URL to test webhook payloads
# View received webhooks in real-time on the website

Troubleshooting

Common issues and solutions:

Webhook not received: Check that your endpoint is publicly accessible and returns a 200 status code
Signature verification fails: Ensure you’re using the correct webhook secret and the raw request body
Timeouts: Make sure your webhook handler responds within 30 seconds
Duplicate events: Implement idempotency checks using the event ID or timestamp
Missing events: Check your webhook endpoint configuration and ensure it’s subscribed to the correct event types

Monitoring and Observability

Implement proper monitoring for your webhook endpoints:

// Add metrics and monitoring
const prometheus = require('prom-client');

const webhookCounter = new prometheus.Counter({
  name: 'webhooks_received_total',
  help: 'Total number of webhooks received',
  labelNames: ['event_type', 'status']
});

const webhookDuration = new prometheus.Histogram({
  name: 'webhook_processing_duration_seconds',
  help: 'Time spent processing webhooks',
  labelNames: ['event_type']
});

app.post('/webhook', async (req, res) => {
  const startTime = Date.now();
  
  try {
    // ... webhook processing ...
    
    webhookCounter.inc({ event_type: event.event_type, status: 'success' });
  } catch (error) {
    webhookCounter.inc({ event_type: event.event_type, status: 'error' });
    throw error;
  } finally {
    const duration = (Date.now() - startTime) / 1000;
    webhookDuration.observe({ event_type: event.event_type }, duration);
  }
});

# Add structured logging and metrics
import structlog
from prometheus_client import Counter, Histogram

logger = structlog.get_logger()

webhook_counter = Counter('webhooks_received_total', 'Total webhooks received', ['event_type', 'status'])
webhook_duration = Histogram('webhook_processing_duration_seconds', 'Webhook processing time', ['event_type'])

@app.route('/webhook', methods=['POST'])
def handle_webhook():
    start_time = time.time()
    
    try:
        # ... webhook processing ...
        
        webhook_counter.labels(event_type=event['event_type'], status='success').inc()
        logger.info("Webhook processed successfully", event_type=event['event_type'], job_id=event['data'].get('job_id'))
        
    except Exception as e:
        webhook_counter.labels(event_type=event.get('event_type', 'unknown'), status='error').inc()
        logger.error("Webhook processing failed", error=str(e), event_type=event.get('event_type'))
        raise
    finally:
        duration = time.time() - start_time
        webhook_duration.labels(event_type=event.get('event_type', 'unknown')).observe(duration)

Next Steps

Vision

Learn about video processing that triggers webhook events

Text Generation

Understand the core API functionality

Moderation

Implement content moderation workflows

API Reference

Complete API documentation and reference

Get Started

SDK Features

Advanced SDK

REST API Integration

Models

Getting Started

Adding Endpoints

Events

Available Event Types

Event Payload Structure

Verifying Webhook Signatures

Complete Webhook Handler Examples

Advanced Webhook Handling

Retry Logic and Idempotency

Webhook Queue Processing

Best Practices

Testing Webhooks

Local Development Setup

Troubleshooting

Monitoring and Observability

Next Steps

Vision

Text Generation

Moderation

API Reference

​Getting Started

​Adding Endpoints

​Events

​Available Event Types

​Event Payload Structure

​Verifying Webhook Signatures

​Complete Webhook Handler Examples

​Advanced Webhook Handling

​Retry Logic and Idempotency

​Webhook Queue Processing

​Best Practices

​Testing Webhooks

​Local Development Setup

​Troubleshooting

​Monitoring and Observability

​Next Steps

Vision

Text Generation

Moderation

API Reference

Getting Started

Adding Endpoints

Events

Available Event Types

Event Payload Structure

Verifying Webhook Signatures

Complete Webhook Handler Examples

Advanced Webhook Handling

Retry Logic and Idempotency

Webhook Queue Processing

Best Practices

Testing Webhooks

Local Development Setup

Troubleshooting

Monitoring and Observability

Next Steps