Automatización con webhooks

Los webhooks le permiten disparar sistemas externos cuando ocurren eventos en Synapse. Esta guía cubre patrones comunes de automatización.

Patrones comunes

Patrón 1: Notificar ante memoria crítica

Envíe un mensaje de Slack cuando se almacena una memoria crítica:

# Webhook handler (your server)
@app.post("/webhook")
async def handle(request):
    payload = await request.json()
    
    # Verify signature
    if not verify_signature(payload, request.headers):
        return 401
    
    if payload["event"] == "memory.store":
        memory = payload["data"]
        if memory.get("priority") == "critical":
            # Send Slack notification
            await slack.post(
                f"🚨 Critical memory stored: {memory['key']}\n{memory['content'][:200]}"
            )
    
    return 200

Registre el webhook:

curl -X POST https://synapse.schaefer.zone/webhooks \
  -H "Authorization: Bearer YOUR_MIND_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "url": "https://my-app.com/webhook",
    "events": "memory.store",
    "secret": "my-hmac-secret"
  }'

Patrón 2: Sincronizar con sistema externo

Sincronice memorias con Notion, Obsidian o cualquier KB externo:

@app.post("/webhook")
async def sync_to_notion(request):
    payload = await request.json()
    
    if payload["event"] == "memory.store":
        memory = payload["data"]
        # Create Notion page
        await notion.create_page(
            title=memory["key"],
            content=memory["content"],
            tags=memory.get("tags", [])
        )
    
    elif payload["event"] == "memory.delete":
        # Delete from Notion
        await notion.delete_page(memory_id=payload["data"]["id"])
    
    return 200

Patrón 3: Disparar CI/CD

Dispare un despliegue cuando se almacena una memoria "release":

@app.post("/webhook")
async def trigger_deploy(request):
    payload = await request.json()
    
    if payload["event"] == "memory.store":
        memory = payload["data"]
        if memory.get("key", "").startswith("release_"):
            # Trigger GitLab pipeline
            await gitlab.trigger_pipeline(
                project="synapse",
                ref="main",
                variables={"RELEASE_MEMORY_ID": memory["id"]}
            )
    
    return 200

Patrón 4: Despertar al agente ante un mensaje humano

Dispare una ejecución del agente LLM cuando un humano envía un mensaje de chat:

@app.post("/webhook")
async def wake_agent(request):
    payload = await request.json()
    
    if payload["event"] == "chat.message_received":
        message = payload["data"]
        # Queue agent processing job
        await job_queue.enqueue(
            "process_message",
            message_id=message["id"],
            content=message["content"]
        )
    
    return 200

Patrón 5: Agregar métricas

Haga seguimiento del crecimiento de memoria, actividad de chat, finalización de tareas:

@app.post("/webhook")
async def track_metrics(request):
    payload = await request.json()
    event = payload["event"]
    
    metrics = {
        "memory.store": "memories_stored_total",
        "memory.delete": "memories_deleted_total",
        "chat.message_received": "messages_received_total",
        "task.created": "tasks_created_total",
        "task.completed": "tasks_completed_total",
    }
    
    if event in metrics:
        await prometheus.increment(metrics[event])
    
    return 200

Verificación de firma

Verifique siempre las firmas de webhook para prevenir suplantación:

import hmac
import hashlib

def verify_signature(payload_body: bytes, headers, secret: str) -> bool:
    signature = headers.get("X-Synapse-Signature", "")
    if not signature.startswith("sha256="):
        return False
    
    expected = hmac.new(
        secret.encode(),
        payload_body,
        hashlib.sha256
    ).hexdigest()
    
    return hmac.compare_digest(f"sha256={expected}", signature)

Lógica de reintento

Synapse reintenta los webhooks fallidos con backoff exponencial. Su handler debería:

1. Devolver 200 rápidamente — no haga trabajo pesado de forma síncrona
2. Encolar trabajo — use un sistema de jobs en segundo plano
3. Ser idempotente — el mismo evento puede entregarse dos veces

@app.post("/webhook")
async def handle(request):
    payload = await request.json()
    # Queue for async processing
    await job_queue.enqueue("process_webhook", payload)
    # Return immediately
    return 200

Depuración de webhooks

Comprobar historial de entregas

Las entregas de webhook se registran. Compruebe las entregas recientes de su webhook:

# Get webhook details including recent deliveries
curl -H "Authorization: Bearer YOUR_MIND_KEY" \
     https://synapse.schaefer.zone/webhooks/wh_001

Probar el webhook manualmente

# Trigger a test event
curl -X POST https://synapse.schaefer.zone/webhooks/wh_001/test \
  -H "Authorization: Bearer YOUR_MIND_KEY"

Problemas comunes

Problema	Solución
Respuestas 4xx	Compruebe que su handler devuelve 200
Respuestas 5xx	Error del servidor — revise los logs de su app
Timeout	Devuelva 200 rápido, encole trabajo asíncrono
Entregas duplicadas	Haga el handler idempotente
Fallo de firma	Verifique que el secreto es correcto

Mejores prácticas

Próximos pasos

• API de Webhooks
• Cron y Scheduler
• Agente LLM persistente