Connessione degli agenti ai webhook
Ricezione degli eventi webhook e attivazione dei workflow degli agenti in risposta
Connessione degli agenti ai webhook è una lezione AI Agents gratuita su CoddyKit. Questa è la lezione 2 di 4. Puoi leggere la lezione completa qui gratuitamente — poi esercitati direttamente nel browser con un editor di codice integrato e un tutor IA disponibile 24/7. Fa parte del percorso di apprendimento AI Agents, e i tuoi progressi si sincronizzano tra il web e l'app CoddyKit. Il corso AI Agents include 4 lezioni in totale.
Che cos'è un webhook
Un webhook è un callback HTTP. Quando si verifica un evento in un servizio esterno, il servizio invia una richiesta POST al Suo endpoint con i dati dell'evento. L'agente elabora il payload e agisce di conseguenza.
I webhook sono basati sul push (gli eventi arrivano quando si verificano), a differenza del polling (che consiste nel controllare ripetutamente).
Endpoint webhook con FastAPI
FastAPI semplifica la creazione di un ricevitore per webhook. Definisca una route POST, analizzi il corpo JSON e deleghi la gestione alla logica dell'agente.
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")}')Verifica della firma dei webhook
Verifichi sempre che le richieste webhook provengano dal mittente previsto. La maggior parte dei servizi firma i payload con HMAC-SHA256 utilizzando un segreto condiviso. Rifiuti le richieste con firme non valide.
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')}Chiavi di idempotenza
I servizi esterni spesso ritentano la consegna dei webhook non riuscita. Una chiave di idempotenza è un ID univoco inviato con ogni evento. Memorizzi le chiavi già elaborate e ignori i duplicati.
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}Strategia di deduplicazione dei retry
Oltre alle chiavi di idempotenza, consideri le finestre di deduplicazione. Se riceve lo stesso contenuto di un evento entro un breve intervallo, è probabile che si tratti di un retry. Confronti gli hash degli eventi per rilevare e scartare i retry.
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)Esecuzione asincrona dell'agente
I gestori dei webhook devono rispondere rapidamente (entro 5 secondi) ed elaborare la logica dell'agente in background. Utilizzi BackgroundTasks in FastAPI per evitare i timeout.
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'}Analisi di payload complessi
Servizi diversi inviano payload con strutture diverse. Scriva funzioni di analisi dedicate per ogni servizio, così l'agente riceverà sempre un oggetto evento normalizzato.
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)L'importanza dei codici di risposta dei webhook
Restituisca lo stato HTTP corretto. Un codice 2xx comunica al mittente che il webhook è stato accettato. Un codice 4xx indica un errore del client (payload non valido). Un codice 5xx o un timeout fanno sì che il mittente ritenti la consegna.
- 200: accettato ed elaborato
- 202: accettato per l'elaborazione asincrona
- 400: richiesta non valida (campi mancanti)
- 401: firma non valida
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'})Testare i webhook in locale
Utilizzi ngrok per esporre il server locale a Internet durante i test. Esegua ngrok http 8000 per ottenere un URL pubblico che crea un tunnel verso la Sua app FastAPI locale.
# 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()Registrazione degli eventi webhook
Registri ogni webhook in ingresso con timestamp, origine, tipo di evento e risultato dell'elaborazione. Questo audit trail è essenziale per eseguire il debug degli eventi persi o dei problemi causati dall'elaborazione duplicata.
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'}
)Limitazione della frequenza dei webhook in ingresso
Protegga l'endpoint webhook dal sovraccarico utilizzando una limitazione della frequenza. La libreria slowapi aggiunge il rate limiting a FastAPI con una quantità minima di codice.
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')}Verifica delle conoscenze: webhook
Verifichi la Sua comprensione delle best practice per gli agenti basati su webhook.
Webhook in produzione
In produzione, combini tutti i pattern: verifica della firma, chiavi di idempotenza, elaborazione in background, logging strutturato e limitazione della frequenza. Esegua il deployment dietro un reverse proxy come nginx per la terminazione TLS e una protezione aggiuntiva.
Impara AI Agents con un tutor IA — gratis
Scrivi ed esegui vero codice nel tuo browser, ricevi aiuto istantaneo da un tutor IA disponibile 24/7, e riprendi da dove hai lasciato sul web o nell'app.
- Corsi
- 60
- Lezioni
- 239
Domande Frequenti
La lezione «Connessione degli agenti ai webhook» è gratuita?
Sì — il testo completo di «Connessione degli agenti ai webhook» è gratuito qui sul web. Per esercitarvi in modo interattivo (un editor di codice integrato e un tutor IA 24/7) e sbloccare il resto del corso AI Agents, passa a CoddyKit PRO. Il corso AI Agents include 4 lezioni in totale.
Cosa imparerò in «Connessione degli agenti ai webhook»?
Ricezione degli eventi webhook e attivazione dei workflow degli agenti in risposta Eserciti AI Agents con codice pratico che esegui direttamente nel browser, e un tutor IA 24/7 risponde alle tue domande mentre lavori sulla lezione.
Ho bisogno di esperienza per iniziare AI Agents?
Non è richiesta alcuna esperienza precedente. AI Agents su CoddyKit è strutturato per principianti e studenti avanzati, quindi puoi iniziare da qui o dall'inizio e procedere al tuo ritmo. Questa è la lezione 2 di 4.
Quanto tempo richiede la lezione «Connessione degli agenti ai webhook»?
La maggior parte delle lezioni CoddyKit richiede circa 5–10 minuti. Ogni lezione è breve e interattiva, quindi fai progressi costanti e riprendi esattamente da dove hai lasciato su web e app.
Posso scrivere ed eseguire codice in questa lezione AI Agents?
Sì. Ogni lezione AI Agents include un editor di codice integrato, quindi scrivi ed esegui codice reale direttamente nel tuo browser e ricevi feedback istantaneo dall'IA — nessuna configurazione locale necessaria.
Tutte le lezioni di questo corso
- Pattern di agenti trigger-azione
- Connessione degli agenti ai webhook
- Agenti pianificati e basati su cron
- Creazione di una pipeline di automazione multi-app