Webhooks Event-Driven: Tutorial Paso a Paso con Código Real | gigstack API 2026

Introducción: ¿Por qué Webhooks para Facturación?

Los webhooks son el puente perfecto entre eventos en tu sistema de pagos y la generación automática de facturas. En lugar de consultar constantemente (polling) si hay nuevos pagos para facturar, los webhooks te notifican instantáneamente cuando algo relevante ocurre.

En este tutorial crearás un servidor webhook en Python que escucha eventos de gigstack API y automatiza acciones basadas en cambios de estado de facturas, pagos o clientes. Ideal para 2026 donde la velocidad de respuesta en facturación es crítica.

Lo que construirás:

• Servidor Flask que recibe webhooks de gigstack
• Validación de firma HMAC para seguridad
• Handlers específicos por tipo de evento
• Retry logic con exponential backoff
• Logging estructurado para debugging

Requisitos previos

• Python 3.9+: Verificado compatible con versiones 2026
• Cuenta gigstack: app.gigstack.pro
• API Key: Disponible en Settings → API
• Servidor accesible: URL pública para recibir webhooks (ngrok para desarrollo)

Paso 1: Setup del proyecto

mkdir gigstack-webhooks
cd gigstack-webhooks
python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate
pip install flask requests python-dotenv

Crea .env:

GIGSTACK_API_KEY=tu_api_key
GIGSTACK_WEBHOOK_SECRET=tu_webhook_secret
PORT=5000

Paso 2: Servidor webhook básico

Crea webhook_server.py:

# webhook_server.py
import os
import hmac
import hashlib
import json
from flask import Flask, request, jsonify
from dotenv import load_dotenv
import logging

load_dotenv()

app = Flask(__name__)
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

WEBHOOK_SECRET = os.getenv('GIGSTACK_WEBHOOK_SECRET').encode()

def verify_signature(payload_body, signature_header):
    """Valida que el webhook venga realmente de gigstack"""
    expected_signature = hmac.new(
        WEBHOOK_SECRET,
        payload_body,
        hashlib.sha256
    ).hexdigest()
    
    return hmac.compare_digest(expected_signature, signature_header)

@app.route('/webhooks/gigstack', methods=['POST'])
def handle_webhook():
    # Obtener firma del header
    signature = request.headers.get('X-Gigstack-Signature')
    
    if not signature:
        logger.warning('⚠️ Webhook sin firma rechazado')
        return jsonify({'error': 'Missing signature'}), 401
    
    # Validar firma HMAC
    payload_body = request.get_data()
    
    if not verify_signature(payload_body, signature):
        logger.warning('⚠️ Firma inválida detectada')
        return jsonify({'error': 'Invalid signature'}), 401
    
    # Parsear evento
    event = request.json
    event_type = event.get('type')
    
    logger.info(f'✅ Webhook recibido: {event_type}')
    
    # Despachar a handler específico
    handler = EVENT_HANDLERS.get(event_type)
    
    if handler:
        try:
            handler(event['data'])
            return jsonify({'status': 'processed'}), 200
        except Exception as e:
            logger.error(f'❌ Error procesando webhook: {e}')
            return jsonify({'error': str(e)}), 500
    else:
        logger.warning(f'⚠️ Evento no manejado: {event_type}')
        return jsonify({'status': 'ignored'}), 200

if __name__ == '__main__':
    port = int(os.getenv('PORT', 5000))
    app.run(host='0.0.0.0', port=port, debug=False)

¿Qué hace este código?

• Valida que cada webhook venga firmado con tu secret (previene ataques)
• Parsea el tipo de evento para despacharlo al handler correcto
• Devuelve 200 inmediatamente para confirmar recepción
• Logea eventos para debugging

Paso 3: Handlers por tipo de evento

Crea event_handlers.py:

# event_handlers.py
import logging
import requests
import os
from datetime import datetime

logger = logging.getLogger(__name__)
API_KEY = os.getenv('GIGSTACK_API_KEY')
BASE_URL = 'https://api.gigstack.io/v2'

HEADERS = {
    'Authorization': f'Bearer {API_KEY}',
    'Content-Type': 'application/json'
}

def handle_invoice_created(data):
    """Se dispara cuando una nueva factura es creada"""
    invoice_id = data.get('_id')
    client_email = data.get('client', {}).get('email')
    total = data.get('total')
    
    logger.info(f'📄 Nueva factura: {invoice_id}')
    logger.info(f'   Cliente: {client_email}')
    logger.info(f'   Total: ${total} MXN')
    
    # Ejemplo: Enviar notificación interna
    send_slack_notification(
        f'💰 Nueva factura por ${total} MXN para {client_email}'
    )

def handle_invoice_stamped(data):
    """Se dispara cuando una factura es timbrada exitosamente"""
    uuid = data.get('uuid')
    folio = data.get('folio')
    pdf_url = data.get('pdf_url')
    
    logger.info(f'✅ Factura timbrada: {uuid}')
    logger.info(f'   Folio: {folio}')
    
    # Ejemplo: Guardar PDF en tu almacenamiento
    download_and_store_pdf(pdf_url, uuid)

def handle_payment_received(data):
    """Se dispara cuando se registra un pago"""
    payment_id = data.get('_id')
    amount = data.get('amount')
    client = data.get('client', {}).get('name')
    
    logger.info(f'💳 Pago recibido: {payment_id}')
    logger.info(f'   Monto: ${amount} MXN')
    logger.info(f'   Cliente: {client}')
    
    # Ejemplo: Generar complemento de pago automáticamente
    if data.get('invoice'):
        generate_payment_complement(data.get('invoice'), payment_id)

def handle_invoice_cancelled(data):
    """Se dispara cuando se cancela una factura"""
    invoice_id = data.get('_id')
    cancellation_reason = data.get('cancellation_reason')
    
    logger.warning(f'🚫 Factura cancelada: {invoice_id}')
    logger.warning(f'   Razón: {cancellation_reason}')
    
    # Ejemplo: Actualizar tu sistema ERP
    sync_cancellation_to_erp(invoice_id)

def handle_client_validated(data):
    """Se dispara cuando se valida un cliente con el SAT"""
    client_id = data.get('_id')
    rfc = data.get('rfc')
    is_valid = data.get('is_valid')
    
    if is_valid:
        logger.info(f'✅ Cliente validado: {rfc}')
    else:
        logger.warning(f'⚠️ Cliente inválido detectado: {rfc}')
        # Ejemplo: Notificar al equipo de ventas
        alert_invalid_client(client_id, rfc)

# Helpers (implementar según tus necesidades)
def send_slack_notification(message):
    """Envía notificación a Slack (ejemplo)"""
    # Implementa con tu webhook de Slack
    pass

def download_and_store_pdf(pdf_url, uuid):
    """Descarga PDF y almacena en S3/Azure/GCS"""
    response = requests.get(pdf_url)
    if response.status_code == 200:
        # Guardar en tu storage preferido
        filename = f'invoices/{uuid}.pdf'
        # Implementa tu lógica de almacenamiento
        logger.info(f'📥 PDF almacenado: {filename}')

def generate_payment_complement(invoice_id, payment_id):
    """Genera complemento de pago automáticamente"""
    endpoint = f'{BASE_URL}/invoices/{invoice_id}/payment-complement'
    payload = {'payment_id': payment_id}
    
    response = requests.post(endpoint, json=payload, headers=HEADERS)
    
    if response.status_code == 201:
        logger.info(f'✅ Complemento de pago generado para {invoice_id}')
    else:
        logger.error(f'❌ Error generando complemento: {response.text}')

def sync_cancellation_to_erp(invoice_id):
    """Sincroniza cancelación con tu sistema ERP"""
    # Implementa tu lógica de sincronización
    pass

def alert_invalid_client(client_id, rfc):
    """Alerta cuando se detecta un cliente con RFC inválido"""
    # Implementa notificación a equipo de ventas
    pass

# Mapeo de eventos a handlers
EVENT_HANDLERS = {
    'invoice.created': handle_invoice_created,
    'invoice.stamped': handle_invoice_stamped,
    'payment.received': handle_payment_received,
    'invoice.cancelled': handle_invoice_cancelled,
    'client.validated': handle_client_validated
}

Paso 4: Integrar handlers en el servidor

Actualiza webhook_server.py para importar los handlers:

# Al inicio del archivo webhook_server.py
from event_handlers import EVENT_HANDLERS

Paso 5: Registrar webhook en gigstack

Crea register_webhook.py:

# register_webhook.py
import os
import requests
from dotenv import load_dotenv

load_dotenv()

API_KEY = os.getenv('GIGSTACK_API_KEY')
BASE_URL = 'https://api.gigstack.io/v2'

def register_webhook(url, events):
    endpoint = f'{BASE_URL}/webhooks'
    
    payload = {
        'url': url,
        'events': events,
        'active': True
    }
    
    headers = {
        'Authorization': f'Bearer {API_KEY}',
        'Content-Type': 'application/json'
    }
    
    response = requests.post(endpoint, json=payload, headers=headers)
    
    if response.status_code == 201:
        webhook_data = response.json()['data']
        print('✅ Webhook registrado exitosamente')
        print(f"   ID: {webhook_data['_id']}")
        print(f"   URL: {webhook_data['url']}")
        print(f"   Secret: {webhook_data['secret']}")
        print('\n⚠️ IMPORTANTE: Guarda el secret en tu .env como GIGSTACK_WEBHOOK_SECRET')
        return webhook_data
    else:
        print(f'❌ Error registrando webhook: {response.text}')
        return None

if __name__ == '__main__':
    # URL pública de tu servidor (usa ngrok para desarrollo)
    webhook_url = 'https://tu-dominio.com/webhooks/gigstack'
    
    # Eventos que quieres escuchar
    events = [
        'invoice.created',
        'invoice.stamped',
        'payment.received',
        'invoice.cancelled',
        'client.validated'
    ]
    
    register_webhook(webhook_url, events)

Ejecuta el registro:

python register_webhook.py

Copia el secret devuelto y agrégalo a tu .env.

Paso 6: Desarrollo local con ngrok

Para probar webhooks en tu máquina local:

# Instala ngrok: https://ngrok.com/download
ngrok http 5000

Ngrok te dará una URL pública temporal como https://abc123.ngrok.io. Usa esta URL cuando registres tu webhook.

Inicia tu servidor:

python webhook_server.py

Paso 7: Testing de webhooks

Crea test_webhook.py para simular eventos:

# test_webhook.py
import requests
import hmac
import hashlib
import json
import os
from dotenv import load_dotenv

load_dotenv()

WEBHOOK_URL = 'http://localhost:5000/webhooks/gigstack'
WEBHOOK_SECRET = os.getenv('GIGSTACK_WEBHOOK_SECRET').encode()

def send_test_event(event_type, data):
    payload = {
        'type': event_type,
        'data': data,
        'timestamp': '2026-01-30T12:00:00Z'
    }
    
    payload_json = json.dumps(payload)
    
    # Generar firma HMAC
    signature = hmac.new(
        WEBHOOK_SECRET,
        payload_json.encode(),
        hashlib.sha256
    ).hexdigest()
    
    headers = {
        'Content-Type': 'application/json',
        'X-Gigstack-Signature': signature
    }
    
    response = requests.post(WEBHOOK_URL, data=payload_json, headers=headers)
    
    print(f'Status: {response.status_code}')
    print(f'Response: {response.json()}')

if __name__ == '__main__':
    # Test: Factura timbrada
    test_data = {
        '_id': '507f1f77bcf86cd799439011',
        'uuid': '123e4567-e89b-12d3-a456-426614174000',
        'folio': 'A-1234',
        'total': 11600.00,
        'pdf_url': 'https://api.gigstack.io/invoices/123.pdf',
        'client': {
            'email': 'cliente@empresa.mx',
            'name': 'Acme Corp'
        }
    }
    
    send_test_event('invoice.stamped', test_data)

Ejecuta el test:

python test_webhook.py

Manejo de errores y retry logic

gigstack reintenta webhooks fallidos automáticamente con exponential backoff. Tu servidor debe:

1. Devolver 200 rápidamente: No hagas procesamiento pesado en el endpoint
2. Ser idempotente: Un evento puede llegar múltiples veces
3. Manejar duplicados: Usa el event_id para deduplicar

Implementa deduplicación:

# Agrega a webhook_server.py
processed_events = set()  # En producción usa Redis o DB

@app.route('/webhooks/gigstack', methods=['POST'])
def handle_webhook():
    event = request.json
    event_id = event.get('id')
    
    # Deduplicación
    if event_id in processed_events:
        logger.info(f'⚠️ Evento duplicado ignorado: {event_id}')
        return jsonify({'status': 'duplicate'}), 200
    
    # ... resto del código
    
    processed_events.add(event_id)
    return jsonify({'status': 'processed'}), 200

Deployment en producción

Opción 1: Railway

# railway.json
{
  "build": {
    "builder": "NIXPACKS"
  },
  "deploy": {
    "startCommand": "python webhook_server.py",
    "restartPolicyType": "ON_FAILURE"
  }
}

Opción 2: Render

# render.yaml
services:
  - type: web
    name: gigstack-webhooks
    env: python
    buildCommand: pip install -r requirements.txt
    startCommand: python webhook_server.py
    envVars:
      - key: GIGSTACK_API_KEY
        sync: false
      - key: GIGSTACK_WEBHOOK_SECRET
        sync: false

Opción 3: Docker

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 5000

CMD ["python", "webhook_server.py"]

Mejores prácticas 2026

1. Usa colas de mensajes: Para procesamiento asíncrono pesado, usa Celery + Redis
2. Monitorea tus webhooks: Implementa alertas si los webhooks fallan consistentemente
3. Valida siempre la firma: Nunca confíes en webhooks sin validación HMAC
4. Logea todo: Los logs son críticos para debugging de eventos asíncronos
5. Versiona tus endpoints: /webhooks/v1/gigstack para facilitar migraciones

Próximos pasos avanzados

• Webhooks bidireccionales: Notificar a gigstack cuando tu sistema cambia
• Fan-out pattern: Un webhook dispara múltiples acciones en paralelo
• Circuit breaker: Detener webhooks temporalmente si tu sistema está saturado
• Event sourcing: Guardar todos los eventos para replay y auditoría

FAQ Técnica: Troubleshooting

¿Por qué mis webhooks no llegan?

Verifica que tu URL sea pública y accesible desde internet. Si estás en desarrollo local, asegúrate de usar ngrok o una alternativa. Revisa los logs en el dashboard de gigstack para ver intentos fallidos.

¿Cómo pruebo webhooks sin generar facturas reales?

Usa tu API key de staging (test mode). Los webhooks funcionan igual pero con datos de prueba. También puedes usar test_webhook.py para simular eventos localmente.

¿Qué hago si un webhook falla por timeout?

gigstack espera respuesta en 10 segundos. Si tu procesamiento toma más, devuelve 200 inmediatamente y procesa en background con una cola (Celery, RabbitMQ, etc.). El webhook solo debe validar y encolar, no procesar.

¿Cómo deduplicar eventos si mi servidor se reinicia?

En producción, usa una base de datos o Redis para trackear event_id procesados. Un set() en memoria se pierde con cada reinicio. Ejemplo: redis.sadd('processed_events', event_id) con TTL de 7 días.

Recursos adicionales

• Documentación Webhooks gigstack: docs.gigstack.io/webhooks
• Flask docs: flask.palletsprojects.com
• HMAC validation: docs.python.org/3/library/hmac
• Ngrok: ngrok.com/docs

Con este tutorial tienes una arquitectura event-driven completa para automatizar facturación. Los webhooks eliminan polling innecesario y hacen tu sistema más reactivo y escalable en 2026.