AI-agenter · Lektion

Anslut agenter till webhooks

Ta emot webhook-händelser och utlösa agentarbetsflöden som svar.

Lektion 2 av 413 steg

Anslut agenter till webhooks är en gratis lektion i AI-agenter på CoddyKit. Detta är lektion 2 av 4. Ni kan läsa hela lektionen gratis nedan och sedan öva praktiskt i webbläsaren med en inbyggd kodredigerare och en AI-handledare som är tillgänglig dygnet runt. Den ingår i lärvägen för AI-agenter, och Era framsteg synkroniseras mellan webben och CoddyKit-appen. Kursen i AI-agenter innehåller totalt 4 lektioner.

Vad är en webhook?

En webhook är en HTTP-callback. När en händelse inträffar i en extern tjänst skickar den en POST-begäran till er endpoint med händelsedata. Er agent bearbetar payloaden och vidtar en åtgärd.

Webhooks är push-baserade (händelser anländer när de inträffar), till skillnad från polling (där ni kontrollerar upprepade gånger).

FastAPI-webhookendpoint

FastAPI gör det enkelt att skapa en webhook-mottagare. Definiera en POST-rutt, tolka JSON-kroppen och lämna över till agentlogiken.

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")}')

Verifiering av webhook-signatur

Kontrollera alltid att webhook-begäranden kommer från den förväntade avsändaren. De flesta tjänster signerar sina payloadar med HMAC-SHA256 och en delad hemlighet. Avvisa begäranden med ogiltiga signaturer.

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')}

Idempotensnycklar

Externa tjänster försöker ofta leverera misslyckade webhookar igen. En idempotensnyckel är ett unikt ID som skickas med varje händelse. Lagra redan behandlade nycklar och hoppa över dubbletter.

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 för deduplicering av återförsök

Utöver idempotensnycklar bör ni överväga dedupliceringsfönster. Om ni tar emot samma händelseinnehåll inom ett kort tidsfönster är det sannolikt ett återförsök. Jämför händelsehashar för att upptäcka och avfärda återförsök.

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)

Köra agenten asynkront

Webhook-hanterare bör svara snabbt (under 5 sekunder) och bearbeta agentlogiken i bakgrunden. Använd BackgroundTasks i FastAPI för att undvika tidsgränser.

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'}

Tolka komplexa payloadar

Olika tjänster skickar payloadar med olika struktur. Skriv särskilda parserfunktioner för varje tjänst så att agenten alltid tar emot ett normaliserat händelseobjekt.

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)

HTTP-svarskoder spelar roll

Returnera rätt HTTP-statuskod. En 2xx-statuskod meddelar avsändaren att webhooken har accepterats. En 4xx-statuskod betyder ett klientfel (felaktig payload). En 5xx-statuskod eller en timeout gör att avsändaren försöker igen.

200: Accepterad och bearbetad
202: Accepterad för asynkron bearbetning
400: Felaktig begäran (fält saknas)
401: Felaktig signatur

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'})

Testa webhooks lokalt

Använd ngrok för att exponera er lokala server mot internet vid testning. Kör ngrok http 8000 för att få en offentlig URL som tunnlar till er lokala FastAPI-app.

# 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()

Loggning av webhook-händelser

Logga varje inkommande webhook med tidsstämpel, källa, händelsetyp och bearbetningsresultat. Detta revisionsspår är avgörande för felsökning av missade händelser eller problem med dubbel bearbetning.

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'}
)

Hastighetsbegränsning av inkommande webhooks

Skydda er webhook-endpoint från att överbelastas genom att använda hastighetsbegränsning. Biblioteket slowapi lägger till hastighetsbegränsning i FastAPI med minimal kod.

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')}

Kunskapskontroll: Webhooks

Testa er förståelse av bästa praxis för webhooks i agenter.

Besvara frågan i den interaktiva lektionen

Webhooks i produktion

I produktion bör ni kombinera alla mönster: signaturverifiering, idempotensnycklar, bakgrundsbehandling, strukturerad loggning och hastighetsbegränsning. Distribuera bakom en reverse proxy som nginx för TLS-terminering och ytterligare skydd.

Gratis att börja

Lär dig AI-agenter med en AI-lärare – gratis

Skriv och kör riktig kod i webbläsaren, få omedelbar hjälp av en AI-lärare dygnet runt och fortsätt där du slutade – på webben eller i appen.

Kurser: 60
Lektioner: 239

Börja lära dig gratis Visa hela kursen

Vanliga frågor

Är lektionen ”Anslut agenter till webhooks” gratis?

Ja – hela texten till ”Anslut agenter till webhooks” kan läsas gratis här på webben. Om Ni vill öva interaktivt med en inbyggd kodredigerare och en AI-handledare som är tillgänglig dygnet runt och låsa upp resten av kursen i AI-agenter, kan Ni uppgradera till CoddyKit PRO. Kursen i AI-agenter innehåller totalt 4 lektioner.

Vad lär jag mig i ”Anslut agenter till webhooks”?

Ta emot webhook-händelser och utlösa agentarbetsflöden som svar. Ni övar på AI-agenter med praktisk kod som körs direkt i webbläsaren, medan en AI-handledare som är tillgänglig dygnet runt svarar på Era frågor under lektionen.

Behöver jag någon erfarenhet för att börja lära mig AI-agenter?

Du behöver inga förkunskaper. Utbildningen i AI-agenter på CoddyKit är upplagd för allt från nybörjare till avancerade elever, så att du kan börja här eller från början och gå fram i din egen takt. Detta är lektion 2 av 4.

Hur lång tid tar lektionen ”Anslut agenter till webhooks”?

De flesta CoddyKit-lektioner tar cirka 5–10 minuter. Varje lektion är kort och interaktiv, så att du gör stadiga framsteg och kan fortsätta precis där du slutade – på webben eller i appen.

Kan jag skriva och köra kod i den här AI-agenter-lektionen?

Ja. Varje AI-agenter-lektion innehåller en inbyggd kodredigerare, så att du kan skriva och köra riktig kod direkt i webbläsaren och få omedelbar AI-feedback – utan lokal installation.

Alla lektioner i den här kursen

Mönster för trigger–åtgärd-agenter
Anslut agenter till webhooks
Schemalagda och cron-baserade agenter
Bygga en automatiseringspipeline för flera appar

← Tillbaka till AI-agenter