Polling a batch endpoint every few seconds to check if ten thousand completions have finished is a waste of compute and a source of flaky logic. For webhooks llm batch jobs, the robust approach is to fire the batch, then let a watcher notify your application via an HTTP callback when the provider marks it complete. Below is a concrete pattern you can ship today using the OpenAI batch API and a small webhook dispatcher.
Step 1: Structure and upload your batch input
LLM providers that support batch inference expect a JSONL file where each line is a standalone request. Keep your schema tight: include a custom id so you can correlate results later.
import json
requests = []
for i, prompt in enumerate(prompts):
requests.append({
"custom_id": f"req-{i}",
"method": "POST",
"url": "/v1/chat/completions",
"body": {
"model": "gpt-4o-mini",
"messages": [{"role": "user", "content": prompt}],
"max_tokens": 200
}
})
with open("batch.jsonl", "w") as f:
for r in requests:
f.write(json.dumps(r) + "\n")Upload the file using the provider SDK. For OpenAI-compatible endpoints this is a standard files call:
from openai import OpenAI
client = OpenAI() # or base_url="https://your-gateway/v1"
file_obj = client.files.create(
file=open("batch.jsonl", "rb"),
purpose="batch"
)
print("uploaded", file_obj.id)Step 2: Submit the batch with a completion window
Create the batch against the chat completions endpoint. The completion_window tells the provider you are fine waiting up to 24h. The API returns a batch id you will track.
batch = client.batches.create(
input_file_id=file_obj.id,
endpoint="/v1/chat/completions",
completion_window="24h"
)
print("batch_id", batch.id)Note: the native batch API does not call your service on completion. That gap is exactly what we close with webhooks llm batch jobs via a watcher process.
Step 3: Build a batch watcher that emits webhooks
Write a small worker that polls the batch status at a sane interval (e.g., 60s) and, on terminal state, downloads the output file and POSTs it to your webhook URL. Sign the payload with HMAC so the receiver can authenticate it.
import time, hmac, hashlib, httpx, json
from openai import OpenAI
WEBHOOK_URL = "https://app.example.com/webhooks/batch"
WEBHOOK_SECRET = b"your-shared-secret"
def sign(body: bytes) -> str:
return hmac.new(WEBHOOK_SECRET, body, hashlib.sha256).hexdigest()
def watch_batch(batch_id: str):
client = OpenAI()
while True:
b = client.batches.retrieve(batch_id)
if b.status == "completed":
out = client.files.content(b.output_file_id)
payload = out.read()
headers = {
"X-Signature": sign(payload),
"Content-Type": "application/json"
}
httpx.post(WEBHOOK_URL, content=payload, headers=headers)
break
elif b.status in ("failed", "expired", "cancelled"):
raise RuntimeError(f"batch {batch_id} ended: {b.status}")
time.sleep(60)If you route requests through an OpenAI-compatible gateway like n4n.ai, the batch lines can carry provider cache-control hints and routing directives; the watcher still downloads results the same way.
Step 4: Implement the webhook receiver endpoint
On the application side, expose a single endpoint that verifies the signature and persists the results. Use FastAPI for brevity.
from fastapi import FastAPI, Request, HTTPException
import hmac, hashlib
app = FastAPI()
SECRET = b"your-shared-secret"
@app.post("/webhooks/batch")
async def batch_webhook(req: Request):
body = await req.body()
sig = req.headers.get("X-Signature", "")
expected = hmac.new(SECRET, body, hashlib.sha256).hexdigest()
if not hmac.compare_digest(sig, expected):
raise HTTPException(status_code=401, detail="bad signature")
# body is JSONL of results; parse and store
for line in body.decode().splitlines():
result = json.loads(line)
process_result(result) # your DB write or queue push
return {"ok": True}Step 5: Make webhook processing idempotent and secure
Batch outputs can be redelivered if your worker retries. Key off custom_id in each result line and use an upsert. Also add a timestamp header and reject stale deliveries older than five minutes to prevent replay.
import time
from fastapi import Header
@app.post("/webhooks/batch")
async def batch_webhook(req: Request, x_timestamp: int = Header(0)):
if abs(time.time() - x_timestamp) > 300:
raise HTTPException(status_code=400, detail="stale")
# ... signature check and upsert by custom_id ...For webhooks llm batch jobs at scale, push the parsed results into a durable queue (SQS, Kafka, or even SQLite) instead of doing heavy work inline. The HTTP handler should return 200 quickly.
Step 6: Verify the pipeline end to end
Start with a batch of three prompts. Run the upload and submit script, then launch the watcher in one terminal and the FastAPI receiver in another.
# terminal 1
uvicorn receiver:app --port 8000
# terminal 2
python watch.py <batch_id>Check the receiver logs for the parsed custom_id values. Confirm the watcher exits after the POST. Then test failure paths: cancel a batch via the provider UI and ensure the watcher raises instead of hanging.
A final production check: run the watcher as a supervised process (systemd, ECS, or a cron with lockfile) so a crash does not silent-drop completion. Because webhooks llm batch jobs decouple notification from polling, you can run multiple watchers with a distributed lock and still get exactly-once delivery to your app.
Step 7: Handle partial output and errors
Providers sometimes return an error_file_id alongside the output. Extend the watcher to fetch both and forward a merged payload:
if b.error_file_id:
err = client.files.content(b.error_file_id).read()
# attach to payload or send separate webhookYour receiver should record errors per custom_id so you can replay only the failed lines. This avoids re-running the entire batch and keeps cost predictable.
Step 8: Clean up files
Batch input and output files count against storage quotas. After a successful webhook delivery, delete the files:
client.files.delete(b.input_file_id)
client.files.delete(b.output_file_id)
if b.error_file_id:
client.files.delete(b.error_file_id)Doing this inside the watcher after the POST succeeds keeps your account tidy and prevents stale data from leaking into later jobs.
The pattern above turns any poll-based LLM batch API into a push-based system with minimal code. You get timely notifications, signature-verified delivery, and a clean separation between inference providers and your application logic.