Skip to main content

Webhooks API

メモリ、チャット、タスクイベントに対する HTTP コールバックの登録 — データ変更時に通知を受信。


Webhooks API

Webhooks を使用すると、Synapse でイベントが発生した際に HTTP コールバックを受信できます。外部自動化のトリガー、通知の送信、他システムへの同期に最適です。

エンドポイント

POST /webhooks

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

レスポンス:

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

GET /webhooks

現在の mind のすべての Webhook を一覧表示します。

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

GET /webhooks/:id

単一の Webhook を取得します。

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

PUT /webhooks/:id

Webhook を更新します(URL、events、secret、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

Webhook を削除します。

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

イベント種別

パターン 発生タイミング
memory.* 任意のメモリイベント
memory.store 新規メモリの保存
memory.update メモリの更新
memory.delete メモリの削除
chat.* 任意のチャットイベント
chat.message_received 人間からの新規メッセージ
task.* 任意のタスクイベント
task.created 新規タスクの作成
task.completed タスクの完了マーク
* 全イベント

Webhook ペイロード

イベント発生時、Synapse はあなたの URL に POST リクエストを送信します。

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

署名検証

secret を設定すると、Synapse は各ペイロードを HMAC-SHA256 で署名します。

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

ハンドラで検証します:

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)

パターン:リアルタイム同期

# 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"])

次のステップ