Python 88.3%
TypeScript 7.6%
Shell 4.1%
1"""OpenAI webhook receiver — Python/FastAPI + official SDK `webhooks.unwrap`, with a manual-HMAC route.2STATUS: UNVERIFIED as a running server (fastapi/uvicorn not installed in this repo; py_compile OK);3the verification logic is LIVE_VERIFIED offline by offline_test.py (2026-09-18).45 pip install openai fastapi uvicorn6 OPENAI_WEBHOOK_SECRET=whsec_... uvicorn server_fastapi:app --port 80007Register https://<public-host>/webhook in the dashboard (Settings -> Project -> Webhooks) or via8POST /v1/webhook_endpoints, then send a test event (POST /v1/webhook_endpoints/{id}/test).9"""10from __future__ import annotations1112import os1314from fastapi import BackgroundTasks, FastAPI, Request, Response15from openai import InvalidWebhookSignatureError, OpenAI1617from verify_manual import InvalidSignature, verify as manual_verify1819SECRET = os.environ["OPENAI_WEBHOOK_SECRET"] # shown once at creation / rotation20client = OpenAI(webhook_secret=SECRET) # OPENAI_API_KEY needed only to call the API back21app = FastAPI()22_seen: set[str] = set() # dedupe on webhook-id (redeliveries possible for up to 72 h)232425def handle(event) -> None:26 if event.type == "response.completed":27 resp = client.responses.retrieve(event.data.id)28 print("response done:", resp.id, (resp.output_text or "")[:80])29 elif event.type in {"batch.completed", "fine_tuning.job.succeeded", "eval.run.succeeded", "video.completed"}:30 print(event.type, event.data.id)31 else:32 print("event", event.type)333435@app.post("/webhook")36async def webhook(request: Request, tasks: BackgroundTasks) -> Response:37 raw = await request.body() # signature covers the RAW body — never re-serialise38 try:39 event = client.webhooks.unwrap(raw, request.headers) # raises on bad signature / stale timestamp (300 s)40 except InvalidWebhookSignatureError:41 return Response("Invalid signature", status_code=400)42 delivery_id = request.headers.get("webhook-id", "")43 if delivery_id in _seen:44 return Response(status_code=200)45 _seen.add(delivery_id)46 tasks.add_task(handle, event) # ack immediately, work in background47 return Response(status_code=200)484950@app.post("/webhook-manual")51async def webhook_manual(request: Request) -> Response:52 raw = await request.body()53 try:54 event = manual_verify(raw, request.headers, SECRET)55 except InvalidSignature:56 return Response("Invalid signature", status_code=400)57 print("verified (manual)", event["type"], event["data"]["id"])58 return Response(status_code=200)59