آموزش و راهنما

Webhook چیست؟ آموزش کامل ساخت وب هوک با Python، FastAPI و API درواره

Webhook روشی برای ارسال خودکار رویداد میان نرم‌افزارهاست. در این آموزش، تفاوت Webhook با API و Polling، ساخت Endpoint، مدیریت Retry و Duplicate و پیاده‌سازی عملی آن با FastAPI و درواره را یاد می‌گیرید.

درواره

درواره

— 7 min read
Share
Webhook چیست؟ آموزش کامل ساخت وب هوک با Python، FastAPI و API درواره

فرض کنید وب‌سایت شما باید بلافاصله بعد از ثبت سفارش، پرداخت موفق، ایجاد تیکت پشتیبانی یا پایان یک پردازش طولانی از این اتفاق باخبر شود.

یک روش این است که برنامه شما هر چند ثانیه یک‌بار از سرویس مقصد بپرسد:

آیا وضعیت تغییر کرده است؟
آیا پرداخت انجام شده است؟
آیا پردازش فایل تمام شده است؟
آیا رویداد جدیدی وجود دارد؟

این روش Polling نام دارد. Polling در بعضی پروژه‌ها مفید است، اما می‌تواند تعداد زیادی درخواست بدون نتیجه ایجاد کند.

Webhook راه متفاوتی ارائه می‌دهد. به‌جای آنکه برنامه شما مرتباً وضعیت را بررسی کند، سرویس مبدأ هنگام وقوع رویداد، یک درخواست HTTP به آدرس برنامه شما می‌فرستد.

در این مقاله یاد می‌گیرید:

  • Webhook یا وب هوک چیست
  • Webhook چگونه کار می‌کند
  • چه تفاوتی با API، Polling و WebSocket دارد
  • Payload و Event چه هستند
  • چگونه یک Webhook Endpoint طراحی کنیم
  • Retry، Duplicate و ترتیب رویدادها را چگونه مدیریت کنیم
  • چگونه یک Webhook واقعی با پایتون و FastAPI بسازیم
  • چگونه داده دریافتی را برای پردازش هوشمند به API درواره ارسال کنیم
  • چگونه Webhook را به‌صورت محلی آزمایش کنیم

Webhook چیست؟

Webhook یک HTTP Callback است که هنگام وقوع یک رویداد، داده‌ای را به‌صورت خودکار از یک نرم‌افزار به نرم‌افزار دیگر ارسال می‌کند.

در این معماری معمولاً دو سیستم وجود دارد:

  • Producer یا سرویس ارسال‌کننده رویداد
  • Consumer یا سرویس دریافت‌کننده رویداد

Consumer یک URL در اختیار Producer قرار می‌دهد:

https://example.com/webhooks/orders

وقتی رویداد موردنظر اتفاق می‌افتد، Producer معمولاً با یک درخواست POST اطلاعات رویداد را به این URL می‌فرستد:

POST /webhooks/orders
Content-Type: application/json

نمونه Payload:

{
  "id": "evt_81f723",
  "type": "order.paid",
  "created_at": "2026-08-06T10:30:00Z",
  "data": {
    "order_id": "ord_1208",
    "amount": 850000,
    "currency": "IRR"
  }
}

برنامه دریافت‌کننده Payload را بررسی می‌کند، عملیات لازم را انجام می‌دهد و با یک کد وضعیت موفق مانند 200 یا 202 دریافت رویداد را تأیید می‌کند.

وب هوک چگونه کار می‌کند؟

چرخه معمول یک Webhook شامل مراحل زیر است:

  1. سرویس دریافت‌کننده یک Webhook URL ایجاد می‌کند.
  2. این URL در سرویس ارسال‌کننده ثبت می‌شود.
  3. رویدادی مانند ایجاد سفارش یا پایان پردازش اتفاق می‌افتد.
  4. سرویس ارسال‌کننده یک Payload می‌سازد.
  5. Payload با درخواست HTTP به Webhook URL ارسال می‌شود.
  6. دریافت‌کننده اعتبار و ساختار درخواست را بررسی می‌کند.
  7. دریافت‌کننده یک پاسخ موفق برمی‌گرداند.
  8. پردازش رویداد در همان لحظه یا به‌صورت غیرهم‌زمان انجام می‌شود.
  9. اگر تحویل ناموفق باشد، سرویس ارسال‌کننده ممکن است درخواست را دوباره ارسال کند.

یک نمونه ساده:

کاربر یک تیکت ثبت می‌کند
        ↓
سیستم پشتیبانی رویداد ticket.created می‌سازد
        ↓
رویداد به Webhook برنامه شما ارسال می‌شود
        ↓
برنامه متن تیکت را پردازش می‌کند
        ↓
نتیجه در سیستم پشتیبانی ذخیره می‌شود

Webhook به ارسال‌کننده اجازه می‌دهد تغییرات را بدون انتظار برای درخواست کلاینت اعلام کند.

اصطلاحات مهم Webhook

Webhook Endpoint

آدرسی عمومی است که درخواست‌های Webhook را دریافت می‌کند:

https://api.example.com/webhooks/tickets

این Endpoint باید متد مورد استفاده سرویس ارسال‌کننده، معمولاً POST، را بپذیرد.

Event

اتفاقی است که باعث ارسال Webhook می‌شود.

نمونه Event Typeها:

order.created
order.paid
order.cancelled
ticket.created
ticket.updated
job.completed
invoice.generated
user.registered

نام رویداد بهتر است:

  • واضح باشد
  • نسخه‌پذیر باشد
  • با سایر رویدادها قرارداد یکسانی داشته باشد
  • وضعیت واقعی رخ‌داده را نشان دهد

Payload

داده‌ای است که همراه Webhook ارسال می‌شود. Payload معمولاً JSON است.

{
  "id": "evt_201",
  "type": "ticket.created",
  "data": {
    "ticket_id": "tkt_841",
    "subject": "مشکل در ورود",
    "description": "پس از وارد کردن رمز عبور وارد پنل نمی‌شوم."
  }
}

Event ID

شناسه یکتای رویداد است:

{
  "id": "evt_201"
}

Event ID برای تشخیص درخواست‌های تکراری اهمیت زیادی دارد.

Delivery Attempt

هر بار تلاش برای تحویل یک رویداد به Webhook Endpoint یک Delivery Attempt محسوب می‌شود.

یک Event می‌تواند چند Delivery Attempt داشته باشد. بنابراین Event ID و Delivery ID الزاماً یک مفهوم نیستند.

Webhook Secret

مقداری محرمانه و مشترک میان ارسال‌کننده و دریافت‌کننده است که معمولاً برای ساخت و بررسی Signature استفاده می‌شود.

تفاوت Webhook با API چیست؟

Webhook و API رقیب یکدیگر نیستند. آن‌ها معمولاً در کنار یکدیگر استفاده می‌شوند.

در یک API معمولی، برنامه شما ارتباط را آغاز می‌کند:

برنامه شما → درخواست API → سرویس مقصد

در Webhook، سرویس مقصد هنگام وقوع رویداد ارتباط را آغاز می‌کند:

سرویس مقصد → درخواست Webhook → برنامه شما
ویژگیAPI معمولیWebhook
آغازکننده ارتباطکلاینتسرویس تولیدکننده رویداد
زمان اجراهنگام درخواست کلاینتهنگام وقوع رویداد
هدف اصلیدریافت یا تغییر دادهاطلاع‌رسانی خودکار رویداد
الگوی ارتباطRequest و ResponseEvent و Callback
نیاز به URL عمومی در برنامه شمامعمولاً خیربله
مناسب برایعملیات درخواستیاعلان تغییرات

برای مثال، برنامه شما با API یک Job ایجاد می‌کند:

POST /v1/jobs

سپس سرویس می‌تواند پس از پایان Job نتیجه را به Webhook برنامه شما ارسال کند:

POST /webhooks/jobs

در این مثال، API برای شروع عملیات و Webhook برای اعلام پایان عملیات استفاده شده است.

تفاوت Webhook و Polling

Polling یعنی برنامه شما در فاصله‌های زمانی مشخص وضعیت را از API دریافت کند.

نمونه:

GET /v1/jobs/job_981

برنامه این درخواست را هر ۱۰ ثانیه تکرار می‌کند تا وضعیت completed شود.

در Webhook، سرویس پس از پایان Job به برنامه شما اطلاع می‌دهد:

POST /webhooks/jobs

{
  "id": "evt_725",
  "type": "job.completed",
  "data": {
    "job_id": "job_981",
    "result_url": "https://example.com/results/result_225"
  }
}
معیارWebhookPolling
زمان دریافت تغییرمعمولاً سریعوابسته به فاصله Polling
تعداد درخواست اضافیکمممکن است زیاد باشد
نیاز به Endpoint عمومیبلهخیر
پیچیدگی دریافت Retryبیشترکمتر
مناسب شبکه محدودهمیشه نهگاهی مناسب‌تر
بازیابی وضعیت ازدست‌رفتهنیازمند طراحیمعمولاً ساده‌تر
مناسب رویداد بلادرنگبلهبا تأخیر

چه زمانی Polling مناسب‌تر است؟

Polling در شرایط زیر ممکن است انتخاب بهتری باشد:

  • امکان ایجاد URL عمومی ندارید
  • سرویس مقصد Webhook ارائه نمی‌کند
  • وضعیت فقط هر چند دقیقه یا ساعت اهمیت دارد
  • تعداد منابع و درخواست‌ها بسیار کم است
  • می‌خواهید بازیابی وضعیت ساده‌تر باشد
  • شبکه ورودی برنامه شما محدود است

چه زمانی Webhook مناسب‌تر است؟

Webhook برای این سناریوها مناسب است:

  • باید سریع از تغییر وضعیت باخبر شوید
  • رویدادها نامنظم هستند
  • می‌خواهید درخواست‌های تکراری را کاهش دهید
  • پردازش سرویس مقصد طولانی است
  • چند سیستم باید با وقوع یک رویداد هماهنگ شوند

در بسیاری از پروژه‌های جدی، بهترین راهکار ترکیب Webhook و API است:

  • Webhook برای اعلان سریع رویداد
  • API برای دریافت وضعیت نهایی و بازیابی اطلاعات

تفاوت Webhook و WebSocket

با وجود شباهت نام، Webhook و WebSocket کاربرد یکسانی ندارند.

Webhook معمولاً یک درخواست HTTP مستقل است که با وقوع رویداد ارسال می‌شود. WebSocket یک اتصال پایدار و دوطرفه میان کلاینت و سرور ایجاد می‌کند.

ویژگیWebhookWebSocket
نوع ارتباطمعمولاً یک‌طرفهدوطرفه
اتصال دائمیخیربله
آغاز پیامسرویس تولیدکننده رویدادهر دو طرف
مناسب Backend-to-Backendبسیار مناسبقابل استفاده
مناسب رابط بلادرنگمحدودبسیار مناسب
نمونه کاربردپایان Job، ثبت سفارشچت زنده، بازی آنلاین، داشبورد زنده

برای اعلان پایان یک پردازش، Webhook معمولاً ساده‌تر است. برای چت زنده‌ای که پیام‌ها باید در هر دو جهت روی اتصال فعال جابه‌جا شوند، WebSocket انتخاب مناسب‌تری است.

یک Payload استاندارد چه ساختاری دارد؟

یک قرارداد مناسب برای Event می‌تواند شامل فیلدهای زیر باشد:

{
  "id": "evt_8c71a0",
  "type": "ticket.created",
  "version": "1",
  "created_at": "2026-08-06T10:30:00Z",
  "data": {
    "ticket_id": "tkt_912",
    "subject": "درخواست راهنمایی",
    "description": "برای انتخاب مدل مناسب به راهنمایی نیاز دارم."
  }
}

فیلد id

شناسه یکتای رویداد است و برای جلوگیری از پردازش تکراری استفاده می‌شود.

فیلد type

نوع رویداد را مشخص می‌کند:

ticket.created

فیلد version

نسخه Schema رویداد را مشخص می‌کند. این فیلد با نسخه API الزاماً یکسان نیست.

فیلد created_at

زمان ایجاد رویداد را با یک قالب استاندارد مانند ISO 8601 مشخص می‌کند:

2026-08-06T10:30:00Z

فیلد data

داده مرتبط با رویداد را نگه می‌دارد.

بهتر است Event Metadata از داده اصلی جدا باشد. این ساختار پردازش و نسخه‌بندی Eventها را ساده‌تر می‌کند.

Thin Payload یا Fat Payload؟

دو رویکرد اصلی برای طراحی Payload وجود دارد.

Thin Payload

Webhook فقط شناسه منبع را ارسال می‌کند:

{
  "id": "evt_101",
  "type": "order.updated",
  "data": {
    "order_id": "ord_981"
  }
}

دریافت‌کننده برای اطلاعات کامل، API را فراخوانی می‌کند:

GET /v1/orders/ord_981

مزایا:

  • Payload کوچک‌تر است
  • داده تازه از API دریافت می‌شود
  • تغییر Schema Webhook ساده‌تر می‌شود
  • اطلاعات کمتری در شبکه ارسال می‌شود

معایب:

  • به یک درخواست API اضافی نیاز دارد
  • اگر منبع حذف شده باشد، دریافت اطلاعات ممکن نیست
  • وابستگی به دسترس‌پذیری API بیشتر می‌شود

Fat Payload

Webhook اطلاعات کامل را ارسال می‌کند:

{
  "id": "evt_101",
  "type": "order.updated",
  "data": {
    "order_id": "ord_981",
    "status": "paid",
    "amount": 850000,
    "currency": "IRR",
    "updated_at": "2026-08-06T10:30:00Z"
  }
}

مزایا:

  • درخواست API اضافی لازم نیست
  • پردازش سریع‌تر آغاز می‌شود
  • Snapshot رویداد حفظ می‌شود

معایب:

  • Payload بزرگ‌تر است
  • نسخه‌بندی پیچیده‌تر می‌شود
  • ممکن است داده هنگام پردازش دیگر تازه نباشد

انتخاب مناسب به کاربرد، حجم داده، حساسیت اطلاعات و نیاز به Snapshot وابسته است.

چرا Webhook ممکن است چند بار ارسال شود؟

تحویل Webhook معمولاً بر مبنای At-least-once Delivery طراحی می‌شود. یعنی سرویس تلاش می‌کند رویداد حداقل یک بار تحویل داده شود، اما احتمال تحویل چندباره وجود دارد.

برای مثال:

  1. Producer رویداد را ارسال می‌کند.
  2. Consumer آن را پردازش می‌کند.
  3. پاسخ Consumer به دلیل اختلال شبکه به Producer نمی‌رسد.
  4. Producer تصور می‌کند تحویل ناموفق بوده است.
  5. همان Event دوباره ارسال می‌شود.

اگر Event مربوط به ثبت اعتبار یا ایجاد سفارش باشد، پردازش دوباره می‌تواند مشکل ایجاد کند.

بنابراین Webhook Handler باید تا حد امکان Idempotent باشد.

جلوگیری از پردازش Duplicate

روش رایج این است که Event ID را در پایگاه داده ذخیره کنید.

شبه‌کد:

اگر event_id قبلاً ذخیره شده است:
    پاسخ موفق برگردان
    دوباره پردازش نکن

event_id را ذخیره کن
رویداد را پردازش کن

نمونه جدول:

CREATE TABLE webhook_events (
    event_id TEXT PRIMARY KEY,
    event_type TEXT NOT NULL,
    status TEXT NOT NULL,
    received_at TEXT NOT NULL,
    processed_at TEXT
);

قرار دادن event_id به‌عنوان Primary Key باعث می‌شود ثبت دوباره همان رویداد با محدودیت پایگاه داده متوقف شود.

استفاده از یک set در حافظه فقط برای آزمایش مناسب است؛ زیرا با Restart شدن برنامه یا اجرای چند سرور اطلاعات آن از بین می‌رود.

آیا ترتیب Webhookها تضمین می‌شود؟

نباید بدون مطالعه مستندات سرویس ارسال‌کننده، ترتیب تحویل رویدادها را تضمین‌شده فرض کنید.

ممکن است رویدادهای زیر ایجاد شوند:

order.created
order.paid
order.cancelled

اما به علت Retry، صف پردازش یا تأخیر شبکه، Consumer آن‌ها را با ترتیب دیگری دریافت کند.

راهکارهای قابل استفاده:

  • ثبت زمان ایجاد Event
  • ثبت نسخه یا Revision منبع
  • مقایسه updated_at
  • دریافت وضعیت نهایی منبع از API
  • جلوگیری از بازگرداندن منبع به وضعیت قدیمی
  • طراحی State Machine برای انتقال وضعیت‌ها

برای مثال، اگر سفارش در وضعیت cancelled است، رسیدن دیرهنگام رویداد order.paid نباید بدون بررسی، وضعیت را به عقب برگرداند.

پاسخ سریع به Webhook

Webhook Endpoint نباید یک پردازش طولانی را قبل از ارسال پاسخ انجام دهد.

طراحی نامناسب:

دریافت Webhook
    ↓
فراخوانی چند API
    ↓
پردازش فایل
    ↓
ارسال ایمیل
    ↓
ثبت گزارش
    ↓
بازگرداندن پاسخ

اگر پردازش طول بکشد، Producer ممکن است Timeout دریافت کرده و همان Event را دوباره ارسال کند.

الگوی مناسب‌تر:

دریافت Webhook
    ↓
بررسی اولیه
    ↓
ثبت Event در پایگاه داده یا Queue
    ↓
پاسخ سریع 2xx
    ↓
پردازش توسط Worker

کدهای متداول:

  • 200 OK: رویداد دریافت و پردازش شده است
  • 202 Accepted: رویداد پذیرفته شده و بعداً پردازش می‌شود
  • 204 No Content: دریافت موفق بدون بدنه پاسخ

رفتار دقیق باید با مستندات Producer هماهنگ باشد.

Retry چگونه کار می‌کند؟

اگر Endpoint در دسترس نباشد یا پاسخ ناموفق برگرداند، بسیاری از سرویس‌ها Webhook را دوباره ارسال می‌کنند.

الگوی رایج Retry:

تلاش اول: بلافاصله
تلاش دوم: با کمی تأخیر
تلاش سوم: با تأخیر بیشتر
تلاش‌های بعدی: Exponential Backoff

نمونه فاصله‌ها:

۱ ثانیه
۲ ثانیه
۴ ثانیه
۸ ثانیه
۱۶ ثانیه

این مقادیر فقط یک مثال‌اند. تعداد تلاش‌ها، زمان نگهداری Event و کدهایی که باعث Retry می‌شوند در هر سرویس متفاوت است.

Producer باید:

  • برای هر Delivery لاگ داشته باشد
  • Timeout مشخص تعیین کند
  • Retry را محدود کند
  • از Exponential Backoff و Jitter استفاده کند
  • امکان مشاهده یا ارسال مجدد رویداد را فراهم کند

Consumer باید:

  • Duplicate را تحمل کند
  • پاسخ را سریع برگرداند
  • شکست پردازش داخلی را ثبت کند
  • وضعیت Event را قابل پیگیری نگه دارد

بررسی اصالت Webhook

عمومی بودن Endpoint به این معنی است که درخواست‌های مختلف می‌توانند به آن برسند. دریافت‌کننده باید بتواند تشخیص دهد درخواست واقعاً از سرویس مورد انتظار آمده است.

یکی از روش‌های رایج، HMAC Signature است.

Producer با استفاده از Secret و بدنه خام درخواست یک Signature می‌سازد:

signature = HMAC(secret, timestamp + "." + raw_body)

سپس اطلاعات را در Header قرار می‌دهد:

X-Webhook-Id: evt_81f723
X-Webhook-Timestamp: 1786000000
X-Webhook-Signature: v1=SIGNATURE_VALUE

Consumer همان محاسبه را انجام می‌دهد و Signatureها را مقایسه می‌کند.

نکات مهم:

  • Signature را روی بدنه خام Request بررسی کنید
  • پیش از Verify کردن، JSON را تغییر یا دوباره Serialize نکنید
  • مقایسه Signature باید Constant-time باشد
  • Timestamp را بررسی کنید
  • Secret را در کد منبع قرار ندهید
  • روش دقیق Verify باید مطابق مستندات همان Producer باشد

نام Headerها و فرمول امضا در تمام سرویس‌ها یکسان نیست. نمونه این مقاله یک قرارداد آموزشی است.

ساخت Webhook با Python و FastAPI

در این پروژه یک Webhook Endpoint برای دریافت رویدادهای تیکت پشتیبانی می‌سازیم. پس از دریافت ticket.created، برنامه متن تیکت را با API هوش مصنوعی درواره خلاصه و دسته‌بندی می‌کند.

ساختار پروژه:

webhook-ai-processor/
├── app.py
├── test_sender.py
├── requirements.txt
└── .env

نصب وابستگی‌ها

فایل requirements.txt:

fastapi
uvicorn[standard]
httpx
python-dotenv

محیط مجازی بسازید:

python -m venv .venv

فعال‌سازی در لینوکس و macOS:

source .venv/bin/activate

فعال‌سازی در Windows PowerShell:

.venv\Scripts\Activate.ps1

نصب بسته‌ها:

pip install -r requirements.txt

تنظیم متغیرهای محیطی

فایل .env:

WEBHOOK_SECRET=replace-with-a-long-random-secret
DARVAREH_API_KEY=YOUR_DARVAREH_API_KEY
DARVAREH_MODEL_ID=YOUR_MODEL_ID

فایل .gitignore:

.env
.venv/
__pycache__/
webhooks.db

پیاده‌سازی Webhook Receiver

فایل app.py:

import hashlib
import hmac
import json
import os
import sqlite3
import time
from datetime import datetime, timezone

import httpx
from dotenv import load_dotenv
from fastapi import BackgroundTasks, FastAPI, HTTPException, Request
from pydantic import BaseModel, Field, ValidationError

load_dotenv()

WEBHOOK_SECRET = os.getenv("WEBHOOK_SECRET")
DARVAREH_API_KEY = os.getenv("DARVAREH_API_KEY")
DARVAREH_MODEL_ID = os.getenv("DARVAREH_MODEL_ID")

DARVAREH_URL = "https://api.darvareh.ir/v1/chat/completions"
DATABASE_PATH = "webhooks.db"
MAX_TIMESTAMP_DIFFERENCE = 300

if not WEBHOOK_SECRET:
    raise RuntimeError("متغیر WEBHOOK_SECRET تنظیم نشده است.")

if not DARVAREH_API_KEY:
    raise RuntimeError("متغیر DARVAREH_API_KEY تنظیم نشده است.")

if not DARVAREH_MODEL_ID:
    raise RuntimeError("متغیر DARVAREH_MODEL_ID تنظیم نشده است.")


app = FastAPI(
    title="Webhook AI Processor",
    version="1.0.0",
)


class TicketData(BaseModel):
    ticket_id: str = Field(min_length=1, max_length=100)
    subject: str = Field(min_length=1, max_length=300)
    description: str = Field(min_length=10, max_length=10_000)


class TicketEvent(BaseModel):
    id: str = Field(min_length=1, max_length=200)
    type: str
    version: str
    created_at: str
    data: TicketData


def get_connection() -> sqlite3.Connection:
    connection = sqlite3.connect(DATABASE_PATH)
    connection.row_factory = sqlite3.Row
    return connection


def initialize_database() -> None:
    with get_connection() as connection:
        connection.execute(
            """
            CREATE TABLE IF NOT EXISTS webhook_events (
                event_id TEXT PRIMARY KEY,
                event_type TEXT NOT NULL,
                status TEXT NOT NULL,
                payload TEXT NOT NULL,
                result TEXT,
                received_at TEXT NOT NULL,
                processed_at TEXT
            )
            """
        )


def verify_signature(
    raw_body: bytes,
    event_id: str,
    timestamp: str,
    signature_header: str,
) -> bool:
    try:
        timestamp_value = int(timestamp)
    except ValueError:
        return False

    current_time = int(time.time())

    if abs(current_time - timestamp_value) > MAX_TIMESTAMP_DIFFERENCE:
        return False

    if not signature_header.startswith("v1="):
        return False

    received_signature = signature_header.removeprefix("v1=")

    signed_content = (
        event_id.encode("utf-8")
        + b"."
        + timestamp.encode("utf-8")
        + b"."
        + raw_body
    )

    expected_signature = hmac.new(
        WEBHOOK_SECRET.encode("utf-8"),
        signed_content,
        hashlib.sha256,
    ).hexdigest()

    return hmac.compare_digest(
        received_signature,
        expected_signature,
    )


def event_exists(event_id: str) -> bool:
    with get_connection() as connection:
        row = connection.execute(
            "SELECT event_id FROM webhook_events WHERE event_id = ?",
            (event_id,),
        ).fetchone()

    return row is not None


def save_event(event: TicketEvent, raw_body: bytes) -> bool:
    try:
        with get_connection() as connection:
            connection.execute(
                """
                INSERT INTO webhook_events (
                    event_id,
                    event_type,
                    status,
                    payload,
                    received_at
                )
                VALUES (?, ?, ?, ?, ?)
                """,
                (
                    event.id,
                    event.type,
                    "received",
                    raw_body.decode("utf-8"),
                    datetime.now(timezone.utc).isoformat(),
                ),
            )
        return True
    except sqlite3.IntegrityError:
        return False


def update_event(
    event_id: str,
    status: str,
    result: dict | None = None,
) -> None:
    with get_connection() as connection:
        connection.execute(
            """
            UPDATE webhook_events
            SET status = ?,
                result = ?,
                processed_at = ?
            WHERE event_id = ?
            """,
            (
                status,
                json.dumps(result, ensure_ascii=False) if result else None,
                datetime.now(timezone.utc).isoformat(),
                event_id,
            ),
        )


async def analyze_ticket(event: TicketEvent) -> None:
    prompt = f"""
تیکت زیر را بررسی کن.

عنوان:
{event.data.subject}

متن:
{event.data.description}

فقط یک JSON معتبر با ساختار زیر برگردان:
{{
  "summary": "خلاصه کوتاه فارسی",
  "category": "technical یا billing یا sales یا general",
  "priority": "low یا medium یا high"
}}
""".strip()

    request_body = {
        "model": DARVAREH_MODEL_ID,
        "messages": [
            {
                "role": "system",
                "content": (
                    "تو دستیار دسته‌بندی تیکت‌های فارسی هستی. "
                    "اطلاعات جدیدی به متن اضافه نکن."
                ),
            },
            {
                "role": "user",
                "content": prompt,
            },
        ],
    }

    headers = {
        "Authorization": f"Bearer {DARVAREH_API_KEY}",
        "Content-Type": "application/json",
    }

    timeout = httpx.Timeout(
        connect=10.0,
        read=60.0,
        write=20.0,
        pool=10.0,
    )

    try:
        async with httpx.AsyncClient(timeout=timeout) as client:
            response = await client.post(
                DARVAREH_URL,
                headers=headers,
                json=request_body,
            )

        response.raise_for_status()
        body = response.json()
        content = body["choices"][0]["message"]["content"]

        update_event(
            event.id,
            "completed",
            {
                "ticket_id": event.data.ticket_id,
                "analysis": content,
                "model": DARVAREH_MODEL_ID,
            },
        )
    except (
        httpx.HTTPError,
        ValueError,
        KeyError,
        IndexError,
        TypeError,
    ) as exc:
        update_event(
            event.id,
            "failed",
            {
                "error": type(exc).__name__,
            },
        )


@app.on_event("startup")
def startup_event() -> None:
    initialize_database()


@app.get("/health")
async def health_check():
    return {"status": "ok"}


@app.post("/webhooks/tickets", status_code=202)
async def receive_ticket_webhook(
    request: Request,
    background_tasks: BackgroundTasks,
):
    raw_body = await request.body()

    event_id = request.headers.get("X-Webhook-Id")
    timestamp = request.headers.get("X-Webhook-Timestamp")
    signature = request.headers.get("X-Webhook-Signature")

    if not event_id or not timestamp or not signature:
        raise HTTPException(
            status_code=400,
            detail={
                "code": "MISSING_WEBHOOK_HEADERS",
                "message": "هدرهای ضروری Webhook ارسال نشده‌اند.",
            },
        )

    if not verify_signature(
        raw_body,
        event_id,
        timestamp,
        signature,
    ):
        raise HTTPException(
            status_code=401,
            detail={
                "code": "INVALID_WEBHOOK_SIGNATURE",
                "message": "امضای Webhook معتبر نیست.",
            },
        )

    if event_exists(event_id):
        return {
            "accepted": True,
            "duplicate": True,
            "event_id": event_id,
        }

    try:
        payload = json.loads(raw_body)
        event = TicketEvent.model_validate(payload)
    except (json.JSONDecodeError, ValidationError) as exc:
        raise HTTPException(
            status_code=422,
            detail={
                "code": "INVALID_WEBHOOK_PAYLOAD",
                "message": "ساختار Payload معتبر نیست.",
            },
        ) from exc

    if event.id != event_id:
        raise HTTPException(
            status_code=400,
            detail={
                "code": "EVENT_ID_MISMATCH",
                "message": "شناسه Header با شناسه Payload یکسان نیست.",
            },
        )

    if event.type != "ticket.created":
        return {
            "accepted": True,
            "ignored": True,
            "event_id": event.id,
        }

    was_inserted = save_event(event, raw_body)

    if not was_inserted:
        return {
            "accepted": True,
            "duplicate": True,
            "event_id": event.id,
        }

    background_tasks.add_task(analyze_ticket, event)

    return {
        "accepted": True,
        "duplicate": False,
        "event_id": event.id,
    }

اجرای برنامه

uvicorn app:app --reload

برنامه روی آدرس زیر اجرا می‌شود:

http://127.0.0.1:8000

Health Check:

curl http://127.0.0.1:8000/health

پاسخ:

{
  "status": "ok"
}

ساخت Sender آزمایشی

چون Endpoint امضای درخواست را بررسی می‌کند، یک برنامه آزمایشی می‌سازیم که Payload را امضا و ارسال کند.

فایل test_sender.py:

import hashlib
import hmac
import json
import os
import time
import uuid

import httpx
from dotenv import load_dotenv

load_dotenv()

WEBHOOK_SECRET = os.getenv("WEBHOOK_SECRET")
WEBHOOK_URL = "http://127.0.0.1:8000/webhooks/tickets"

if not WEBHOOK_SECRET:
    raise RuntimeError("متغیر WEBHOOK_SECRET تنظیم نشده است.")

event_id = f"evt_{uuid.uuid4().hex}"
timestamp = str(int(time.time()))

payload = {
    "id": event_id,
    "type": "ticket.created",
    "version": "1",
    "created_at": "2026-08-06T10:30:00Z",
    "data": {
        "ticket_id": "tkt_912",
        "subject": "انتخاب مدل مناسب",
        "description": (
            "برای خلاصه‌سازی متن‌های فارسی طولانی به یک مدل "
            "سریع و اقتصادی نیاز دارم."
        ),
    },
}

raw_body = json.dumps(
    payload,
    ensure_ascii=False,
    separators=(",", ":"),
).encode("utf-8")

signed_content = (
    event_id.encode("utf-8")
    + b"."
    + timestamp.encode("utf-8")
    + b"."
    + raw_body
)

signature = hmac.new(
    WEBHOOK_SECRET.encode("utf-8"),
    signed_content,
    hashlib.sha256,
).hexdigest()

headers = {
    "Content-Type": "application/json",
    "X-Webhook-Id": event_id,
    "X-Webhook-Timestamp": timestamp,
    "X-Webhook-Signature": f"v1={signature}",
}

response = httpx.post(
    WEBHOOK_URL,
    headers=headers,
    content=raw_body,
    timeout=20.0,
)

print("Status:", response.status_code)
print("Response:", response.json())

Sender را اجرا کنید:

python test_sender.py

پاسخ مورد انتظار:

Status: 202
Response: {
  "accepted": true,
  "duplicate": false,
  "event_id": "evt_..."
}

بعد از دریافت رویداد، FastAPI پردازش پس‌زمینه را اجرا و متن تیکت را برای تحلیل به درواره ارسال می‌کند.

آدرس پایه API درواره:

https://api.darvareh.ir/v1

Endpoint استفاده‌شده در این پروژه:

https://api.darvareh.ir/v1/chat/completions

برای انتخاب Model ID و مشاهده قیمت به‌روز مدل‌ها، صفحه مدل‌ها و قیمت‌های درواره را بررسی کنید.

آزمایش Duplicate Delivery

برای آزمایش Duplicate کافی است Sender را طوری تغییر دهید که event_id ثابتی داشته باشد:

event_id = "evt_duplicate_test_001"

بار اول پاسخ باید مشابه زیر باشد:

{
  "accepted": true,
  "duplicate": false,
  "event_id": "evt_duplicate_test_001"
}

بار دوم:

{
  "accepted": true,
  "duplicate": true,
  "event_id": "evt_duplicate_test_001"
}

در بار دوم، رویداد دوباره برای پردازش هوش مصنوعی ارسال نمی‌شود.

محدودیت BackgroundTasks در FastAPI

BackgroundTasks برای مثال‌های ساده و پردازش‌های سبک مناسب است، اما جایگزین Queue پایدار در محیط Production نیست.

اگر برنامه بعد از ارسال پاسخ متوقف شود، Task درون حافظه ممکن است از بین برود. همچنین چند Instance برنامه مدیریت متمرکزی روی Taskها ندارند.

برای محیط Production بهتر است از Queue و Worker استفاده شود:

Webhook Endpoint
       ↓
Database یا Message Queue
       ↓
Worker
       ↓
API درواره
       ↓
ثبت نتیجه

ابزارهای متداول:

  • Celery
  • Redis
  • RabbitMQ
  • Kafka
  • Dramatiq
  • RQ
  • سرویس‌های Queue مدیریت‌شده

اصل مهم این است که Event پیش از ارسال پاسخ موفق، در یک محل پایدار ثبت شود.

تست Webhook روی Localhost

سرویس بیرونی نمی‌تواند مستقیماً به این آدرس دسترسی پیدا کند:

http://127.0.0.1:8000/webhooks/tickets

این آدرس فقط روی کامپیوتر خودتان قابل دسترسی است.

برای آزمایش می‌توانید از یک Tunnel موقت استفاده کنید که URL عمومی را به Localhost متصل می‌کند. ابزارهایی مانند ngrok و Cloudflare Tunnel برای این کار شناخته شده‌اند.

نمونه URL عمومی:

https://example-tunnel.example/webhooks/tickets

این URL به برنامه محلی زیر هدایت می‌شود:

http://127.0.0.1:8000/webhooks/tickets

برای مشاهده خام درخواست‌ها نیز ابزارهایی مانند Webhook.site در مرحله توسعه مفیدند. با این حال، داده واقعی، محرمانه یا اطلاعات کاربران را به Endpointهای آزمایشی عمومی ارسال نکنید.

طراحی Webhook در محیط Production

یک مسیر جدا برای هر Integration بسازید

نمونه:

/webhooks/support
/webhooks/orders
/webhooks/jobs

یا در صورت اتصال به چند Provider:

/webhooks/provider-a
/webhooks/provider-b

این طراحی باعث می‌شود:

  • قراردادهای مختلف با هم ترکیب نشوند
  • Secretها جدا باشند
  • مانیتورینگ ساده‌تر شود
  • تغییر یک Integration روی دیگری اثر نگذارد

Raw Body را نگه دارید

Signature اغلب روی بایت‌های خام درخواست محاسبه می‌شود. اگر JSON را Parse و دوباره Serialize کنید، فاصله‌ها یا ترتیب فیلدها ممکن است تغییر کند و Signature نامعتبر شود.

ابتدا:

Raw Body را بخوان
Signature را Verify کن
سپس JSON را Parse کن

Event را پیش از پردازش ثبت کنید

ثبت اولیه Event باعث می‌شود در صورت توقف Worker بتوانید پردازش را ادامه دهید.

وضعیت‌های مناسب:

received
processing
completed
failed
ignored

پاسخ Producer را از نتیجه پردازش جدا کنید

پاسخ 202 Accepted یعنی رویداد دریافت و برای پردازش پذیرفته شده است. این پاسخ الزاماً به معنی موفقیت نهایی تحلیل، ارسال ایمیل یا ذخیره خروجی نیست.

خطای داخلی را عمومی نکنید

پاسخ نامناسب:

{
  "error": "sqlite3.OperationalError at /srv/app.py line 184"
}

پاسخ مناسب‌تر:

{
  "error": {
    "code": "WEBHOOK_PROCESSING_ERROR",
    "message": "پردازش درخواست انجام نشد.",
    "request_id": "req_8210"
  }
}

جزئیات کامل خطا باید در لاگ داخلی ثبت شود.

Eventهای ناشناخته را مدیریت کنید

ممکن است Producer در آینده Event Type جدیدی اضافه کند. برنامه نباید برای هر Event ناشناخته از کار بیفتد.

می‌توانید Event را ثبت و نادیده بگیرید:

{
  "accepted": true,
  "ignored": true,
  "event_id": "evt_901"
}

رفتار دقیق به قرارداد شما با Producer بستگی دارد.

نسخه‌بندی Webhook

Schema رویداد ممکن است در آینده تغییر کند. برای مدیریت این تغییرات می‌توان نسخه را در Payload قرار داد:

{
  "type": "ticket.created",
  "version": "2"
}

یا نسخه را در URL قرار داد:

/webhooks/v1/tickets
/webhooks/v2/tickets

تغییرات زیر ممکن است Breaking Change باشند:

  • حذف یک فیلد
  • تغییر نام فیلد
  • تغییر نوع داده
  • تغییر معنای Event
  • تغییر ساختار آبجکت‌های تو‌در‌تو
  • اجباری کردن یک فیلد جدید

اضافه کردن فیلد اختیاری معمولاً سازگارتر است، به شرط آنکه Consumer فیلدهای ناشناخته را تحمل کند.

مانیتورینگ Webhook

برای هر Delivery اطلاعات زیر را ثبت کنید:

  • Event ID
  • Event Type
  • Delivery ID
  • زمان دریافت
  • مدت پاسخ
  • کد وضعیت
  • تعداد Retry
  • نتیجه Signature Verification
  • وضعیت پردازش
  • مدت پردازش
  • شناسه درخواست داخلی
  • خطای نهایی بدون اطلاعات محرمانه

شاخص‌های مهم:

Delivery Success Rate
Processing Success Rate
Duplicate Rate
Average Response Time
Processing Latency
Retry Count
Oldest Unprocessed Event
Queue Depth

فقط ثبت لاگ کافی نیست. برای افزایش خطاها، رشد Queue یا توقف Worker هشدار تعریف کنید.

کاربرد Webhook در پروژه‌های هوش مصنوعی

پردازش فایل طولانی

کاربر فایلی آپلود می‌کند و Job ساخته می‌شود. پس از پایان پردازش، Webhook نتیجه را به Backend اعلام می‌کند.

تولید تصویر یا ویدئو

برخی عملیات تولید رسانه طولانی‌اند. اگر سرویس مورد استفاده از Callback پشتیبانی کند، نتیجه می‌تواند پس از پایان به Webhook ارسال شود.

تحلیل تیکت پشتیبانی

رویداد ticket.created باعث اجرای خلاصه‌سازی، تشخیص موضوع یا اولویت‌بندی تیکت می‌شود.

تحلیل بازخورد مشتری

هر بازخورد جدید برای دسته‌بندی موضوع، لحن یا استخراج نکات کلیدی به مدل هوش مصنوعی ارسال می‌شود.

پردازش اسناد

بعد از ورود سند جدید، Workflow استخراج متن، خلاصه‌سازی و تولید Metadata شروع می‌شود.

خودکارسازی محتوا

انتشار یک محتوا می‌تواند Workflow تولید خلاصه، کپشن شبکه اجتماعی یا دسته‌بندی موضوعی را فعال کند.

برای این کاربردها می‌توانید Backend خود را به API درواره متصل کنید. پشتیبانی از Webhook، Callback یا پردازش Async باید برای Endpoint و سرویس مورد استفاده به‌طور جداگانه از مستندات آن بررسی شود.

اشتباهات رایج در استفاده از Webhook

پردازش طولانی قبل از پاسخ

این کار احتمال Timeout و ارسال Duplicate را افزایش می‌دهد.

فرض کردن تحویل دقیقاً یک‌بار

Webhook ممکن است چند بار تحویل داده شود. Consumer باید Duplicate را مدیریت کند.

فرض کردن ترتیب رویدادها

Eventها ممکن است خارج از ترتیب ایجاد دریافت شوند.

نداشتن Event ID

بدون شناسه یکتا، تشخیص Duplicate دشوار می‌شود.

استفاده از حافظه برای Deduplication

حافظه با Restart شدن یا افزایش تعداد سرورها قابل اتکا نیست.

Parse کردن JSON پیش از Signature Verification

تغییر نمایش بدنه می‌تواند بررسی Signature را خراب کند.

قرار دادن Secret در کد

Secret باید در Environment Variable یا Secret Manager نگهداری شود.

Retry تمام خطاها

خطای ساختار Payload با Retry برطرف نمی‌شود. سیاست Retry باید بر اساس نوع پاسخ تعریف شود.

نداشتن امکان Replay

در پروژه‌های مهم بهتر است اپراتور بتواند Event ناموفق را پس از رفع مشکل دوباره پردازش کند.

استفاده از Payload به‌عنوان تنها منبع حقیقت

گاهی Payload فقط یک اعلان است. برای عملیات حساس یا وضعیت‌های متغیر ممکن است لازم باشد وضعیت نهایی از API مبدأ دریافت شود.

چک‌لیست پیاده‌سازی Webhook

پیش از انتشار Webhook Endpoint این موارد را بررسی کنید:

  • URL عمومی و HTTPS آماده است
  • متد و Content-Type صحیح تعریف شده‌اند
  • Event ID یکتا وجود دارد
  • Event Type مشخص است
  • نسخه Schema تعیین شده است
  • Timestamp در Payload یا Header وجود دارد
  • Signature روی Raw Body بررسی می‌شود
  • Secret در کد منبع قرار ندارد
  • Timestamp قدیمی رد می‌شود
  • Duplicateها شناسایی می‌شوند
  • Event پیش از پاسخ در محل پایدار ثبت می‌شود
  • پاسخ موفق سریع برگردانده می‌شود
  • پردازش طولانی در Worker انجام می‌شود
  • Eventهای ناشناخته مدیریت می‌شوند
  • ترتیب رویدادها تضمین‌شده فرض نمی‌شود
  • Retry و Backoff تعریف شده‌اند
  • وضعیت‌های failed قابل مشاهده‌اند
  • امکان Replay کنترل‌شده وجود دارد
  • Timeout مشخص است
  • لاگ‌ها اطلاعات محرمانه ندارند
  • مانیتورینگ و هشدار فعال است
  • نمونه Payload در مستندات وجود دارد
  • تغییرات Schema نسخه‌بندی می‌شوند
  • تست Duplicate، Timeout و Payload نامعتبر انجام شده است

پرسش‌های متداول

Webhook به زبان ساده چیست؟

Webhook آدرسی است که یک نرم‌افزار هنگام وقوع رویداد، اطلاعات آن رویداد را به‌صورت خودکار به آن ارسال می‌کند.

آیا Webhook همان API است؟

خیر. API و Webhook معمولاً مکمل یکدیگرند. در API، برنامه شما درخواست را آغاز می‌کند؛ در Webhook، سرویس دیگر هنگام وقوع رویداد به برنامه شما درخواست می‌فرستد.

آیا Webhook همیشه از POST استفاده می‌کند؟

بیشتر Webhookها از POST استفاده می‌کنند، اما رفتار دقیق به قرارداد سرویس ارسال‌کننده وابسته است.

آیا Webhook حتماً JSON است؟

خیر. JSON رایج‌ترین قالب است، اما بعضی سرویس‌ها از Form Data، XML یا فرمت‌های دیگر استفاده می‌کنند.

آیا Webhook به URL عمومی نیاز دارد؟

برای دریافت Webhook از یک سرویس اینترنتی، Endpoint باید از شبکه آن سرویس قابل دسترسی باشد. هنگام توسعه می‌توان از Tunnel موقت استفاده کرد.

چرا Webhook من چند بار دریافت می‌شود؟

ممکن است پاسخ موفق به Producer نرسیده باشد یا Producer طبق سیاست Retry دوباره تلاش کرده باشد. Handler باید بر اساس Event ID از پردازش تکراری جلوگیری کند.

برای پاسخ Webhook از ۲۰۰ استفاده کنیم یا ۲۰۲؟

اگر رویداد کاملاً پردازش شده است، 200 مناسب است. اگر رویداد ثبت شده و پردازش بعداً انجام می‌شود، 202 Accepted می‌تواند معنای دقیق‌تری داشته باشد. مستندات Producer را نیز بررسی کنید.

اگر بعد از ارسال پاسخ، پردازش شکست بخورد چه می‌شود؟

Consumer باید وضعیت Event را failed ثبت کند و سازوکار Retry داخلی یا Replay داشته باشد. Producer پس از دریافت پاسخ موفق معمولاً از شکست داخلی شما باخبر نمی‌شود.

آیا FastAPI برای ساخت Webhook مناسب است؟

بله. FastAPI از دریافت Raw Body، Headerها، Validation، پردازش Async و مستندات OpenAPI پشتیبانی می‌کند. برای محیط Production باید Queue، پایگاه داده و مانیتورینگ نیز در نظر گرفته شود.

آیا می‌توان Webhook را به هوش مصنوعی متصل کرد؟

بله. Webhook می‌تواند ورود یک Event را اعلام کند و Backend شما داده آن را برای خلاصه‌سازی، دسته‌بندی یا تحلیل به API هوش مصنوعی ارسال کند.

آدرس API درواره چیست؟

Base URL درواره:

https://api.darvareh.ir/v1

Endpoint مربوط به Chat Completions:

https://api.darvareh.ir/v1/chat/completions

مدل مناسب درواره را از کجا انتخاب کنیم؟

شناسه مدل‌ها و قیمت به‌روز آن‌ها در صفحه مدل‌های درواره نمایش داده می‌شود.

جمع‌بندی

Webhook یکی از مهم‌ترین الگوهای ارتباط رویدادمحور میان نرم‌افزارهاست. به‌جای آنکه برنامه شما دائماً برای بررسی تغییرات Polling انجام دهد، سرویس مبدأ هنگام وقوع رویداد یک درخواست HTTP به Endpoint شما ارسال می‌کند.

ساخت یک Webhook اولیه ساده است، اما پیاده‌سازی قابل اتکا فقط به ایجاد یک مسیر POST محدود نمی‌شود. باید مواردی مانند Signature Verification، Duplicate Delivery، Retry، ترتیب نامطمئن رویدادها، پاسخ سریع، صف پردازش، نسخه‌بندی و مانیتورینگ را نیز در نظر بگیرید.

برای پروژه‌های هوش مصنوعی، Webhook می‌تواند نقطه شروع Workflowهایی مانند خلاصه‌سازی تیکت، تحلیل بازخورد، پردازش سند یا دریافت نتیجه Jobهای طولانی باشد. در این معماری، Backend شما رویداد را دریافت و اعتبارسنجی می‌کند و در صورت نیاز داده را برای پردازش به API درواره می‌فرستد.

برای شروع استفاده از API مدل‌های هوش مصنوعی، به وب‌سایت درواره مراجعه کنید. مدل‌های قابل استفاده و قیمت به‌روز آن‌ها نیز در صفحه مدل‌های درواره در دسترس است.

منابع

مقالات مرتبط

برای مطالعه شرایط استفاده و محدودیت‌های مسئولیت، صفحه «سلب مسئولیت» را مشاهده کنید.

Read more

اتوماسیون هوش مصنوعی چیست؟ کاربردها و آموزش ساخت AI Automation

اتوماسیون هوش مصنوعی چیست؟ کاربردها و آموزش ساخت AI Automation

اتوماسیون هوش مصنوعی با ترکیب گردش‌کارهای خودکار و مدل‌های هوش مصنوعی، پردازش متن، دسته‌بندی، استخراج اطلاعات و تصمیم‌های پیشنهادی را خودکار می‌کند. در این راهنما، معماری و ساخت نمونه عملی آن با API درواره را می‌آموزید.
Agentic Commerce چیست؟ آینده خرید با ایجنت هوش مصنوعی

Agentic Commerce چیست؟ آینده خرید با ایجنت هوش مصنوعی

Agentic Commerce شیوه‌ای جدید برای خرید اینترنتی است که در آن ایجنت هوش مصنوعی می‌تواند نیاز کاربر را بفهمد، محصولات را جست‌وجو و مقایسه کند و فرایند خرید را پیش ببرد. در این راهنما با معماری، UCP، ACP و پیاده‌سازی آن با API درواره آشنا می‌شوید.