Webhook API

ElevenLabs webhook integration for emergency call processing

Webhook Endpoint

Process emergency calls from ElevenLabs AI voice agents for access911.

Endpoint

POST https://your-api-gateway-url/elevenlabs-webhook

Request

Headers

Header	Type	Required	Description
Content-Type	string	Yes	application/json
elevenlabs-signature	string	No	Webhook signature for verification

Body

The webhook receives a JSON payload from ElevenLabs with the following structure:

{
  "type": "post_call_transcription",
  "event_timestamp": 1703123456,
  "data": {
    "conversation_id": "conv_123456789",
    "agent_id": "agent_emergency_911",
    "status": "completed",
    "transcript": [
      {
        "speaker": "caller",
        "text": "Hello, I need help. There's a fire at my house.",
        "timestamp": 1703123400
      },
      {
        "speaker": "agent",
        "text": "I understand you have an emergency. Can you tell me your location?",
        "timestamp": 1703123405
      }
    ],
    "analysis": {
      "transcript_summary": "Caller reported house fire and provided location details.",
      "call_successful": true,
      "data_collection_results": {
        "emergency_type": {
          "value": "structure_fire",
          "rationale": "Caller explicitly mentioned house fire"
        },
        "location": {
          "value": "1234 Main Street, Anytown",
          "rationale": "Caller provided specific address"
        },
        "latitude": {
          "value": 34.0522,
          "rationale": "Geocoded from provided address"
        },
        "longitude": {
          "value": -118.2437,
          "rationale": "Geocoded from provided address"
        },
        "severity": {
          "value": "critical",
          "rationale": "Structure fire is always critical priority"
        }
      }
    },
    "metadata": {
      "call_duration_secs": 180,
      "language": "en-US"
    }
  }
}

Request Fields

Field	Type	Required	Description
type	string	Yes	Event type (post_call_transcription)
event_timestamp	integer	Yes	Unix timestamp of the event
data	object	Yes	Call data and analysis
data.conversation_id	string	Yes	Unique conversation identifier
data.agent_id	string	Yes	ElevenLabs agent identifier
data.status	string	Yes	Call status (completed, failed, etc.)
data.transcript	array	Yes	Conversation transcript
data.analysis	object	Yes	AI analysis results
data.analysis.transcript_summary	string	Yes	Summary of the conversation
data.analysis.call_successful	boolean	Yes	Whether the call was successful
data.analysis.data_collection_results	object	Yes	Extracted emergency metadata
data.metadata	object	No	Additional call metadata

Data Collection Results

The data_collection_results object contains structured emergency metadata:

Field	Type	Description
emergency_type	object	Type of emergency with value and rationale
location	object	Location information with value and rationale
latitude	object	Latitude coordinate with value and rationale
longitude	object	Longitude coordinate with value and rationale
severity	object	Emergency severity with value and rationale

Response

Success Response

{
  "status": "success",
  "message": "Webhook received"
}

Error Response

{
  "status": "error",
  "message": "Error description"
}

HTTP Status Codes

Code	Description
200	Success - Webhook processed successfully
400	Bad Request - Invalid webhook payload
401	Unauthorized - Invalid webhook signature
500	Internal Server Error - Processing error

Webhook Signature Verification

Optional webhook signature verification for security:

Signature Header Format

elevenlabs-signature: t=1703123456,v0=sha256=abc123...

Verification Process

1. Extract timestamp and signature from header
2. Verify timestamp is within 30 minutes
3. Calculate expected signature using webhook secret
4. Compare signatures using constant-time comparison

Example Verification

import hmac
import hashlib
import time

def verify_webhook_signature(signature_header, body, secret):
    if not signature_header or not secret:
        return True  # Skip verification if not configured
    
    # Parse signature header
    headers_parts = signature_header.split(",")
    timestamp = None
    signature = None
    
    for part in headers_parts:
        if part.startswith("t="):
            timestamp = part[2:]
        elif part.startswith("v0="):
            signature = part
    
    # Verify timestamp (within 30 minutes)
    if timestamp:
        req_timestamp = int(timestamp) * 1000
        tolerance = int(time.time() * 1000) - (30 * 60 * 1000)
        if req_timestamp <= tolerance:
            return False
    
    # Verify signature
    if timestamp and signature:
        message = f"{timestamp}.{body.decode('utf-8')}"
        expected_digest = 'v0=' + hmac.new(
            key=secret.encode("utf-8"),
            msg=message.encode("utf-8"),
            digestmod=hashlib.sha256
        ).hexdigest()
        return hmac.compare_digest(signature, expected_digest)
    
    return False

Data Storage

DynamoDB Storage

Emergency calls are stored in DynamoDB with this structure:

{
  "conversation_id": "conv_123456789",
  "timestamp": 1703123456,
  "agent_id": "agent_emergency_911",
  "summary": "Caller reported house fire and provided location details.",
  "call_successful": true,
  "duration_secs": 180,
  "transcript_length": 15,
  "created_at": "2023-12-21T10:30:56.789Z",
  "emergency_type": "structure_fire",
  "location": "1234 Main Street, Anytown",
  "latitude": 34.0522,
  "longitude": -118.2437,
  "severity": "critical"
}

S3 Storage

Full webhook payloads are stored in S3 for audit and analysis:

s3://your-bucket/calls/conv_123456789/conv_123456789_20231221_103056.json

Error Handling

Common Errors

Error	Description	Solution
Invalid JSON	Malformed request body	Check JSON syntax and structure
Missing fields	Required fields not provided	Ensure all required fields are present
Invalid signature	Webhook signature verification failed	Check webhook secret and signature format
Database error	DynamoDB or S3 operation failed	Check AWS credentials and permissions
Geocoding error	Location geocoding failed	Verify location data format

Error Response Format

{
  "status": "error",
  "message": "Error description",
  "timestamp": "2023-12-21T10:30:56.789Z",
  "error_code": "INVALID_SIGNATURE"
}

Rate Limits

• Request rate: 1000 requests per minute
• Burst limit: 500 requests per minute
• Timeout: 30 seconds per request

Security Considerations

Webhook Security

• Use HTTPS: Always use HTTPS for webhook endpoints
• Verify Signatures: Implement webhook signature verification
• Validate Input: Validate all incoming data
• Rate Limiting: Implement rate limiting to prevent abuse

Data Security

• Encryption: Use encryption at rest and in transit
• Access Control: Implement proper access controls
• Audit Logging: Log all webhook processing activities
• Data Retention: Implement appropriate data retention policies

Testing

Local Testing

Test webhook integration locally:

# Start local webhook server
python webhook_server.py

# Test webhook endpoint
curl -X POST http://localhost:8000/elevenlabs-webhook \
  -H "Content-Type: application/json" \
  -d '{
    "type": "post_call_transcription",
    "event_timestamp": 1703123456,
    "data": {
      "conversation_id": "test_conv_123",
      "agent_id": "test_agent",
      "status": "completed",
      "transcript": [],
      "analysis": {
        "transcript_summary": "Test emergency call",
        "call_successful": true,
        "data_collection_results": {
          "emergency_type": {"value": "test"},
          "location": {"value": "Test Location"},
          "severity": {"value": "moderate"}
        }
      }
    }
  }'

Webhook Validation

Validate webhook configuration:

def test_webhook_endpoint():
    # Test webhook signature verification
    # Test data extraction
    # Test DynamoDB storage
    # Test S3 storage
    pass

Monitoring

Key Metrics

• Webhook requests: Number of webhook requests received
• Processing time: Time to process each webhook
• Success rate: Percentage of successfully processed webhooks
• Error rate: Number of failed webhook processing attempts

CloudWatch Metrics

import boto3

cloudwatch = boto3.client('cloudwatch')

def publish_webhook_metrics(metric_name, value, unit='Count'):
    cloudwatch.put_metric_data(
        Namespace='DispatchAI/Webhook',
        MetricData=[
            {
                'MetricName': metric_name,
                'Value': value,
                'Unit': unit,
                'Timestamp': datetime.utcnow()
            }
        ]
    )

Best Practices

Performance Optimization

• Process Quickly: Process webhooks quickly to ensure real-time updates
• Use Async Processing: Use asynchronous processing for non-critical operations
• Implement Caching: Cache frequently accessed data
• Optimize Database: Use batch operations and connection pooling

Error Handling

• Implement Retry Logic: Retry failed operations with exponential backoff
• Log Errors: Log all errors for debugging and monitoring
• Handle Failures Gracefully: Don’t fail the entire webhook for non-critical errors
• Monitor Error Rates: Track error rates and types

Security

• Verify Signatures: Always verify webhook signatures in production
• Validate Input: Validate all incoming data
• Use HTTPS: Always use HTTPS for webhook endpoints
• Implement Rate Limiting: Prevent abuse and DoS attacks

SDK Examples

Python

import requests
import json
import hmac
import hashlib

class DispatchAIWebhook:
    def __init__(self, webhook_url, secret=None):
        self.webhook_url = webhook_url
        self.secret = secret
    
    def verify_signature(self, signature_header, body):
        if not self.secret:
            return True
        
        # Implementation of signature verification
        # ... (see verification example above)
        return True
    
    def process_webhook(self, payload, signature_header=None):
        if not self.verify_signature(signature_header, json.dumps(payload)):
            raise ValueError("Invalid webhook signature")
        
        response = requests.post(
            self.webhook_url,
            json=payload,
            headers={'Content-Type': 'application/json'}
        )
        
        if response.status_code != 200:
            raise Exception(f"Webhook error: {response.status_code}")
        
        return response.json()

# Usage
webhook = DispatchAIWebhook(
    "https://your-api-gateway-url/elevenlabs-webhook",
    secret="your_webhook_secret"
)

result = webhook.process_webhook(payload)
print(f"Webhook processed: {result['status']}")

JavaScript

class DispatchAIWebhook {
  constructor(webhookUrl, secret = null) {
    this.webhookUrl = webhookUrl;
    this.secret = secret;
  }
  
  async processWebhook(payload, signatureHeader = null) {
    if (!this.verifySignature(signatureHeader, JSON.stringify(payload))) {
      throw new Error('Invalid webhook signature');
    }
    
    const response = await fetch(this.webhookUrl, {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json'
      },
      body: JSON.stringify(payload)
    });
    
    if (!response.ok) {
      throw new Error(`Webhook error: ${response.status}`);
    }
    
    return await response.json();
  }
  
  verifySignature(signatureHeader, body) {
    // Implementation of signature verification
    // ... (see verification example above)
    return true;
  }
}

// Usage
const webhook = new DispatchAIWebhook(
  'https://your-api-gateway-url/elevenlabs-webhook',
  'your_webhook_secret'
);

webhook.processWebhook(payload)
  .then(result => console.log(`Webhook processed: ${result.status}`))
  .catch(error => console.error('Error:', error));

Production Use: Ensure proper security configuration and monitoring before deploying to production emergency response systems.