AIでWebhook統合 — イベント駆動型自動化の構築

外部サービスで何かが起きたら、Webhookで通知を受け取り、AIが判断して自動アクションを実行する。Discord Bot、GitHub Webhook、Stripeの支払い通知など、イベント駆動型自動化の実装方法を解説する。

Webhook受信サーバーの基本構成

FastAPIで軽量なWebhook受信サーバーを立てる。

from fastapi import FastAPI, Request, HTTPException
import hmac, hashlib

app = FastAPI()

@app.post("/webhook/{service}")
async def receive_webhook(service: str, request: Request):
    body = await request.body()
    headers = request.headers

    if service == "github":
        verify_github_signature(body, headers)
    elif service == "discord":
        verify_discord_signature(body, headers)

    payload = await request.json()
    await process_event(service, payload)
    return {"status": "ok"}

GitHub Webhookの実装

署名検証

GitHub WebhookはX-Hub-Signature-256ヘッダーでHMAC-SHA256署名を送る。必ず検証する。

import os

GITHUB_WEBHOOK_SECRET = os.environ["GITHUB_WEBHOOK_SECRET"]

def verify_github_signature(body, headers):
    signature = headers.get("X-Hub-Signature-256", "")
    expected = "sha256=" + hmac.new(
        GITHUB_WEBHOOK_SECRET.encode(),
        body,
        hashlib.sha256
    ).hexdigest()
    if not hmac.compare_digest(signature, expected):
        raise HTTPException(status_code=401, detail="Invalid signature")

イベント処理

async def handle_github_event(payload):
    action = payload.get("action")

    if "pull_request" in payload:
        pr = payload["pull_request"]
        if action == "opened":
            await ai_review_pr(pr["number"], pr["diff_url"])
        elif action == "closed" and pr.get("merged"):
            await update_changelog(pr["title"], pr["body"])

    elif "issue" in payload and action == "opened":
        await auto_label_issue(payload["issue"])

Discord Webhookの実装

Discord Botからのメッセージ受信

from nacl.signing import VerifyKey
from nacl.exceptions import BadSignatureError

DISCORD_PUBLIC_KEY = os.environ["DISCORD_PUBLIC_KEY"]

def verify_discord_signature(body, headers):
    signature = headers.get("X-Signature-Ed25519", "")
    timestamp = headers.get("X-Signature-Timestamp", "")
    verify_key = VerifyKey(bytes.fromhex(DISCORD_PUBLIC_KEY))
    try:
        verify_key.verify(timestamp.encode() + body, bytes.fromhex(signature))
    except BadSignatureError:
        raise HTTPException(status_code=401, detail="Invalid signature")

Discordへの通知送信

import httpx

DISCORD_WEBHOOK_URL = os.environ["DISCORD_WEBHOOK_URL"]

async def send_discord_notification(title, description, color=0x00ff00):
    embed = {
        "embeds": [{
            "title": title,
            "description": description,
            "color": color
        }]
    }
    async with httpx.AsyncClient() as client:
        await client.post(DISCORD_WEBHOOK_URL, json=embed)

セキュリティの必須事項

対策	理由
署名検証を必ず実装する	偽造リクエストを排除
HTTPS必須	通信内容の盗聴防止
レート制限を設ける	DDoS攻撃の緩和
ペイロードサイズを制限する	メモリ枯渇攻撃の防止
シークレットは環境変数に格納	ソースコードへの漏洩防止

from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)

@app.post("/webhook/{service}")
@limiter.limit("60/minute")
async def receive_webhook(service, request: Request):
    body = await request.body()
    if len(body) > 16384:
        raise HTTPException(status_code=413, detail="Payload too large")

AI処理パイプラインとの接続

Webhook受信後、AIに判断させてアクションを自動実行する。

import subprocess

async def ai_review_pr(pr_number, diff_url):
    result = subprocess.run(
        ["claude", "-p", f"PR #{pr_number}のdiffをレビューして。{diff_url}"],
        capture_output=True, text=True, timeout=120
    )
    await post_github_comment(pr_number, result.stdout)

Webhook統合により、外部イベントをトリガーとしたAI自動化パイプラインが完成する。署名検証を省略せず、セキュリティを最優先で設計すること。