AI Agents · Ders

Aracıları Web Kancalarına Bağlama

Web kancası etkinliklerini alma ve karşılığında aracı iş akışlarını tetikleme.

2. ders / 413 adım

Aracıları Web Kancalarına Bağlama, CoddyKit'te ücretsiz bir AI Agents dersidir. Bu, 4 dersinin 2. dersidir. Aşağıdan dersin tamamını ücretsiz okuyabilir, sonra tarayıcıda yerleşik kod editörü ve 7/24 yapay zeka koçu ile uygulamalı olarak pratik yapabilirsin. Bu, AI Agents öğrenme yolunun bir parçasıdır ve ilerlemeniz web ve CoddyKit uygulaması arasında senkronize olur. AI Agents kursu toplamda 4 dersten oluşur.

Web Kancası Nedir?

web kancası, bir HTTP geri çağrısıdır. Harici bir hizmette olay meydana geldiğinde hizmet, olay verilerini içeren bir POST isteğini uç noktanıza gönderir. Ajanınız yükü işler ve eyleme geçer.

Web kancaları, düzenli sorgulamanın aksine gönderim tabanlıdır; olaylar gerçekleştiğinde gelir, siz de hizmeti tekrar tekrar kontrol etmezsiniz.

FastAPI Web Kancası Uç Noktası

FastAPI, bir web kancası alıcısı oluşturmayı kolaylaştırır. Bir POST rotası tanımlayın, JSON gövdesini ayrıştırın ve işlemi ajan mantığınıza devredin.

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

Web Kancası İmzası Doğrulama

Web kancası isteklerinin beklenen göndericiden geldiğini her zaman doğrulayın. Çoğu hizmet, yüklerini paylaşılan bir gizli anahtar kullanarak HMAC-SHA256 ile imzalar. Geçersiz imzaları içeren istekleri reddedin.

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

İdempotentlik Anahtarları

Harici hizmetler, başarısız web kancası teslimlerini çoğu zaman yeniden dener. idempotentlik anahtarı, her olayla birlikte gönderilen benzersiz bir ID'dir. İşlenen anahtarları saklayın ve yinelenenleri atlayın.

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}

Yeniden Denemeleri Tekilleştirme Stratejisi

İdempotentlik anahtarlarının yanı sıra tekilleştirme pencerelerini de değerlendirin. Aynı olay içeriğini kısa bir süre içinde alırsanız bunun bir yeniden deneme olma olasılığı yüksektir. Yeniden denemeleri algılayıp kaldırmak için olay karmalarını karşılaştırın.

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)

Ajanı Eşzamansız Çalıştırma

Web kancası işleyicileri hızlı yanıt vermeli (5 saniyenin altında) ve ajan mantığını arka planda işlemelidir. Zaman aşımlarını önlemek için FastAPI'de BackgroundTasks kullanın.

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

Karmaşık Yükleri Ayrıştırma

Farklı hizmetler farklı yük biçimleri gönderir. Ajanınızın her zaman normalleştirilmiş bir olay nesnesi alması için her hizmete özel ayrıştırıcı işlevler yazın.

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)

Web Kancası Yanıt Kodları Önemlidir

Doğru HTTP durumunu döndürün. 2xx, göndericiye web kancasının kabul edildiğini bildirir. 4xx, istemci hatası olduğunu gösterir (hatalı yük). 5xx veya zaman aşımı, göndericinin yeniden denemesine neden olur.

200: Kabul edildi ve işlendi
202: Eşzamansız işleme için kabul edildi
400: Hatalı istek (eksik alanlar)
401: Hatalı imza

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

Web Kancalarını Yerel Olarak Sınama

Yerel sunucunuzu sınama amacıyla internete açmak için ngrok kullanın. Yerel FastAPI uygulamanıza tünel oluşturan genel erişime açık bir URL almak için ngrok http 8000 komutunu çalıştırın.

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

Web Kancası Olaylarının Günlük Kaydı

Gelen her web kancasını timestamp, kaynak, olay türü ve işleme sonucuyla birlikte günlüğe kaydedin. Bu denetim izi, kaçırılan olaylarda veya yinelenen işleme sorunlarında hata ayıklamak için gereklidir.

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

Gelen Web Kancalarında Hız Sınırlama

Hız sınırlama kullanarak web kancası uç noktanızın aşırı yüklenmesini önleyin. slowapi kitaplığı, minimum kodla FastAPI'ye hız sınırlama ekler.

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

Bilgi Kontrolü: Web Kancaları

Ajanlar için web kancalarında en iyi uygulamaları anlayıp anlamadığınızı sınayın.

İnteraktif derste cevapla

Üretimde Web Kancaları

Üretimde tüm kalıpları bir arada kullanın: imza doğrulama, idempotentlik anahtarları, arka plan işleme, yapılandırılmış günlük kaydı ve hız sınırlama. TLS sonlandırması ve ek koruma için nginx gibi bir ters vekil sunucunun arkasına dağıtın.

Başlamak ücretsiz

Yapay zeka eğitmeniyle AI Agents öğren — ücretsiz

Tarayıcında gerçek kod yaz ve çalıştır, 7/24 yapay zeka eğitmeninden anında yardım al; web'de ya da uygulamada kaldığın yerden devam et.

Kurslar: 60
Dersler: 239

Ücretsiz öğrenmeye başla Tüm kursu görüntüle

Sıkça Sorulan Sorular

“Aracıları Web Kancalarına Bağlama” dersi ücretsiz mi?

Evet — “Aracıları Web Kancalarına Bağlama” dersin tüm metni burada web'de ücretsiz olarak okunabilir. Etkileşimli olarak pratik yapmak (yerleşik kod editörü ve 7/24 yapay zeka koçu) ve AI Agents kursunun geri kalanını açmak için CoddyKit PRO'ya yükselt. AI Agents kursu toplamda 4 dersten oluşur.

“Aracıları Web Kancalarına Bağlama” dersinde ne öğreneceğim?

Web kancası etkinliklerini alma ve karşılığında aracı iş akışlarını tetikleme. AI Agents ile uygulamalı kodu tarayıcıda doğrudan çalıştırarak pratik yaparsın ve 7/24 yapay zeka koçu dersi çalışırken sorularını yanıtlar.

AI Agents öğrenmeye başlamak için deneyim gerekli mi?

Önceden deneyim gerekmez. CoddyKit'te AI Agents, başlangıçtan ileri seviyeye kadar yapılandırıldığı için buradan başlayabilir veya başından başlayıp kendi hızında ilerleme yapabilirsin. Bu, 4 dersinin 2. dersidir.

“Aracıları Web Kancalarına Bağlama” dersi ne kadar sürer?

Çoğu CoddyKit dersi yaklaşık 5–10 dakika sürer. Her biri kısa ve etkileşimli olduğu için sabit ilerleme yaparsın ve web ile uygulama arasında tam olarak bıraktığın yerden devam edebilirsin.

Bu AI Agents dersinde kod yazıp çalıştırabilir miyim?

Evet. Her AI Agents dersi yerleşik bir kod editörü içerir, bu sayede tarayıcıda gerçek kod yazıp çalıştırabilir ve anlık yapay zeka geri bildirimi alırsın — yerel kurulum gerekli değildir.

Bu kursun tüm dersleri

Tetikleyici-Eylem Aracı Kalıpları
Aracıları Web Kancalarına Bağlama
Zamanlama ve Cron Tabanlı Aracılar
Birden Fazla Uygulama İçin Otomasyon Hattı Oluşturma

← AI Agents Sayfasına Dön