webhook-setup

Webhook のセットアップ

このスキルを使用すると、AI エージェントが本番環境に対応した webhook 受信エンドポイントを構築し、webhook の送信元を設定できます。エージェントは、イベントペイロードを受け付ける HTTP エンドポイントを実装し、送信元を認証するために暗号署名を検証し、リトライを安全に処理できるようイベントを冪等に処理して、種類に応じた適切なハンドラーにイベントを振り分けます。その結果、リプレイ攻撃、順序が前後した配信、プロバイダーのタイムアウトなど、実環境で発生する障害にも対応できる、信頼性の高いイベント駆動型の統合が実現します。

ワークフロー

1. webhook エンドポイントを設計する： 推測されにくく、変更のない URL パス（例：/webhooks/stripe、/webhooks/github）に HTTP POST エンドポイントを作成します。受信を確認するため、エンドポイントは速やかに（ほとんどのプロバイダーでは 5 秒以内に）200 OK レスポンスを返す必要があります。時間のかかる処理は、ジョブキューを使って非同期で実行します。HTTPS のみを使用してください。ほとんどのプロバイダーは、HTTP のエンドポイントを拒否します。

2. 署名検証を実装する： すべての webhook プロバイダーは、HMAC-SHA256、RSA、または同様の方式でペイロードに署名します。イベントを処理する前に、プロバイダーの署名シークレットを使って署名を検証します。タイミング攻撃を防ぐため、一定時間で比較する関数を使って署名を比較します。署名がないリクエストや無効な署名のリクエストは、直ちに 401 Unauthorized レスポンスで拒否します。検証には、解析済みの JSON ではなく、署名前のバイト列と異なる場合がある生のリクエストボディを読み取ります。

3. イベントを解析し、種類別に振り分ける： 検証済みのペイロードを解析し、イベントの種類（例：payment_intent.succeeded、push）を抽出します。レジストリまたは switch 文を使い、各イベントの種類を専用のハンドラー関数に振り分けます。認識できないイベントの種類は警告レベルで記録し、未処理のイベントに対するプロバイダーのリトライが延々と続かないように 200 OK を返します。

4. イベントを冪等に処理する： プロバイダーは、タイムリーな 200 レスポンスを受け取れないと webhook の配信を再試行するため、ハンドラーが同じイベントを複数回受信することがあります。処理済みイベント ID をデータベースのテーブルに保存し、処理前に重複がないか確認します。データベーストランザクションを使って、イベントを処理済みとして記録する操作と副作用の実行を原子的に行います。

5. 非同期処理とデッドレターキューを追加する： メールの送信や複数のレコードの更新など、重い処理が必要なイベントについては、webhook を直ちに確認応答し、バックグラウンド処理のためにイベントをキューに登録します。リトライを使い切って失敗したイベントは、手動で調査できるようデッドレターキュー（DLQ）に移します。DLQ の深さを監視し、アラートを設定します。

6. プロバイダー側で webhook を設定する： webhook プロバイダーにエンドポイント URL を登録し、必要なイベントの種類を選択します（必要最小限を登録します）。署名シークレットを控えておきます。プロバイダーのテスト／ping 機能を使って webhook をテストします。プロバイダーのダッシュボードで配信失敗を監視するよう設定します。

対応テクノロジー

• Web フレームワーク： Express.js、Fastify、Flask、FastAPI、Django、Rails、Spring Boot
• キューシステム： Bull/BullMQ（Redis）、Celery（Python）、Sidekiq（Ruby）、SQS、RabbitMQ
• プロバイダー： Stripe、GitHub、Slack、Twilio、SendGrid、Shopify、PayPal、Paddle
• 監視： Svix（webhook インフラストラクチャ）、Hookdeck、ngrok（ローカル開発）
• データベース： PostgreSQL、MySQL、Redis（冪等性の追跡用）

使用方法

webhook プロバイダー（Stripe、GitHub など）、処理したいイベント、サーバーフレームワークをエージェントに指定します。エージェントは、署名検証、イベントの振り分け、冪等な処理、エラー処理を備えた完全な webhook 受信エンドポイントを生成します。ローカル開発では、エージェントがテスト用に ngrok または同様のトンネルを設定できます。

例

例 1：Stripe webhook 受信エンドポイント（Express.js）

const express = require("express");
const crypto = require("crypto");
const { Queue } = require("bullmq");

const app = express();
const eventQueue = new Queue("webhook-events", { connection: { host: "localhost" } });

// CRITICAL: Use raw body for signature verification — must be before json parser
app.post("/webhooks/stripe", express.raw({ type: "application/json" }), async (req, res) => {
  const signature = req.headers["stripe-signature"];
  const webhookSecret = process.env.STRIPE_WEBHOOK_SECRET;

  // Step 1: Verify signature
  let event;
  try {
    event = verifyStripeSignature(req.body, signature, webhookSecret);
  } catch (err) {
    console.error(`Signature verification failed: ${err.message}`);
    return res.status(401).json({ error: "Invalid signature" });
  }

  // Step 2: Idempotency check — skip if already processed
  const alreadyProcessed = await checkIdempotency(event.id);
  if (alreadyProcessed) {
    console.log(`Event ${event.id} already processed, skipping`);
    return res.status(200).json({ received: true, deduplicated: true });
  }

  // Step 3: Acknowledge immediately, process async
  try {
    await eventQueue.add(event.type, {
      eventId: event.id,
      type: event.type,
      data: event.data.object,
      created: event.created,
    });
    await markEventReceived(event.id);
    res.status(200).json({ received: true });
  } catch (err) {
    console.error(`Failed to enqueue event ${event.id}:`, err);
    res.status(500).json({ error: "Processing failed" });
  }
});

// Stripe signature verification (manual implementation)
function verifyStripeSignature(payload, signatureHeader, secret) {
  const elements = signatureHeader.split(",").reduce((acc, part) => {
    const [key, value] = part.split("=");
    acc[key.trim()] = value;
    return acc;
  }, {});

  const timestamp = elements["t"];
  const expectedSig = elements["v1"];

  // Protect against replay attacks: reject events older than 5 minutes
  const tolerance = 300; // 5 minutes
  const currentTime = Math.floor(Date.now() / 1000);
  if (currentTime - parseInt(timestamp) > tolerance) {
    throw new Error("Webhook timestamp too old — possible replay attack");
  }

  // Compute expected signature
  const signedPayload = `${timestamp}.${payload}`;
  const computedSig = crypto
    .createHmac("sha256", secret)
    .update(signedPayload)
    .digest("hex");

  // Constant-time comparison to prevent timing attacks
  if (!crypto.timingSafeEqual(Buffer.from(expectedSig), Buffer.from(computedSig))) {
    throw new Error("Signature mismatch");
  }

  return JSON.parse(payload);
}

// Idempotency tracking with PostgreSQL
const { Pool } = require("pg");
const db = new Pool();

async function checkIdempotency(eventId) {
  const result = await db.query(
    "SELECT 1 FROM processed_webhook_events WHERE event_id = $1",
    [eventId]
  );
  return result.rows.length > 0;
}

async function markEventReceived(eventId) {
  await db.query(
    "INSERT INTO processed_webhook_events (event_id, received_at) VALUES ($1, NOW()) ON CONFLICT (event_id) DO NOTHING",
    [eventId]
  );
}

// Event handler workers (BullMQ)
const { Worker } = require("bullmq");

const worker = new Worker("webhook-events", async (job) => {
  const { eventId, type, data } = job.data;

  const handlers = {
    "payment_intent.succeeded": handlePaymentSucceeded,
    "payment_intent.payment_failed": handlePaymentFailed,
    "customer.subscription.deleted": handleSubscriptionCanceled,
    "invoice.payment_failed": handleInvoicePaymentFailed,
  };

  const handler = handlers[type];
  if (!handler) {
    console.log(`Unhandled event type: ${type}`);
    return;
  }

  await handler(data, eventId);
  await db.query("UPDATE processed_webhook_events SET processed_at = NOW() WHERE event_id = $1", [eventId]);
}, { connection: { host: "localhost" } });

async function handlePaymentSucceeded(paymentIntent, eventId) {
  const orderId = paymentIntent.metadata.order_id;
  await db.query("UPDATE orders SET status = 'paid', payment_id = $1 WHERE id = $2", [
    paymentIntent.id, orderId,
  ]);
  console.log(`Order ${orderId} marked as paid (event: ${eventId})`);
}

async function handlePaymentFailed(paymentIntent, eventId) {
  const orderId = paymentIntent.metadata.order_id;
  await db.query("UPDATE orders SET status = 'payment_failed' WHERE id = $1", [orderId]);
  // Trigger customer notification
  console.log(`Payment failed for order ${orderId} (event: ${eventId})`);
}

async function handleSubscriptionCanceled(subscription, eventId) {
  await db.query("UPDATE users SET subscription_status = 'canceled' WHERE stripe_customer_id = $1", [
    subscription.customer,
  ]);
}

async function handleInvoicePaymentFailed(invoice, eventId) {
  console.log(`Invoice ${invoice.id} payment failed, attempt ${invoice.attempt_count}`);
}

app.listen(3000, () => console.log("Webhook receiver listening on port 3000"));

例 2：CI トリガー用 GitHub webhook（Python/FastAPI）

import hashlib
import hmac
import os
import logging
from datetime import datetime, timezone
from fastapi import FastAPI, Request, HTTPException, BackgroundTasks
from pydantic import BaseModel

app = FastAPI()
logger = logging.getLogger(__name__)

GITHUB_WEBHOOK_SECRET = os.environ["GITHUB_WEBHOOK_SECRET"]

# In-memory idempotency store (use Redis or PostgreSQL in production)
processed_deliveries: set[str] = set()


def verify_github_signature(payload_body: bytes, signature_header: str | None) -> None:
    """Verify the GitHub webhook signature using HMAC-SHA256."""
    if not signature_header:
        raise HTTPException(status_code=401, detail="Missing X-Hub-Signature-256 header")

    expected_signature = "sha256=" + hmac.new(
        GITHUB_WEBHOOK_SECRET.encode(),
        payload_body,
        hashlib.sha256,
    ).hexdigest()

    if not hmac.compare_digest(expected_signature, signature_header):
        raise HTTPException(status_code=401, detail="Invalid webhook signature")


@app.post("/webhooks/github")
async def github_webhook(request: Request, background_tasks: BackgroundTasks):
    # Step 1: Read raw body and verify signature
    body = await request.body()
    signature = request.headers.get("X-Hub-Signature-256")
    verify_github_signature(body, signature)

    # Step 2: Idempotency check using delivery ID
    delivery_id = request.headers.get("X-GitHub-Delivery")
    if delivery_id in processed_deliveries:
        logger.info(f"Duplicate delivery {delivery_id}, skipping")
        return {"status": "duplicate", "delivery_id": delivery_id}

    # Step 3: Parse event and route
    event_type = request.headers.get("X-GitHub-Event")
    payload = await request.json()

    logger.info(f"Received GitHub event: {event_type} (delivery: {delivery_id})")

    # Acknowledge immediately, process in background
    processed_deliveries.add(delivery_id)
    background_tasks.add_task(process_github_event, event_type, payload, delivery_id)

    return {"status": "accepted", "event": event_type, "delivery_id": delivery_id}


async def process_github_event(event_type: str, payload: dict, delivery_id: str):
    """Route and process GitHub webhook events."""
    handlers = {
        "push": handle_push,
        "pull_request": handle_pull_request,
        "issues": handle_issue,
        "ping": handle_ping,
    }

    handler = handlers.get(event_type)
    if not handler:
        logger.warning(f"Unhandled GitHub event type: {event_type}")
        return

    try:
        await handler(payload)
        logger.info(f"Successfully processed {event_type} (delivery: {delivery_id})")
    except Exception as e:
        logger.error(f"Failed to process {event_type} (delivery: {delivery_id}): {e}")
        # In production, move to dead letter queue for retry
        raise


async def handle_push(payload: dict):
    """Trigger CI pipeline on push to main branch."""
    ref = payload.get("ref", "")
    repo = payload["repository"]["full_name"]
    commits = payload.get("commits", [])

    if ref != "refs/heads/main":
        logger.info(f"Ignoring push to {ref} on {repo}")
        return

    commit_messages = [c["message"].split("\n")[0] for c in commits]
    logger.info(f"Push to main on {repo}: {len(commits)} commits")

    # Check if any commits modify CI-relevant files
    changed_files = set()
    for commit in commits:
        changed_files.update(commit.get("added", []))
        changed_files.update(commit.get("modified", []))

    # Trigger appropriate CI jobs
    if any(f.startswith("src/") for f in changed_files):
        await trigger_ci_pipeline(repo, "build-and-test", payload["after"])
    if any(f.startswith("infrastructure/") for f in changed_files):
        await trigger_ci_pipeline(repo, "infrastructure-plan", payload["after"])

    logger.info(f"CI pipelines triggered for {repo}@{payload['after'][:8]}")


async def handle_pull_request(payload: dict):
    """Run checks on pull request events."""
    action = payload["action"]
    pr = payload["pull_request"]
    repo = payload["repository"]["full_name"]

    if action in ("opened", "synchronize"):
        logger.info(f"PR #{pr['number']} {action} on {repo}: {pr['title']}")
        await trigger_ci_pipeline(repo, "pr-checks", pr["head"]["sha"])
    elif action == "closed" and pr.get("merged"):
        logger.info(f"PR #{pr['number']} merged on {repo}")
        await trigger_ci_pipeline(repo, "deploy-staging", pr["merge_commit_sha"])


async def handle_issue(payload: dict):
    """Log issue events."""
    action = payload["action"]
    issue = payload["issue"]
    logger.info(f"Issue #{issue['number']} {action}: {issue['title']}")


async def handle_ping(payload: dict):
    """Respond to GitHub's test ping."""
    logger.info(f"Ping received. Hook ID: {payload.get('hook_id')}, Zen: {payload.get('zen')}")


async def trigger_ci_pipeline(repo: str, pipeline: str, commit_sha: str):
    """Trigger a CI pipeline (stub — replace with actual CI API call)."""
    logger.info(f"Triggering {pipeline} for {repo}@{commit_sha[:8]}")
    # In production: POST to Jenkins, GitHub Actions, CircleCI, etc.

ベストプラクティス

• すばやく応答し、後で処理します。 3～5秒以内に 200 OK を返します。ほとんどのプロバイダーは10～30秒でタイムアウトし、再試行するため、処理が重複する原因になります。実際の処理には、イベントをバックグラウンドジョブキューに登録してください。
• Webhook ペイロードを処理する前に、必ず署名を検証します。 プロバイダーの署名シークレットと、定時間比較関数を使用してください。開発環境であっても、検証を省略しないでください。
• イベント ID を使って冪等性を実装します。 プロバイダーの配信 ID／イベント ID をデータベースに保存し、処理前に重複がないか確認してください。原子的な重複排除には INSERT ... ON CONFLICT DO NOTHING を使用します。
• すべての再試行後も処理に失敗するイベントには、デッドレターキューを使用します。 DLQ の深さを監視し、アラートを設定してください。バグを修正した後に DLQ からイベントを再生するためのツールも用意します。
• 処理するイベントだけをサブスクライブします。 処理しないイベントを受信すると、帯域幅を無駄にし、ログにノイズが増えます。ほとんどのプロバイダーでは、Webhook の設定時に特定のイベントタイプを選択できます。
• イベントのタイムスタンプを確認して、リプレイ攻撃を防ぎます。 5分より古いイベントは拒否してください（許容範囲に応じて設定可能）。Stripe、GitHub、その他ほとんどのプロバイダーでは、この目的で署名ヘッダーにタイムスタンプが含まれています。

エッジケース

• 順序が前後した配信： Webhook は順序どおりに届かない場合があります（例：invoice.created より先に invoice.paid が届く）。順番どおりに配信されると想定するのではなく、イベントのタイムスタンプとリソースの現在の状態（API 経由で取得）を使って判断してください。
• デプロイ中のプロバイダーによる再試行： デプロイ中にサーバーが停止していると、プロバイダーは再試行します。サーバー復旧後に同じイベントを受信しても、冪等性ロジックで適切に処理できるようにしてください。常に少なくとも1つのインスタンスが利用可能になるよう、ローリングデプロイを使用します。
• ペイロードサイズの上限： プロバイダーによっては大きなペイロードを送信します（特に差分の多い GitHub のプッシュイベント）。エンドポイントに適切なリクエストボディサイズの上限（例：10MB）を設定し、413 Payload Too Large を適切に処理してください。
• シークレットのローテーション： Webhook 署名シークレットをローテーションする際は、移行期間中、古いシークレットと新しいシークレットの両方をサポートしてください。まず新しいシークレットで検証し、失敗した場合は古いシークレットで検証します。すべての配信で新しいシークレットが使われていることを確認した後、古いシークレットを削除してください。
• エンドポイント URL の変更： Webhook URL を変更する必要がある場合は、古い URL の登録を解除する前に、新しい URL を登録してください。プロバイダーから新しい URL への配信が成功したと確認できるまで、両方のエンドポイントを並行して稼働させます。
• ローカルでのテスト： ngrok、Cloudflare Tunnel、またはプロバイダーの CLI（例：stripe listen --forward-to localhost:3000/webhooks/stripe）を使って、Webhook をローカルの開発サーバーに転送します。