{"title":"أتمتة الـ Webhooks","slug":"webhook-automation","category":"guides","summary":"شغّل الأنظمة الخارجية عند تغير الذكريات — مزامنة، إشعارات، أتمتة.","audience":["human","llm"],"tags":["guide","webhooks","automation","integration"],"difficulty":"intermediate","updated":"2026-06-27","word_count":309,"read_minutes":2,"lang":"ar","translated":true,"requested_lang":"ar","content_markdown":"\n# أتمتة الـ Webhooks\n\nتتيح لك الـ Webhooks تشغيل الأنظمة الخارجية عند إطلاق أحداث Synapse. يغطي هذا الدليل\nأنماط أتمتة شائعة.\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، أو أي KB خارجي:\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 # شغّل pipeline GitLab\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: إيقاظ الوكيل عند رسالة الإنسان\n\nشغّل تشغيل وكيل LLM عندما يرسل الإنسان رسالة دردشة:\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 # أدرج مهمة معالجة الوكيل في الطابور\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\nيعيد Synapse محاولة الـ webhooks الفاشلة مع تراجع أُسي. يجب أن يفعل معالجك:\n\n1. **يُرجع 200 بسرعة** — لا تفعل عملًا ثقيلًا تزامنيًا\n2. **أدرج العمل في الطابور** — استخدم نظام مهام خلفي\n3. **كن idempotent** — قد يُسلَّم نفس الحدث مرتين\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## تصحيح الـ Webhooks\n\n### تحقق من سجل التسليم\n\nتُسجَّل تسليمات الـ webhook. تحقق من التسليمات الحديثة لـ 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| تسليمات مكررة | اجعل المعالج idempotent |\n| عدم تطابق التوقيع | تحقق أن السر صحيح |\n\n## أفضل الممارسات\n\n> [!TIP]\n> - **تحقق من التوقيعات دائمًا** — لا تتخطى هذا أبدًا\n> - **أرجع 200 بسرعة** — لا تحجب Synapse\n> - **كن idempotent** — تعامل مع التسليمات المكررة\n> - **استخدم أحداثًا محددة** — `memory.store` لا `*`\n> - **راقب إخفاقات التسليم** — أعد تنبيهًا\n\n## الخطوات التالية\n\n- [Webhooks API](/docs/api/webhooks)\n- [Cron والمجدول](/docs/api/cron)\n- [وكيل LLM الدائم](/docs/guides/persistent-llm-agent)\n","content_html":"

أتمتة الـ Webhooks

\n

تتيح لك الـ Webhooks تشغيل الأنظمة الخارجية عند إطلاق أحداث Synapse. يغطي هذا الدليل\nأنماط أتمتة شائعة.

\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، أو أي KB خارجي:

\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            # شغّل pipeline GitLab\n            await gitlab.trigger_pipeline(\n                project="synapse",\n                ref="main",\n                variables={"RELEASE_MEMORY_ID": memory["id"]}\n            )\n    \n    return 200

النمط 4: إيقاظ الوكيل عند رسالة الإنسان

\n

شغّل تشغيل وكيل LLM عندما يرسل الإنسان رسالة دردشة:

\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        # أدرج مهمة معالجة الوكيل في الطابور\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 محاولة الـ webhooks الفاشلة مع تراجع أُسي. يجب أن يفعل معالجك:

\n
    \n
  1. يُرجع 200 بسرعة — لا تفعل عملًا ثقيلًا تزامنيًا
    2. \n
  2. أدرج العمل في الطابور — استخدم نظام مهام خلفي
    3. \n
  3. كن idempotent — قد يُسلَّم نفس الحدث مرتين
    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

تصحيح الـ Webhooks

\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 بسرعة، أدرج العمل غير المتزامن
تسليمات مكررةاجعل المعالج idempotent
عدم تطابق التوقيعتحقق أن السر صحيح
\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"]}