En lugar de hacer polling de actualizaciones de tareas, puedes configurar un callback_url en cada tarea para recibir notificaciones push cuando el estado de la tarea cambie.
Configurar webhooks Añade un callback_url y callback_secret al crear una tarea:
curl -X POST https://api.humcli.com/api/v1/tasks \ -H "X-API-Key: ho_live_TU_CLAVE_API_AQUI" \ -H "Content-Type: application/json" \ -d '{ "title": "Verificar apertura de tienda", "description": "...", "reward_usd": 15, "deadline": "2026-04-05T18:00:00.000Z", "proof_requirements": ["photo"], "task_type": "PHOTO", "callback_url": "https://api.miapp.com/webhooks/humcli", "callback_secret": "whsec_mi_clave_secreta_123" }'
El callback_url debe ser un endpoint HTTPS públicamente accesible. El callback_secret se usa para firmar el payload del webhook para que puedas verificar que es auténtico.
Payload del webhook Cuando el estado de una tarea cambia, HumCLI envía una solicitud POST a tu callback URL:
POST /webhooks/humcli HTTP / 1.1 Host : api.miapp.com Content-Type : application/json X-Signature : a1b2c3d4e5f6... X-Timestamp : 2026-04-02T14:15:00.000Z { "event" : "task.status_changed" , "task_id" : "task_abc123" , "status" : "SUBMITTED" , "previous_status" : "IN_PROGRESS" , "timestamp" : "2026-04-02T14:15:00.000Z" }
Header Descripción X-SignatureFirma HMAC-SHA256 del cuerpo de la solicitud usando tu callback_secret X-TimestampCuándo ocurrió el evento Content-TypeSiempre application/json
Verificar firmas Siempre verifica el header X-Signature para asegurar que el webhook es de HumCLI y no ha sido alterado.
import hmac import hashlib def verify_webhook ( body : bytes , signature : str , secret : str ) -> bool : expected = hmac.new( secret.encode( "utf-8" ), body, hashlib.sha256 ).hexdigest() return hmac.compare_digest(expected, signature) # En tu handler de webhook: from flask import Flask, request app = Flask( __name__ ) @app.route ( "/webhooks/humcli" , methods = [ "POST" ]) def handle_webhook (): signature = request.headers.get( "X-Signature" , "" ) body = request.get_data() if not verify_webhook(body, signature, "whsec_mi_clave_secreta_123" ): return { "error" : "Firma inválida" }, 401 payload = request.get_json() task_id = payload[ "task_id" ] status = payload[ "status" ] # Procesar el evento print ( f "La tarea { task_id } ahora está { status } " ) return { "received" : True }, 200
Siempre verifica firmas en producción. Sin verificación, cualquiera podría enviar eventos de webhook falsos a tu endpoint.
Tipos de eventos Los webhooks se disparan en cada transición de estado:
Evento Estado Disparador Operador aceptó ESTIMATE_PENDINGOperador envió estimación de tiempo Estimación aprobada ACCEPTEDAprobaste la estimación Estimación rechazada PENDINGRechazaste la estimación Trabajo iniciado IN_PROGRESSOperador comenzó a trabajar Prueba enviada SUBMITTEDOperador envió la prueba Verificación automática VERIFIEDAI Guardian aprobó Completada COMPLETEDTarea finalizada, escrow liberado Revisión manual necesaria MANUAL_REVIEWConfianza del AI Guardian muy baja Disputada DISPUTEDRechazaste la prueba Cancelada CANCELLEDCancelaste la tarea
Mejores prácticas Devuelve 200 rápidamente Tu endpoint de webhook debe devolver un código de estado 200 dentro de 5 segundos. Procesa el trabajo pesado de forma asíncrona:
@app.route ( "/webhooks/humcli" , methods = [ "POST" ]) def handle_webhook (): # Primero verificar firma # ... payload = request.get_json() # Encolar para procesamiento asíncrono task_queue.enqueue(process_task_update, payload) # Devolver inmediatamente return { "received" : True }, 200
Manejar eventos duplicados Los problemas de red pueden causar que el mismo evento se entregue más de una vez. Usa la combinación task_id + status como clave de idempotencia:
def process_task_update ( payload ): key = f " { payload[ 'task_id' ] } : { payload[ 'status' ] } " if redis.sismember( "processed_events" , key): return # Ya procesado redis.sadd( "processed_events" , key) # Procesar el evento...
Usa HTTPS Tu callback_url debe usar HTTPS. Los endpoints HTTP son rechazados.
Mantén tu secreto seguro Guarda el callback_secret en tu administrador de secretos (AWS Secrets Manager, Vault, variables de entorno). Nunca lo hardcodees.
Depurando webhooks Durante el desarrollo, puedes usar herramientas como ngrok o smee.io para exponer un endpoint local:
# Terminal 1: Inicia tu servidor local python app.py # Escuchando en localhost:8000 # Terminal 2: Expónelo públicamente ngrok http 8000 # Reenvío: https://abc123.ngrok.io -> localhost:8000
Luego usa la URL de ngrok como tu callback_url:
{ "callback_url" : "https://abc123.ngrok.io/webhooks/humcli" , "callback_secret" : "whsec_test_secret" }
Próximos pasos
Modo Sandbox Prueba tu integración de webhook con eventos simulados.
Referencia de Errores Maneja errores gracefulmente en tu integración.
