Webhook Nedir? FastAPI ile Sağlam Bir Webhook Altyapısı

Kapıya her dakika bakmak yerine zil takmak

Webhook, bir sistemde bir olay olduğunda başka bir sisteme otomatik gönderilen HTTP isteğidir. Kısacası “bir şey olunca sana haber veririm” sözleşmesi.

Bir kargo beklediğinizi düşünün. İki seçeneğiniz var: her beş dakikada bir kapıyı açıp kargocu geldi mi diye bakmak ya da bir kapı zili takıp beklemek. İlki polling, ikincisi webhook.

Polling’de istemci sürekli sorar: “Ödeme tamamlandı mı?” Cevap çoğu zaman “hayır”dır. Bu hem sunucuyu boşuna yorar hem de olayı ancak bir sonraki sorguda öğrenmenize yol açar. Webhook’ta ise olay olduğu anda karşı taraf sizin belirlediğiniz bir URL’e POST atar. Gereksiz trafik yok, gecikme yok.

	Polling	Webhook
Kim başlatır?	Alıcı sürekli sorar	Gönderen olay olunca bildirir
Gecikme	Sorgu aralığı kadar	Neredeyse anlık
Boşa giden istek	Çok	Yok
Alıcıdan beklenen	Bir cron/zamanlayıcı	Herkese açık bir HTTPS endpoint
Zorluk	Basit ama israfçı	Güvenlik, tekrar deneme ve sıra yönetimi gerekir

Webhook’a bazen “reverse API” denmesi de bu yüzden: normalde siz API’yi çağırırsınız, burada API sizi çağırır.

Webhook nasıl çalışır?

Akış dört adımdan oluşur ve iki tarafın da sorumluluğu vardır.

Kayıt: Alıcı, gönderen sisteme bir URL verir (örn. https://api.siteniz.com/webhooks/stripe) ve hangi olayları dinlemek istediğini seçer. Genelde bu sırada ortak bir gizli anahtar (secret) üretilir.
Olay: Gönderen tarafta bir şey olur: ödeme alınır, bir commit push edilir, bir SMS gelir.
Teslimat: Gönderen, olayı JSON olarak bu URL’e POST eder. İsteğin header’ında, gövdenin secret ile imzalanmış hali (HMAC) bulunur.
Onay: Alıcı imzayı doğrular ve hızlıca 2xx döner. 2xx gelmezse ya da istek zaman aşımına uğrarsa gönderen bir süre sonra tekrar dener.

Alıcının altın kuralı şu: önce onayla, sonra işle. Uzun süren işi (mail atmak, rapor üretmek) HTTP isteğinin içinde yapmayın. Olayı kaydedin, 200 dönün, işi arka planda bitirin.

Gerçek hayattan örnekler

Servis	Tipik olay	Ne işe yarar?
Stripe	`invoice.paid`, `customer.subscription.deleted`	Abonelik durumunu kendi veritabanınızla senkron tutmak
GitHub	`push`, `pull_request`	CI/CD tetiklemek, Slack’e bildirim atmak
Twilio / WhatsApp	gelen mesaj	Kullanıcıya otomatik cevap vermek
Shopify	`orders/create`	Siparişi depoya veya muhasebeye iletmek

Stripe ile çalışıyorsanız şunu bilmekte fayda var: ödeme sonrası kullanıcının tarayıcısı “başarılı” sayfasına hiç düşmeyebilir. Sekmeyi kapatabilir, interneti kopabilir. Bu yüzden abonelik durumunun tek doğru kaynağı redirect değil, webhook’tur.

FastAPI ile webhook alıcısı yazalım

Tek dosyalık bir örnek yerine, gerçek bir projeye taşıyabileceğiniz küçük ama düzenli bir yapı kuralım. Senaryo: bir ödeme servisi bize payment.succeeded ve payment.failed olaylarını gönderiyor.

webhook-receiver/
├── app/
│   ├── main.py            # FastAPI uygulaması
│   ├── config.py          # secret ve ayarlar (.env'den)
│   ├── security.py        # HMAC imza doğrulama
│   ├── schemas.py         # Pydantic olay modeli
│   ├── store.py           # idempotency (aynı olayı iki kez işleme)
│   ├── handlers.py        # olay tipi -> fonksiyon eşleşmesi
│   └── routers/
│       └── webhooks.py    # endpoint
└── sender.py              # gönderen taraf (sonraki bölüm)

Kurulum:

pip install fastapi uvicorn pydantic-settings httpx
uvicorn app.main:app --reload

1. Ayarlar: `config.py`

Secret’ı asla koda gömmeyin. pydantic-settings, WEBHOOK_SECRET ortam değişkenini otomatik okur.

from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    webhook_secret: str = "whsec_degistir_beni"
    signature_tolerance_seconds: int = 300  # 5 dk'dan eski istekleri reddet


settings = Settings()

2. İmza doğrulama: `security.py`

Yazının en önemli kısmı burası. Webhook URL’iniz internete açık; imza kontrolü yoksa herkes size sahte bir “ödeme alındı” olayı gönderebilir.

Stripe’ın kullandığı yapıya benzer bir şema kullanıyoruz: X-Webhook-Signature: t=<timestamp>,v1=<imza>. İmza, timestamp.gövde metninin HMAC-SHA256 ile hash’lenmiş hali.

import hashlib
import hmac
import time

from fastapi import HTTPException, Request

from app.config import settings


def compute_signature(secret: str, timestamp: str, body: bytes) -> str:
    signed_payload = f"{timestamp}.".encode() + body
    return hmac.new(secret.encode(), signed_payload, hashlib.sha256).hexdigest()


async def verify_signature(request: Request) -> bytes:
    """Header formatı: X-Webhook-Signature: t=1731590000,v1=<hex>"""
    header = request.headers.get("X-Webhook-Signature")
    if not header:
        raise HTTPException(status_code=401, detail="İmza yok")

    try:
        parts = dict(item.split("=", 1) for item in header.split(","))
        timestamp, received_sig = parts["t"], parts["v1"]
    except (ValueError, KeyError):
        raise HTTPException(status_code=401, detail="İmza formatı hatalı")

    # Replay saldırısına karşı: eski istekleri kabul etme
    if abs(time.time() - int(timestamp)) > settings.signature_tolerance_seconds:
        raise HTTPException(status_code=401, detail="İstek çok eski")

    body = await request.body()  # HAM gövde: JSON'a çevirmeden imzala
    expected_sig = compute_signature(settings.webhook_secret, timestamp, body)

    # == yerine compare_digest: timing attack'e karşı sabit süreli karşılaştırma
    if not hmac.compare_digest(expected_sig, received_sig):
        raise HTTPException(status_code=401, detail="Geçersiz imza")

    return body

Burada üç detay var ve üçü de gerçek projelerde sıkça atlanıyor:

Ham gövde (request.body()): İmzayı JSON’u parse edip tekrar string’e çevirdikten sonra hesaplarsanız, boşluklar veya key sırası değişir ve imza asla tutmaz.
hmac.compare_digest: Normal == ilk farklı karakterde durur. Saldırgan cevap süresini ölçerek imzayı karakter karakter tahmin edebilir.
Timestamp kontrolü: Geçerli bir isteği yakalayan biri onu saatler sonra tekrar göndermeye çalışabilir (replay). Timestamp imzanın içinde olduğu için değiştirilemez.

3. Olay modeli: `schemas.py`

from datetime import datetime
from typing import Any

from pydantic import BaseModel


class WebhookEvent(BaseModel):
    id: str              # olayın benzersiz kimliği -> idempotency anahtarı
    type: str            # "payment.succeeded" gibi
    created_at: datetime
    data: dict[str, Any]

4. Idempotency: `store.py`

Webhook gönderen sistemlerin neredeyse hepsi “at-least-once” çalışır: bir olayı en az bir kez, bazen birden fazla kez gönderir. Sizin 200’ünüz ağda kaybolursa aynı olay tekrar gelir. Müşteriye iki kez fatura maili gitmemesi için olay ID’sini kaydediyoruz.

"""Demo için bellek içi. Production'da Redis (SET NX + TTL)
ya da Postgres'te UNIQUE(event_id) kısıtı kullanın."""

_processed: set[str] = set()


def mark_if_new(event_id: str) -> bool:
    """Olay ilk kez geliyorsa True döner ve işaretler."""
    if event_id in _processed:
        return False
    _processed.add(event_id)
    return True

5. Olay yönlendirme: `handlers.py`

Olay tipleri arttıkça dev bir if/elif bloğu yazmamak için küçük bir dekoratörle bir kayıt defteri (registry) kuruyoruz. Yeni bir olay eklemek, yeni bir fonksiyon yazmak kadar kolay.

import logging
from collections.abc import Callable

from app.schemas import WebhookEvent

logger = logging.getLogger("webhooks")

HANDLERS: dict[str, Callable[[WebhookEvent], None]] = {}


def on(event_type: str):
    """Bir fonksiyonu belirli bir olay tipine bağlayan dekoratör."""
    def decorator(func: Callable[[WebhookEvent], None]):
        HANDLERS[event_type] = func
        return func
    return decorator


@on("payment.succeeded")
def handle_payment_succeeded(event: WebhookEvent) -> None:
    logger.info("Ödeme alındı: %s", event.data.get("amount"))
    # aboneliği aktifleştir, fatura maili gönder...


@on("payment.failed")
def handle_payment_failed(event: WebhookEvent) -> None:
    logger.warning("Ödeme başarısız: %s", event.data.get("customer_id"))


def dispatch(event: WebhookEvent) -> None:
    handler = HANDLERS.get(event.type)
    if handler is None:
        logger.info("Bilinmeyen olay tipi atlandı: %s", event.type)
        return
    handler(event)

Bilinmeyen bir olay tipinde hata fırlatmıyoruz, sadece logluyoruz. Gönderen taraf yarın yeni bir olay tipi eklerse sisteminiz patlamamalı.

6. Endpoint: `routers/webhooks.py`

Tüm parçalar burada buluşuyor. İmza kontrolü bir FastAPI dependency’si olduğu için fonksiyonun içine girmeden çalışıyor.

from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException
from pydantic import ValidationError

from app.handlers import dispatch
from app.schemas import WebhookEvent
from app.security import verify_signature
from app.store import mark_if_new

router = APIRouter(prefix="/webhooks", tags=["webhooks"])


@router.post("/payments", status_code=200)
async def receive_payment_webhook(
    background_tasks: BackgroundTasks,
    body: bytes = Depends(verify_signature),
):
    try:
        event = WebhookEvent.model_validate_json(body)
    except ValidationError:
        raise HTTPException(status_code=400, detail="Geçersiz payload")

    # Aynı olay ikinci kez gelirse yine 200 dön ama tekrar işleme
    if not mark_if_new(event.id):
        return {"status": "duplicate"}

    # Önce onayla, sonra işle
    background_tasks.add_task(dispatch, event)
    return {"status": "accepted"}

Dikkat: tekrar eden olaya hata değil 200 dönüyoruz. 409 dönseydik gönderen taraf bunu başarısızlık sanar ve denemeye devam ederdi.

7. Uygulama: `main.py`

import logging

from fastapi import FastAPI

from app.routers import webhooks

logging.basicConfig(level=logging.INFO)

app = FastAPI(title="Webhook Receiver")
app.include_router(webhooks.router)


@app.get("/health")
def health():
    return {"ok": True}

Test edelim

Aşağıdaki küçük test, dört senaryonun hepsini kontrol ediyor. Bir sonraki bölümdeki sign fonksiyonunu kullanıyor.

import json, time, uuid
from fastapi.testclient import TestClient
from app.main import app
from app.config import settings
from sender import sign

client = TestClient(app)

def post(event, sig=None):
    body = json.dumps(event).encode()
    return client.post("/webhooks/payments", content=body,
        headers={"X-Webhook-Signature": sig or sign(settings.webhook_secret, body)})

ev = {"id": "evt_" + uuid.uuid4().hex, "type": "payment.succeeded",
      "created_at": "2026-11-14T12:00:00Z", "data": {"amount": 4900}}

print(post(ev).json())                                       # {'status': 'accepted'}
print(post(ev).json())                                       # {'status': 'duplicate'}
print(post(ev, sig=f"t={int(time.time())},v1=sahte").json()) # {'detail': 'Geçersiz imza'}
print(post(ev, sig="t=1000,v1=sahte").json())                # {'detail': 'İstek çok eski'}

Production notu: BackgroundTasks aynı process’te çalışır. Sunucu iş bitmeden yeniden başlarsa iş kaybolur ve olay “işlendi” olarak işaretli kalır. Ciddi bir sistemde olayı önce veritabanına status='pending' ile yazın, işi Celery, ARQ veya Dramatiq gibi bir kuyruğa verin ve başarılı olunca status='done' yapın.

Madalyonun diğer yüzü: webhook göndermek

Kendi SaaS’ınızın müşterilerine webhook göndermesi gerektiğinde iş tersine döner. Artık imzalamak ve karşı tarafın sunucusu çöktüğünde tekrar denemek sizin sorumluluğunuzdadır.

import hashlib
import hmac
import json
import time

import httpx

RETRYABLE_STATUS = {408, 429, 500, 502, 503, 504}


def sign(secret: str, body: bytes) -> str:
    timestamp = str(int(time.time()))
    sig = hmac.new(
        secret.encode(), f"{timestamp}.".encode() + body, hashlib.sha256
    ).hexdigest()
    return f"t={timestamp},v1={sig}"


def send_webhook(url: str, secret: str, event: dict, max_attempts: int = 5) -> bool:
    body = json.dumps(event, separators=(",", ":")).encode()

    for attempt in range(1, max_attempts + 1):
        headers = {
            "Content-Type": "application/json",
            "X-Webhook-Signature": sign(secret, body),  # her denemede taze timestamp
            "X-Webhook-Id": event["id"],
        }
        try:
            response = httpx.post(url, content=body, headers=headers, timeout=10)
            if 200 <= response.status_code < 300:
                return True
            if response.status_code not in RETRYABLE_STATUS:
                return False  # 400/401 gibi hatalar tekrar denenince düzelmez
        except httpx.TransportError:
            pass  # bağlantı hatası, timeout -> tekrar dene

        if attempt < max_attempts:
            time.sleep(2 ** attempt)  # 2, 4, 8, 16 sn: exponential backoff

    return False  # burada olayı dead-letter tablosuna yazın

Bu kodda dikkat edilmesi gereken birkaç karar var:

Exponential backoff: Karşı tarafın sunucusu zaten zorlanıyorsa her saniye istek atmak onu daha da boğar. Bekleme süresini her denemede ikiye katlıyoruz.
Her denemede yeni imza: Timestamp imzanın parçası. Eski imzayı tekrar kullanırsanız, alıcının 5 dakikalık toleransına takılırsınız.
Her hatayı tekrar denemiyoruz: 401 imza hatasıdır, bir saat sonra da 401 olacaktır. Sadece geçici hatalar (5xx, 429, timeout) tekrar denemeye değer.
Dead-letter: Tüm denemeler bittiğinde olayı kaybetmeyin. Ayrı bir tabloya yazın, müşteriye panelde “tekrar gönder” butonu sunun.

Gerçek bir sistemde time.sleep ile döngüde beklemek yerine, her denemeyi bir kuyruğa gecikmeli iş olarak koyarsınız. Stripe başarısız webhook’ları üç güne yayılan bir takvimle tekrar dener. Yukarıdaki döngü ise mantığı göstermek için sade tutuldu.

Bilmekte fayda olan ekstralar

Sıralama garantisi yoktur

Olaylar gönderildiği sırayla gelmeyebilir. subscription.updated olayı, subscription.created olayından önce ulaşabilir. Çözüm: olayın created_at alanına bakın ve elinizdeki kayıttan daha eskiyse yok sayın. Ya da olayı sadece bir “bir şey değişti” sinyali olarak kullanıp güncel durumu API’den tekrar çekin. Stripe’ta bu ikinci yöntem çok yaygındır.

Thin vs fat payload

Fat payload: Olayın tüm verisi gövdede gelir. Hızlıdır ama veri bayatlayabilir ve hassas bilgi taşır.
Thin payload: Sadece {"type": "order.updated", "id": "ord_123"} gelir, detayı siz API’den çekersiniz. Her zaman güncel veriyle çalışırsınız, ekstra bir istek pahasına.

Secret rotasyonu

Secret’ı değiştirmeniz gerektiğinde (sızıntı, çalışan ayrılışı vb.) bir geçiş dönemi tanıyın. Gönderen bir süre iki imza birden gönderir (v1=eski,v1=yeni), alıcı ikisinden biri tutarsa kabul eder.

Güvenlik kontrol listesi

Endpoint sadece HTTPS üzerinden erişilebilir
HMAC imzası ham gövde üzerinden ve compare_digest ile doğrulanıyor
Timestamp toleransı var (replay koruması)
Secret ortam değişkeninde, koda veya repoya gömülü değil
Payload boyutuna üst sınır konmuş (örn. 1 MB)
Gönderen tarafta: müşterinin verdiği URL’in localhost, 169.254.169.254 gibi iç adreslere gitmesi engellenmiş (SSRF koruması)

Son madde özellikle webhook gönderen bir ürün yapıyorsanız kritik. Kullanıcı URL olarak bulut sağlayıcınızın metadata adresini girerse, sunucunuz kendi gizli bilgilerini ona okuyabilir.

Geliştirirken işe yarayan araçlar

Araç	Ne için?
webhook.site	Gelen isteği görmek için anında geçici bir URL
ngrok / Cloudflare Tunnel	Localhost’taki FastAPI’nizi internete açmak
Stripe CLI (`stripe listen --forward-to localhost:8000/...`)	Stripe olaylarını doğrudan local’e yönlendirmek ve tetiklemek
Svix, Hookdeck	Retry, log, imza gibi her şeyi hazır sunan webhook altyapı servisleri

Sık yapılan hatalar

İmzayı parse edilmiş JSON üzerinden doğrulamak.
Endpoint içinde 30 saniyelik iş yapıp gönderenin timeout’una takılmak. Sonuç: aynı olay tekrar gelir, iş iki kez yapılır.
Tekrar eden olaya 4xx dönmek ve sonsuz retry döngüsüne girmek.
Gelen her isteği loglamamak. Bir şeyler ters gittiğinde “olay hiç geldi mi?” sorusunun cevabı yalnızca loglardadır.

Son söz

Webhook’un özü tek cümle: “bir şey olunca sana bir HTTP isteği atarım.” Ama güvenilir bir webhook sistemi, o cümlenin etrafına örülen dört alışkanlıktan oluşur: imzayı doğrula, hızlıca onayla, aynı olayı iki kez işleme, başarısız olanı kaybetme.

Bu dördünü oturttuğunuzda Stripe’tan, GitHub’dan ya da kendi servisinizden gelen olayları aynı güvenle karşılayabilirsiniz. Yukarıdaki yapıyı kopyalayıp handlers.py’a kendi olaylarınızı ekleyerek başlayabilirsiniz.