---
title: "webhook.receive sample: Python + FastAPI"
description: "Operator Contract v1.4.1 webhook.receive module, Python + FastAPI. Generated from the contract."
---

# webhook.receive: Python + FastAPI

<!-- GENERATED by scripts/connect-samples/generate.ts from docs/public/openapi/connect-v1.yaml. Do not edit by hand. -->

The endpoint OneHazel delivers events to. Acknowledge fast, deduplicate on `deliveryId`, process after acknowledging.

Generated from Operator Contract v1.4.1. Key `python/fastapi/webhook.receive`; the `connect_samples` MCP tool returns this page for that key.

## What it implements

| Operation | Purpose | Idempotency key |
|---|---|---|
| `POST /webhook` | Receive a OneHazel event. | no |

Every delivery is checked before it is parsed: `X-OneHazel-Signature` is `sha256=<hex>`, HMAC-SHA256 of the RAW
body keyed by the subscription's `whsec_` signing secret, compared in constant time. There is no timestamp, no
bearer and no `X-OneHazel-Env`, so the signature is the only proof of origin and deduplicating on `deliveryId`
is the only replay defence. A bad signature answers `401` with the flat `{ code, message, retryable }` envelope.

## Run it

```bash
pip install fastapi uvicorn
ONEHAZEL_SIGNING_SECRET=<whsec signing secret> uvicorn onehazel_webhook_receive:app --port 3000
# base URL to register: https://<your host>/onehazel/v1 (ONEHAZEL_BASE_PATH changes the path)
```

Environment: `ONEHAZEL_SIGNING_SECRET` is the subscription's `whsec_` signing secret. During a rotation set
`ONEHAZEL_SIGNING_SECRET_PREVIOUS` to the old one until it stops being used.

## Code

`onehazel_webhook_receive.py`

```python
"""OneHazel webhook.receive module, Operator Contract v1.4.1. Generated; regenerate rather than edit."""

from __future__ import annotations

import hashlib
import hmac
import json
import os
import re
import secrets
import sys
import threading
import time
from typing import Any, Callable, Optional

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

# Generated from the OneHazel Operator Contract v1.4.1: operations, request fields,
# error codes with their HTTP status, and the signature tolerance. Do not hand-edit; regenerate.
CONTRACT: dict[str, Any] = json.loads(r'''{
  "version": "1.4.1",
  "auth": {
    "scheme": "sha256",
    "bearer": false,
    "envHeader": false
  },
  "errorStatus": {
    "unauthorized": 401,
    "signature_invalid": 401,
    "idempotency_key_conflict": 409,
    "launch_token_consumed": 409,
    "round_closed": 409,
    "currency_unsupported": 422,
    "currency_mismatch": 422,
    "player_blocked": 422,
    "rollback_amount_mismatch": 422,
    "insufficient_funds": 402,
    "internal_error": 500,
    "unavailable": 500,
    "invalid_request": 400,
    "session_unknown": 404,
    "session_expired": 404,
    "launch_token_unknown": 404,
    "launch_token_expired": 404,
    "player_unknown": 404,
    "game_unknown": 404,
    "transaction_unknown": 404,
    "freeround_unknown": 404
  },
  "retryable": [
    "internal_error",
    "unavailable"
  ],
  "ops": [
    {
      "path": "/webhook",
      "operationId": "webhookReceive",
      "summary": "Receive a OneHazel event.",
      "idempotent": false,
      "fields": [
        {
          "name": "type",
          "type": "string",
          "required": true
        },
        {
          "name": "data",
          "type": "object",
          "required": true
        },
        {
          "name": "operatorId",
          "type": "string",
          "required": true
        },
        {
          "name": "timestamp",
          "type": "string",
          "required": true
        },
        {
          "name": "deliveryId",
          "type": "string",
          "required": true
        }
      ],
      "closed": false
    }
  ]
}''')

# ───────────────────────────── configuration ─────────────────────────────


def _required_env(name: str) -> str:
    value = os.environ.get(name, "")
    if not value:
        # Never print the value of a secret, only its name.
        print(f"{name} is not set", file=sys.stderr)
        sys.exit(1)
    return value


# During a rotation OneHazel signs with the new secret; the old one stays valid for 1 hour.
SIGNING_SECRETS = [s for s in (_required_env("ONEHAZEL_SIGNING_SECRET"), os.environ.get("ONEHAZEL_SIGNING_SECRET_PREVIOUS", "")) if s]
BASE_PATH = os.environ.get("ONEHAZEL_BASE_PATH", "/onehazel/v1").rstrip("/")

Reply = tuple[int, dict[str, Any]]
Header = Callable[[str], Optional[str]]

# ───────────────────────────── errors ─────────────────────────────


class ContractError(Exception):
    """Raised anywhere in a handler; becomes the flat {code, message, retryable} envelope."""

    def __init__(self, code: str, message: str) -> None:
        super().__init__(message)
        self.code = code
        self.message = message


def error_reply(code: str, message: str) -> Reply:
    return CONTRACT["errorStatus"][code], {"code": code, "message": message, "retryable": code in CONTRACT["retryable"]}


# ───────────────────────────── request validation ─────────────────────────────

_TYPES: dict[str, Callable[[Any], bool]] = {
    "integer": lambda v: isinstance(v, int) and not isinstance(v, bool),
    "string": lambda v: isinstance(v, str),
    "boolean": lambda v: isinstance(v, bool),
    "array": lambda v: isinstance(v, list),
    "object": lambda v: isinstance(v, dict),
}


def check_fields(value: Any, fields: list[dict[str, Any]], closed: bool, where: str) -> dict[str, Any]:
    if not isinstance(value, dict):
        raise ContractError("invalid_request", f"{where or 'body'} must be a JSON object")
    if closed:
        known = {f["name"] for f in fields}
        for key in value:
            if key not in known:
                raise ContractError("invalid_request", f"unknown field {where}{key}")
    for f in fields:
        name = where + f["name"]
        v = value.get(f["name"])
        if v is None:
            if f["required"]:
                raise ContractError("invalid_request", f"{name} is required")
            continue
        if not _TYPES.get(f["type"], lambda _v: True)(v):
            raise ContractError("invalid_request", f"{name} must be {f['type']}")
        if isinstance(v, int) and not isinstance(v, bool):
            if "minimum" in f and v < f["minimum"]:
                raise ContractError("invalid_request", f"{name} is below {f['minimum']}")
            if "exclusiveMinimum" in f and v <= f["exclusiveMinimum"]:
                raise ContractError("invalid_request", f"{name} must be greater than {f['exclusiveMinimum']}")
        if isinstance(v, str):
            if "minLength" in f and len(v) < f["minLength"]:
                raise ContractError("invalid_request", f"{name} is too short")
            if "maxLength" in f and len(v) > f["maxLength"]:
                raise ContractError("invalid_request", f"{name} is too long")
            if "pattern" in f and not re.search(f["pattern"], v):
                raise ContractError("invalid_request", f"{name} is malformed")
            if "enum" in f and v not in f["enum"]:
                raise ContractError("invalid_request", f"{name} is not an allowed value")
        if "fields" in f:
            check_fields(v, f["fields"], False, f"{name}.")
    return value


def parse_body(raw_body: bytes) -> Any:
    try:
        return json.loads(raw_body)
    except ValueError:
        raise ContractError("invalid_request", "body is not valid JSON") from None


def find_op(path: str) -> Optional[dict[str, Any]]:
    return next((o for o in CONTRACT["ops"] if o["path"] == path), None)


# ───────────────────────────── delivery authentication ─────────────────────────────
#
# Deliveries carry `X-OneHazel-Signature: sha256=<hex HMAC-SHA256(whsec, raw body)>`, the
# OneHazel webhook engine's format. There is no timestamp, no bearer and no X-OneHazel-Env,
# so the signature is the only proof of origin and deduplicating on deliveryId is the only
# replay defence. Verify over the RAW bytes, before the body is parsed.


def verify_webhook_signature(raw_body: bytes, header: Optional[str], keys: list[str]) -> bool:
    match = re.fullmatch(r"sha256=([0-9a-f]{64})", (header or "").strip())
    if not match:
        return False
    ok = False
    for key in keys:
        expected = hmac.new(key.encode(), raw_body, hashlib.sha256).hexdigest()
        # No early exit: every secret is compared.
        ok = hmac.compare_digest(expected, match.group(1)) or ok
    return ok


def authenticate_webhook(header: Header, raw_body: bytes) -> None:
    if not verify_webhook_signature(raw_body, header("x-onehazel-signature"), SIGNING_SECRETS):
        raise ContractError("signature_invalid", "bad X-OneHazel-Signature")


# ───────────────────────────── deliveries ─────────────────────────────
#
# Acknowledge within 5 s, deduplicate on deliveryId (deliveries are at-least-once),
# and do the real work AFTER acknowledging, on a worker thread. The seen-set lives in
# memory here; keep it in your database (a unique index on deliveryId) so it survives
# restarts.

import queue

SEEN: set[str] = set()
_LOCK = threading.Lock()
_INBOX: "queue.Queue[dict[str, Any]]" = queue.Queue()


def process_event(event: dict[str, Any]) -> None:
    """Your event handling. Conformance deliveries use the type `cert.conformance`: never act on players for them."""
    print(f"{event['type']} {event['deliveryId']}")


def _worker() -> None:
    while True:
        event = _INBOX.get()
        try:
            process_event(event)
        except Exception as e:  # noqa: BLE001
            print(f"event processing failed: {type(e).__name__}", file=sys.stderr)


threading.Thread(target=_worker, daemon=True).start()


def handle(path: str, header: Header, raw_body: bytes) -> Reply:
    """The whole module: one call per request, framework-agnostic."""
    try:
        op = find_op(path)
        if op is None:
            return 404, {"code": "invalid_request", "message": "unknown operation", "retryable": False}
        authenticate_webhook(header, raw_body)
        event = check_fields(parse_body(raw_body), op["fields"], op["closed"], "")
        if header("x-onehazel-delivery") != event["deliveryId"] or header("x-onehazel-event") != event["type"]:
            raise ContractError("invalid_request", "X-OneHazel-Delivery and X-OneHazel-Event must match the body")
        with _LOCK:
            if event["deliveryId"] in SEEN:
                return 200, {"received": True, "duplicate": True}
            SEEN.add(event["deliveryId"])
        _INBOX.put(event)
        return 200, {"received": True, "duplicate": False}
    except ContractError as e:
        return error_reply(e.code, e.message)
    except Exception as e:  # noqa: BLE001
        print(f"webhook handler failed: {type(e).__name__}", file=sys.stderr)
        return error_reply("internal_error", "internal error")


# ───────────────────────────── FastAPI wiring ─────────────────────────────

app = FastAPI()


@app.post(BASE_PATH + "/{operation:path}")
async def onehazel(operation: str, request: Request) -> JSONResponse:
    # Read the RAW body: the signature is computed over the exact bytes OneHazel sent.
    raw = await request.body()
    status, body = handle("/" + operation, lambda name: request.headers.get(name), raw)
    return JSONResponse(body, status_code=status)
```

## Before you go live

- Replace the in-memory (or file) seen-set with a unique index on `deliveryId` in your database, so dedup survives restarts.
- Run conformance (`connect_run_conformance`) until every scenario passes.