Menghubungkan Agen ke Webhook
Terima acara webhook dan picu alur kerja agen sebagai respons.
Menghubungkan Agen ke Webhook adalah pelajaran AI Agents gratis di CoddyKit. Ini adalah pelajaran 2 dari 4. Kamu bisa membaca pelajaran lengkapnya di bawah secara gratis — lalu praktikkan langsung di browser dengan editor kode bawaan dan tutor AI 24/7. Ini adalah bagian dari jalur belajar AI Agents, dan progresmu tersinkronisasi di web dan aplikasi CoddyKit. Kursus AI Agents mencakup 4 pelajaran total.
Apa Itu Webhook
Webhook adalah panggilan balik HTTP. Saat suatu peristiwa terjadi di layanan eksternal, layanan tersebut mengirim permintaan POST ke titik akhir Anda bersama data peristiwa. Agen Anda memproses muatan tersebut dan bertindak.
Webhook berbasis push (peristiwa tiba saat terjadi), berbeda dengan pemeriksaan berkala (Anda memeriksanya berulang kali).
Titik Akhir Webhook FastAPI
FastAPI memudahkan pembuatan penerima webhook. Tentukan rute POST, uraikan body JSON, lalu serahkan pemrosesannya kepada logika agen Anda.
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")}')Verifikasi Tanda Tangan Webhook
Selalu verifikasi bahwa permintaan webhook berasal dari pengirim yang diharapkan. Sebagian besar layanan menandatangani muatannya dengan HMAC-SHA256 menggunakan rahasia bersama. Tolak permintaan dengan tanda tangan yang tidak valid.
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')}Kunci Idempotensi
Layanan eksternal sering mencoba kembali pengiriman webhook yang gagal. Kunci idempotensi adalah ID unik yang dikirim bersama setiap peristiwa. Simpan kunci yang telah diproses dan lewati peristiwa duplikat.
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}Strategi Deduplicasi Percobaan Ulang
Selain kunci idempotensi, pertimbangkan jendela deduplikasi. Jika Anda menerima konten peristiwa yang sama dalam jangka waktu singkat, kemungkinan besar itu adalah percobaan ulang. Bandingkan nilai hash peristiwa untuk mendeteksi dan mengabaikan percobaan ulang.
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)Menjalankan Agen Secara Asinkron
Penangan webhook harus merespons dengan cepat (kurang dari 5 detik) dan memproses logika agen di latar belakang. Gunakan BackgroundTasks di FastAPI untuk menghindari batas waktu.
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'}Menguraikan Muatan yang Kompleks
Setiap layanan mengirim bentuk muatan yang berbeda. Tulis fungsi pengurai khusus untuk setiap layanan agar agen Anda selalu menerima objek peristiwa yang seragam.
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)Kode Respons Webhook Itu Penting
Kembalikan status HTTP yang benar. Kode 2xx menunjukkan kepada pengirim bahwa webhook diterima. Kode 4xx berarti kesalahan klien (muatan tidak valid). Kode 5xx atau batas waktu menyebabkan pengirim mencoba kembali.
- 200: Diterima dan diproses
- 202: Diterima untuk pemrosesan asinkron
- 400: Permintaan tidak valid (bidang tidak ada)
- 401: Tanda tangan tidak valid
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'})Menguji Webhook Secara Lokal
Gunakan ngrok untuk mengekspos server lokal Anda ke internet untuk pengujian. Jalankan ngrok http 8000 untuk mendapatkan URL publik yang membuat terowongan ke aplikasi FastAPI lokal Anda.
# 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()Pencatatan Peristiwa Webhook
Catat setiap webhook masuk beserta timestamp, sumber, jenis peristiwa, dan hasil pemrosesannya. Jejak audit ini penting untuk menelusuri kesalahan peristiwa yang terlewat atau masalah pemrosesan ganda.
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'}
)Pembatasan Laju Webhook Masuk
Lindungi titik akhir webhook agar tidak kewalahan dengan menerapkan pembatasan laju. Pustaka slowapi menambahkan pembatasan laju ke FastAPI dengan sedikit kode.
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')}Uji Pemahaman: Webhook
Uji pemahaman Anda tentang praktik terbaik webhook untuk agen.
Webhook dalam Produksi
Dalam lingkungan produksi, gabungkan semua pola: verifikasi tanda tangan, kunci idempotensi, pemrosesan di latar belakang, pencatatan terstruktur, dan pembatasan laju. Terapkan layanan di balik proksi terbalik seperti nginx untuk terminasi TLS dan perlindungan tambahan.
Pertanyaan yang Sering Diajukan
Apakah pelajaran “Menghubungkan Agen ke Webhook” gratis?
Ya — teks lengkap “Menghubungkan Agen ke Webhook” gratis dibaca di sini di web. Untuk praktiknya secara interaktif (editor kode bawaan dan tutor AI 24/7) dan buka sisa kursus AI Agents, upgrade ke CoddyKit PRO. Kursus AI Agents mencakup 4 pelajaran total.
Apa yang akan aku pelajari di “Menghubungkan Agen ke Webhook”?
Terima acara webhook dan picu alur kerja agen sebagai respons. Kamu berlatih AI Agents dengan kode praktik yang langsung kamu jalankan di browser, dan tutor AI 24/7 menjawab pertanyaanmu saat kamu mengerjakan pelajaran ini.
Apakah aku perlu pengalaman untuk memulai AI Agents?
Tidak diperlukan pengalaman sebelumnya. AI Agents di CoddyKit dirancang untuk pemula hingga pelajar tingkat lanjut, jadi kamu bisa memulai di sini atau dari awal dan belajar sesuai kecepatan kamu sendiri. Ini adalah pelajaran 2 dari 4.
Berapa lama pelajaran “Menghubungkan Agen ke Webhook” memakan waktu?
Sebagian besar pelajaran CoddyKit memakan waktu sekitar 5–10 menit. Setiap pelajaran ringkas dan interaktif, jadi kamu membuat kemajuan stabil dan melanjutkan dari tempat kamu tinggalkan di web dan aplikasi.
Bisakah aku menulis dan menjalankan kode dalam pelajaran AI Agents ini?
Ya. Setiap pelajaran AI Agents menyertakan editor kode bawaan, jadi kamu menulis dan menjalankan kode nyata langsung di browser dan mendapatkan umpan balik AI instan — tidak diperlukan penyiapan lokal.
Semua pelajaran dalam kursus ini
- Pola Agen Pemicu-Tindakan
- Menghubungkan Agen ke Webhook
- Agen Berbasis Penjadwalan dan Cron
- Membangun Pipeline Otomatisasi Multi-Aplikasi