0Pricing
AI Agents · Lekcja

Łączenie agentów z webhookami

Odbieranie zdarzeń webhooków i uruchamianie w odpowiedzi przepływów pracy agenta.

Łączenie agentów z webhookami to bezpłatna lekcja AI Agents na CoddyKit. To lekcja 2 z 4. Możesz przeczytać całą lekcję poniżej za darmo — a potem ćwiczyć ją interaktywnie w przeglądarce z wbudowanym edytorem kodu i tutorem AI dostępnym 24/7. To część ścieżki edukacyjnej AI Agents, a Twój postęp synchronizuje się między webem a aplikacją CoddyKit. Kurs AI Agents zawiera 4 lekcji w sumie.

Czym jest webhook?

Webhook to wywołanie zwrotne HTTP. Gdy w zewnętrznej usłudze wystąpi zdarzenie, wysyła ona żądanie POST do endpointu wraz z danymi zdarzenia. Agent przetwarza payload i wykonuje działanie.

Webhooki działają w modelu push (zdarzenia docierają w chwili wystąpienia), w przeciwieństwie do odpytywania (wielokrotnego sprawdzania).

Endpoint webhooka w FastAPI

FastAPI ułatwia utworzenie odbiornika webhooków. Należy zdefiniować trasę POST, przetworzyć ciało JSON i przekazać je do logiki agenta.

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

Weryfikowanie podpisu webhooka

Należy zawsze sprawdzać, czy żądania webhooka pochodzą od oczekiwanego nadawcy. Większość usług podpisuje payloady za pomocą HMAC-SHA256 z użyciem wspólnego sekretu. Należy odrzucać żądania z nieprawidłowymi podpisami.

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

Klucze idempotencji

Zewnętrzne usługi często ponawiają nieudane dostarczanie webhooków. Klucz idempotencji to unikatowy identyfikator wysyłany z każdym zdarzeniem. Należy przechowywać przetworzone klucze i pomijać duplikaty.

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 deduplikacji ponowień

Oprócz kluczy idempotencji należy rozważyć okna deduplikacji. Jeśli w krótkim odstępie czasu zostanie odebrana ta sama treść zdarzenia, prawdopodobnie jest to ponowienie. Należy porównywać skróty zdarzeń, aby wykrywać i odrzucać ponowienia.

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)

Asynchroniczne uruchamianie agenta

Obsługa webhooka powinna szybko zwracać odpowiedź (w czasie poniżej 5 sekund), a logikę agenta przetwarzać w tle. Należy użyć BackgroundTasks w FastAPI, aby uniknąć przekroczenia limitu czasu.

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

Parsowanie złożonych payloadów

Różne usługi wysyłają payloady o różnej strukturze. Należy napisać osobne funkcje parsera dla każdej usługi, aby agent zawsze otrzymywał znormalizowany obiekt zdarzenia.

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)

Kody odpowiedzi webhooka mają znaczenie

Należy zwracać właściwy status HTTP. Kod 2xx informuje nadawcę, że webhook został przyjęty. Kod 4xx oznacza błąd po stronie klienta (nieprawidłowy payload). Kod 5xx lub przekroczenie limitu czasu powoduje ponowienie próby przez nadawcę.

  • 200: Przyjęto i przetworzono
  • 202: Przyjęto do przetwarzania asynchronicznego
  • 400: Nieprawidłowe żądanie (brakujące pola)
  • 401: Nieprawidłowy podpis
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'})

Testowanie webhooków lokalnie

Należy użyć ngrok, aby na potrzeby testów udostępnić lokalny serwer w internecie. Uruchomienie ngrok http 8000 zapewnia publiczny adres URL, który tworzy tunel do lokalnej aplikacji FastAPI.

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

Rejestrowanie zdarzeń webhooków

Należy rejestrować każdy przychodzący webhook wraz ze znacznikiem czasu, źródłem, typem zdarzenia i wynikiem przetwarzania. Taki ślad audytowy jest niezbędny do debugowania problemów z pominiętymi zdarzeniami lub duplikowaniem przetwarzania.

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

Ograniczanie liczby żądań przychodzących webhooków

Należy chronić endpoint webhooka przed przeciążeniem, stosując ograniczanie liczby żądań. Biblioteka slowapi dodaje ograniczanie liczby żądań do FastAPI przy użyciu niewielkiej ilości kodu.

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

Sprawdzenie wiedzy: webhooki

Sprawdź swoją wiedzę na temat najlepszych praktyk dotyczących webhooków w agentach.

Webhooki na produkcji

W środowisku produkcyjnym należy połączyć wszystkie wzorce: weryfikowanie podpisu, klucze idempotencji, przetwarzanie w tle, ustrukturyzowane rejestrowanie i ograniczanie liczby żądań. Aplikację należy wdrożyć za serwerem reverse proxy, takim jak nginx, aby zapewnić terminację TLS i dodatkową ochronę.

Często zadawane pytania

Czy lekcja „Łączenie agentów z webhookami” jest bezpłatna?

Tak — pełny tekst „Łączenie agentów z webhookami” jest dostępny za darmo tutaj w sieci. Aby ćwiczyć ją interaktywnie (wbudowany edytor kodu i tutor AI dostępny 24/7) i odblokować resztę kursu AI Agents, przejdź na CoddyKit PRO. Kurs AI Agents zawiera 4 lekcji w sumie.

Co nauczysz się w „Łączenie agentów z webhookami”?

Odbieranie zdarzeń webhooków i uruchamianie w odpowiedzi przepływów pracy agenta. Ćwiczysz AI Agents z praktycznym kodem, który uruchamiasz bezpośrednio w przeglądarce, a tutor AI dostępny 24/7 odpowiada na Twoje pytania podczas pracy nad lekcją.

Czy potrzebuję doświadczenia, aby zacząć AI Agents?

Nie wymagamy żadnego doświadczenia. AI Agents w CoddyKit jest strukturyzowany dla początkujących i zaawansowanych użytkowników, więc możesz zacząć tutaj lub od początku i uczyć się w swoim tempie. To lekcja 2 z 4.

Ile czasu zajmuje lekcja „Łączenie agentów z webhookami”?

Większość lekcji CoddyKit trwa około 5–10 minut. Każda lekcja to mały, interaktywny krok, dzięki czemu robisz systematyczne postępy i zawsze wracasz dokładnie do tego samego miejsca — na webie i w aplikacji.

Czy mogę pisać i uruchamiać kod w tej lekcji AI Agents?

Tak. Każda lekcja AI Agents zawiera wbudowany edytor kodu, więc piszesz i uruchamiasz prawdziwy kod bezpośrednio w przeglądarce i od razu otrzymujesz sprzężenie zwrotne od AI — bez konfiguracji na komputerze.

Wszystkie lekcje w tym kursie

  1. Wzorce agentów trigger-action
  2. Łączenie agentów z webhookami
  3. Agenci planowani i oparte na Cronie
  4. Budowanie potoku automatyzacji obejmującego wiele aplikacji
← Powrót do AI Agents