ربط الوكلاء بـ Webhooks

ما هو Webhook؟

نقطة نهاية Webhook في FastAPI

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

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

مفاتيح عدم تكرار التنفيذ

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}

استراتيجية إزالة تكرار عمليات إعادة المحاولة

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)

تشغيل الوكيل بشكل غير متزامن

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

تحليل الحمولات المعقدة

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

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 محليًا

# 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

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

تحديد معدل Webhooks الواردة

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

اختبار المعرفة: Webhooks

Webhooks في بيئة الإنتاج