webhook_server.py

"""A webhook receiver, with no framework.

    export VISION_WEBHOOK_SECRET=whsec_…      # dashboard → Webhooks
    python examples/webhook_server.py

The endpoint must be HTTPS and must not resolve to a private address, so a bare localhost
receiver is unreachable by design — put a tunnel (ngrok, cloudflared) in front of it while
developing.
"""

import json
import os
from http.server import BaseHTTPRequestHandler, HTTPServer

from visionapi import WebhookSignatureError, verify_webhook

SECRET = os.environ.get("VISION_WEBHOOK_SECRET")
if not SECRET:
    raise SystemExit("Set VISION_WEBHOOK_SECRET (dashboard → Webhooks).")


class Handler(BaseHTTPRequestHandler):
    def do_POST(self) -> None:  # noqa: N802 - BaseHTTPRequestHandler's naming
        if self.path != "/hooks/vision":
            self.send_response(404)
            self.end_headers()
            return

        # Read the raw bytes. The signature covers exactly what arrived — parsing first
        # and re-serializing changes key order and whitespace, and the HMAC stops matching.
        body = self.rfile.read(int(self.headers.get("Content-Length", 0)))

        try:
            event = verify_webhook(body, self.headers.get("X-Vision-Signature"), SECRET)
        except WebhookSignatureError as err:
            print("rejected delivery:", err)
            self.send_response(400)
            self.end_headers()
            return

        # Any 2xx is success. Acknowledge immediately and do the work afterwards — a slow
        # handler looks like a failed delivery and earns a retry at +1 m, +5 m, +15 m, +40 m.
        self.send_response(202)
        self.end_headers()

        delivery = self.headers.get("X-Vision-Delivery")
        if event["event"] == "task.failed":
            print(f"task {event['task_id']} failed: {event['error']['code']} (delivery {delivery})")
            return

        print(f"task {event['task_id']} completed — {event['credits_used']} credits")
        print(json.dumps(event.get("result"))[:300])

    def log_message(self, *args) -> None:  # quieter than the default access log
        pass


print("listening on http://localhost:3000/hooks/vision")
HTTPServer(("", 3000), Handler).serve_forever()