Webhook Integration for AI Results

You will build a webhook system that delivers AI inference results to external services asynchronously. By the end, you will have a webhook dispatcher on your GPU server that sends inference results to registered endpoints with retry logic, signature verification, and delivery tracking.

Why Webhooks for AI

Long-running inference tasks should not force clients to poll for results. Webhooks push results to the client as soon as generation completes, integrating your AI pipeline with Slack, email, CRM systems, or any service that accepts HTTP callbacks.

Webhook Dispatcher

Build a dispatcher that sends inference results to registered webhook URLs with proper error handling.

import hashlib
import hmac
import json
import time
import httpx
import asyncio
from datetime import datetime

class WebhookDispatcher:
    def __init__(self, signing_secret: str, max_retries: int = 3):
        self.secret = signing_secret
        self.max_retries = max_retries
        self.client = httpx.AsyncClient(timeout=10)

    def sign_payload(self, payload: str, timestamp: str) -> str:
        message = f"{timestamp}.{payload}"
        return hmac.new(
            self.secret.encode(), message.encode(), hashlib.sha256
        ).hexdigest()

    async def deliver(self, url: str, data: dict) -> bool:
        payload = json.dumps(data)
        timestamp = str(int(time.time()))
        signature = self.sign_payload(payload, timestamp)

        headers = {
            "Content-Type": "application/json",
            "X-Webhook-Timestamp": timestamp,
            "X-Webhook-Signature": f"sha256={signature}",
        }

        for attempt in range(self.max_retries):
            try:
                resp = await self.client.post(url, content=payload, headers=headers)
                if resp.status_code < 300:
                    return True
                if resp.status_code >= 400 and resp.status_code < 500:
                    return False  # Client error, do not retry
            except httpx.RequestError:
                pass
            await asyncio.sleep(2 ** attempt)
        return False

API Integration

Integrate the webhook dispatcher with your FastAPI inference server.

from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel

app = FastAPI()
dispatcher = WebhookDispatcher(signing_secret="your-secret-key")

class InferenceRequest(BaseModel):
    messages: list[dict]
    max_tokens: int = 256
    webhook_url: str | None = None

@app.post("/api/v1/generate")
async def generate(req: InferenceRequest, background_tasks: BackgroundTasks):
    # Run inference
    from openai import OpenAI
    client = OpenAI(base_url="http://localhost:8000/v1", api_key="not-needed")
    response = client.chat.completions.create(
        model="meta-llama/Llama-3.1-8B-Instruct",
        messages=req.messages, max_tokens=req.max_tokens
    )

    result = {
        "content": response.choices[0].message.content,
        "model": response.model,
        "tokens": response.usage.total_tokens,
        "completed_at": datetime.utcnow().isoformat()
    }

    # Deliver via webhook if URL provided
    if req.webhook_url:
        background_tasks.add_task(dispatcher.deliver, req.webhook_url, result)
        return {"status": "processing", "delivery": "webhook"}

    return result

Webhook Receiver Example

Build a receiver that verifies signatures and processes incoming inference results.

from flask import Flask, request, jsonify
import hmac
import hashlib

app = Flask(__name__)
WEBHOOK_SECRET = "your-secret-key"

@app.route("/webhook/inference", methods=["POST"])
def receive_webhook():
    timestamp = request.headers.get("X-Webhook-Timestamp")
    signature = request.headers.get("X-Webhook-Signature", "").replace("sha256=", "")

    payload = request.get_data(as_text=True)
    expected = hmac.new(
        WEBHOOK_SECRET.encode(), f"{timestamp}.{payload}".encode(), hashlib.sha256
    ).hexdigest()

    if not hmac.compare_digest(signature, expected):
        return jsonify({"error": "Invalid signature"}), 401

    data = request.get_json()
    print(f"Received inference result: {data['content'][:100]}...")
    # Process the result -- store in database, send notification, etc.

    return jsonify({"status": "received"}), 200

Webhook Registration and Management

from pydantic import BaseModel, HttpUrl

class WebhookRegistration(BaseModel):
    url: HttpUrl
    events: list[str] = ["inference.completed"]
    secret: str

webhooks_store = {}

@app.post("/api/v1/webhooks")
async def register_webhook(reg: WebhookRegistration):
    webhook_id = str(uuid.uuid4())
    webhooks_store[webhook_id] = reg.dict()
    return {"id": webhook_id, "status": "registered"}

@app.get("/api/v1/webhooks")
async def list_webhooks():
    return {"webhooks": webhooks_store}

@app.delete("/api/v1/webhooks/{webhook_id}")
async def delete_webhook(webhook_id: str):
    webhooks_store.pop(webhook_id, None)
    return {"status": "deleted"}

Production Considerations

For production webhook delivery, use a Redis queue to persist delivery attempts and handle retries across server restarts. Store delivery logs in the ELK stack for debugging failed deliveries. For Celery-based async processing, webhooks integrate naturally as post-task callbacks.

Add Prometheus metrics for webhook delivery success rates and latency. The self-hosting guide covers infrastructure, and our tutorials section has more integration patterns. Set up the inference backend with the vLLM production guide.