Menyambungkan Ejen kepada Webhook
Menerima acara webhook dan mencetuskan aliran kerja ejen sebagai respons.
Menyambungkan Ejen kepada Webhook ialah pelajaran Ejen AI percuma di CoddyKit. Ini ialah pelajaran 2 daripada 4. Anda boleh membaca keseluruhan pelajaran di bawah secara percuma — kemudian berlatih secara praktikal dalam pelayar menggunakan penyunting kod terbina dalam dan tutor kecerdasan buatan 24/7. Pelajaran ini merupakan sebahagian daripada laluan pembelajaran Ejen AI, dan kemajuan anda disegerakkan merentas web serta aplikasi CoddyKit. Kursus Ejen AI merangkumi sejumlah 4 pelajaran.
Apakah Panggilan Balik Web?
Panggilan balik web ialah panggilan balik HTTP. Apabila sesuatu peristiwa berlaku dalam perkhidmatan luaran, perkhidmatan itu menghantar permintaan POST kepada titik akhir anda bersama data peristiwa. Ejen anda memproses muatan tersebut dan bertindak.
Panggilan balik web berasaskan penghantaran segera (peristiwa tiba apabila berlaku), berbeza daripada tinjauan (anda menyemak berulang kali).
Titik Akhir Panggilan Balik Web FastAPI
FastAPI memudahkan penciptaan penerima panggilan balik web. Tentukan laluan POST, huraikan isi JSON dan serahkan tugas kepada logik ejen 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")}')Pengesahan Tandatangan Panggilan Balik Web
Sentiasa sahkan bahawa permintaan panggilan balik web datang daripada penghantar yang dijangka. Kebanyakan perkhidmatan menandatangani muatannya dengan HMAC-SHA256 menggunakan rahsia yang dikongsi. Tolak permintaan yang mempunyai tandatangan tidak sah.
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
Perkhidmatan luaran sering mencuba semula penghantaran panggilan balik web yang gagal. Kunci idempotensi ialah ID unik yang dihantar bersama setiap peristiwa. Simpan kunci yang telah diproses dan langkau pendua.
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 Penyahduplikatan Percubaan Semula
Selain kunci idempotensi, pertimbangkan tetingkap penyahduplikatan. Jika anda menerima kandungan peristiwa yang sama dalam tempoh yang singkat, kemungkinan besar itu ialah percubaan semula. Bandingkan nilai cincangan peristiwa untuk mengesan dan membuang percubaan semula.
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 Ejen Secara Tak Segerak
Pengendali panggilan balik web hendaklah memberikan respons dengan pantas (kurang daripada 5 saat) dan memproses logik ejen di latar belakang. Gunakan BackgroundTasks dalam FastAPI untuk mengelakkan tamat masa.
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'}Menghuraikan Muatan Kompleks
Perkhidmatan yang berbeza menghantar bentuk muatan yang berbeza. Tulis fungsi penghuraian khusus untuk setiap perkhidmatan supaya ejen anda sentiasa menerima objek peristiwa yang ternormal.
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)Kod Respons Panggilan Balik Web Penting
Kembalikan status HTTP yang betul. Kod 2xx memberitahu penghantar bahawa panggilan balik web telah diterima. Kod 4xx bermaksud ralat klien (muatan tidak sah). Kod 5xx atau tamat masa menyebabkan penghantar mencuba semula.
- 200: Diterima dan diproses
- 202: Diterima untuk pemprosesan tak segerak
- 400: Permintaan tidak sah (medan tiada)
- 401: Tandatangan tidak sah
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 Panggilan Balik Web Secara Setempat
Gunakan ngrok untuk mendedahkan pelayan setempat anda kepada Internet bagi tujuan pengujian. Jalankan ngrok http 8000 untuk mendapatkan URL awam yang membuat terowong ke aplikasi FastAPI setempat 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()Pengelogan Peristiwa Panggilan Balik Web
Catat setiap panggilan balik web yang masuk bersama timestamp, sumber, jenis peristiwa dan hasil pemprosesan. Jejak audit ini penting untuk menyahpepijat peristiwa yang terlepas atau masalah pemprosesan pendua.
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'}
)Pengehadan Kadar Panggilan Balik Web Masuk
Lindungi titik akhir panggilan balik web daripada dibebani dengan menggunakan pengehadan kadar. Pustaka slowapi menambahkan pengehadan kadar pada FastAPI dengan kod yang minimum.
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')}Semakan Pengetahuan: Panggilan Balik Web
Uji pemahaman anda tentang amalan terbaik panggilan balik web untuk ejen.
Panggilan Balik Web dalam Persekitaran Pengeluaran
Dalam persekitaran pengeluaran, gabungkan semua corak: pengesahan tandatangan, kunci idempotensi, pemprosesan latar belakang, pengelogan berstruktur dan pengehadan kadar. Gunakan proksi songsang seperti nginx untuk penamatan TLS dan perlindungan tambahan.
Pelajari Ejen AI dengan tutor kecerdasan buatan — percuma
Tulis dan jalankan kod sebenar dalam pelayar anda, dapatkan bantuan segera daripada tutor kecerdasan buatan yang tersedia 24/7, dan sambung semula dari tempat anda berhenti di web atau dalam aplikasi.
- Kursus
- 60
- Pelajaran
- 239
Soalan Lazim
Adakah pelajaran “Menyambungkan Ejen kepada Webhook” percuma?
Ya — teks penuh “Menyambungkan Ejen kepada Webhook” boleh dibaca secara percuma di web ini. Untuk berlatih secara interaktif menggunakan penyunting kod terbina dalam dan tutor kecerdasan buatan 24/7, serta membuka kunci baki kursus Ejen AI, tingkat taraf kepada CoddyKit PRO. Kursus Ejen AI merangkumi sejumlah 4 pelajaran.
Apakah yang akan saya pelajari dalam “Menyambungkan Ejen kepada Webhook”?
Menerima acara webhook dan mencetuskan aliran kerja ejen sebagai respons. Anda berlatih Ejen AI menggunakan kod praktikal yang dijalankan terus dalam pelayar, manakala tutor kecerdasan buatan 24/7 menjawab soalan anda semasa anda mengikuti pelajaran.
Adakah saya memerlukan pengalaman untuk memulakan Ejen AI?
Tiada pengalaman terdahulu diperlukan. Pembelajaran Ejen AI di CoddyKit disusun untuk pelajar daripada peringkat pemula hingga lanjutan, jadi anda boleh bermula di sini atau dari awal dan belajar mengikut kadar anda sendiri. Ini ialah pelajaran 2 daripada 4.
Berapa lamakah pelajaran “Menyambungkan Ejen kepada Webhook” diambil?
Kebanyakan pelajaran CoddyKit mengambil masa kira-kira 5–10 minit. Setiap pelajaran ringkas dan interaktif, jadi anda boleh membuat kemajuan secara berterusan dan menyambung tepat dari tempat anda berhenti di web atau aplikasi.
Bolehkah saya menulis dan menjalankan kod dalam pelajaran Ejen AI ini?
Ya. Setiap pelajaran Ejen AI menyertakan penyunting kod terbina dalam, jadi anda boleh menulis dan menjalankan kod sebenar terus dalam pelayar serta menerima maklum balas kecerdasan buatan serta-merta — tanpa memerlukan persediaan setempat.
Semua pelajaran dalam kursus ini
- Corak Ejen Pencetus-Tindakan
- Menyambungkan Ejen kepada Webhook
- Ejen Berjadual dan Berasaskan Cron
- Membina Saluran Automasi Berbilang Aplikasi