Webhooks API

Webhook cho phép bạn nhận HTTP callback khi sự kiện xảy ra trong Synapse. Hoàn hảo để kích hoạt tự động hóa bên ngoài, gửi thông báo hoặc đồng bộ với hệ thống khác.

Các endpoint

POST /webhooks

Đăng ký webhook mới.

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.*",
    "secret": "my-hmac-secret-min-8-chars"
  }'

Phản hồi:

{
  "id": "wh_001",
  "url": "https://my-app.com/webhook",
  "events": ["memory.*"],
  "enabled": true,
  "created_at": "2026-06-27T..."
}

GET /webhooks

Liệt kê tất cả webhook cho mind hiện tại.

curl -H "Authorization: Bearer YOUR_MIND_KEY" \
     https://synapse.schaefer.zone/webhooks

GET /webhooks/:id

Lấy một webhook đơn lẻ.

curl -H "Authorization: Bearer YOUR_MIND_KEY" \
     https://synapse.schaefer.zone/webhooks/wh_001

PUT /webhooks/:id

Cập nhật webhook (URL, sự kiện, secret hoặc cờ enabled).

curl -X PUT https://synapse.schaefer.zone/webhooks/wh_001 \
  -H "Authorization: Bearer YOUR_MIND_KEY" \
  -H "Content-Type: application/json" \
  -d '{"enabled": false}'

DELETE /webhooks/:id

Xóa một webhook.

curl -X DELETE -H "Authorization: Bearer YOUR_MIND_KEY" \
     https://synapse.schaefer.zone/webhooks/wh_001

Loại sự kiện

Payload Webhook

Khi sự kiện kích hoạt, Synapse POST đến URL của bạn:

{
  "event": "memory.store",
  "timestamp": "2026-06-27T...",
  "mind_id": "m_xyz789",
  "data": {
    "id": "mem_001",
    "category": "fact",
    "key": "user_name",
    "content": "Michael Schäfer"
  }
}

Xác minh chữ ký

Nếu bạn đặt secret, Synapse ký mỗi payload bằng HMAC-SHA256:

X-Synapse-Signature: sha256=<hex-hmac>

Xác minh trong trình xử lý của bạn:

import hmac, hashlib

def verify_signature(payload_body: bytes, signature: str, secret: str) -> bool:
    expected = hmac.new(
        secret.encode(),
        payload_body,
        hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(f"sha256={expected}", signature)

Mẫu: Đồng bộ thời gian thực

# Your webhook handler
@app.post("/webhook")
async def handle_webhook(request):
    body = await request.body()
    signature = request.headers.get("X-Synapse-Signature", "")
    if not verify_signature(body, signature, WEBHOOK_SECRET):
        return 401

    event = json.loads(body)
    if event["event"] == "memory.store":
        # Sync to another system
        sync_to_external(event["data"])
    elif event["event"] == "chat.message_received":
        # Trigger agent wake-up
        notify_agent(event["data"])

Bước tiếp theo

• Cron & Scheduler
• Hướng dẫn tự động hóa Webhook