웹훅을 사용하면 Synapse 이벤트가 발생할 때 외부 시스템을 트리거할 수\n있습니다. 이 가이드는 일반적인 자동화 패턴을 다룹니다.
\n중요 메모리가 저장될 때 Slack 메시지를 전송:
\n# Webhook handler (your server)\n\nasync def handle(request):\n payload = await request.json()\n \n # Verify signature\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 # Send Slack notification\n await slack.post(\n f"🚨 Critical memory stored: {memory['key']}\\n{memory['content'][:200]}"\n )\n \n return 200웹훅 등록:
\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 }'메모리를 Notion, Obsidian 또는 외부 KB에 동기화:
\n\nasync def sync_to_notion(request):\n payload = await request.json()\n \n if payload["event"] == "memory.store":\n memory = payload["data"]\n # Create Notion page\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 # Delete from Notion\n await notion.delete_page(memory_id=payload["data"]["id"])\n \n return 200"release" 메모리가 저장될 때 배포 트리거:
\n\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 # Trigger 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사람이 채팅 메시지를 보낼 때 LLM 에이전트 실행 트리거:
\n\nasync def wake_agent(request):\n payload = await request.json()\n \n if payload["event"] == "chat.message_received":\n message = payload["data"]\n # Queue agent processing job\n await job_queue.enqueue(\n "process_message",\n message_id=message["id"],\n content=message["content"]\n )\n \n return 200메모리 증가, 채팅 활동, 작업 완료 추적:
\n\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스푸핑을 방지하기 위해 항상 웹훅 서명을 검증하십시오:
\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)Synapse는 실패한 웹훅을 지수 백오프로 재시도합니다. 귀하의 핸들러는\n다음을 수행해야 합니다:
\n\nasync def handle(request):\n payload = await request.json()\n # Queue for async processing\n await job_queue.enqueue("process_webhook", payload)\n # Return immediately\n return 200웹훅 전송은 기록됩니다. 웹훅의 최근 전송을 확인하십시오:
\n# Get webhook details including recent deliveries\ncurl -H "Authorization: Bearer YOUR_MIND_KEY" \\\n https://synapse.schaefer.zone/webhooks/wh_001# Trigger a test event\ncurl -X POST https://synapse.schaefer.zone/webhooks/wh_001/test \\\n -H "Authorization: Bearer YOUR_MIND_KEY"| 문제 | \n해결 | \n
|---|---|
| 4xx 응답 | \n핸들러가 200을 반환하는지 확인 | \n
| 5xx 응답 | \n서버 오류 — 앱 로그 확인 | \n
| 시간 초과 | \n200을 빨리 반환, 작업을 비동기로 큐잉 | \n
| 중복 전송 | \n핸들러를 멱등으로 만들기 | \n
| 서명 불일치 | \n시크릿이 올바른지 확인 | \n