Webhooks para Logística: Notificaciones en Tiempo Real
Webhooks para Logística: Notificaciones en Tiempo Real
Introducción
Los webhooks transforman la gestión de transporte al proporcionar notificaciones instantáneas de eventos críticos. Esta guía técnica muestra cómo implementar webhooks para logística.
---
Qué son los Webhooks
Los webhooks son notificaciones HTTP automáticas que se envían cuando ocurre un evento específico. En logística, permiten recibir alertas instantáneas sobre:
- Cambios de estado de envío - Incidencias y retrasos - Entregas completadas - Actualizaciones de ubicación - Alertas de documentación
---
Eventos Disponibles para Transporte
| Evento | Descripción | Payload | | -------------------- | ---------------------- | ------------------- | | `envio.creado` | Nuevo envío registrado | ID, origen, destino | | `envio.en_ruta` | Vehículo en movimiento | GPS, ETA | | `envio.parada` | Parada registrada | Ubicación, tiempo | | `envio.entregado` | Entrega completada | Firma, hora | | `envio.incidente` | Incidencia reportada | Tipo, descripción | | `documento.generado` | CMR/eCMR creado | Documento PDF | | `pago.procesado` | Pago completado | Importe, referencia |
---
Configuración de Webhooks
Registrar Endpoint
```bash curl -X POST "https://api.cargoffer.com/v1/webhooks" \ -H "Authorization: Bearer TU_API_KEY" \ -H "Content-Type: application/json" \ -d '{ "url": "https://tu-sistema.com/webhooks/logistica", "events": ["envio.", "documento."], "secret": "tu_secret_key" }' ```
Estructura del Payload
```json { "event": "envio.entregado", "timestamp": "2026-02-20T15:30:00Z", "data": { "envio_id": "ENV-2026-001234", "estado": "entregado", "cliente": "Empresa SL", "conductor": "Juan García", "vehiculo": "1234-ABC", "ubicacion_entrega": { "lat": 40.4168, "lon": -3.7038, "direccion": "Calle Gran Vía 1, Madrid" }, "firma_digital": "https://cdn.cargoffer.com/firmas/2026/02/abc123.pdf", "foto_entrega": "https://cdn.cargoffer.com/fotos/2026/02/img456.jpg", "incidencias": [] }, "signature": "sha256=abc123..." } ```
---
Verificar Firma
Python
```python import hmac import hashlib import json
def verify_webhook_signature(payload, signature, secret): """Verifica que el webhook viene de Cargoffer"""
expected = hmac.new( secret.encode(), payload.encode(), hashlib.sha256 ).hexdigest()
return hmac.compare_digest(f"sha256={expected}", signature) ```
Node.js
```javascript const crypto = require("crypto");
function verifyWebhook(payload, signature, secret) { const expected = crypto .createHmac("sha256", secret) .update(payload) .digest("hex");
return crypto.timingSafeEqual( Buffer.from(`sha256=${expected}`), Buffer.from(signature), ); } ```
---
Manejo de Webhooks
Servidor Express.js
```javascript const express = require("express"); const crypto = require("crypto");
const app = express(); app.use(express.raw({ type: "application/json" }));
const WEBHOOK_SECRET = process.env.WEBHOOK_SECRET;
app.post("/webhooks/logistica", (req, res) => { // Verificar firma const signature = req.headers["x-cargoffer-signature"]; const isValid = verifyWebhook(req.body, signature, WEBHOOK_SECRET);
if (!isValid) { return res.status(401).json({ error: "Invalid signature" }); }
const payload = JSON.parse(req.body.toString());
// Procesar evento switch (payload.event) { case "envio.entregado": handleEntrega(payload.data); break; case "envio.incidente": handleIncidente(payload.data); break; case "documento.generado": handleDocumento(payload.data); break; default: console.log("Evento no manejado:", payload.event); }
// Responder rápido para evitar timeout res.status(200).json({ received: true }); });
function handleEntrega(data) { // Actualizar sistema ERP // Notificar cliente // Generar factura }
function handleIncidente(data) { // Alertar equipo operaciones // Registrar incidencia // Notificar seguro si aplica }
app.listen(3000); ```
Python/FastAPI
```python from fastapi import FastAPI, Request, Header from pydantic import BaseModel from typing import Optional import hmac import hashlib
app = FastAPI()
class WebhookPayload(BaseModel): event: str timestamp: str data: dict
def verify_signature(payload: str, signature: str, secret: str) -> bool: expected = hmac.new( secret.encode(), payload.encode(), hashlib.sha256 ).hexdigest() return hmac.compare_digest(f"sha256={expected}", signature)
@app.post("/webhooks/logistica") async def receive_webhook( request: Request, x_cargoffer_signature: str = Header(None) ): payload = await request.body()
if not verify_signature(payload, x_cargoffer_signature, WEBHOOK_SECRET): raise HTTPException(status_code=401, detail="Invalid signature")
data = WebhookPayload.parse_raw(payload)
# Procesar según evento if data.event == "envio.entregado": await handle_entrega(data.data) elif data.event == "envio.incidente": await handle_incidente(data.data)
return {"status": "received"} ```
---
Casos de Uso
Notificaciones a Cliente
```python async def notificar_cliente(envio_id, evento): """Envía notificación cuando cambia estado de envío"""
cliente = await get_cliente(envio_id)
mensajes = { "envio.en_ruta": f"Tu envío {envio_id} está en camino. ETA: {evento.eta}", "envio.entregado": f"✅ Entregado: Tu mercancía ha llegado a destino", "envio.incidente": f"⚠️ Incidencia: Se ha reportado un problema con {envio_id}" }
# Enviar email, SMS, push notification await send_notification( cliente.email, cliente.telefono, mensajes[evento] ) ```
Integración con ERP
```python async def sincronizar_erp(evento): """Sincroniza estado con sistema ERP"""
if evento.event == "envio.entregado": # Marcar como entregado en ERP await erp.update_orden( orden_id=evento.data["referencia"], estado="entregado", fecha_entrega=evento.timestamp )
# Generar factura automática await erp.generar_factura( cliente=evento.data["cliente"], importe=evento.data["importe"] ) ```
Alertas de Seguridad
```python async def evaluar_alertas(evento): """Evalúa si requiere alertas especiales"""
# Incidencias críticas if evento.data.get("tipo") == "accidente": await notify_seguridad( tipo="accidente", vehiculo=evento.data["vehiculo"], ubicacion=evento.data["ubicacion"] )
# Retrasos importantes if evento.data.get("retraso_minutos", 0) > 60: await notify_cliente( tipo="retraso", motivo=evento.data.get("causa") ) ```
---
Mejores Prácticas
Retry Logic
```python import asyncio from datetime import datetime, timedelta
async def handle_webhook_with_retry(evento, max_retries=3): """Maneja webhook con reintentos automáticos"""
for intento in range(max_retries): try: await procesar_evento(evento) return {"status": "success"} except Exception as e: if intento == max_retries - 1: # Notificar a equipo técnico await notify_admin(f"Webhook fallido tras {max_retries}: {e}") raise await asyncio.sleep(2 intento) # Exponential backoff ```
Logging y Monitoreo
```python import logging
logger = logging.getLogger("webhooks")
@app.post("/webhooks/logistica") async def webhook_handler(request: Request): logger.info(f"Recibido webhook: {event} - {timestamp}")
try: await process_event(data) logger.info(f"Procesado OK: {event_id}") except Exception as e: logger.error(f"Error: {event_id} - {str(e)}", exc_info=True) raise
return {"status": "ok"} ```
Monitorización
```python
Métricas a监控ar
METRICAS = [ "webhooks.recibidos.total", "webhooks.procesados.ok", "webhooks.procesados.error", "webhooks.tiempo_respuesta.p50", "webhooks.tiempo_respuesta.p99" ] ```
---
Configuración en Dashboard
1. Acceder a Settings > Webhooks 2. Crear nuevo webhook con URL pública 3. Seleccionar eventos a monitorear 4. Configurar secret para firma 5. Probar con evento de prueba
---
FAQ
Cuál es el timeout máximo?
60 segundos. Responde inmediatamente y procesa async.
Los webhooks se reintentan?
Sí, hasta 3 veces con exponential backoff.
Puedo recibir webhooks en staging?
Sí, usa ngrok para desarrollo local.
---
Conclusión
Webhooks son esenciales para operativa logística en tiempo real. Permiten automatizar procesos y mantener informados a clientes instantáneamente.