エージェントとWebhookの接続
Webhookイベントを受信し、それに応じてエージェントのワークフローを起動します。
「エージェントとWebhookの接続」はCoddyKit上の無料AI Agentsレッスンです。 これはレッスン2/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。 これはAI Agents学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 AI Agentsコースには全4レッスンが含まれています。
Webhookとは
webhookはHTTPコールバックです。外部サービスでイベントが発生すると、イベントデータを含むPOSTリクエストがエンドポイントに送信されます。エージェントはペイロードを処理してアクションを実行します。
webhookはプッシュベース（イベント発生時に届く）であるのに対し、ポーリングは繰り返し確認する方式です。
FastAPI Webhookエンドポイント
FastAPIを使うと、webhookの受信処理を簡単に作成できます。POSTルートを定義し、JSON本文を解析して、エージェントのロジックに処理を引き渡します。
from fastapi import FastAPI, Request
from pydantic import BaseModel
app = FastAPI()
class WebhookPayload(BaseModel):
event: str
data: dict
@app.post('/webhook')
async def receive_webhook(payload: WebhookPayload):
print(f'Received event: {payload.event}')
print(f'Data: {payload.data}')
# Route to the right agent handler
if payload.event == 'email.received':
await handle_email_event(payload.data)
elif payload.event == 'file.uploaded':
await handle_file_event(payload.data)
return {'status': 'accepted'}
async def handle_email_event(data: dict):
print(f'Processing email from: {data.get("from")}')
async def handle_file_event(data: dict):
print(f'Processing file: {data.get("filename")}')Webhook署名の検証
webhookリクエストが想定した送信元から来たことを必ず検証してください。ほとんどのサービスは、共有シークレットを使ってHMAC-SHA256でペイロードに署名します。署名が無効なリクエストは拒否してください。
import hmac
import hashlib
from fastapi import FastAPI, Request, HTTPException
app = FastAPI()
WEBHOOK_SECRET = 'your-webhook-secret-here'
def verify_signature(payload_bytes: bytes, signature_header: str) -> bool:
expected = hmac.new(
WEBHOOK_SECRET.encode(),
payload_bytes,
hashlib.sha256
).hexdigest()
received = signature_header.replace('sha256=', '')
return hmac.compare_digest(expected, received)
@app.post('/webhook/verified')
async def verified_webhook(request: Request):
payload_bytes = await request.body()
signature = request.headers.get('X-Signature', '')
if not verify_signature(payload_bytes, signature):
raise HTTPException(status_code=401, detail='Invalid signature')
# Safe to process
import json
data = json.loads(payload_bytes)
return {'status': 'verified', 'event': data.get('event')}冪等性キー
外部サービスは、失敗したwebhookの配信を再試行することがよくあります。冪等性キーは、各イベントとともに送信される一意のIDです。処理済みのキーを保存し、重複をスキップしてください。
from fastapi import FastAPI, Request, HTTPException
import redis
import json
app = FastAPI()
r = redis.Redis(host='localhost', port=6379, decode_responses=True)
@app.post('/webhook/idempotent')
async def idempotent_webhook(request: Request):
payload = await request.json()
# Extract idempotency key from header or payload
idempotency_key = request.headers.get('Idempotency-Key') or payload.get('event_id')
if not idempotency_key:
raise HTTPException(status_code=400, detail='Missing idempotency key')
redis_key = f'webhook:processed:{idempotency_key}'
# Check if already processed
if r.exists(redis_key):
print(f'Duplicate event {idempotency_key}, skipping')
return {'status': 'duplicate', 'idempotency_key': idempotency_key}
# Process event
# ... agent logic here ...
# Mark as processed (expire after 24h)
r.setex(redis_key, 86400, '1')
return {'status': 'processed', 'idempotency_key': idempotency_key}再試行の重複排除戦略
冪等性キーに加えて、重複排除の時間枠も検討してください。短い時間枠内に同じイベント内容を受信した場合、それは再試行である可能性が高いです。イベントハッシュを比較して再試行を検出し、破棄します。
import hashlib
import json
from datetime import datetime
# In-memory store; use Redis in production
recent_hashes = {}
DEDUP_WINDOW_SECONDS = 300 # 5 minutes
def is_duplicate(payload: dict) -> bool:
# Hash the event content
content = json.dumps(payload, sort_keys=True)
event_hash = hashlib.md5(content.encode()).hexdigest()
now = datetime.utcnow().timestamp()
# Clean up old entries
expired = [h for h, ts in recent_hashes.items() if now - ts > DEDUP_WINDOW_SECONDS]
for h in expired:
del recent_hashes[h]
if event_hash in recent_hashes:
return True
recent_hashes[event_hash] = now
return False
# Test
payload = {'event': 'payment.completed', 'amount': 100}
print('First:', is_duplicate(payload)) # False
print('Second:', is_duplicate(payload)) # True (duplicate)エージェントの非同期実行
webhookハンドラーはすばやく（5秒未満で）応答し、エージェントのロジックはバックグラウンドで処理してください。タイムアウトを避けるには、FastAPIのBackgroundTasksを使用します。
from fastapi import FastAPI, BackgroundTasks
import asyncio
app = FastAPI()
async def run_agent_job(event: str, data: dict):
print(f'Agent starting for event: {event}')
await asyncio.sleep(2) # Simulate LLM call
print(f'Agent finished for event: {event}')
@app.post('/webhook/async')
async def async_webhook(request_data: dict, background_tasks: BackgroundTasks):
event = request_data.get('event', 'unknown')
data = request_data.get('data', {})
# Respond immediately
background_tasks.add_task(run_agent_job, event, data)
return {'status': 'accepted', 'message': 'Processing in background'}複雑なペイロードの解析
サービスごとにペイロードの形式は異なります。各サービス専用のパーサー関数を作成し、エージェントが常に正規化されたイベントオブジェクトを受け取るようにしてください。
from dataclasses import dataclass
from typing import Optional
@dataclass
class NormalizedEvent:
event_type: str
source: str
resource_id: str
metadata: dict
def parse_github_webhook(payload: dict) -> NormalizedEvent:
return NormalizedEvent(
event_type='github.' + payload.get('action', 'unknown'),
source='github',
resource_id=str(payload.get('repository', {}).get('id', '')),
metadata={
'repo': payload.get('repository', {}).get('full_name'),
'sender': payload.get('sender', {}).get('login')
}
)
def parse_stripe_webhook(payload: dict) -> NormalizedEvent:
return NormalizedEvent(
event_type=payload.get('type', 'unknown'),
source='stripe',
resource_id=payload.get('id', ''),
metadata={'amount': payload.get('data', {}).get('object', {}).get('amount')}
)
# Usage
github_payload = {'action': 'opened', 'repository': {'id': 123, 'full_name': 'user/repo'}, 'sender': {'login': 'alice'}}
event = parse_github_webhook(github_payload)
print(event)Webhookのレスポンスコードは重要
正しいHTTPステータスを返してください。2xxはwebhookを受け付けたことを送信元に伝えます。4xxはクライアントエラー（不正なペイロード）を意味します。5xxまたはタイムアウトが発生すると、送信元は再試行します。
- 200：受け付けて処理済み
- 202：非同期処理のために受け付け済み
- 400：不正なリクエスト（必須フィールドが不足）
- 401：不正な署名
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
app = FastAPI()
@app.post('/webhook/proper-responses')
async def proper_webhook(request: Request):
try:
payload = await request.json()
except Exception:
raise HTTPException(status_code=400, detail='Invalid JSON body')
required_fields = ['event', 'data']
for field in required_fields:
if field not in payload:
raise HTTPException(status_code=400, detail=f'Missing field: {field}')
event = payload['event']
known_events = ['email.received', 'file.uploaded', 'payment.completed']
if event not in known_events:
# Acknowledge unknown events gracefully - do not retry
return JSONResponse(status_code=200, content={'status': 'ignored', 'reason': 'unknown event'})
# Start background processing
return JSONResponse(status_code=202, content={'status': 'accepted'})Webhookのローカルテスト
テストではngrokを使って、ローカルサーバーをインターネットに公開します。ngrok http 8000を実行すると、ローカルのFastAPIアプリにトンネル接続する公開URLを取得できます。
# Start your FastAPI app
# uvicorn main:app --reload --port 8000
# In another terminal, start ngrok:
# ngrok http 8000
# You get: https://abc123.ngrok.io
# Now configure your webhook in Stripe/GitHub/etc. to:
# https://abc123.ngrok.io/webhook
# Test with curl:
import subprocess
def test_webhook_locally():
test_payload = '{"event": "email.received", "data": {"from": "test@example.com"}}'
# In real usage you would run this in terminal:
# curl -X POST http://localhost:8000/webhook \
# -H 'Content-Type: application/json' \
# -d '{"event": "email.received", "data": {"from": "test@example.com"}}'
print('Test payload:', test_payload)
print('Send to: http://localhost:8000/webhook')
test_webhook_locally()Webhookイベントのログ記録
受信したすべてのwebhookについて、タイムスタンプ、送信元、イベント種別、処理結果をログに記録してください。この監査証跡は、イベントの取りこぼしや重複処理の問題をデバッグするうえで不可欠です。
import logging
import json
from datetime import datetime
import sys
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s %(levelname)s %(message)s',
stream=sys.stdout
)
logger = logging.getLogger('webhook')
def log_webhook_event(event_id: str, event_type: str, source: str, status: str, details: dict = None):
logger.info(json.dumps({
'timestamp': datetime.utcnow().isoformat(),
'event_id': event_id,
'event_type': event_type,
'source': source,
'status': status,
'details': details or {}
}))
# Usage in webhook handler
log_webhook_event(
event_id='evt_123',
event_type='email.received',
source='gmail',
status='processed',
details={'from': 'user@example.com', 'action_taken': 'reply_sent'}
)受信Webhookのレート制限
レート制限を使用して、webhookエンドポイントが過負荷にならないよう保護してください。slowapiライブラリを使うと、最小限のコードでFastAPIにレート制限を追加できます。
from fastapi import FastAPI, Request
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded
limiter = Limiter(key_func=get_remote_address)
app = FastAPI()
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
@app.post('/webhook/limited')
@limiter.limit('100/minute')
async def rate_limited_webhook(request: Request):
payload = await request.json()
return {'status': 'accepted', 'event': payload.get('event')}理解度チェック：Webhook
エージェントでwebhookを扱う際のベストプラクティスについて、理解度を確認しましょう。
本番環境でのWebhook
本番環境では、署名検証、冪等性キー、バックグラウンド処理、構造化ログ、レート制限のパターンをすべて組み合わせます。TLS終端と追加の保護のために、nginxなどのリバースプロキシの背後にデプロイしてください。
よくある質問
「エージェントとWebhookの接続」レッスンは無料ですか？
はい。「エージェントとWebhookの接続」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し（組み込みコードエディタと24時間対応のAIチューター）、AI Agentsコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 AI Agentsコースには全4レッスンが含まれています。
「エージェントとWebhookの接続」で何を学びますか？
Webhookイベントを受信し、それに応じてエージェントのワークフローを起動します。 ブラウザで直接実行するハンズオンコードでAI Agentsを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。
AI Agentsを始めるのに経験は必要ですか？
事前経験は必要ありません。CoddyKitのAI Agentsは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン2/4です。
「エージェントとWebhookの接続」レッスンにはどのくらい時間がかかりますか？
ほとんどのCoddyKitレッスンは約5～10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。
このAI Agentsレッスンでコードを書いて実行できますか？
はい。すべてのAI Agentsレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。
このコースのすべてのレッスン
- トリガー・アクション型エージェントのパターン
- エージェントとWebhookの接続
- スケジューリングとCronベースのエージェント
- 複数アプリの自動化パイプラインを構築する