Long-running inference jobs need a reliable way to push results back to your service without blocking the request thread. A webhook receiver for llm completions lets the provider call you when the generation finishes, turning a synchronous API into an event-driven pipeline that scales with your workload instead of your timeout budget.
Prerequisites
- Python 3.11 or newer
fastapi,uvicorn,httpx(install viapip install fastapi uvicorn httpx)ngrokfor exposing your local receiver to the internet (free tier is sufficient)- A shared HMAC secret agreed with your LLM provider or gateway
- An OpenAI-compatible LLM endpoint (or a gateway that fronts one)
You should be comfortable with async def, basic HMAC verification, and JSON. No frontend or ORM required for this tutorial.
The payload contract
Define the shape of the callback before writing code. A minimal completion webhook looks like this:
{
"job_id": "job_8f2c1a",
"status": "completed",
"model": "mistralai/mixtral-8x7b-instruct",
"choices": [
{ "index": 0, "text": "The capital of France is Paris." }
],
"usage": { "prompt_tokens": 12, "completion_tokens": 8, "total_tokens": 20 }
}If the job failed, status is failed and an error field carries a string. Your webhook receiver for llm completions must accept both shapes without raising.
Step 1: Scaffold the FastAPI receiver
FastAPI gives you async request handling with minimal boilerplate. Create receiver.py:
import asyncio
import json
import logging
from fastapi import FastAPI, Request, Response
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("webhook")
app = FastAPI()
@app.post("/webhooks/llm-completions")
async def llm_webhook(request: Request):
payload = await request.json()
logger.info("Received payload: %s", payload.get("job_id"))
return Response(status_code=202)Run it:
uvicorn receiver:app --port 8000Test with a curl:
curl -X POST http://localhost:8000/webhooks/llm-completions \
-H "content-type: application/json" \
-d '{"job_id":"test_1","status":"completed","model":"x","choices":[]}'Expected output in the uvicorn log: INFO:webhook:Received payload: test_1. curl returns an empty body with HTTP 202.
Step 2: Verify the signature
Never trust unsigned callbacks. The provider should send an X-Signature header: HMAC-SHA256 of the raw body keyed by your shared secret, hex-encoded.
import hmac
import hashlib
SECRET = b"your-32-byte-secret"
async def verify_signature(request: Request) -> bool:
body = await request.body()
sig = request.headers.get("X-Signature", "")
expected = hmac.new(SECRET, body, hashlib.sha256).hexdigest()
return hmac.compare_digest(sig, expected)Wire it into the handler:
@app.post("/webhooks/llm-completions")
async def llm_webhook(request: Request):
if not await verify_signature(request):
logger.warning("Bad signature")
return Response(status_code=401)
payload = await request.json()
logger.info("Verified job %s", payload.get("job_id"))
return Response(status_code=202)Replay the curl without the header and you get 401. Generate a valid sig with:
SIG=$(python3 -c 'import hmac,hashlib; print(hmac.new(b"your-32-byte-secret", b"{\"job_id\":\"test_1\"}", hashlib.sha256).hexdigest())')
curl -X POST http://localhost:8000/webhooks/llm-completions \
-H "content-type: application/json" -H "X-Signature: $SIG" \
-d '{"job_id":"test_1","status":"completed"}'Now you get 202.
Step 3: Ack fast, process later
A webhook receiver for llm completions should not run downstream work (database writes, fan-out, embeddings) inside the HTTP handler. Return 202 immediately and hand off to a background worker.
queue: asyncio.Queue = asyncio.Queue()
@app.post("/webhooks/llm-completions")
async def llm_webhook(request: Request):
if not await verify_signature(request):
return Response(status_code=401)
body = await request.body()
await queue.put(body)
return Response(status_code=202)
@app.on_event("startup")
async def startup():
asyncio.create_task(worker())
async def worker():
while True:
raw = await queue.get()
payload = json.loads(raw)
logger.info("Processing job %s status=%s", payload["job_id"], payload["status"])
# Persist to DB, trigger next step, etc.
queue.task_done()This keeps p99 handler latency under 10 ms even if processing takes seconds. Size the queue with asyncio.Queue(maxsize=1000) to apply backpressure under load.
Step 4: Idempotency
Providers retry webhooks on network errors or non-2xx responses. Use job_id + status as a dedupe key. An in-memory set works for a single instance; use Redis in production.
seen = set()
async def worker():
while True:
raw = await queue.get()
payload = json.loads(raw)
key = (payload["job_id"], payload["status"])
if key in seen:
logger.info("Duplicate %s, skipping", key)
queue.task_done()
continue
seen.add(key)
# process payload
queue.task_done()Step 5: Firing the async job
To exercise the receiver you need a caller that emits completions. If you use an OpenRouter-class gateway such as n4n.ai, you get one OpenAI-compatible endpoint covering 240+ models with automatic fallback, so you can submit the initial request without writing provider-specific retry logic. The request includes your public webhook URL:
import httpx
async def submit_job(prompt: str, callback_url: str):
async with httpx.AsyncClient() as client:
r = await client.post(
"https://api.n4n.ai/v1/completions",
headers={"Authorization": "Bearer $KEY"},
json={
"model": "mistralai/mixtral-8x7b-instruct",
"prompt": prompt,
"callback_url": callback_url,
"metadata": {"job_id": "job_8f2c1a"}
}
)
return r.json()Swap callback_url for your ngrok URL once exposed. The gateway will POST to that URL when inference finishes.
Step 6: Expose locally with ngrok
Start the receiver, then in a second shell:
ngrok http 8000Copy the https://xxxx.ngrok.io URL. Point your provider or the submit_job call to https://xxxx.ngrok.io/webhooks/llm-completions. Trigger a job and watch the uvicorn log:
INFO:webhook:Verified job job_8f2c1a
INFO:webhook:Processing job job_8f2c1a status=completedThe ngrok inspector shows the inbound POST with a 202 response. That round trip proves the webhook receiver for llm completions works end to end.
Step 7: Failure handling and retries
Return 500 only when you genuinely cannot process. The provider should back off and retry. For transient errors, re-queue with a cap:
async def worker():
while True:
raw = await queue.get()
try:
payload = json.loads(raw)
# process, may raise on DB outage
except Exception as e:
logger.error("Processing failed: %s", e)
retry = payload.get("retries", 0) + 1
if retry < 5:
payload["retries"] = retry
await queue.put(json.dumps(payload).encode())
else:
logger.critical("Dropping job %s", payload.get("job_id"))
finally:
queue.task_done()Without a retry cap you will create a retry storm during an outage.
Production notes
- Use Redis
SET key NXor PostgresINSERT ... ON CONFLICTfor idempotency across multiple workers. - Rotate the HMAC secret quarterly and support two active secrets during overlap.
- Emit metrics: webhook arrivals, signature failures, queue depth, processing lag.
- Parse the
usageblock and forward per-token counts to your metering system; if your gateway already does per-token metering, reconcile against it.
A webhook receiver for llm completions is a small surface area with outsized reliability impact. Get the signature check and the fast-ack pattern right, and the rest is straightforward business logic.