Best Practices

1
from extend_ai import Extend
2
from extend_ai.wrapper.errors import WebhookSignatureVerificationError
3
4
client = Extend(token="YOUR_API_KEY")
5
6
@app.post('/webhook')
7
async def handle_webhook(request):
8
    try:
9
        event = client.webhooks.verify_and_parse(
10
            body=request.body.decode(),
11
            headers=dict(request.headers),
12
            signing_secret="wss_your_signing_secret"
13
        )
14
        
15
        # Queue the event for async processing, then respond immediately
16
        await message_queue.send({
17
            "type": "webhook-event",
18
            "payload": event
19
        })
20
        
21
        return {"status": "ok"}
22
    except WebhookSignatureVerificationError:
23
        return {"error": "Invalid signature"}, 401

1
from extend_ai import Extend
2
from extend_ai.wrapper.errors import WebhookSignatureVerificationError
3
4
client = Extend(token="YOUR_API_KEY")
5
6
# ❌ Bad: Synchronous work can cause timeouts and duplicates
7
@app.post('/webhook')
8
async def handle_webhook_bad(request):
9
    event = client.webhooks.verify_and_parse(body, headers, secret)
10
    
11
    await send_to_multiple_apis(event["payload"])
12
    await generate_pdf_report(event["payload"])
13
    await enrich_data_from_third_party(event["payload"])
14
    await send_email_notifications(event["payload"])
15
    
16
    return {"status": "ok"}
17
18
# ✅ Good: Enqueue then respond
19
@app.post('/webhook')
20
async def handle_webhook_good(request):
21
    try:
22
        event = client.webhooks.verify_and_parse(
23
            body=request.body.decode(),
24
            headers=dict(request.headers),
25
            signing_secret="wss_your_signing_secret"
26
        )
27
        
28
        await job_queue.add("process-webhook", event)
29
        return {"status": "ok"}
30
    except WebhookSignatureVerificationError:
31
        return {"error": "Invalid signature"}, 401

1
async def process_webhook(event):
2
    event_id = event["eventId"]
3
4
    # Check if already processed
5
    if await redis.get(f"processed:{event_id}"):
6
        return
7
8
    # Process the event
9
    await handle_event(event)
10
11
    # Mark as processed (expire after 7 days)
12
    await redis.setex(f"processed:{event_id}", 86400 * 7, "true")

1
from extend_ai import Extend
2
from extend_ai.wrapper.errors import WebhookSignatureVerificationError
3
4
client = Extend(token="YOUR_API_KEY")
5
6
@app.post('/webhook')
7
async def handle_webhook(request):
8
    try:
9
        event = client.webhooks.verify_and_parse(
10
            body=request.body.decode(),
11
            headers=dict(request.headers),
12
            signing_secret="wss_your_signing_secret"
13
        )
14
15
        queued = await message_queue.send(event)
16
        if not queued.success:
17
            # Return 503 to trigger a retry from Extend
18
            return {"error": "Service temporarily unavailable"}, 503
19
20
        return {"status": "ok"}
21
    except WebhookSignatureVerificationError:
22
        return {"error": "Invalid signature"}, 401