Webhooks Guide

1. Event occurs in CallIntel (call completed, transcript ready, etc.)
   ↓
2. System triggers webhook event
   ↓
3. HTTP POST request sent to your configured webhook URL
   ↓
4. Your application receives and processes the data
   ↓
5. Your application sends HTTP 200 response
   ↓
6. Webhook delivery marked as successful

curl -X POST https://py.callai.rejoicehub.com/v1/webhooks \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "url": "https://your-domain.com/webhooks/callintel",
    "events": [
      "call.completed",
      "call.transcript_ready",
      "batch.completed"
    ],
    "active": true
  }'

{
  "event_type": "call.completed",
  "timestamp": "2024-12-16T10:30:45Z",
  "webhook_id": "wh_1a2b3c4d5e",
  "data": {
    "id": "call_123456789",
    "call_from": "+1234567890",
    "call_to": "+0987654321",
    "agent_id": "agent_abc123",
    "agent_name": "Support Agent",
    "status": "completed",
    "direction": "outbound",
    "call_type": "voice",
    "start_time": "2024-12-16T10:25:00Z",
    "end_time": "2024-12-16T10:30:45Z",
    "duration_seconds": 345,
    "recording_url": "https://storage.callintel.com/recordings/call_123.mp3",
    "transcript_url": "https://storage.callintel.com/transcripts/call_123.txt",
    "custom_metadata": {
      "campaign_id": "camp_001",
      "customer_id": "cust_789"
    }
  }
}

{
  "event_type": "call.transcript_ready",
  "timestamp": "2024-12-16T10:31:00Z",
  "webhook_id": "wh_1a2b3c4d5e",
  "data": {
    "id": "call_123456789",
    "call_id": "call_123456789",
    "transcript": "Agent: Hello, how can I help you today? Customer: I'd like to check my order status...",
    "summary": "Customer inquired about order #12345. Agent confirmed order is in transit and will arrive by Dec 20.",
    "sentiment": "positive",
    "intent": "order_tracking",
    "actionable_items": [
      "Send customer tracking link",
      "Follow up if not delivered by Dec 20"
    ],
    "keywords": ["order", "tracking", "delivery"],
    "confidence_score": 0.95
  }
}

{
  "event_type": "batch.completed",
  "timestamp": "2024-12-16T11:45:00Z",
  "webhook_id": "wh_1a2b3c4d5e",
  "data": {
    "id": "batch_987654321",
    "batch_name": "Q4 Customer Survey",
    "agent_id": "agent_abc123",
    "agent_name": "Survey Agent",
    "status": "completed",
    "total_numbers": 1000,
    "successfully_called": 856,
    "failed_calls": 144,
    "avg_duration_seconds": 287,
    "start_time": "2024-12-16T10:00:00Z",
    "end_time": "2024-12-16T11:45:00Z",
    "success_rate": 0.856,
    "cost_usd": 45.20,
    "custom_metadata": {
      "campaign_type": "survey",
      "department": "customer_success"
    }
  }
}

import hmac
import hashlib
from flask import Flask, request

app = Flask(__name__)
WEBHOOK_SECRET = "your_webhook_secret"

@app.route('/webhooks/callintel', methods=['POST'])
def handle_webhook():
    # Get signature from headers
    signature = request.headers.get('X-Callintel-Signature')
    
    # Calculate expected signature
    payload = request.get_data(as_text=True)
    expected_signature = hmac.new(
        WEBHOOK_SECRET.encode(),
        payload.encode(),
        hashlib.sha256
    ).hexdigest()
    
    # Verify signature
    if not hmac.compare_digest(signature, expected_signature):
        return {'error': 'Invalid signature'}, 401
    
    # Process webhook
    data = request.get_json()
    process_event(data)
    
    return {'status': 'success'}, 200

def process_event(event):
    webhook_id = event.get('webhook_id')
    
    # Check if already processed
    if WebhookLog.query.filter_by(webhook_id=webhook_id).exists():
        return  # Already processed
    
    # Process event
    event_data = event.get('data')
    # ... your logic here ...
    
    # Log webhook
    log = WebhookLog(webhook_id=webhook_id, processed_at=datetime.now())
    db.session.add(log)
    db.session.commit()

def handle_webhook(event):
    try:
        # Process event
        result = process_event(event)
        return {'status': 'success'}, 200
    except TemporaryError as e:
        # Return 5xx for CallIntel to retry
        return {'error': 'Temporary error'}, 503
    except PermanentError as e:
        # Log error and return 200 to prevent retries
        logger.error(f"Permanent error: {e}")
        return {'status': 'error', 'logged': True}, 200

@app.route('/webhooks/callintel', methods=['POST'])
def handle_webhook():
    event = request.get_json()
    
    # Acknowledge immediately
    result = {'status': 'received'}
    
    # Process asynchronously
    process_event_async.delay(event)
    
    return result, 202  # Accepted

def log_webhook(event, status, response):
    log = WebhookLog(
        webhook_id=event.get('webhook_id'),
        event_type=event.get('event_type'),
        status=status,
        response=response,
        received_at=datetime.now()
    )
    db.session.add(log)
    db.session.commit()

# Send test webhook
curl -X POST https://your-domain.com/webhooks/callintel \
  -H "Content-Type: application/json" \
  -H "X-Callintel-Signature: test_signature" \
  -d '{
    "event_type": "call.completed",
    "timestamp": "2024-12-16T10:30:45Z",
    "webhook_id": "wh_test_123",
    "data": {
      "id": "call_test_123",
      "status": "completed",
      "duration_seconds": 300
    }
  }'

curl -X GET "https://py.callai.rejoicehub.com/v1/webhooks/logs?limit=50" \
  -H "Authorization: Bearer YOUR_API_KEY"

# Test URL accessibility
curl -X POST https://your-domain.com/webhooks/callintel \
  -H "Content-Type: application/json" \
  -d '{"test": true}'

# ✗ WRONG - Will timeout
@app.route('/webhooks/callintel', methods=['POST'])
def handle_webhook():
    event = request.get_json()
    process_long_task(event)  # Takes 30 seconds
    return {'status': 'success'}, 200

# ✓ CORRECT - Responds quickly
@app.route('/webhooks/callintel', methods=['POST'])
def handle_webhook():
    event = request.get_json()
    queue.enqueue(process_long_task, event)
    return {'status': 'accepted'}, 202

def process_event(event):
    webhook_id = event['webhook_id']
    
    # Use atomic operation to prevent duplicates
    created = WebhookLog.create_if_not_exists(webhook_id)
    
    if not created:
        return  # Already processed
    
    # Process event...

curl -X POST https://py.callai.rejoicehub.com/v1/webhooks \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "url": "https://your-domain.com/webhooks",
    "events": ["call.completed"],
    "filters": {
      "direction": "outbound",
      "status": "completed",
      "min_duration": 60
    }
  }'

# Production webhook
curl -X POST https://py.callai.rejoicehub.com/v1/webhooks \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -d '{
    "url": "https://py.callai.rejoicehub.com/webhooks",
    "events": ["call.completed", "batch.completed"]
  }'

# Analytics webhook
curl -X POST https://py.callai.rejoicehub.com/v1/webhooks \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -d '{
    "url": "https://analytics.example.com/webhooks",
    "events": ["call.completed"]
  }'

# Backup webhook
curl -X POST https://py.callai.rejoicehub.com/v1/webhooks \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -d '{
    "url": "https://backup.example.com/webhooks",
    "events": ["call.completed", "call.transcript_ready"]
  }'

curl -X PUT https://py.callai.rejoicehub.com/v1/webhooks/wh_123 \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -d '{
    "retry_policy": {
      "max_attempts": 5,
      "backoff_multiplier": 2,
      "initial_delay_seconds": 60,
      "max_delay_seconds": 3600
    }
  }'