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β
- Include
webhook_urlwhen creating an analysis - CallCov processes the audio
- When complete, we POST results to your webhook URL
- 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β
- Always validate signatures before processing
- Respond quickly (< 30 seconds)
- Process asynchronously if business logic is slow
- Use HTTPS for your webhook URL
- Handle duplicates (store processed webhook IDs)
- Log all webhooks for debugging
Retry Logicβ
If your webhook endpoint fails, CallCov automatically retries:
| Attempt | Delay |
|---|---|
| 1st retry | 1 minute |
| 2nd retry | 5 minutes |
| 3rd retry | 15 minutes |
| 4th retry | 1 hour |
| 5th retry | 6 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
Relatedβ
- Create Analysis - Include webhook_url when submitting
- Error Handling - Handle webhook failures gracefully
- Testing Guide - Test webhooks in development