Webhooks and Callbacks for Async LLM

Why Webhooks? Long-running LLM operations benefit from async patterns where the server notifies the client when complete.

Webhook Flow

1. Client submits request with callback URL
2. Server returns immediately with task ID
3. Server processes request
4. Server POSTs result to callback URL

Implementation

API Side

from fastapi import FastAPI, BackgroundTasks
import httpx

app = FastAPI()

@app.post("/api/v1/completions")
async def create_completion(
    prompt: str,
    callback_url: str,
    background_tasks: BackgroundTasks
):
    task_id = generate_task_id()

    # Start async processing
    background_tasks.add_task(
        process_and_callback,
        task_id, prompt, callback_url
    )

    return {"task_id": task_id, "status": "processing"}

async def process_and_callback(task_id, prompt, callback_url):
    try:
        result = await llm.generate(prompt)
        payload = {
            "task_id": task_id,
            "status": "completed",
            "result": result
        }
    except Exception as e:
        payload = {
            "task_id": task_id,
            "status": "failed",
            "error": str(e)
        }

    async with httpx.AsyncClient() as client:
        await client.post(callback_url, json=payload)

Client Side

from flask import Flask, request

app = Flask(__name__)

@app.route("/webhook/llm-callback", methods=["POST"])
def handle_callback():
    data = request.json
    task_id = data["task_id"]

    if data["status"] == "completed":
        result = data["result"]
        process_result(task_id, result)
    else:
        handle_error(task_id, data["error"])

    return {"received": True}

Security

import hmac
import hashlib

# Sign webhooks
def sign_payload(payload, secret):
    return hmac.new(
        secret.encode(),
        json.dumps(payload).encode(),
        hashlib.sha256
    ).hexdigest()

# Verify on client side
def verify_signature(payload, signature, secret):
    expected = sign_payload(payload, secret)
    return hmac.compare_digest(expected, signature)

Retry Logic

async def send_with_retry(url, payload, max_retries=3):
    for attempt in range(max_retries):
        try:
            response = await httpx.post(url, json=payload)
            if response.status_code == 200:
                return
        except Exception:
            pass
        await asyncio.sleep(2 ** attempt)

    # Store in dead letter queue
    await store_failed_callback(url, payload)

Best Practices

• Sign payloads for security
• Implement retry with exponential backoff
• Support idempotent callbacks
• Log all webhook attempts
• Provide webhook testing endpoints
• Include task ID for correlation