Get notified the instant
something happens. No polling.

curl -X POST https://api.purpleflea.com/v1/webhooks \ -H "Authorization: Bearer sk_live_..." \ -H "Content-Type: application/json" \ -d '{ "url": "https://your-agent.example.com/webhooks/purpleflea", "events": [ "bet.placed", "bet.won", "bet.lost", "agent.registered", "balance.low", "position.opened", "position.closed", "position.liquidated", "funding.paid", "pnl.alert", "swap.initiated", "swap.completed", "swap.failed", "balance.received", "domain.registered", "domain.expiring", "dns.updated", "referral.earned" ], "secret": "whsec_your_random_secret_here" }'

{ "id": "wh_01HXYZ9AB3CDEF", "url": "https://your-agent.example.com/webhooks/purpleflea", "events": ["bet.placed", "bet.won", "bet.lost", "agent.registered", "balance.low", "position.opened", "position.closed", "position.liquidated", "funding.paid", "pnl.alert", "swap.initiated", "swap.completed", "swap.failed", "balance.received", "domain.registered", "domain.expiring", "dns.updated", "referral.earned"], "status": "active", "created_at": "2025-11-14T10:22:00Z", "last_delivery_at": null, "total_deliveries": 0, "failed_deliveries": 0 }

import hashlib import hmac import json import time import logging from typing import Any from fastapi import FastAPI, Request, HTTPException, Header from fastapi.responses import JSONResponse app = FastAPI(title="Purple Flea Webhook Receiver") logger = logging.getLogger(__name__) WEBHOOK_SECRET = "whsec_your_random_secret_here" TIMESTAMP_TOLERANCE_SECONDS = 300 # reject replays older than 5 min processed_event_ids: set[str] = set() # use Redis in production def verify_signature( body: bytes, timestamp: str, signature_header: str, ) -> bool: """Verify HMAC-SHA256 signature from X-PurpleFlea-Signature header.""" if not signature_header or not signature_header.startswith("sha256="): return False received_sig = signature_header.removeprefix("sha256=") signed_payload = timestamp.encode() + b"." + body computed = hmac.new( WEBHOOK_SECRET.encode("utf-8"), signed_payload, hashlib.sha256, ).hexdigest() return hmac.compare_digest(computed, received_sig) def check_timestamp(timestamp_str: str) -> None: """Raise if the timestamp is missing or too old (replay protection).""" try: event_time = int(timestamp_str) except (ValueError, TypeError): raise HTTPException(status_code=400, detail="invalid timestamp") age = abs(time.time() - event_time) if age > TIMESTAMP_TOLERANCE_SECONDS: raise HTTPException(status_code=400, detail="timestamp too old") @app.post("/webhooks/purpleflea") async def receive_webhook( request: Request, x_purpleflea_signature: str = Header(default=""), x_purpleflea_timestamp: str = Header(default=""), ) -> JSONResponse: body = await request.body() # 1. Timestamp replay guard check_timestamp(x_purpleflea_timestamp) # 2. Signature verification — use raw bytes before any parsing if not verify_signature(body, x_purpleflea_timestamp, x_purpleflea_signature): logger.warning("Webhook signature mismatch — possible forgery") raise HTTPException(status_code=401, detail="invalid signature") # 3. Parse payload try: event: dict[str, Any] = json.loads(body) except json.JSONDecodeError: raise HTTPException(status_code=400, detail="invalid json") event_id = event.get("id", "") event_type = event.get("event", "") data = event.get("data", {}) # 4. Idempotency — skip already-processed events (retries re-deliver same id) if event_id in processed_event_ids: logger.info(f"Duplicate event ignored: {event_id}") return JSONResponse({"received": True, "duplicate": True}) processed_event_ids.add(event_id) # 5. Respond 200 immediately; heavy work runs after logger.info(f"Event received: {event_type} ({event_id})") # 6. Route by event type if event_type == "position.liquidated": market = data.get("market_id") loss = data.get("loss_amount", 0) logger.error(f"LIQUIDATED {market} — loss=${loss:.2f}. Triggering rebalance.") # await rebalance_portfolio(data) elif event_type == "pnl.alert": pnl = data.get("unrealized_pnl", 0) pct = data.get("pnl_pct", 0) logger.warning(f"PNL ALERT: {data.get('market_id')} {pct:.1f}% (${pnl:.2f})") elif event_type == "funding.paid": amt = data.get("amount", 0) direction = data.get("direction") logger.info(f"FUNDING: {direction} ${amt:.4f} on {data.get('market_id')}") elif event_type == "bet.won": logger.info(f"BET WON: ${data.get('profit', 0):.2f} on {data.get('game_name')}") elif event_type == "bet.lost": logger.info(f"BET LOST: ${data.get('loss_amount', 0):.2f} on {data.get('game_id')}") elif event_type == "balance.low": bal = data.get("current_balance", 0) threshold = data.get("threshold", 0) logger.warning(f"LOW BALANCE: ${bal:.2f} (threshold ${threshold:.2f})") # await top_up_balance(data) elif event_type == "swap.completed": logger.info( f"SWAP: {data.get('from_amount')} {data.get('from_token')} " f"-> {data.get('to_amount', 0):.2f} {data.get('to_token')}" ) elif event_type == "swap.failed": logger.error( f"SWAP FAILED: {data.get('from_token')}->{data.get('to_token')} " f"reason={data.get('failure_reason')}" ) elif event_type == "balance.received": logger.info( f"RECEIVED: {data.get('amount')} {data.get('token')} on {data.get('chain')}" ) elif event_type == "domain.expiring": days = data.get("days_remaining") logger.warning(f"DOMAIN EXPIRING: {data.get('domain')} in {days} days") # await renew_domain(data) elif event_type == "dns.updated": logger.info( f"DNS: {data.get('action')} {data.get('record_type')} " f"{data.get('record_name')} on {data.get('domain')}" ) elif event_type == "referral.earned": commission = data.get("commission_usd", 0) cumulative = data.get("cumulative_earned_usd", 0) logger.info(f"REFERRAL: +${commission:.2f} (total ${cumulative:.2f})") if cumulative >= 50.0: pass # await trigger_withdrawal() elif event_type == "agent.registered": logger.info( f"NEW AGENT: {data.get('username')} ({data.get('agent_id')}) " f"via referral {data.get('referral_code')}" ) else: logger.debug(f"Unhandled event type: {event_type}") return JSONResponse({"received": True}) if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8080)

import express from 'express'; import crypto from 'crypto'; const app = express(); const WEBHOOK_SECRET = 'whsec_your_random_secret_here'; const TIMESTAMP_TOLERANCE_MS = 300_000; // 5 minutes // Must use raw body for signature verification — do NOT use express.json() here app.use('/webhooks/purpleflea', express.raw({ type: 'application/json' })); app.use(express.json()); const processedEventIds = new Set(); // use Redis in production function verifySignature(rawBody, timestamp, signatureHeader) { if (!signatureHeader?.startsWith('sha256=')) return false; const receivedSig = signatureHeader.slice('sha256='.length); const signedPayload = `${timestamp}.${rawBody}`; const computed = crypto .createHmac('sha256', WEBHOOK_SECRET) .update(signedPayload) .digest('hex'); try { return crypto.timingSafeEqual(Buffer.from(computed), Buffer.from(receivedSig)); } catch { return false; // length mismatch guard } } function checkTimestamp(timestampStr) { const ts = parseInt(timestampStr, 10); if (isNaN(ts)) return false; return Math.abs(Date.now() - ts * 1000) <= TIMESTAMP_TOLERANCE_MS; } app.post('/webhooks/purpleflea', (req, res) => { const timestamp = req.headers['x-purpleflea-timestamp'] ?? ''; const signature = req.headers['x-purpleflea-signature'] ?? ''; const rawBody = req.body; // Buffer from express.raw() if (!checkTimestamp(timestamp)) { return res.status(400).json({ error: 'timestamp missing or too old' }); } if (!verifySignature(rawBody, timestamp, signature)) { console.error('Webhook signature mismatch'); return res.status(401).json({ error: 'invalid signature' }); } // Respond 200 immediately — process asynchronously res.json({ received: true }); let event; try { event = JSON.parse(rawBody); } catch { console.error('Failed to parse webhook body'); return; } const { id: eventId, event: eventType, data } = event; // Idempotency guard if (processedEventIds.has(eventId)) { console.log(`Duplicate event skipped: ${eventId}`); return; } processedEventIds.add(eventId); // Evict old IDs after 1 hour to prevent unbounded growth setTimeout(() => processedEventIds.delete(eventId), 3_600_000); handleEvent(eventType, data, eventId).catch(console.error); }); async function handleEvent(eventType, data, eventId) { console.log(`[${eventId}] ${eventType}`); switch (eventType) { case 'position.liquidated': console.error(`LIQUIDATED: ${data.market_id} loss=$${data.loss_amount}`); // await rebalancePortfolio(data); break; case 'pnl.alert': console.warn(`PNL ALERT: ${data.market_id} ${data.pnl_pct}% ($${data.unrealized_pnl})`); break; case 'funding.paid': console.log(`FUNDING: ${data.direction} $${data.amount} on ${data.market_id}`); break; case 'bet.won': console.log(`BET WON: $${data.profit} on ${data.game_name}`); break; case 'bet.lost': console.log(`BET LOST: $${data.loss_amount} on ${data.game_id}`); break; case 'balance.low': console.warn(`LOW BALANCE: $${data.current_balance} (threshold $${data.threshold})`); break; case 'swap.completed': console.log(`SWAP: ${data.from_amount} ${data.from_token} -> ${data.to_amount} ${data.to_token}`); break; case 'swap.failed': console.error(`SWAP FAILED: ${data.from_token}->${data.to_token} reason=${data.failure_reason}`); break; case 'balance.received': console.log(`RECEIVED: ${data.amount} ${data.token} on ${data.chain}`); break; case 'domain.expiring': console.warn(`DOMAIN EXPIRING: ${data.domain} in ${data.days_remaining} days`); break; case 'dns.updated': console.log(`DNS: ${data.action} ${data.record_type} ${data.record_name} on ${data.domain}`); break; case 'referral.earned': console.log(`REFERRAL: +$${data.commission_usd} (total $${data.cumulative_earned_usd})`); if (data.cumulative_earned_usd >= 50) { // await triggerWithdrawal(); } break; case 'agent.registered': console.log(`NEW AGENT: ${data.username} (${data.agent_id})`); break; default: console.log(`Unhandled event: ${eventType}`); } } app.listen(8080, () => console.log('Webhook receiver listening on :8080'));