Agenttien yhdistäminen webhookeihin
Webhook-tapahtumien vastaanottaminen ja agenttityönkulkujen käynnistäminen niiden perusteella
Agenttien yhdistäminen webhookeihin on ilmainen Tekoälyagentit-oppitunti CoddyKitissä. Tämä on oppitunti 2/4. Voit lukea koko oppitunnin alta ilmaiseksi ja harjoitella sen jälkeen käytännössä selaimessa sisäänrakennetulla koodieditorilla ja ympäri vuorokauden käytettävissä olevan tekoälytuutorin avulla. Oppitunti kuuluu Tekoälyagentit-oppimispolkuun, ja edistymisesi synkronoituu verkon ja CoddyKit-sovelluksen välillä. Tekoälyagentit-kurssilla on yhteensä 4 oppituntia.
Mikä on webhook?
Webhook on HTTP-takaisinkutsu. Kun ulkoisessa palvelussa tapahtuu tapahtuma, palvelu lähettää päätepisteeseenne POST-pyynnön, joka sisältää tapahtuman tiedot. Agentti käsittelee hyötykuorman ja toimii sen perusteella.
Webhookit ovat push-pohjaisia (tapahtumat saapuvat tapahtuessaan), kun taas pollingissa tietoja tarkistetaan toistuvasti.
FastAPI-webhook-päätepiste
FastAPI helpottaa webhook-vastaanottimen luomista. Määritelkää POST-reitti, jäsentäkää JSON-runko ja siirtäkää käsittely agenttilogiikalle.
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-allekirjoituksen tarkistaminen
Varmistakaa aina, että webhook-pyynnöt tulevat odotetulta lähettäjältä. Useimmat palvelut allekirjoittavat hyötykuormansa HMAC-SHA256:lla käyttäen jaettua salaisuutta. Hylätkää pyynnöt, joiden allekirjoitus ei kelpaa.
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')}Idempotenssiavaimet
Ulkoiset palvelut yrittävät usein toimittaa epäonnistuneen webhookin uudelleen. Idempotenssiavain on jokaisen tapahtuman mukana lähetettävä yksilöllinen tunnus. Tallentakaa käsitellyt avaimet ja ohittakaa kaksoiskappaleet.
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}Uudelleenyritysten deduplikointistrategia
Idempotenssiavainten lisäksi kannattaa käyttää deduplikointi-ikkunoita. Jos vastaanotatte saman tapahtuman sisällön lyhyen ajan kuluessa, kyseessä on todennäköisesti uudelleenyritys. Vertailkaa tapahtumien tiivisteitä havaitaksenne ja hylätäksenne uudelleenyritykset.
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)Agentin suorittaminen asynkronisesti
Webhook-käsittelijöiden tulee vastata nopeasti (alle viidessä sekunnissa) ja käsitellä agenttilogiikka taustalla. Käyttäkää FastAPIn BackgroundTasks-toimintoa aikakatkaisujen välttämiseksi.
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'}Monimutkaisten hyötykuormien jäsentäminen
Eri palvelut lähettävät rakenteeltaan erilaisia hyötykuormia. Kirjoittakaa jokaiselle palvelulle oma jäsennysfunktio, jotta agentti saa aina normalisoidun tapahtumaobjektin.
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)Webhookin vastauskoodit ovat tärkeitä
Palauttakaa oikea HTTP-tilakoodi. 2xx kertoo lähettäjälle, että webhook hyväksyttiin. 4xx tarkoittaa asiakasvirhettä (virheellinen hyötykuorma). 5xx tai aikakatkaisu saa lähettäjän yrittämään uudelleen.
- 200: Hyväksytty ja käsitelty
- 202: Hyväksytty asynkronista käsittelyä varten
- 400: Virheellinen pyyntö (puuttuvia kenttiä)
- 401: Virheellinen allekirjoitus
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'})Webhookien testaaminen paikallisesti
Käyttäkää ngrok-työkalua paikallisen palvelimenne julkaisemiseen internetiin testausta varten. Suorittakaa ngrok http 8000, niin saatte julkisen URL-osoitteen, joka tunneloituu paikalliseen FastAPI-sovellukseenne.
# 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-tapahtumien lokikirjaus
Kirjatkaa jokainen saapuva webhook lokiin yhdessä aikaleiman, lähteen, tapahtumatyypin ja käsittelytuloksen kanssa. Tämä tarkastusjälki on välttämätön ohitettujen tapahtumien tai päällekkäisen käsittelyn ongelmien selvittämisessä.
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'}
)Saapuvien webhookien nopeusrajoitus
Suojatkaa webhook-päätepisteenne ylikuormitukselta nopeusrajoituksella. slowapi-kirjasto lisää FastAPIin nopeusrajoituksen vähäisellä koodimäärällä.
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')}Tietotesti: webhookit
Testatkaa, miten hyvin ymmärrätte agenttien webhookien parhaat käytännöt.
Webhookit tuotannossa
Tuotannossa yhdistäkää kaikki mallit: allekirjoitusten tarkistaminen, idempotenssiavaimet, taustakäsittely, rakenteinen lokikirjaus ja nopeusrajoitus. Ottakaa käyttöön nginxin kaltainen käänteisvälityspalvelin TLS:n päättämistä ja lisäsuojausta varten.
Opi Tekoälyagentit tekoälytuutorin avulla — ilmaiseksi
Kirjoita ja suorita oikeaa koodia selaimessa, saa välitöntä apua tekoälytuutorilta ympäri vuorokauden ja jatka siitä, mihin jäit, verkossa tai sovelluksessa.
- Kurssit
- 60
- Oppitunnit
- 239
Usein kysytyt kysymykset
Onko oppitunti ”Agenttien yhdistäminen webhookeihin” ilmainen?
Kyllä – oppitunnin ”Agenttien yhdistäminen webhookeihin” koko tekstin voi lukea täällä verkossa ilmaiseksi. Jos haluat harjoitella interaktiivisesti sisäänrakennetulla koodieditorilla ja ympäri vuorokauden käytettävissä olevan tekoälytuutorin avulla sekä avata koko Tekoälyagentit-kurssin, päivitä CoddyKit PROhon. Tekoälyagentit-kurssilla on yhteensä 4 oppituntia.
Mitä opin oppitunnilla ”Agenttien yhdistäminen webhookeihin”?
Webhook-tapahtumien vastaanottaminen ja agenttityönkulkujen käynnistäminen niiden perusteella Harjoittelet Tekoälyagentit-aihetta koodilla, jonka suoritat suoraan selaimessa. Ympäri vuorokauden käytettävissä oleva tekoälytuutori vastaa kysymyksiisi oppitunnin aikana.
Tarvitsenko kokemusta aloittaakseni Tekoälyagentit-opiskelun?
Aiempi kokemus ei ole tarpeen. CoddyKitin Tekoälyagentit-oppimispolku sopii vasta-alkajista edistyneisiin, joten voit aloittaa tästä tai alusta ja edetä omaan tahtiisi. Tämä on oppitunti 2/4.
Kuinka kauan ”Agenttien yhdistäminen webhookeihin”-oppitunnin suorittaminen kestää?
Useimmat CoddyKitin oppitunnit kestävät noin 5–10 minuuttia. Jokainen oppitunti on lyhyt ja interaktiivinen, joten edistyt tasaisesti ja voit jatkaa siitä, mihin jäit – sekä verkossa että sovelluksessa.
Voinko kirjoittaa ja suorittaa koodia tällä Tekoälyagentit-oppitunnilla?
Kyllä. Jokainen Tekoälyagentit-oppitunti sisältää sisäänrakennetun koodieditorin, joten voit kirjoittaa ja suorittaa oikeaa koodia suoraan selaimessa ja saada välitöntä palautetta tekoälyltä – paikallista asennusta ei tarvita.
Kaikki tämän kurssin oppitunnit
- Laukaisin–toiminto-agenttien mallit
- Agenttien yhdistäminen webhookeihin
- Ajastus- ja cron-pohjaiset agentit
- Monisovellusten automaatioputken rakentaminen