> ## Documentation Index
> Fetch the complete documentation index at: https://developer.klikit.io/llms.txt
> Use this file to discover all available pages before exploring further.

# Python receiver (reference)

> Minimal Flask server that verifies, dedupes, and acknowledges klikit webhook events.

A copy-paste-runnable webhook receiver in Python + Flask. Around 80 lines.
Demonstrates the four things you have to get right: raw-body capture,
constant-time signature compare, idempotency on `x-klikit-event-id`, and
fast 2xx acknowledgement.

<Tip>
  The source also lives in the partner-api repo at
  [examples/webhook-receiver/python](https://github.com/klikit/partner-api/tree/develop/examples/webhook-receiver/python)
  so you can clone it directly.
</Tip>

## Run it

```bash theme={null}
pip install -r requirements.txt
KLIKIT_WEBHOOK_SECRET=<your secret> python server.py
```

Listens on `:8080` and exposes `POST /webhooks/klikit`.

## Source

```python server.py theme={null}
"""Klikit webhook receiver — minimal reference (Python + Flask)."""
import hashlib
import hmac
import logging
import os
import threading

from flask import Flask, jsonify, request

SECRET = os.environ.get("KLIKIT_WEBHOOK_SECRET")
if not SECRET:
    raise SystemExit("KLIKIT_WEBHOOK_SECRET not set")

SECRET_BYTES = SECRET.encode("utf-8")

app = Flask(__name__)
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s")

# Replace with your own persistent store (Redis SETNX, Postgres unique index, etc).
_seen_event_ids: set[str] = set()
_seen_lock = threading.Lock()


@app.post("/webhooks/klikit")
def receive():
    sig_header = request.headers.get("x-klikit-signature")
    event_id = request.headers.get("x-klikit-event-id")
    event_type = request.headers.get("x-klikit-event-type")

    if not sig_header or not event_id or not event_type:
        return jsonify(error="missing klikit headers"), 400

    # IMPORTANT: compute HMAC over the raw body bytes, not the parsed JSON.
    raw_body = request.get_data(cache=True)
    expected = hmac.new(SECRET_BYTES, raw_body, hashlib.sha256).hexdigest()

    if not hmac.compare_digest(sig_header, expected):
        return jsonify(error="invalid signature"), 401

    with _seen_lock:
        if event_id in _seen_event_ids:
            return jsonify(status="duplicate-ignored"), 200
        _seen_event_ids.add(event_id)

    payload = request.get_json(silent=True) or {}

    # Process asynchronously so klikit's 10s timeout isn't tied to downstream work.
    threading.Thread(
        target=_handle_event, args=(event_type, payload), daemon=True
    ).start()

    return jsonify(status="received"), 200


def _handle_event(event_type: str, payload: dict) -> None:
    brand_id = payload.get("brand_id")
    branch_id = payload.get("branch_id")
    orders = payload.get("orders") or []
    logging.info(
        "[%s] brand=%s branch=%s orders=%d",
        event_type, brand_id, branch_id, len(orders),
    )

    if event_type == "klikit.order.created.v2":
        pass  # push to POS / kitchen display
    elif event_type == "klikit.order.status.updated":
        pass  # update order record
    elif event_type == "klikit.order.cart.updated":
        pass  # replace cart state
    else:
        logging.warning("unknown event type: %s", event_type)


if __name__ == "__main__":
    app.run(host="0.0.0.0", port=int(os.environ.get("PORT", "8080")))
```

```text requirements.txt theme={null}
flask>=3.0
```

## Hardening for production

The reference above runs as-is, but two things need real-world replacements
before you put it in production:

* **`_seen_event_ids` is in-memory.** Replace with Redis (`SET key NX EX
  86400`) or a Postgres unique index. An in-memory set evaporates on every
  restart, letting duplicates back in.
* **`threading.Thread` is a single-process worker.** For real throughput,
  push to a durable queue (SQS, Celery, RQ) and process from a separate
  worker so an OOM in your processor can't drop events.
* **Use a real WSGI server** (`gunicorn`, `uvicorn` with an ASGI shim, etc.)
  in production. Flask's dev server is not suitable for live traffic.