Guía de Webhooks

Los webhooks proporcionan notificaciones en tiempo real cuando tu análisis se completa, eliminando la necesidad de hacer polling para obtener resultados.

¿Por Qué Usar Webhooks?

Sin webhooks (polling):

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

Con webhooks:

# Submit once, get notified when ready
submit_analysis(audio_url, webhook_url="https://your-app.com/webhooks")
# Continue with other work...
# Webhook llega cuando el análisis está completo!

Cómo Funcionan los Webhooks

1. Incluye webhook_url al crear un análisis
2. CallCov procesa el audio
3. Cuando se complete, hacemos POST de los resultados a tu URL de webhook
4. Tu servidor responde con 200 OK

Configurar Webhooks

Paso 1: Crea un Endpoint de Webhook

Tu endpoint de webhook debe:

• Aceptar requests POST
• Responder con código de estado 200-299
• Responder dentro de 30 segundos

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)

Paso 2: Envía Análisis con URL de Webhook

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']}")

Paso 3: Recibe Webhook

Cuando el análisis se completa, tu endpoint recibe:

{
  "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: Hola, gracias por llamar...",
    "segments": [
      {
        "speaker": "Agent",
        "text": "Hola, gracias por llamar.",
        "start": 0.0,
        "end": 2.5
      }
    ]
  },
  "results": {
    "compliance": { ... },
    "quality": { ... },
    "coaching": { ... }
  }
}

Seguridad de Webhooks

Verifica la Autenticidad del Webhook

CallCov incluye una firma en el header X-CallCov-Signature. Verifícala para asegurar que el webhook proviene de nosotros:

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)

Mejores Prácticas

1. Siempre valida las firmas antes de procesar
2. Responde rápidamente (< 30 segundos)
3. Procesa asincrónicamente si la lógica de negocio es lenta
4. Usa HTTPS para tu URL de webhook
5. Maneja duplicados (almacena IDs de webhooks procesados)
6. Registra todos los webhooks para debugging

Lógica de Reintento

Si tu endpoint de webhook falla, CallCov automáticamente reintenta:

Intento	Retraso
1er reintento	1 minuto
2do reintento	5 minutos
3er reintento	15 minutos
4to reintento	1 hora
5to reintento	6 horas

Después de 5 intentos fallidos, dejamos de reintentar. Aún puedes obtener los resultados a través de la API.

Probar Webhooks Localmente

Opción 1: ngrok

Usa ngrok para exponer tu servidor local:

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

# Terminal 2: Expose it publicly
ngrok http 5000

Usa la URL de ngrok como tu webhook_url:

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

Opción 2: LocalTunnel

npm install -g localtunnel
lt --port 5000

Opción 3: Servicio Echo de Modo de Prueba

CallCov proporciona un servicio echo para 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'
    }
)

Ve tu webhook en:

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

Patrón de Procesamiento Asíncrono

Para lógica de negocio de larga duración, responde inmediatamente y procesa asíncronamente:

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

Monitorear Webhooks

Rastrea la salud de webhooks en tu aplicación:

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

Problemas Comunes

Webhook No Recibido

Causas:

• URL no accesible públicamente
• Firewall bloqueando las IPs de CallCov
• Problemas con el certificado HTTPS
• Endpoint devolviendo estado no-200

Soluciones:

• Prueba con ngrok o servicio webhook-test
• Verifica los logs del servidor para requests entrantes
• Verifica que el certificado HTTPS sea válido
• Asegúrate de que el endpoint devuelva 200 OK

Webhooks Duplicados

CallCov puede enviar el mismo webhook múltiples veces. Manéjalos idempotentemente:

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

Relacionado

• Crear Análisis - Incluye webhook_url al enviar
• Manejo de Errores - Maneja fallos de webhook con elegancia
• Guía de Testing - Prueba webhooks en desarrollo