Python×FastAPIでWebhookサーバーを自作する方法！LINE・GitHub・Stripeの通知を受け取るイベント駆動アプリ入門
「Webhookって名前はよく聞くけど、実際どうやって受け取るの？」
そんな疑問を持っている方、多いんじゃないでしょうか。LINEのメッセージ通知、GitHubのプッシュイベント、Stripeの決済完了通知……これらはすべてWebhookという仕組みで飛んできます。
今回は、PythonのWebフレームワークFastAPIを使って、自前のWebhookサーバーをゼロから作る方法を解説します。「難しそう」が「あ、こういうことか！」に変わる記事を目指しました。ぜひ最後まで読んでみてください 😊
🔔 Webhookとは？ポーリングとの違いをざっくり理解しよう
まず「Webhookって何？」をざっくり説明します。
イメージとしてはこんな感じです。
- 📞 ポーリング：「まだ？まだ？まだ？」と自分から何度も問い合わせる方式
- 📬 Webhook：「何かあったらこっちに連絡して」と電話番号を渡しておく方式
つまりWebhookは、イベントが起きたときだけ相手から通知を受け取る仕組みです。効率がよく、リアルタイムに近い処理ができるため、LINEやGitHub・Stripeなどの外部サービスが広く採用しています。
受け取る側（私たちのサーバー）は特定のURLでHTTP POSTリクエストを待ち受けていればOKです。FastAPIはそのURLエンドポイントを爆速で作れるので、Webhookサーバーとの相性が抜群なんですよね。
🛠️ 環境を整えよう
対象読者：PythonとHTTPの基礎を知っている初〜中級者の方を想定しています。FastAPI自体が初めてでも大丈夫です。
まず必要なライブラリをインストールしておきましょう。
pip install fastapi uvicorn python-dotenv httpxポイントをまとめるとこんな感じです。
fastapi：Webフレームワーク本体uvicorn：ASGIサーバー（FastAPIを動かすために必要）python-dotenv：環境変数の管理（シークレットトークン管理に使う）httpx：外部APIへのHTTPリクエスト送信用
📦 Webhookサーバーの基本構造を作る
まずはWebhookの「受け取り口」を作ります。FastAPIでPOSTエンドポイントを定義するだけでOKです。
# main.py
from fastapi import FastAPI, Request
import uvicorn
app = FastAPI()
@app.post("/webhook")
async def receive_webhook(request: Request):
# リクエストボディをJSON形式で受け取る
body = await request.json()
print(f"Webhookを受信しました: {body}")
return {"status": "ok"}
if __name__ == "__main__":
uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)
これがWebhookサーバーの最小構成です。POST /webhookにリクエストが来たらボディを受け取って返答する、シンプルな構造ですよね。
サーバーを起動してみましょう。
python main.pyブラウザで http://localhost:8000/docs を開くと、FastAPIが自動生成したSwagger UIが確認できます。エンドポイントが正しく登録されているかここで確認しておきましょう 👀
🔐 セキュリティ対策｜署名検証を実装しよう
実務でWebhookを使う場合、「本当に正規のサービスから来たリクエストか」を検証する処理が必須です。悪意のある第三者が偽のリクエストを送ってくる可能性があるからです。
多くのサービスはリクエストヘッダーにHMAC署名を付けて送ってきます。受け取った側はシークレットキーを使って署名を検証します。
# セキュリティ検証付きWebhookエンドポイントの例
import hmac
import hashlib
from fastapi import FastAPI, Request, HTTPException
import os
from dotenv import load_dotenv
load_dotenv()
app = FastAPI()
WEBHOOK_SECRET = os.getenv("WEBHOOK_SECRET", "your-secret-key")
def verify_signature(payload: bytes, signature: str, secret: str) -> bool:
"""HMAC-SHA256で署名を検証する"""
expected = hmac.new(
secret.encode(),
payload,
hashlib.sha256
).hexdigest()
# タイミング攻撃対策にhmac.compare_digestを使う
return hmac.compare_digest(f"sha256={expected}", signature)
@app.post("/webhook/secure")
async def receive_secure_webhook(request: Request):
# ヘッダーから署名を取得
signature = request.headers.get("X-Hub-Signature-256", "")
body = await request.body()
# 署名の検証
if not verify_signature(body, signature, WEBHOOK_SECRET):
raise HTTPException(status_code=403, detail="Invalid signature")
import json
payload = json.loads(body)
print(f"検証OK！ペイロード: {payload}")
return {"status": "verified"}
ここが重要です：
hmac.compare_digestを使うことでタイミング攻撃を防いでいます- シークレットキーは.envファイルで管理してコードに直書きしない
- 署名が不正なら即座に
403を返す
💬 実践①：LINE Messaging APIのWebhookを受け取る
LINEのBotはユーザーがメッセージを送ると、登録したWebhook URLにイベントデータを送ってきます。これを受け取ってオウム返しするシンプルなBotを作ってみましょう。
LINE Developersコンソールで「Webhookの送信」を有効化し、Webhook URLに自分のサーバーのURLを設定しておく必要があります（後述のngrokで公開します）。
# LINE Webhook受信エンドポイント
import httpx
from fastapi import FastAPI, Request
import os
app = FastAPI()
LINE_CHANNEL_ACCESS_TOKEN = os.getenv("LINE_CHANNEL_ACCESS_TOKEN")
async def reply_to_line(reply_token: str, message: str):
"""LINEに返信メッセージを送信する"""
headers = {
"Authorization": f"Bearer {LINE_CHANNEL_ACCESS_TOKEN}",
"Content-Type": "application/json",
}
body = {
"replyToken": reply_token,
"messages": [{"type": "text", "text": message}],
}
async with httpx.AsyncClient() as client:
response = await client.post(
"https://api.line.me/v2/bot/message/reply",
headers=headers,
json=body,
)
return response.status_code
@app.post("/webhook/line")
async def line_webhook(request: Request):
data = await request.json()
events = data.get("events", [])
for event in events:
# テキストメッセージのイベントのみ処理
if event["type"] == "message" and event["message"]["type"] == "text":
reply_token = event["replyToken"]
user_message = event["message"]["text"]
# オウム返し
await reply_to_line(reply_token, f"受け取りました：{user_message}")
return {"status": "ok"}
LINEのWebhookはリクエストを受け取ったら200 OKを素早く返すことが重要です。重い処理は非同期で別タスクに分離するとよいですよ。
🐙 実践②：GitHubのプッシュイベントを受け取る
GitHubのWebhookを使うと、リポジトリへのプッシュやPRのオープンなど、様々なイベントをトリガーにできます。CI/CDや自動通知の入り口になる部分ですよね。
# GitHub Pushイベントを受け取るエンドポイント
from fastapi import FastAPI, Request, Header
from typing import Optional
app = FastAPI()
@app.post("/webhook/github")
async def github_webhook(
request: Request,
x_github_event: Optional[str] = Header(None),
):
payload = await request.json()
print(f"GitHubイベント種別: {x_github_event}")
if x_github_event == "push":
# プッシュしたブランチ名を取得
branch = payload.get("ref", "").replace("refs/heads/", "")
pusher = payload.get("pusher", {}).get("name", "unknown")
commits = payload.get("commits", [])
print(f"🔔 {pusher}さんが '{branch}' ブランチに {len(commits)} コミットをプッシュしました")
for commit in commits:
print(f" - {commit['message']} ({commit['id'][:7]})")
elif x_github_event == "pull_request":
action = payload.get("action")
pr_title = payload.get("pull_request", {}).get("title")
print(f"📝 PRが{action}されました: {pr_title}")
return {"status": "received", "event": x_github_event}
x_github_eventヘッダーでイベントの種別を判定するのがポイントです。push・pull_request・issuesなど、GitHubからは様々なイベントが飛んできます。
💳 実践③：Stripeの決済完了通知を受け取る
Stripeの決済Webhookは、EC系アプリ開発では頻出パターンです。「支払い成功」「サブスクリプション更新」などのイベントを受け取って、DBのステータスを更新するような処理に使います。
# Stripe Webhook受信エンドポイント
from fastapi import FastAPI, Request, HTTPException
import hmac
import hashlib
import time
import os
app = FastAPI()
STRIPE_WEBHOOK_SECRET = os.getenv("STRIPE_WEBHOOK_SECRET")
def verify_stripe_signature(payload: bytes, sig_header: str, secret: str) -> bool:
"""Stripeの署名を検証する（タイムスタンプ付き）"""
try:
# ヘッダーからタイムスタンプと署名を分解
elements = dict(item.split("=", 1) for item in sig_header.split(","))
timestamp = elements.get("t", "")
signature = elements.get("v1", "")
# タイムスタンプが5分以上古ければ拒否（リプレイ攻撃対策）
if abs(time.time() - int(timestamp)) > 300:
return False
signed_payload = f"{timestamp}.{payload.decode()}"
expected_sig = hmac.new(
secret.encode(), signed_payload.encode(), hashlib.sha256
).hexdigest()
return hmac.compare_digest(expected_sig, signature)
except Exception:
return False
@app.post("/webhook/stripe")
async def stripe_webhook(request: Request):
body = await request.body()
sig_header = request.headers.get("Stripe-Signature", "")
if not verify_stripe_signature(body, sig_header, STRIPE_WEBHOOK_SECRET):
raise HTTPException(status_code=400, detail="Invalid Stripe signature")
import json
event = json.loads(body)
event_type = event.get("type")
if event_type == "payment_intent.succeeded":
payment_intent = event["data"]["object"]
amount = payment_intent.get("amount", 0)
print(f"✅ 決済成功！ 金額: {amount // 100}円")
# ここでDBのステータス更新やメール送信を行う
elif event_type == "customer.subscription.deleted":
print("❌ サブスクリプションがキャンセルされました")
return {"status": "ok"}
Stripeの署名検証ではタイムスタンプのチェックが特徴的です。5分以上前のリクエストは弾くことで、過去のWebhookを再送する「リプレイ攻撃」を防いでいます。
🌐 ローカルサーバーを外部公開する｜ngrokの使い方
ローカルで動いているサーバーにLINEやGitHubからWebhookを届けるには、外部からアクセスできるURLが必要です。開発中はngrokを使うのが定番です。
ngrokをインストール後、こうするだけです。
# FastAPIサーバーを起動
python main.py
# 別のターミナルでngrokを起動（8000番ポートを公開）
ngrok http 8000ngrokが起動すると https://xxxx.ngrok-free.app のようなURLが発行されます。このURLをLINEやGitHubのWebhook設定画面に貼り付ければOKです。
確認しておきましょう：ngrokの無料プランではURLが再起動のたびに変わります。開発用と割り切って使うのがよいですよ。
⚡ 非同期処理でWebhookをもっと堅牢にする
Webhookの鉄則として、受け取ったらすぐに200を返し、重い処理は非同期でバックグラウンドに回すという設計があります。FastAPIにはBackgroundTasksという便利な仕組みがあります。
from fastapi import FastAPI, Request, BackgroundTasks
import asyncio
app = FastAPI()
async def heavy_process(payload: dict):
"""時間のかかる処理（DB更新・メール送信など）"""
await asyncio.sleep(2) # 重い処理のシミュレーション
print(f"バックグラウンド処理完了: {payload.get('event_type')}")
@app.post("/webhook/async")
async def async_webhook(request: Request, background_tasks: BackgroundTasks):
payload = await request.json()
# 重い処理をバックグラウンドに登録（すぐに200を返せる）
background_tasks.add_task(heavy_process, payload)
return {"status": "accepted"} # すぐに返答！
BackgroundTasksを使うことで、レスポンスを先に返しながら処理を続けることができます。外部サービスはタイムアウト（だいたい5〜30秒）があるので、この設計は実務でかなり重要ですよね。
📋 まとめ
今回学んだことをまとめます。
- ✅ Webhookは「イベントが起きたときだけ通知を受け取る」仕組み
- ✅ FastAPIでPOSTエンドポイントを作るだけでWebhookサーバーが完成する
- ✅ HMAC署名検証でなりすましリクエストをブロックする
- ✅ LINE・GitHub・Stripeそれぞれに固有の検証方法がある
- ✅ BackgroundTasksで重い処理を非同期化して即座に200を返す設計が大事
- ✅ 開発中はngrokでローカルサーバーを一時公開できる
「Webhookって難しそう」から「あ、POSTリクエストを受け取るだけか」という感覚に変わってきたんじゃないでしょうか 😊
FastAPIとWebhookを組み合わせると、イベント駆動なアプリ開発の入り口がぐっと広がります。LINEの自動応答Bot、GitHubプッシュ時の自動デプロイ通知、Stripeの決済ログ記録……どれも今回の知識でベースが作れます。
ぜひローカル環境で動かしてみてください！わからなかったところは何度でも読み返してもらえると嬉しいです。一緒に学んでいきましょう 🚀
こちらも読まれています
📚 関連商品・おすすめ書籍
※本記事にはアフィリエイトリンクが含まれます。
PR
