Skip to main content

Webhooks Guide

Webhooks provide real-time notifications when your analysis completes, eliminating the need to poll for results.

Why Use Webhooks?​

Without webhooks (polling):

# Poll every 10 seconds until complete
while True:
analysis = get_analysis(analysis_id)
if analysis['status'] == 'completed':
break
time.sleep(10) # Inefficient!

With webhooks:

# Submit once, get notified when ready
submit_analysis(audio_url, webhook_url="https://your-app.com/webhooks")
# Continue with other work...
# Webhook arrives when analysis is complete!

How Webhooks Work​

  1. Include webhook_url when creating an analysis
  2. CallCov processes the audio
  3. When complete, we POST results to your webhook URL
  4. Your server responds with 200 OK

Setting Up Webhooks​

Step 1: Create a Webhook Endpoint​

Your webhook endpoint must:

  • Accept POST requests
  • Respond with 200-299 status code
  • Respond within 30 seconds
from flask import Flask, request, jsonify
app = Flask(__name__)
@app.route('/webhooks/analysis', methods=['POST'])
def handle_analysis_webhook():
data = request.json
# Validate webhook (recommended)
if not validate_webhook(request):
return jsonify({"error": "Invalid webhook"}), 401
# Process the analysis results
analysis_id = data['id']
status = data['status']
if status == 'completed':
# Extract insights
results = data['results']
compliance = results['compliance']
quality = results['quality']
coaching = results['coaching']
# Your business logic here
store_analysis_results(analysis_id, results)
notify_manager_if_compliance_issue(compliance)
update_agent_scorecard(quality)
elif status == 'failed':
# Handle failures
error = data.get('error_message')
log_error(f"Analysis {analysis_id} failed: {error}")
return jsonify({"received": True}), 200
if __name__ == '__main__':
app.run(port=5000)

Step 2: Submit Analysis with Webhook URL​

import requests
API_KEY = "your_api_key_here"
API_URL = "https://api.callcov.com/api/v1"
response = requests.post(
f"{API_URL}/analysis",
headers={"X-API-Key": API_KEY},
json={
"audio_url": "https://example.com/call.wav",
"agent_id": "agent_001",
"webhook_url": "https://your-app.com/webhooks/analysis"
}
)
print(f"Analysis ID: {response.json()['id']}")

Step 3: Receive Webhook​

When analysis completes, your endpoint receives:

{
"id": "550e8400-e29b-41d4-a716-446655440000",
"object": "analysis",
"created": 1642248000,
"status": "completed",
"call": {
"agent_id": "agent_001",
"contact_id": "customer_12345"
},
"audio": {
"url": "https://s3.amazonaws.com/callcov/...",
"duration_seconds": 125.5,
"format": "wav"
},
"transcript": {
"text": "Agent: Hello, thank you for calling...",
"segments": [
{
"speaker": "Agent",
"text": "Hello, thank you for calling.",
"start": 0.0,
"end": 2.5
}
]
},
"results": {
"compliance": { ... },
"quality": { ... },
"coaching": { ... }
}
}

Webhook Security​

Verify Webhook Authenticity​

CallCov includes a signature in the X-CallCov-Signature header. Verify it to ensure the webhook came from us:

import hmac
import hashlib

def validate_webhook(request):
"""Verify webhook signature"""
signature = request.headers.get('X-CallCov-Signature')
if not signature:
return False

# Get raw request body
payload = request.get_data()

# Your webhook secret (from CallCov dashboard)
webhook_secret = os.environ.get('CALLCOV_WEBHOOK_SECRET')

# Compute expected signature
expected_signature = hmac.new(
webhook_secret.encode('utf-8'),
payload,
hashlib.sha256
).hexdigest()

# Compare signatures (timing-safe)
return hmac.compare_digest(signature, expected_signature)

Best Practices​

  1. Always validate signatures before processing
  2. Respond quickly (< 30 seconds)
  3. Process asynchronously if business logic is slow
  4. Use HTTPS for your webhook URL
  5. Handle duplicates (store processed webhook IDs)
  6. Log all webhooks for debugging

Retry Logic​

If your webhook endpoint fails, CallCov automatically retries:

AttemptDelay
1st retry1 minute
2nd retry5 minutes
3rd retry15 minutes
4th retry1 hour
5th retry6 hours

After 5 failed attempts, we stop retrying. You can still fetch results via the API.

Testing Webhooks Locally​

Option 1: ngrok​

Use ngrok to expose your local server:

# Terminal 1: Start your webhook server
python webhook_server.py

# Terminal 2: Expose it publicly
ngrok http 5000

Use the ngrok URL as your webhook_url:

https://abc123.ngrok.io/webhooks/analysis

Option 2: LocalTunnel​

npm install -g localtunnel
lt --port 5000

Option 3: Test Mode Echo Service​

CallCov provides an echo service for testing:

response = requests.post(
'https://api.callcov.com/api/v1/analysis/',
headers={'Authorization': 'Bearer sk_test_abc123...'},
json={
'audio_url': 'https://example.com/test-call.wav',
'webhook_url': 'https://webhook-test.callcov.com/echo/YOUR_UNIQUE_ID'
}
)

View your webhook at:

https://webhook-test.callcov.com/view/YOUR_UNIQUE_ID

Async Processing Pattern​

For long-running business logic, respond immediately and process async:

from flask import Flask, request, jsonify
from celery import Celery

app = Flask(__name__)
celery = Celery('tasks', broker='redis://localhost:6379')

@celery.task
def process_analysis_async(data):
"""Process analysis in background"""
# Your long-running business logic
store_in_database(data)
update_dashboards(data)
send_email_notifications(data)
generate_reports(data)

@app.route('/webhooks/analysis', methods=['POST'])
def handle_webhook():
data = request.json

# Validate signature
if not validate_webhook(request):
return jsonify({"error": "Invalid signature"}), 401

# Queue for async processing
process_analysis_async.delay(data)

# Respond immediately
return jsonify({"received": True}), 200

Monitoring Webhooks​

Track webhook health in your application:

import logging
from datetime import datetime

webhook_logger = logging.getLogger('webhooks')

@app.route('/webhooks/analysis', methods=['POST'])
def handle_webhook():
start_time = datetime.now()

try:
data = request.json

# Log receipt
webhook_logger.info(f"Received webhook for analysis {data['id']}")

# Validate
if not validate_webhook(request):
webhook_logger.warning(f"Invalid signature for {data['id']}")
return jsonify({"error": "Invalid"}), 401

# Process
process_analysis(data)

# Log success
duration = (datetime.now() - start_time).total_seconds()
webhook_logger.info(f"Processed {data['id']} in {duration}s")

return jsonify({"received": True}), 200

except Exception as e:
# Log error
webhook_logger.error(f"Webhook error: {str(e)}", exc_info=True)
# Still return 200 to prevent retries for application errors
return jsonify({"error": "Internal error"}), 500

Common Issues​

Webhook Not Received​

Causes:

  • URL not publicly accessible
  • Firewall blocking CallCov's IPs
  • HTTPS certificate issues
  • Endpoint returning non-200 status

Solutions:

  • Test with ngrok or webhook-test service
  • Check server logs for incoming requests
  • Verify HTTPS certificate is valid
  • Ensure endpoint returns 200 OK

Duplicate Webhooks​

CallCov may send the same webhook multiple times. Handle idempotently:

processed_webhooks = set()

@app.route('/webhooks/analysis', methods=['POST'])
def handle_webhook():
data = request.json
analysis_id = data['id']

# Check if already processed
if analysis_id in processed_webhooks:
return jsonify({"received": True}), 200

# Process
process_analysis(data)

# Mark as processed
processed_webhooks.add(analysis_id)

return jsonify({"received": True}), 200