Reward the First 200, Not the Signup: A Signed Webhook Handler in Python

Marcos Placona put the API activation problem plainly: "Developers are signing up. Keys are being issued. And then most of them never make a successful request." His funnel counts keys that got "at least one successful response, not a 401 or a 400."

So if you give new developers free credits, the signup is the wrong trigger. A better one is the first 2xx response on a key that belongs to the account. Real programs already pay for usage, not signups: Coinbase Developer Platform's referral program only converts a referral "once the developer actually uses the platform."

This post builds that trigger in two parts, using only the Python standard library:

1. API side: detect an account's first 2xx exactly once, and emit a signed event.
2. Billing side: a webhook receiver that verifies the signature, rejects stale or replayed events, and grants the credit once.

Why sign the event at all

Granting credits is a write to money-like state. Stripe's webhook docs warn that without verification, "an attacker could send fake webhook events to your endpoint to trigger actions like fulfilling orders, granting account access or modifying records." The same docs give the three guards used below:

• An HMAC-SHA256 signature, checked with a constant-time comparison
• A timestamp tolerance (their libraries default to 5 minutes; never 0, which disables the recency check)
• A log of processed event IDs, because the same event can arrive more than once

The code

"""Grant a one-time reward on an account's first successful (2xx) API call,
and receive it through a signed webhook. Stdlib only (sqlite3, hmac)."""
import hashlib, hmac, json, sqlite3, time, uuid

SECRET = b"whsec_replace_me"
TOLERANCE = 300  # seconds; never 0, a 0 tolerance would switch the recency check off

db = sqlite3.connect(":memory:")
db.executescript("""
CREATE TABLE first_success (account_id TEXT PRIMARY KEY, at INTEGER);
CREATE TABLE processed_events (event_id TEXT PRIMARY KEY);
CREATE TABLE credits (account_id TEXT, amount INTEGER, reason TEXT);
""")

# --- 1. API side: called from your request log / middleware -----------------
def on_api_response(account_id, status):
    """Return a signed event the first time this account gets a 2xx, else None."""
    if not 200 <= status < 300:
        return None  # a 401 or 400 is not activation
    cur = db.execute("INSERT OR IGNORE INTO first_success VALUES (?, ?)",
                     (account_id, int(time.time())))
    if cur.rowcount == 0:
        return None  # already had a first success; the reward is one-time
    event = {"id": str(uuid.uuid4()), "type": "api.first_success",
             "account_id": account_id}
    return sign(json.dumps(event))

def sign(payload, ts=None):
    ts = ts or int(time.time())
    mac = hmac.new(SECRET, f"{ts}.{payload}".encode(), hashlib.sha256).hexdigest()
    return {"body": payload, "headers": {"X-Signature": f"t={ts},v1={mac}"}}

# --- 2. Billing side: the webhook receiver ----------------------------------
def receive(body, header):
    parts = dict(p.split("=", 1) for p in header.split(","))
    ts, got = int(parts["t"]), parts["v1"]
    if abs(time.time() - ts) > TOLERANCE:
        return 400, "stale timestamp"
    want = hmac.new(SECRET, f"{ts}.{body}".encode(), hashlib.sha256).hexdigest()
    if not hmac.compare_digest(want, got):  # constant-time compare
        return 400, "bad signature"
    event = json.loads(body)
    if db.execute("INSERT OR IGNORE INTO processed_events VALUES (?)",
                  (event["id"],)).rowcount == 0:
        return 200, "duplicate, ignored"
    db.execute("INSERT INTO credits VALUES (?, ?, ?)",
               (event["account_id"], 100, event["type"]))
    db.commit()
    return 200, "credited"

if __name__ == "__main__":
    print(on_api_response("acct_1", 401))            # None: auth failed
    msg = on_api_response("acct_1", 200)              # first 2xx -> event
    print(on_api_response("acct_1", 200))            # None: already rewarded
    print(receive(msg["body"], msg["headers"]["X-Signature"]))   # credited
    print(receive(msg["body"], msg["headers"]["X-Signature"]))   # duplicate
    forged = msg["body"].replace("acct_1", "acct_2")
    print(receive(forged, msg["headers"]["X-Signature"]))        # bad signature
    old = sign(msg["body"], ts=int(time.time()) - 3600)
    print(receive(old["body"], old["headers"]["X-Signature"]))   # stale
    print(db.execute("SELECT * FROM credits").fetchall())

What the demo prints

None
None
(200, 'credited')
(200, 'duplicate, ignored')
(400, 'bad signature')
(400, 'stale timestamp')
[('acct_1', 100, 'api.first_success')]

Line by line: a 401 is not activation, so no event. The first 200 creates one event. A second 200 creates nothing, because the first_success primary key makes the reward one-time. The receiver credits the event once, ignores the replay, rejects a body edited to point at another account, and rejects an event signed an hour ago.

Wiring it to real traffic

on_api_response needs two things you already have: the account that owns the key, and the status code you returned. The cheapest place to call it is the middleware that writes your request log, right after the response status is known. If you'd rather not touch the hot path, run it as a small job over the request log every minute; the INSERT OR IGNORE makes reprocessing the same rows harmless.

On the billing side, receive takes the raw request body as a string. Verify the signature against the exact bytes you received, before any JSON parsing or re-serialising, or a harmless change in whitespace will break the check.

Notes for production

• Keep the receiver fast. Stripe's docs say an endpoint must "quickly return a successful status code (2xx) before any complex logic that could cause a timeout." Write the event to a queue, then credit from a worker.
• Don't rely on order. Events can arrive out of order, so a reward should depend only on its own event, not on a sequence.
• Pick the unit your product meters. 100 above is a placeholder. Use credits, runs or minutes, whatever your billing already understands.
• Measure it. Placona's advice is to set a target for time to first successful call and measure it after each fix. The first_success.at column, minus the signup time, gives you that number for free.

Posted from the Mantis team account under the Stickly organization. Stickly is a widget that lets SaaS founders reward their own users with product credits through signed webhooks, founded by Tarun Kumar. Stickly does not detect API calls itself; the code above is the part you run on your side. This post was drafted with AI assistance; the code was run and its output pasted unedited.