Receive a Referral Reward Webhook in Python with Flask

To receive a WarpLink referral webhook in Python, read the raw bytes with request.get_data(), verify the X-WarpLink-Signature HMAC-SHA256 over timestamp.body with the whole whsec_ secret, insert the event id so a retry does nothing, and grant the reward. This guide gives a complete Flask receiver you can run and test locally.

You get a server that verifies each delivery, stores handled event ids in SQLite, handles referral.claimed, referral.flagged, and referral.updated, and answers well inside the 10-second limit. A script signs a sample event so you can test without a phone. The Node.js version is here.

What arrives

Every delivery is a POST with Content-Type: application/json and three headers:

The body has four fields: id, type, created, and data. For a referral event, data holds referral_id, app_id, link (id, slug, url), installer_id, platform, match (method, exact, confidence), tapped_at, installed_at, claimed_at, claimed_via, and flags. The webhooks guide lists the complete payloads.

The signature is the lowercase hex HMAC-SHA256 of ${timestamp}.${rawBody}, keyed with the whole signing secret including its whsec_ prefix. For 24 hours after a rotation the header holds two signatures, separated by a comma.

Step 1: Create the endpoint and copy the secret

Open the app in the dashboard, go to Referral handoff, choose the Webhook tab, and add an endpoint with a public https address. Copy the signing secret, which starts with whsec_ and is shown in full once.

export WARPLINK_WEBHOOK_SECRET="whsec_your_secret_here"

Step 2: Set up the project

mkdir warplink-webhook && cd warplink-webhook
python3 -m venv .venv
source .venv/bin/activate
pip install flask

Step 3: Write the verifier

The order follows the docs: check the timestamp is a plain integer within 300 seconds, compute the HMAC over the raw bytes, and compare in constant time against every listed signature. Save this as verify.py:

import hashlib
import hmac
import re
import time

TOLERANCE_SECONDS = 300


def verify_signature(
    secret: str, raw_body: bytes, signature_header: str, timestamp: str
) -> bool:
    if not re.fullmatch(r"\d{1,12}", timestamp):
        return False
    if abs(time.time() - int(timestamp)) > TOLERANCE_SECONDS:
        return False

    signed = timestamp.encode() + b"." + raw_body
    expected = hmac.new(secret.encode(), signed, hashlib.sha256).hexdigest()

    return any(
        hmac.compare_digest(signature.strip().lower(), expected)
        for signature in signature_header.split(",")
    )

hmac.compare_digest runs in constant time. The any call accepts a match on any listed signature, which is what makes a secret rotation painless.

Step 4: Write the receiver

Save this as app.py. It verifies the raw bytes first, records the event id in SQLite with INSERT OR IGNORE, and answers fast:

import json
import os
import sqlite3

from flask import Flask, request

from verify import verify_signature

SECRET = os.environ["WARPLINK_WEBHOOK_SECRET"]  # whsec_...

db = sqlite3.connect("webhooks.db", check_same_thread=False)
db.execute("CREATE TABLE IF NOT EXISTS events (id TEXT PRIMARY KEY)")
db.execute(
    "CREATE TABLE IF NOT EXISTS rewards ("
    "installer_id TEXT PRIMARY KEY, link_id TEXT, event_id TEXT)"
)

app = Flask(__name__)


def first_time(event_id: str) -> bool:
    cursor = db.execute("INSERT OR IGNORE INTO events (id) VALUES (?)", (event_id,))
    db.commit()
    return cursor.rowcount == 1


def set_reward(event: dict) -> None:
    data = event["data"]
    db.execute(
        "INSERT OR REPLACE INTO rewards (installer_id, link_id, event_id) VALUES (?, ?, ?)",
        (data["installer_id"], data["link"]["id"], event["id"]),
    )
    db.commit()


def handle_event(event: dict) -> None:
    if not first_time(event["id"]):
        return  # a retry or a resend

    kind = event["type"]
    if kind == "referral.claimed":
        set_reward(event)
        print(f"reward for installer {event['data']['installer_id']}")
    elif kind == "referral.updated":
        set_reward(event)  # the credit moves to the new link
        print(f"credit moved from {event['data']['previous']['link']['id']}")
    elif kind == "referral.flagged":
        print(f"late claim {event['data']['referral_id']}: {event['data']['flags']}")
    elif kind == "webhook.test":
        print("test event received")


@app.post("/webhooks/warplink")
def warplink_webhook():
    raw_body = request.get_data()
    valid = verify_signature(
        SECRET,
        raw_body,
        request.headers.get("X-WarpLink-Signature", ""),
        request.headers.get("X-WarpLink-Timestamp", ""),
    )
    if not valid:
        return "invalid signature", 400

    handle_event(json.loads(raw_body))
    return "", 204

Start it:

flask --app app run --port 3000

Four choices are worth naming:

• request.get_data(), not request.json. The signature covers the exact bytes we sent. Parse the JSON from those bytes after the check.
• 400 for a bad signature. Any 4xx ends the event as failed, so a forged request is never retried. Only a 5xx or a timeout retries.
• Dedupe in the database. Delivery is at least once. The PRIMARY KEY on the event id makes INSERT OR IGNORE an atomic "have I seen this?" check, even with several workers.
• Keep the handler short. We wait 10 seconds for a reply. For slow reward systems, enqueue the event and return at once.

Step 5: Test without a phone

The dashboard's Send test button reaches a public address. For a local server, sign a sample event yourself. Save this as send_test.py:

import hashlib
import hmac
import json
import os
import time
import urllib.error
import urllib.request

SECRET = os.environ["WARPLINK_WEBHOOK_SECRET"]

event = {
    "id": "evt_local_0001",
    "type": "referral.claimed",
    "created": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
    "data": {
        "referral_id": "ref_local_0001",
        "app_id": "3f6c1c0e-6a7e-4d0e-9b1e-0a5a8f0c2b11",
        "link": {
            "id": "b7a1f2e4-52c8-4a38-8f64-1d7a0c9e3a55",
            "slug": "invite-ana",
            "url": "https://acme.aplnk.to/invite-ana",
        },
        "installer_id": "d1c2f9a0-7b43-4f0e-a6d5-3c2b8e1f7a90",
        "platform": "ios",
        "match": {"method": "install_token", "exact": True, "confidence": 1},
        "tapped_at": "2026-10-19T15:24:41Z",
        "installed_at": "2026-10-19T15:29:55Z",
        "claimed_at": "2026-10-19T15:30:01Z",
        "claimed_via": "sdk",
        "flags": [],
    },
}

raw_body = json.dumps(event).encode()
timestamp = str(int(time.time()))
signature = hmac.new(
    SECRET.encode(), timestamp.encode() + b"." + raw_body, hashlib.sha256
).hexdigest()


def send() -> None:
    request = urllib.request.Request(
        "http://localhost:3000/webhooks/warplink",
        data=raw_body,
        method="POST",
        headers={
            "Content-Type": "application/json",
            "X-WarpLink-Signature": signature,
            "X-WarpLink-Timestamp": timestamp,
            "X-WarpLink-Event-Id": event["id"],
        },
    )
    try:
        with urllib.request.urlopen(request) as response:
            print(response.status)
    except urllib.error.HTTPError as error:
        print(error.code)


send()
send()  # the second delivery is a repeat and grants nothing

Run it in a second terminal with the same secret:

python3 send_test.py

You see 204 twice, and the server prints one reward for installer ... line. The second 204 is the dedupe working: same event id, no second reward. Change a byte of raw_body after signing and the answer is 400.

To reach your machine from the dashboard, expose port 3000 through any https tunnel, add that address as the endpoint, and press Send test. The test event has an id that starts evt_test_, goes out once, and is never retried.

Step 6: Move to production

• Use your own database. Put the event insert and the reward write in one transaction, so a crash between them cannot lose or double a reward.
• Run behind a production server such as gunicorn, and keep the endpoint on public https.
• Handle referral.updated. It carries previous, the old link and match. Take the credit from the previous link's owner and give it to the new one. See exact versus estimated matches.
• Rotate without downtime. The old secret keeps signing beside the new one for 24 hours. The verifier already accepts either signature.
• Resend after an outage. A failed event can be sent again with the same id for 7 days, from the Referrals report or with POST /v1/webhook-events/{id}/resend.

For the rules that decide whether a claim should be paid, read verify a referral before you grant the reward.

Frequently Asked Questions

How do I verify a WarpLink webhook signature in Python?

Read the raw bytes with request.get_data(), build the timestamp, a period, and the body, and compute hmac.new with the whole whsec_ secret, the signed bytes, and hashlib.sha256. Compare the hex digest to each signature in X-WarpLink-Signature with hmac.compare_digest, after checking the timestamp is within 300 seconds.

Should I use request.json or request.get_data() in Flask?

Use request.get_data(). The signature covers the exact bytes we sent, and a parsed and re-serialized body does not match. Parse the JSON from the same bytes after the check passes.

How do I stop a retry from paying a reward twice?

Insert the event id into a table with the id as primary key before granting the reward. If the insert adds no row, the event is a repeat. The id is the same on every retry and every resend.

Does the secret include the whsec_ prefix?

Yes. The HMAC key is the full secret string including the whsec_ prefix, encoded as UTF-8.

How many times will WarpLink retry a failed delivery?

Up to 5 attempts over 20 minutes, at once and then after 30 seconds, 2 minutes, 8 minutes, and 20 minutes from the first. A 5xx answer or a timeout retries. A 4xx or 3xx answer ends the event as failed.

Related Guides

• Reference: Receive referral webhooks and the Webhooks API.
• Same receiver in Node.js: Receive a referral reward webhook in Node.js.
• The whole path: Referrals and WarpLink SDK 1.2.0.
• Launch pass: A referral link checklist for launch.
• Use case: App referral programs.

How WarpLink Helps

We do the matching, keep the record, and send a signed event the moment an install is credited. Your receiver stays small because the hard part, proving which tap led to which sign-up, is already done.

One workflow, three pillars. Deep linking carries the invitation, install attribution credits the link that produced the new user, and real-time analytics shows referrals beside every other link. Sub-10ms redirects on the way in, one SDK on the way out.

Create a free WarpLink account and get deep linking, install attribution, and real-time analytics in one SDK, with 10,000 clicks a month on the free plan.