> ## Documentation Index
> Fetch the complete documentation index at: https://docs.animusai.co/llms.txt
> Use this file to discover all available pages before exploring further.
# Webhooks
> Webhooks are a way to consume events from Animus in real-time. They allow your application to receive notifications when certain events happen, such as when video processing jobs complete. This is particularly useful for long-running operations where you don't want to continuously poll for status updates.
## Getting Started
To start using webhooks, you'll need to:
1. **Set up an endpoint** - Create a URL that can receive POST requests
2. **Configure the webhook** - Add your endpoint in the Consumer App Portal
3. **Verify signatures** - Implement signature verification for security
4. **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:
```json theme={null}
{
"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](https://docs.svix.com/receiving/verifying-payloads/why).
Our webhook partner Svix offers a set of useful libraries that make verifying webhooks very simple:
```javascript JavaScript theme={null}
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);
```
```python Python theme={null}
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)
```
```go Go theme={null}
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](https://docs.svix.com/receiving/introduction).
For more instructions and examples of how to verify signatures, check out their [webhook verification documentation](https://docs.svix.com/receiving/verifying-payloads/how).
## Complete Webhook Handler Examples
Here are complete examples of webhook handlers in different frameworks:
```javascript Node.js/Express theme={null}
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');
});
```
```python Flask theme={null}
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)
```
```python FastAPI theme={null}
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
```
```go Go/Gin theme={null}
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:
```javascript JavaScript theme={null}
// 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;
}
}
```
```python Python theme={null}
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:
```javascript JavaScript theme={null}
// 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}`);
}
});
```
```python Python theme={null}
# 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
1. **Always verify signatures** - This ensures the webhook is actually from Animus and not a malicious actor
2. **Respond quickly** - Return a 200 status code as soon as possible. Do heavy processing asynchronously
3. **Handle retries** - If your endpoint is down, we'll retry the webhook. Make sure your handler is idempotent
4. **Log events** - Keep logs of received webhooks for debugging and monitoring
5. **Use HTTPS** - Always use HTTPS endpoints for security
6. **Handle failures gracefully** - Your webhook handler should be robust and handle unexpected data
7. **Implement rate limiting** - Protect your endpoint from potential abuse
8. **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
```bash ngrok theme={null}
# 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
```
```bash Webhook.site theme={null}
# 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:
```javascript JavaScript theme={null}
// 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);
}
});
```
```python Python theme={null}
# 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
Learn about video processing that triggers webhook events
Understand the core API functionality
Implement content moderation workflows
Complete API documentation and reference