FastAPIで安全にWebhook受信APIを実装する：署名検証・再送対策・冪等性・非同期処理まで含めた実務ガイド

greeden

4時間前

FastAPIで安全にWebhook受信APIを実装する：署名検証・再送対策・冪等性・非同期処理まで含めた実務ガイド

要約

Webhook受信APIは「ただPOSTを受けるエンドポイント」ではありません。実務では、送信元の真正性確認、改ざん防止、再送への耐性、タイムアウト回避、監査しやすさまで含めて設計する必要があります。Stripe は署名検証に公式ライブラリの利用を推奨しており、GitHub もシークレットを使った署名検証を強く勧めています。
FastAPIでは、Request から生のリクエスト本文を取得して署名検証を行い、検証後にイベント種別ごとの処理へ流す形が安全です。本文を改変してから検証すると失敗しやすく、Stripe も未加工のリクエスト本文を使うことを案内しています。
実運用では、Webhook受信APIの本体で重い処理を完了させるのではなく、まず検証して記録し、素早くACKを返し、後続処理はバックグラウンドやジョブキューへ逃がすのが安定します。FastAPIの BackgroundTasks はレスポンス後の軽い後処理に使えます。
この記事では、FastAPIでの安全なWebhook実装を、基本設計 → 署名検証 → 冪等性 → 再送・順不同対策 → 監査ログ → テストの順で整理し、StripeやGitHubの考え方も踏まえながら、プロバイダ依存を減らした実務パターンとしてまとめます。

誰が読んで得をするか

個人開発・学習者さん

決済完了やGitHubイベントをFastAPIで受け取りたいけれど、Webhookの安全な作り方がまだ曖昧な方。
「POSTを受けてDB更新すればよいのでは？」と思っているけれど、署名検証や再送対策まで含めると何をすべきか知りたい方。

この方には、まずWebhookは“認証付きの外部イベント受信口”だという理解を固めながら、最小構成から安全側へ寄せていく型をお渡しします。StripeやGitHubは、いずれもシークレットを使った署名検証を重視しています。

小規模チームのバックエンドエンジニアさん

決済、GitHub、外部SaaSなど複数のWebhookを扱い始め、エンドポイント設計や再送処理が散らかってきた方。
タイムアウトや二重処理を避けつつ、共通化できる部分を整理したい方。

この方には、**「受信・検証・記録・ACK・後続処理」**という役割分担と、署名検証・イベントID保存・ジョブ投入の分け方が役に立つはずです。一般的なWebhookベストプラクティスでも、真正性確認、再送、冪等性、速やかな応答が重要視されています。

SaaS開発チーム・スタートアップの皆さま

決済連携や外部イベント連携が事業上重要で、取りこぼしや二重処理が売上や顧客体験に直結する方。
将来的にイベント種別が増えても壊れにくい、監査しやすい、再送に強い受信基盤をFastAPI上で整えたい方。

この方には、イベント受信を業務処理から分離し、監査ログ・冪等性テーブル・ジョブキューへ流す設計が特に重要です。GitHubやStripeも、Webhookは署名検証のうえで処理し、失敗や再送を前提に設計することを示しています。

アクセシビリティ評価

最初に要約を置き、そのあとに「なぜ危ないか」「どう守るか」「FastAPIでどう書くか」を段階的に並べています。
専門用語は初出で簡潔に説明し、以降は同じ表現で統一しています。
コードは短めに分割し、役割ごとに見られるようにしています。
目標レベルはAA相当です。

1. Webhook受信APIは、なぜ普通のPOST APIより慎重に作るべきなのか

Webhookは、外部サービスがあなたのサーバへ能動的に送ってくる通知です。たとえば「決済成功」「請求失敗」「リポジトリ更新」などのイベントが、外からやってきます。GitHubはWebhook配信を受け取るサーバ側で署名検証をしてから処理することを勧めており、Stripeも Stripe-Signature ヘッダとエンドポイントシークレットを使った検証を公式に案内しています。

普通の社内APIなら、認証トークン付きのクライアントが自分から呼びます。けれどWebhookは、送信元が本物かどうかをこちらで確かめないといけないという点が大きく違います。さらに、Webhookは再送や順不同が起こり得るため、「1回しか来ない」「順番どおり来る」という前提で書くと、実務で壊れやすくなります。Webhookのベストプラクティスでも、認証・署名・再送・冪等性が繰り返し重要視されています。

2. まず決めるべき基本方針：受信口の責務を小さくする

Webhook受信APIの責務は、できるだけ小さく保つのが安全です。おすすめの流れは、次の5段階です。

生のリクエスト本文とヘッダを受け取る
署名や送信元の真正性を検証する
イベントIDなどを記録して、重複でないか確認する
可能ならすぐに 2xx を返す
本当に重い業務処理は後段へ渡す

この形にすると、Webhook受信口は「外部イベントの玄関」になり、業務ロジックの本体とは分離できます。Stripeのドキュメントでも、Webhookエンドポイントは署名を検証してイベントを安全に扱う前提で説明されており、FastAPIではレスポンス後の軽い後処理に BackgroundTasks を利用できます。

3. 署名検証の考え方：シークレット＋生の本文が基本

GitHubは、Webhookシークレットを使った X-Hub-Signature-256 の検証を案内しています。GitHubの配送は、シークレットが設定されていれば、SHA-256のHMAC値をヘッダに載せて送られます。Stripeも同様に、Stripe-Signature ヘッダとエンドポイントシークレット、そして未加工のリクエスト本文を使って署名検証するよう案内しています。

ここでとても大切なのが、JSONとしてパースする前の生本文を検証に使うことです。Stripeは署名エラーの原因として、本文が改変されているケースを明示しており、未加工の本文取得を重要なポイントとして説明しています。FastAPIでは Request オブジェクトから生本文へアクセスできます。

4. FastAPIで生の本文を安全に受け取る最小例

まずは、FastAPIでWebhook受信口の骨組みを作ります。ここではプロバイダ非依存の形で、ヘッダと生本文を受け取ります。

from fastapi import APIRouter, Request, Header, HTTPException, status

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

@router.post("/generic")
async def receive_webhook(
    request: Request,
    x_signature: str | None = Header(default=None),
):
    raw_body = await request.body()

    if not x_signature:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="missing signature header",
        )

    # ここで署名検証
    # 署名OKならJSONパース
    payload = await request.json()

    return {"received": True}

この段階ではまだ署名検証を実装していませんが、ポイントは明確です。
await request.body() で生本文を先に取ること、そして署名確認が済む前に業務処理へ進まないことです。FastAPIの Request から生のリクエスト本文へアクセスできることは、Webhook実装で非常に重要です。

5. HMAC署名を自前で検証する基本パターン

GitHubのWebhook検証に近い形として、HMAC-SHA256で署名を照合する最小パターンを示します。実際のプロバイダが公式ライブラリを出しているなら、まずはそれを使うのが推奨です。Stripeも公式ライブラリ利用を勧めています。

import hashlib
import hmac

def verify_hmac_sha256(raw_body: bytes, header_signature: str, secret: str) -> bool:
    digest = hmac.new(
        key=secret.encode("utf-8"),
        msg=raw_body,
        digestmod=hashlib.sha256,
    ).hexdigest()

    # 例: "sha256=<hex>" 形式を想定
    expected = f"sha256={digest}"
    return hmac.compare_digest(expected, header_signature)

FastAPI側では次のように使えます。

from fastapi import Request, Header, HTTPException, status

WEBHOOK_SECRET = "replace-me"

@router.post("/github-like")
async def receive_github_like_webhook(
    request: Request,
    x_hub_signature_256: str | None = Header(default=None),
):
    raw_body = await request.body()

    if not x_hub_signature_256:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="missing signature",
        )

    if not verify_hmac_sha256(raw_body, x_hub_signature_256, WEBHOOK_SECRET):
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="invalid signature",
        )

    payload = await request.json()
    return {"ok": True, "event_type": payload.get("action")}

この考え方は多くのWebhookに共通しますが、プロバイダごとに署名ヘッダ形式や時刻署名の有無が異なるため、本番では必ず公式仕様に合わせてください。GitHubは X-Hub-Signature-256、Stripeは Stripe-Signature を使います。

6. Stripeのような公式ライブラリ前提の検証は、なるべく公式に寄せる

StripeはWebhook署名検証について、公式ライブラリの利用を推奨しています。また、必要なものは Stripe-Signature ヘッダ、エンドポイントシークレット、未加工のリクエスト本文だと明記しています。

そのため、Stripe連携では概念的に次のような流れになります。

# これは概念例です。実際は stripe ライブラリの推奨手順に合わせてください。
@router.post("/stripe")
async def receive_stripe_webhook(
    request: Request,
    stripe_signature: str | None = Header(default=None, alias="Stripe-Signature"),
):
    raw_body = await request.body()

    if not stripe_signature:
        raise HTTPException(status_code=400, detail="missing Stripe-Signature")

    # stripe.Webhook.construct_event(raw_body, stripe_signature, endpoint_secret)
    # のような公式検証へ渡すイメージ
    event = {"type": "payment_intent.succeeded"}  # ここは説明用

    return {"received": True, "type": event["type"]}

ここで大切なのは、「署名検証が通ってからイベント種別を見る」順番です。
イベント種別に応じて処理を変えたくなる気持ちはありますが、まずは本物かどうかを確認するのが先です。

7. 冪等性：同じイベントが2回来ても壊れないようにする

Webhookでは、同じイベントが再送されることがあります。ネットワークの揺らぎや受信側のタイムアウトで、送信元は「届いたか分からない」ため再送します。そこで必要なのが冪等性です。つまり、同じイベントを複数回受けても、最終状態が1回目と変わらないようにする設計です。Webhookのベストプラクティスでも、冪等性は中核的なテーマです。

もっとも実務的なのは、プロバイダが付けるイベントIDや配信IDを保存して、処理済みかどうかを確認する方法です。GitHubでは配信ごとの識別子や各種ヘッダがあり、StripeでもイベントオブジェクトにIDがあります。

7.1 処理済みイベントを保存するモデル例

from pydantic import BaseModel
from datetime import datetime

class ProcessedWebhookEvent(BaseModel):
    provider: str
    event_id: str
    received_at: datetime

7.2 疑似的な重複チェック例

from fastapi import HTTPException, status

processed_event_ids: set[str] = set()

def ensure_not_processed(provider: str, event_id: str) -> None:
    key = f"{provider}:{event_id}"
    if key in processed_event_ids:
        raise HTTPException(
            status_code=status.HTTP_200_OK,
            detail="already processed",
        )
    processed_event_ids.add(key)

本番ではもちろんDBやRedisに保存する方が安全です。
ただ、考え方としては「イベントIDを一意制約で保持し、二重処理を止める」という形が分かりやすいです。

8. 順不同や遅延への備え：イベント順序を信用しすぎない

Webhookは必ずしも時系列どおりに届くとは限りません。
そのため、「Aのあとに必ずBが来る」と決め打ちすると壊れやすいです。Webhook一般のベストプラクティスでも、受信側は順不同や再送を前提に設計することが推奨されています。

実務では、次のような考え方が安定します。

Webhookは「状態更新のきっかけ」として使う
必要なら送信元APIへ再照会して現在の正状態を取得する
受信イベントだけで全状態を決め打ちしない

たとえば決済連携なら、「決済成功イベントが来たら、そのまま請求状態を決める」のではなく、「イベントを受けたうえで必要に応じて最新の支払い状態を再取得する」方が堅牢なことがあります。StripeもWebhookを非同期イベントの受信口として位置づけています。

9. ACKは早く返す：重い処理は後段へ逃がす

Webhook送信側は、短時間で 2xx が返ることを期待していることが多いです。ここで受信APIが重い集計や大量DB更新をしていると、タイムアウトして再送を呼びやすくなります。そのため、検証・最低限の永続化・ACK を先に済ませ、後続処理はバックグラウンドやキューへ渡すのが定番です。FastAPIでは BackgroundTasks を使ってレスポンス後に軽い処理を走らせることができます。

9.1 FastAPIの `BackgroundTasks` を使う例

from fastapi import BackgroundTasks, Request, Header

def process_webhook_event(provider: str, event_id: str, payload: dict) -> None:
    # 実際はDB更新やジョブキュー投入
    print(provider, event_id, payload.get("type"))

@router.post("/generic-async")
async def receive_webhook_async(
    request: Request,
    background_tasks: BackgroundTasks,
    x_signature: str | None = Header(default=None),
):
    raw_body = await request.body()

    if not x_signature:
        raise HTTPException(status_code=400, detail="missing signature")

    # 署名検証はここで
    payload = await request.json()

    provider = "generic"
    event_id = payload.get("id")
    if not event_id:
        raise HTTPException(status_code=400, detail="missing event id")

    background_tasks.add_task(process_webhook_event, provider, event_id, payload)
    return {"received": True}

ただし、BackgroundTasks は同一プロセス内の軽い後処理向きです。
重い処理や再試行が必要な処理は、以前の記事で扱った Celery などのジョブキューへ逃がす方が安定します。FastAPI公式でも BackgroundTasks はレスポンス後の処理に使えることが示されています。

10. イベントルーティング：if文を増やしすぎない形にする

Webhookイベントが増えてくると、1つのエンドポイント内に if event_type == ... が大量に並び始めます。
そこで、イベント種別ごとのハンドラ関数を辞書で束ねると整理しやすくなります。

from collections.abc import Callable

def handle_payment_succeeded(payload: dict) -> None:
    ...

def handle_payment_failed(payload: dict) -> None:
    ...

HANDLERS: dict[str, Callable[[dict], None]] = {
    "payment.succeeded": handle_payment_succeeded,
    "payment.failed": handle_payment_failed,
}

def dispatch_event(event_type: str, payload: dict) -> None:
    handler = HANDLERS.get(event_type)
    if handler:
        handler(payload)

これをWebhook後段処理から呼ぶようにしておくと、イベント追加時の差分が小さくなります。
さらに、プロバイダごとに stripe_handlers.py、github_handlers.py のように分けておくと、責務が見えやすくなります。

11. 監査ログとイベントログ：通常ログと分けて考える

Webhookは外部から入ってくる重要イベントなので、監査しやすいログを別で持っておくと後々とても助かります。

最低限、次の情報を残せると便利です。

provider（Stripe、GitHubなど）
event_id
event_type
received_at
signature_verified
request_id
processing_status
error_message（失敗時）

StripeやGitHubのような外部イベントは、「届いた」「検証した」「処理した／保留した」が追えるだけで、障害調査がかなり楽になります。GitHubも各種ヘッダや配送識別情報を持っており、配信の追跡に役立ちます。

12. エラー時の返し方：何を2xxにして、何を4xx/5xxにするか

Webhookでは、受信側の返し方が再送挙動に影響することがあります。
一般的な考え方としては、次の整理が分かりやすいです。

署名不正、本文不正、必須ヘッダ欠落
→ 4xx（こちらの受信条件を満たしていない）
一時的な内部障害、DB障害
→ 5xx（再送される前提で設計）
既に処理済みの重複イベント
→ 2xx でOKとすることが多い

ここでの判断はプロバイダの再送ポリシーとも関係するため、最終的には各サービスの仕様に合わせます。ただ、**「重複は成功扱いにしてよい」「署名不正は絶対に成功扱いしない」**という考え方は、多くの場面で有効です。GitHubやStripeのドキュメントはいずれも、まず正当な配信かを確認することを重視しています。

13. Webhookを受けるURLは、できれば専用に分ける

複数サービスのWebhookを1つの受信口へまとめることもできますが、実務ではプロバイダごとにURLを分ける方が安全で運用しやすいです。

例:

/webhooks/stripe
/webhooks/github
/webhooks/internal

こうしておくと、

シークレット管理が分かれる
署名方式の違いを吸収しやすい
ログや監視を分けやすい
事故時の切り分けが早い

という利点があります。GitHubとStripeではヘッダ名も署名形式も違うため、最初から分ける方が自然です。

14. テスト戦略：最低限守りたい5本

Webhookは「受けて終わり」に見えますが、壊れやすいのでテストがとても大切です。
まずは、次の5本があるだけでもかなり安心です。

正しい署名で受信できる
署名不正なら拒否される
同じイベントIDを2回送っても二重処理しない
重い後続処理が失敗しても受信口の責務が壊れない
必須ヘッダや必須イベントIDがなければ弾く

FastAPIの通常のAPIテストと同じように、署名ヘッダと生本文を意識して TestClient で確認できます。GitHubやStripeが示す「署名ヘッダ＋生本文＋シークレット」の考え方に沿ってテストデータを作ると、実務で役立つテストになります。

15. 読者別ロードマップ

個人開発・学習者さん

まずはプロバイダごとにWebhook URLを分ける
Request.body() で生本文を受ける形を作る
シークレット署名検証を入れる
イベントIDを保存して重複処理を防ぐ
重い処理は BackgroundTasks かジョブキューへ渡す

小規模チームのエンジニアさん

「受信・検証・ACK・後続処理」の責務分離をチームで共有する
署名検証をプロバイダ別の共通関数にまとめる
イベント保存テーブルや監査ログを整える
冪等性キー（イベントID）の永続化を入れる
テストで「重複」「署名不正」「順不同に近いケース」を守る

SaaS開発チーム・スタートアップの皆さま

Webhook受信をドメインイベント入口として設計し直す
監査ログ、イベントテーブル、ジョブキューを分離する
プロバイダ別に再送・障害時ポリシーを決める
監視項目として「受信数」「検証失敗数」「重複数」「処理失敗数」を持つ
決済や契約変更のWebhookは、請求状態や監査ログの記事とつなげて全体設計する

参考リンク

FastAPI
Stripe
GitHub
実務的なベストプラクティス
- Webhook Security Best Practices
- Best Practices for Receiving Webhooks

まとめ

Webhook受信APIは、普通のPOST APIより慎重に作るべき入口です。署名検証、冪等性、再送対策、監査しやすさまで含めて考えると、実務でかなり壊れにくくなります。
FastAPIでは、Request から生本文を取得して署名を確認し、素早くACKを返し、後続処理をバックグラウンドやジョブキューへ渡す形が相性のよい基本パターンです。
StripeやGitHubのような主要サービスも、シークレットを使った署名検証を前提にしています。まずは公式仕様に寄せ、プロバイダごとの差は受信口や共通関数で吸収するのがおすすめです。
最初から完璧なイベント基盤を作る必要はありませんが、「受信・検証・記録・ACK・後続処理」を分けるだけで、将来の複雑化にかなり強くなります。

次の記事としては、この流れと相性が良い「FastAPIで実装する外部APIクライアント設計（再試行・タイムアウト・サーキットブレーカ）」や、「FastAPIのジョブキュー設計をWebhook後続処理にどうつなぐか」が自然につながります。

FastAPIで安全にWebhook受信APIを実装する：署名検証・再送対策・冪等性・非同期処理まで含めた実務ガイド

要約

誰が読んで得をするか

個人開発・学習者さん

小規模チームのバックエンドエンジニアさん

SaaS開発チーム・スタートアップの皆さま

アクセシビリティ評価

1. Webhook受信APIは、なぜ普通のPOST APIより慎重に作るべきなのか

2. まず決めるべき基本方針：受信口の責務を小さくする

3. 署名検証の考え方：シークレット＋生の本文が基本

4. FastAPIで生の本文を安全に受け取る最小例

5. HMAC署名を自前で検証する基本パターン

6. Stripeのような公式ライブラリ前提の検証は、なるべく公式に寄せる

7. 冪等性：同じイベントが2回来ても壊れないようにする

7.1 処理済みイベントを保存するモデル例

7.2 疑似的な重複チェック例

8. 順不同や遅延への備え：イベント順序を信用しすぎない

9. ACKは早く返す：重い処理は後段へ逃がす

9.1 FastAPIの BackgroundTasks を使う例

10. イベントルーティング：if文を増やしすぎない形にする

11. 監査ログとイベントログ：通常ログと分けて考える

12. エラー時の返し方：何を2xxにして、何を4xx/5xxにするか

13. Webhookを受けるURLは、できれば専用に分ける

14. テスト戦略：最低限守りたい5本

15. 読者別ロードマップ

個人開発・学習者さん

小規模チームのエンジニアさん

SaaS開発チーム・スタートアップの皆さま

参考リンク

まとめ

共有:

9.1 FastAPIの `BackgroundTasks` を使う例