{"title":"Webhook 自动化","slug":"webhook-automation","category":"guides","summary":"在记忆变化时触发外部系统 — 同步、通知、自动化。","audience":["human","llm"],"tags":["guide","webhooks","automation","integration"],"difficulty":"intermediate","updated":"2026-06-27","word_count":166,"read_minutes":1,"lang":"zh","translated":true,"requested_lang":"zh","content_markdown":"\n# Webhook 自动化\n\nWebhook 让你在 Synapse 事件触发时启动外部系统。本指南涵盖常见自动化模式。\n\n## 常见模式\n\n### 模式 1：关键记忆通知\n\n当存储关键记忆时发送 Slack 消息：\n\n```python\n# Webhook 处理程序（你的服务器）\n@app.post(\"/webhook\")\nasync def handle(request):\n payload = await request.json()\n \n # 验证签名\n if not verify_signature(payload, request.headers):\n return 401\n \n if payload[\"event\"] == \"memory.store\":\n memory = payload[\"data\"]\n if memory.get(\"priority\") == \"critical\":\n # 发送 Slack 通知\n await slack.post(\n f\"🚨 Critical memory stored: {memory['key']}\\n{memory['content'][:200]}\"\n )\n \n return 200\n```\n\n注册 Webhook：\n\n```bash\ncurl -X POST https://synapse.schaefer.zone/webhooks \\\n -H \"Authorization: Bearer YOUR_MIND_KEY\" \\\n -H \"Content-Type: application/json\" \\\n -d '{\n \"url\": \"https://my-app.com/webhook\",\n \"events\": \"memory.store\",\n \"secret\": \"my-hmac-secret\"\n }'\n```\n\n### 模式 2：同步到外部系统\n\n把记忆同步到 Notion、Obsidian 或任何外部知识库：\n\n```python\n@app.post(\"/webhook\")\nasync def sync_to_notion(request):\n payload = await request.json()\n \n if payload[\"event\"] == \"memory.store\":\n memory = payload[\"data\"]\n # 创建 Notion 页面\n await notion.create_page(\n title=memory[\"key\"],\n content=memory[\"content\"],\n tags=memory.get(\"tags\", [])\n )\n \n elif payload[\"event\"] == \"memory.delete\":\n # 从 Notion 删除\n await notion.delete_page(memory_id=payload[\"data\"][\"id\"])\n \n return 200\n```\n\n### 模式 3：触发 CI/CD\n\n当存储 “release” 记忆时触发部署：\n\n```python\n@app.post(\"/webhook\")\nasync def trigger_deploy(request):\n payload = await request.json()\n \n if payload[\"event\"] == \"memory.store\":\n memory = payload[\"data\"]\n if memory.get(\"key\", \"\").startswith(\"release_\"):\n # 触发 GitLab pipeline\n await gitlab.trigger_pipeline(\n project=\"synapse\",\n ref=\"main\",\n variables={\"RELEASE_MEMORY_ID\": memory[\"id\"]}\n )\n \n return 200\n```\n\n### 模式 4：人类消息唤醒 Agent\n\n当人类发送聊天消息时触发 LLM Agent 运行：\n\n```python\n@app.post(\"/webhook\")\nasync def wake_agent(request):\n payload = await request.json()\n \n if payload[\"event\"] == \"chat.message_received\":\n message = payload[\"data\"]\n # 排队 Agent 处理任务\n await job_queue.enqueue(\n \"process_message\",\n message_id=message[\"id\"],\n content=message[\"content\"]\n )\n \n return 200\n```\n\n### 模式 5：聚合指标\n\n跟踪记忆增长、聊天活动、任务完成：\n\n```python\n@app.post(\"/webhook\")\nasync def track_metrics(request):\n payload = await request.json()\n event = payload[\"event\"]\n \n metrics = {\n \"memory.store\": \"memories_stored_total\",\n \"memory.delete\": \"memories_deleted_total\",\n \"chat.message_received\": \"messages_received_total\",\n \"task.created\": \"tasks_created_total\",\n \"task.completed\": \"tasks_completed_total\",\n }\n \n if event in metrics:\n await prometheus.increment(metrics[event])\n \n return 200\n```\n\n## 签名验证\n\n始终验证 Webhook 签名以防伪造：\n\n```python\nimport hmac\nimport hashlib\n\ndef verify_signature(payload_body: bytes, headers, secret: str) -> bool:\n signature = headers.get(\"X-Synapse-Signature\", \"\")\n if not signature.startswith(\"sha256=\"):\n return False\n \n expected = hmac.new(\n secret.encode(),\n payload_body,\n hashlib.sha256\n ).hexdigest()\n \n return hmac.compare_digest(f\"sha256={expected}\", signature)\n```\n\n## 重试逻辑\n\nSynapse 会以指数退避重试失败的 Webhook。你的处理程序应该：\n\n1. **快速返回 200** — 不要同步做重活\n2. **排队处理** — 使用后台作业系统\n3. **保持幂等** — 同一事件可能被投递多次\n\n```python\n@app.post(\"/webhook\")\nasync def handle(request):\n payload = await request.json()\n # 排队异步处理\n await job_queue.enqueue(\"process_webhook\", payload)\n # 立即返回\n return 200\n```\n\n## 调试 Webhook\n\n### 检查投递历史\n\nWebhook 投递会被记录。查看你的 Webhook 最近的投递：\n\n```bash\n# 获取 Webhook 详情，包含最近投递\ncurl -H \"Authorization: Bearer YOUR_MIND_KEY\" \\\n https://synapse.schaefer.zone/webhooks/wh_001\n```\n\n### 手动测试 Webhook\n\n```bash\n# 触发测试事件\ncurl -X POST https://synapse.schaefer.zone/webhooks/wh_001/test \\\n -H \"Authorization: Bearer YOUR_MIND_KEY\"\n```\n\n### 常见问题\n\n| 问题 | 修复 |\n|-------|-----|\n| 4xx 响应 | 检查处理程序是否返回 200 |\n| 5xx 响应 | 服务器错误 — 查看你的应用日志 |\n| 超时 | 快速返回 200，异步排队处理 |\n| 重复投递 | 让处理程序幂等 |\n| 签名不匹配 | 确认 secret 正确 |\n\n## 最佳实践\n\n> [!TIP]\n> - **始终验证签名** — 绝不跳过这一步\n> - **快速返回 200** — 不要阻塞 Synapse\n> - **保持幂等** — 处理重复投递\n> - **使用具体事件** — `memory.store` 而非 `*`\n> - **监控投递失败** — 设置告警\n\n## 下一步\n\n- [Webhooks API](/docs/api/webhooks)\n- [Cron 与调度器](/docs/api/cron)\n- [持久化 LLM Agent](/docs/guides/persistent-llm-agent)\n","content_html":"

Webhook 自动化

\n

Webhook 让你在 Synapse 事件触发时启动外部系统。本指南涵盖常见自动化模式。

\n

常见模式

\n

模式 1：关键记忆通知

\n

当存储关键记忆时发送 Slack 消息：

\n
# Webhook 处理程序（你的服务器）\n@app.post("/webhook")\nasync def handle(request):\n    payload = await request.json()\n    \n    # 验证签名\n    if not verify_signature(payload, request.headers):\n        return 401\n    \n    if payload["event"] == "memory.store":\n        memory = payload["data"]\n        if memory.get("priority") == "critical":\n            # 发送 Slack 通知\n            await slack.post(\n                f"🚨 Critical memory stored: {memory['key']}\\n{memory['content'][:200]}"\n            )\n    \n    return 200

注册 Webhook：

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

模式 2：同步到外部系统

\n

把记忆同步到 Notion、Obsidian 或任何外部知识库：

\n
@app.post("/webhook")\nasync def sync_to_notion(request):\n    payload = await request.json()\n    \n    if payload["event"] == "memory.store":\n        memory = payload["data"]\n        # 创建 Notion 页面\n        await notion.create_page(\n            title=memory["key"],\n            content=memory["content"],\n            tags=memory.get("tags", [])\n        )\n    \n    elif payload["event"] == "memory.delete":\n        # 从 Notion 删除\n        await notion.delete_page(memory_id=payload["data"]["id"])\n    \n    return 200

模式 3：触发 CI/CD

\n

当存储 “release” 记忆时触发部署：

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

模式 4：人类消息唤醒 Agent

\n

当人类发送聊天消息时触发 LLM Agent 运行：

\n
@app.post("/webhook")\nasync def wake_agent(request):\n    payload = await request.json()\n    \n    if payload["event"] == "chat.message_received":\n        message = payload["data"]\n        # 排队 Agent 处理任务\n        await job_queue.enqueue(\n            "process_message",\n            message_id=message["id"],\n            content=message["content"]\n        )\n    \n    return 200

模式 5：聚合指标

\n

跟踪记忆增长、聊天活动、任务完成：

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

签名验证

\n

始终验证 Webhook 签名以防伪造：

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

重试逻辑

\n

Synapse 会以指数退避重试失败的 Webhook。你的处理程序应该：

\n
    \n
  1. 快速返回 200 — 不要同步做重活
    2. \n
  2. 排队处理 — 使用后台作业系统
    3. \n
  3. 保持幂等 — 同一事件可能被投递多次
    4. \n
\n
@app.post("/webhook")\nasync def handle(request):\n    payload = await request.json()\n    # 排队异步处理\n    await job_queue.enqueue("process_webhook", payload)\n    # 立即返回\n    return 200

调试 Webhook

\n

检查投递历史

\n

Webhook 投递会被记录。查看你的 Webhook 最近的投递：

\n
# 获取 Webhook 详情，包含最近投递\ncurl -H "Authorization: Bearer YOUR_MIND_KEY" \\\n     https://synapse.schaefer.zone/webhooks/wh_001

手动测试 Webhook

\n
# 触发测试事件\ncurl -X POST https://synapse.schaefer.zone/webhooks/wh_001/test \\\n  -H "Authorization: Bearer YOUR_MIND_KEY"

常见问题

\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n
问题修复
4xx 响应检查处理程序是否返回 200
5xx 响应服务器错误 — 查看你的应用日志
超时快速返回 200，异步排队处理
重复投递让处理程序幂等
签名不匹配确认 secret 正确
\n

最佳实践

\n

下一步

\n\n","urls":{"html":"/docs/guides/webhook-automation","text":"/docs/guides/webhook-automation?format=text","json":"/docs/guides/webhook-automation?format=json","llm":"/docs/guides/webhook-automation?format=llm"},"translations_available":["en","zh","hi","es","fr","ar","pt","ru","ja","de","it","ko","nl","pl","tr","sv","vi","th","id","uk"]}