Webhooks

Learn how to use webhooks for real-time notifications and event-driven processing with both Python and JavaScript SDKs.

Overview

Webhooks allow you to receive real-time notifications when transcription jobs complete, fail, or reach certain milestones. Instead of polling for status updates, VerbalisAI will send HTTP POST requests to your specified endpoint.

Webhook Events

VerbalisAI sends webhooks for the following events:

Event	Description
`transcription.started`	Transcription job has begun processing
`transcription.progress`	Progress update (25%, 50%, 75% completion)
`transcription.completed`	Transcription job completed successfully
`transcription.failed`	Transcription job failed
`file.uploaded`	File upload completed
`usage.limit_warning`	Usage approaching limit (80%, 90%)
`usage.limit_exceeded`	Usage limit exceeded

Setting Up Webhooks

Python SDK

from verbalisai import VerbalisAI
import asyncio

async def transcribe_with_webhook():
    client = VerbalisAI()
    
    # Start transcription with webhook
    transcription = await client.transcriptions.create(
        audio_url="https://example.com/long-audio.mp3",
        model="pro",
        
        # Webhook configuration
        webhook_url="https://yourserver.com/webhooks/verbalisai",
        webhook_auth_header_name="Authorization",
        webhook_auth_header_value="Bearer your-secret-token",
        
        # Don't wait for completion - use webhook instead
        wait_until_complete=False
    )
    
    print(f"Transcription started: {transcription.id}")
    print("Webhook will be called when processing completes")
    
    return transcription

asyncio.run(transcribe_with_webhook())

JavaScript SDK

import { VerbalisAI } from '@verbalisai/sdk';

async function transcribeWithWebhook() {
  const client = new VerbalisAI();
  
  // Start transcription with webhook
  const transcription = await client.transcriptions.create({
    audioUrl: 'https://example.com/long-audio.mp3',
    model: 'pro',
    
    // Webhook configuration
    webhookUrl: 'https://yourserver.com/webhooks/verbalisai',
    webhookAuthHeaderName: 'Authorization',
    webhookAuthHeaderValue: 'Bearer your-secret-token',
    
    // Don't wait for completion - use webhook instead
    waitUntilComplete: false
  });
  
  console.log(`Transcription started: ${transcription.id}`);
  console.log('Webhook will be called when processing completes');
  
  return transcription;
}

transcribeWithWebhook();

Webhook Payload Structure

Transcription Completed

{
  "event": "transcription.completed",
  "timestamp": "2025-06-16T15:30:00Z",
  "data": {
    "transcription_id": "clx1234567890abcdef",
    "status": "completed",
    "audio_url": "https://example.com/audio.mp3",
    "duration": 125.5,
    "model": "pro",
    "text": "Complete transcription text...",
    "topics": ["technology", "AI", "transcription"],
    "summary": {
      "text": "Brief summary of the audio content...",
      "type": "bullets"
    },
    "segments": [
      {
        "id": 0,
        "text": "Hello, this is the beginning of the audio.",
        "start": 0.0,
        "end": 3.2,
        "speaker_id": "speaker_1"
      }
    ],
    "processing_time": 45.2,
    "credits_used": 1.25
  }
}

Transcription Failed

{
  "event": "transcription.failed",
  "timestamp": "2025-06-16T15:30:00Z",
  "data": {
    "transcription_id": "clx1234567890abcdef",
    "status": "failed",
    "audio_url": "https://example.com/audio.mp3",
    "error": {
      "code": "AUDIO_FORMAT_UNSUPPORTED",
      "message": "The audio format is not supported",
      "details": "Expected MP3, WAV, FLAC, M4A, or OGG format"
    },
    "processing_time": 5.1
  }
}

Progress Update

{
  "event": "transcription.progress",
  "timestamp": "2025-06-16T15:25:00Z",
  "data": {
    "transcription_id": "clx1234567890abcdef",
    "status": "processing",
    "progress": 50,
    "estimated_completion": "2025-06-16T15:35:00Z"
  }
}

Webhook Server Implementation

Express.js Server (JavaScript)

import express from 'express';
import crypto from 'crypto';
import { VerbalisAI } from '@verbalisai/sdk';

const app = express();
const client = new VerbalisAI();

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

// Webhook endpoint
app.post('/webhooks/verbalisai', async (req, res) => {
  try {
    // Verify webhook signature
    const signature = req.headers['x-verbalisai-signature'];
    const payload = req.body;
    
    if (!verifyWebhookSignature(payload, signature)) {
      return res.status(401).json({ error: 'Invalid signature' });
    }
    
    const event = JSON.parse(payload.toString());
    
    // Process the webhook event
    await handleWebhookEvent(event);
    
    // Acknowledge receipt
    res.status(200).json({ success: true });
    
  } catch (error) {
    console.error('Webhook error:', error);
    res.status(500).json({ error: 'Internal server error' });
  }
});

function verifyWebhookSignature(payload, signature) {
  const secret = process.env.WEBHOOK_SECRET;
  if (!secret) return true; // Skip verification if no secret configured
  
  const expectedSignature = crypto
    .createHmac('sha256', secret)
    .update(payload)
    .digest('hex');
  
  return crypto.timingSafeEqual(
    Buffer.from(signature, 'hex'),
    Buffer.from(expectedSignature, 'hex')
  );
}

async function handleWebhookEvent(event) {
  console.log(`Received webhook: ${event.event}`);
  
  switch (event.event) {
    case 'transcription.completed':
      await handleTranscriptionCompleted(event.data);
      break;
      
    case 'transcription.failed':
      await handleTranscriptionFailed(event.data);
      break;
      
    case 'transcription.progress':
      await handleTranscriptionProgress(event.data);
      break;
      
    default:
      console.log('Unknown event type:', event.event);
  }
}

async function handleTranscriptionCompleted(data) {
  console.log(`Transcription ${data.transcription_id} completed`);
  
  // Example: Save to database
  await saveTranscriptionToDatabase(data);
  
  // Example: Send notification
  await sendCompletionNotification(data);
  
  // Example: Trigger next workflow step
  await triggerPostProcessing(data);
}

async function handleTranscriptionFailed(data) {
  console.error(`Transcription ${data.transcription_id} failed:`, data.error);
  
  // Example: Log error for monitoring
  await logTranscriptionError(data);
  
  // Example: Send alert
  await sendFailureAlert(data);
  
  // Example: Retry with different settings
  if (data.error.code === 'AUDIO_QUALITY_LOW') {
    await retryWithDifferentModel(data);
  }
}

async function handleTranscriptionProgress(data) {
  console.log(`Transcription ${data.transcription_id} progress: ${data.progress}%`);
  
  // Example: Update progress in real-time dashboard
  await updateProgressInDashboard(data);
  
  // Example: Send WebSocket update to client
  await broadcastProgressUpdate(data);
}

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

Flask Server (Python)

from flask import Flask, request, jsonify
import hmac
import hashlib
import json
from verbalisai import VerbalisAI
import asyncio

app = Flask(__name__)
client = VerbalisAI()

@app.route('/webhooks/verbalisai', methods=['POST'])
def webhook_handler():
    try:
        # Verify webhook signature
        signature = request.headers.get('X-VerbalisAI-Signature')
        payload = request.get_data()
        
        if not verify_webhook_signature(payload, signature):
            return jsonify({'error': 'Invalid signature'}), 401
        
        event = json.loads(payload)
        
        # Process the webhook event asynchronously
        asyncio.run(handle_webhook_event(event))
        
        # Acknowledge receipt
        return jsonify({'success': True}), 200
        
    except Exception as e:
        print(f'Webhook error: {e}')
        return jsonify({'error': 'Internal server error'}), 500

def verify_webhook_signature(payload, signature):
    secret = os.getenv('WEBHOOK_SECRET')
    if not secret:
        return True  # Skip verification if no secret configured
    
    expected_signature = hmac.new(
        secret.encode(),
        payload,
        hashlib.sha256
    ).hexdigest()
    
    return hmac.compare_digest(signature, expected_signature)

async def handle_webhook_event(event):
    print(f"Received webhook: {event['event']}")
    
    event_type = event['event']
    data = event['data']
    
    if event_type == 'transcription.completed':
        await handle_transcription_completed(data)
    elif event_type == 'transcription.failed':
        await handle_transcription_failed(data)
    elif event_type == 'transcription.progress':
        await handle_transcription_progress(data)
    else:
        print(f"Unknown event type: {event_type}")

async def handle_transcription_completed(data):
    print(f"Transcription {data['transcription_id']} completed")
    
    # Example: Save to database
    await save_transcription_to_database(data)
    
    # Example: Send notification
    await send_completion_notification(data)
    
    # Example: Trigger next workflow step
    await trigger_post_processing(data)

async def handle_transcription_failed(data):
    print(f"Transcription {data['transcription_id']} failed: {data['error']}")
    
    # Example: Log error for monitoring
    await log_transcription_error(data)
    
    # Example: Send alert
    await send_failure_alert(data)

async def handle_transcription_progress(data):
    print(f"Transcription {data['transcription_id']} progress: {data['progress']}%")
    
    # Example: Update progress in real-time dashboard
    await update_progress_in_dashboard(data)

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=3000)

Advanced Webhook Patterns

Webhook Queue Processing

import Queue from 'bull';
import redis from 'redis';

// Create Redis connection
const redisClient = redis.createClient();

// Create webhook processing queue
const webhookQueue = new Queue('webhook processing', {
  redis: { port: 6379, host: '127.0.0.1' }
});

// Webhook endpoint - just queue the event
app.post('/webhooks/verbalisai', async (req, res) => {
  try {
    const event = JSON.parse(req.body);
    
    // Add to queue for async processing
    await webhookQueue.add('process-webhook', event, {
      attempts: 3,
      backoff: 'exponential',
      delay: 1000
    });
    
    res.status(200).json({ queued: true });
    
  } catch (error) {
    console.error('Queue error:', error);
    res.status(500).json({ error: 'Failed to queue webhook' });
  }
});

// Process webhooks from queue
webhookQueue.process('process-webhook', async (job) => {
  const event = job.data;
  
  try {
    await handleWebhookEvent(event);
    console.log(`Processed webhook: ${event.event}`);
  } catch (error) {
    console.error('Webhook processing failed:', error);
    throw error; // This will trigger retry
  }
});

// Monitor queue
webhookQueue.on('completed', (job) => {
  console.log(`Webhook job ${job.id} completed`);
});

webhookQueue.on('failed', (job, err) => {
  console.error(`Webhook job ${job.id} failed:`, err);
});

Idempotent Webhook Processing

const processedWebhooks = new Set();

async function handleWebhookEvent(event) {
  // Create idempotency key
  const idempotencyKey = `${event.event}_${event.data.transcription_id}_${event.timestamp}`;
  
  // Check if already processed
  if (processedWebhooks.has(idempotencyKey)) {
    console.log('Webhook already processed, skipping');
    return;
  }
  
  try {
    // Process the event
    await processEvent(event);
    
    // Mark as processed
    processedWebhooks.add(idempotencyKey);
    
    // Optional: Store in database for persistence
    await storeProcessedWebhook(idempotencyKey);
    
  } catch (error) {
    console.error('Webhook processing failed:', error);
    throw error;
  }
}

Webhook Retry with Exponential Backoff

import asyncio
import aiohttp
from tenacity import retry, stop_after_attempt, wait_exponential

class WebhookProcessor:
    def __init__(self):
        self.session = None
    
    async def __aenter__(self):
        self.session = aiohttp.ClientSession()
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.session.close()
    
    @retry(
        stop=stop_after_attempt(5),
        wait=wait_exponential(multiplier=1, min=4, max=60)
    )
    async def send_webhook(self, url, payload, headers=None):
        """Send webhook with retry logic"""
        try:
            async with self.session.post(url, json=payload, headers=headers) as response:
                if response.status >= 200 and response.status < 300:
                    return await response.json()
                else:
                    raise aiohttp.ClientResponseError(
                        request_info=response.request_info,
                        history=response.history,
                        status=response.status
                    )
        except Exception as e:
            print(f"Webhook delivery failed: {e}")
            raise

# Usage
async def deliver_webhook(url, event_data):
    async with WebhookProcessor() as processor:
        try:
            result = await processor.send_webhook(url, event_data)
            print("Webhook delivered successfully")
            return result
        except Exception as e:
            print(f"Webhook delivery failed after all retries: {e}")

Webhook Security

Signature Verification

import crypto from 'crypto';

function verifyWebhookSignature(payload, signature, secret) {
  // VerbalisAI uses HMAC-SHA256
  const expectedSignature = crypto
    .createHmac('sha256', secret)
    .update(payload, 'utf8')
    .digest('hex');
  
  // Secure comparison to prevent timing attacks
  return crypto.timingSafeEqual(
    Buffer.from(signature, 'hex'),
    Buffer.from(expectedSignature, 'hex')
  );
}

// Usage in Express middleware
function webhookVerification(req, res, next) {
  const signature = req.headers['x-verbalisai-signature'];
  const secret = process.env.WEBHOOK_SECRET;
  
  if (!secret) {
    console.warn('Webhook secret not configured - skipping verification');
    return next();
  }
  
  if (!signature) {
    return res.status(401).json({ error: 'Missing signature header' });
  }
  
  const payload = req.body;
  if (!verifyWebhookSignature(payload, signature, secret)) {
    return res.status(401).json({ error: 'Invalid webhook signature' });
  }
  
  next();
}

app.use('/webhooks', webhookVerification);

IP Whitelisting

const VERBALISAI_IPS = [
  '192.168.1.100',
  '192.168.1.101',
  // Add VerbalisAI's webhook IP addresses
];

function ipWhitelist(req, res, next) {
  const clientIP = req.ip || req.connection.remoteAddress;
  
  if (!VERBALISAI_IPS.includes(clientIP)) {
    console.warn(`Webhook from unauthorized IP: ${clientIP}`);
    return res.status(403).json({ error: 'Unauthorized IP address' });
  }
  
  next();
}

app.use('/webhooks', ipWhitelist);

Testing Webhooks

Local Development with ngrok

# Install ngrok
npm install -g ngrok

# Start your webhook server
node webhook-server.js

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

# Use the ngrok URL as your webhook endpoint
# https://abc123.ngrok.io/webhooks/verbalisai

Webhook Testing Server

import express from 'express';

const app = express();
app.use(express.json());

// Simple webhook receiver for testing
app.post('/test-webhook', (req, res) => {
  console.log('Received webhook:');
  console.log('Headers:', req.headers);
  console.log('Body:', JSON.stringify(req.body, null, 2));
  
  res.status(200).json({ 
    received: true, 
    timestamp: new Date().toISOString() 
  });
});

app.listen(3001, () => {
  console.log('Test webhook server running on port 3001');
});

Mock Webhook Events

import requests
import json

def send_mock_webhook(webhook_url, event_type="transcription.completed"):
    """Send a mock webhook for testing"""
    
    mock_events = {
        "transcription.completed": {
            "event": "transcription.completed",
            "timestamp": "2025-06-16T15:30:00Z",
            "data": {
                "transcription_id": "test_12345",
                "status": "completed",
                "text": "This is a test transcription",
                "duration": 30.5,
                "topics": ["testing", "webhooks"]
            }
        },
        "transcription.failed": {
            "event": "transcription.failed",
            "timestamp": "2025-06-16T15:30:00Z",
            "data": {
                "transcription_id": "test_12345",
                "status": "failed",
                "error": {
                    "code": "TEST_ERROR", 
                    "message": "This is a test error"
                }
            }
        }
    }
    
    payload = mock_events.get(event_type)
    if not payload:
        raise ValueError(f"Unknown event type: {event_type}")
    
    response = requests.post(
        webhook_url,
        json=payload,
        headers={'Content-Type': 'application/json'}
    )
    
    print(f"Mock webhook sent: {response.status_code}")
    return response

# Usage
send_mock_webhook("http://localhost:3000/webhooks/verbalisai")

Best Practices

Error Handling

Always return a 2xx status code to acknowledge receipt
Implement retry logic with exponential backoff
Log webhook failures for monitoring
Use idempotency keys to handle duplicate events

Security

Always verify webhook signatures
Use HTTPS endpoints
Consider IP whitelisting
Store webhook secrets securely

Performance

Process webhooks asynchronously using queues
Avoid long-running operations in webhook handlers
Use database transactions for consistency
Monitor webhook processing times

Monitoring

Track webhook delivery success rates
Monitor processing times and queue depths
Set up alerts for webhook failures
Log all webhook events for debugging

Ready to learn about streaming? Check out the Streaming guide for real-time audio processing and live transcription features.

Getting Started

Python SDK

JavaScript SDK

Advanced Usage

SDK - Webhooks

Webhooks

Overview

Webhook Events

Setting Up Webhooks

Python SDK

JavaScript SDK

Webhook Payload Structure

Transcription Completed

Transcription Failed

Progress Update

Webhook Server Implementation

Express.js Server (JavaScript)

Flask Server (Python)

Advanced Webhook Patterns

Webhook Queue Processing

Idempotent Webhook Processing

Webhook Retry with Exponential Backoff

Webhook Security

Signature Verification

IP Whitelisting

Testing Webhooks

Local Development with ngrok

Webhook Testing Server

Mock Webhook Events

Best Practices

Error Handling

Security

Performance

Monitoring

Getting Started

Python SDK

JavaScript SDK

Advanced Usage

​Webhooks

​Overview

​Webhook Events

​Setting Up Webhooks

​Python SDK

​JavaScript SDK

​Webhook Payload Structure

​Transcription Completed

​Transcription Failed

​Progress Update

​Webhook Server Implementation

​Express.js Server (JavaScript)

​Flask Server (Python)

​Advanced Webhook Patterns

​Webhook Queue Processing

​Idempotent Webhook Processing

​Webhook Retry with Exponential Backoff

​Webhook Security

​Signature Verification

​IP Whitelisting

​Testing Webhooks

​Local Development with ngrok

​Webhook Testing Server

​Mock Webhook Events

​Best Practices

​Error Handling

​Security

​Performance

​Monitoring

Webhooks

Overview

Webhook Events

Setting Up Webhooks

Python SDK

JavaScript SDK

Webhook Payload Structure

Transcription Completed

Transcription Failed

Progress Update

Webhook Server Implementation

Express.js Server (JavaScript)

Flask Server (Python)

Advanced Webhook Patterns

Webhook Queue Processing

Idempotent Webhook Processing

Webhook Retry with Exponential Backoff

Webhook Security

Signature Verification

IP Whitelisting

Testing Webhooks

Local Development with ngrok

Webhook Testing Server

Mock Webhook Events

Best Practices

Error Handling

Security

Performance

Monitoring