0Pricing
AI Agents · Lektion

Agenten mit Webhooks verbinden

Webhook-Ereignisse empfangen und daraufhin Agenten-Workflows auslösen.

Agenten mit Webhooks verbinden ist eine kostenlose AI Agents-Lektion auf CoddyKit. Dies ist Lektion 2 von 4. Du kannst die komplette Lektion unten kostenlos lesen – dann übst du sie direkt im Browser mit einem integrierten Code-Editor und einem KI-Tutor rund um die Uhr. Sie ist Teil des AI Agents-Lernpfads, und dein Fortschritt wird über Web und CoddyKit-App synchronisiert. Der AI Agents-Kurs umfasst insgesamt 4 Lektionen.

Was ist ein Webhook?

Ein Webhook ist ein HTTP-Callback. Wenn ein Ereignis in einem externen Dienst eintritt, sendet dieser eine POST-Anfrage mit den Ereignisdaten an Ihren Endpunkt. Ihr Agent verarbeitet den Payload und reagiert darauf.

Webhooks sind Push-basiert (Ereignisse treffen ein, sobald sie auftreten), während Polling bedeutet, dass Sie wiederholt prüfen.

FastAPI-Webhook-Endpunkt

Mit FastAPI lässt sich problemlos ein Webhook-Empfänger erstellen. Definieren Sie eine POST-Route, analysieren Sie den JSON-Body und übergeben Sie die Verarbeitung an Ihre Agentenlogik.

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

Überprüfung von Webhook-Signaturen

Überprüfen Sie stets, ob Webhook-Anfragen vom erwarteten Absender stammen. Die meisten Dienste signieren ihre Payloads mit HMAC-SHA256 unter Verwendung eines gemeinsamen geheimen Schlüssels. Lehnen Sie Anfragen mit ungültigen Signaturen ab.

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

Idempotenzschlüssel

Externe Dienste versuchen häufig, fehlgeschlagene Webhook-Zustellungen erneut zu senden. Ein Idempotenzschlüssel ist eine eindeutige ID, die mit jedem Ereignis gesendet wird. Speichern Sie bereits verarbeitete Schlüssel und überspringen Sie Duplikate.

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}

Strategie zur Deduplizierung von Wiederholungen

Berücksichtigen Sie zusätzlich zu Idempotenzschlüsseln Zeitfenster für die Deduplizierung. Wenn Sie innerhalb eines kurzen Zeitfensters denselben Ereignisinhalt empfangen, handelt es sich wahrscheinlich um eine Wiederholung. Vergleichen Sie Ereignis-Hashes, um Wiederholungen zu erkennen und zu verwerfen.

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)

Asynchrone Ausführung des Agenten

Webhook-Handler sollten schnell antworten (in weniger als 5 Sekunden) und die Agentenlogik im Hintergrund verarbeiten. Verwenden Sie BackgroundTasks in FastAPI, um Timeouts zu vermeiden.

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

Komplexe Payloads analysieren

Verschiedene Dienste senden Payloads in unterschiedlichen Strukturen. Schreiben Sie für jeden Dienst eigene Parserfunktionen, damit Ihr Agent stets ein normalisiertes Ereignisobjekt erhält.

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)

Webhook-Antwortcodes sind wichtig

Geben Sie den korrekten HTTP-Status zurück. Ein 2xx-Status teilt dem Absender mit, dass der Webhook akzeptiert wurde. Ein 4xx-Status weist auf einen Clientfehler hin (fehlerhafter Payload). Ein 5xx-Status oder ein Timeout veranlasst den Absender zu einem erneuten Zustellversuch.

  • 200: Akzeptiert und verarbeitet
  • 202: Zur asynchronen Verarbeitung akzeptiert
  • 400: Ungültige Anfrage (fehlende Felder)
  • 401: Ungültige 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'})

Webhooks lokal testen

Verwenden Sie ngrok, um Ihren lokalen Server für Tests im Internet bereitzustellen. Führen Sie ngrok http 8000 aus, um eine öffentliche URL zu erhalten, die zu Ihrer lokalen FastAPI-App tunnelt.

# 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-Ereignisse protokollieren

Protokollieren Sie jeden eingehenden Webhook mit Zeitstempel, Quelle, Ereignistyp und Verarbeitungsergebnis. Dieser Prüfpfad ist für die Fehlersuche bei verpassten Ereignissen oder Problemen mit doppelter Verarbeitung unverzichtbar.

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

Rate-Limiting für eingehende Webhooks

Schützen Sie Ihren Webhook-Endpunkt mit Rate-Limiting davor, überlastet zu werden. Die Bibliothek slowapi fügt FastAPI mit minimalem Code Rate-Limiting hinzu.

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

Wissensüberprüfung: Webhooks

Testen Sie Ihr Verständnis der Best Practices für Webhooks in Agenten.

Webhooks in der Produktion

In der Produktion sollten Sie alle Muster kombinieren: Signaturüberprüfung, Idempotenzschlüssel, Hintergrundverarbeitung, strukturiertes Logging und Rate-Limiting. Stellen Sie die Anwendung hinter einem Reverse Proxy wie nginx bereit, um die TLS-Terminierung und zusätzlichen Schutz zu ermöglichen.

Häufig gestellte Fragen

Ist die Lektion „Agenten mit Webhooks verbinden“ kostenlos?

Ja — der vollständige Text von „Agenten mit Webhooks verbinden“ ist hier im Web kostenlos zu lesen. Um sie interaktiv zu üben (integrierter Code-Editor und 24/7 KI-Tutor) und den Rest des AI Agents-Kurses freizuschalten, upgrade auf CoddyKit PRO. Der AI Agents-Kurs umfasst insgesamt 4 Lektionen.

Was lerne ich in „Agenten mit Webhooks verbinden“?

Webhook-Ereignisse empfangen und daraufhin Agenten-Workflows auslösen. Du übst AI Agents mit praktischem Code, den du direkt im Browser ausführst, und ein 24/7 KI-Tutor beantwortet deine Fragen während du die Lektion bearbeitest.

Brauche ich Erfahrung, um AI Agents zu starten?

Keine Vorkenntnisse erforderlich. AI Agents auf CoddyKit ist für Anfänger bis fortgeschrittene Lernende strukturiert, sodass du hier starten oder von Anfang an beginnen und in deinem eigenen Tempo voranschreiten kannst. Dies ist Lektion 2 von 4.

Wie lange dauert die Lektion „Agenten mit Webhooks verbinden“?

Die meisten CoddyKit-Lektionen dauern etwa 5–10 Minuten. Jede ist kompakt und interaktiv, sodass du stetig Fortschritte machst und genau dort weitermachst, wo du aufgehört hast – im Web und in der App.

Kann ich in dieser AI Agents-Lektion Code schreiben und ausführen?

Ja. Jede AI Agents-Lektion enthält einen integrierten Code-Editor, sodass du echten Code direkt in deinem Browser schreibst und ausführst und sofort KI-Feedback erhältst — ohne lokale Einrichtung erforderlich.

Alle Lektionen in diesem Kurs

  1. Muster für Trigger-Action-Agenten
  2. Agenten mit Webhooks verbinden
  3. Zeitplanung und Cron-basierte Agenten
  4. Eine Multi-App-Automatisierungspipeline erstellen
← Zurück zu AI Agents