Receiving Kairon webhooks

{
  "id": "evt_01k5x2q3r4s5t6v7w8x9y0z1a2",
  "type": "lead.stage.changed",
  "created": 1790000000,
  "data": {
    "object": {
      "id": "lead_00000000000000000000000000",
      "object": "lead",
      "created_at": "2026-08-14T09:22:41.310Z",
      "stage": "meeting_booked",
      "full_name": "Sample Person",
      "first_name": "Sample",
      "last_name": "Person",
      "headline": "Head of Revenue Operations at Example Co",
      "summary": "This is a test event from Kairon. No such person exists.",
      "location": "Madrid, Community of Madrid, Spain",
      "avatar_url": null,
      "linkedin_url": "https://www.linkedin.com/in/sample-person",
      "linkedin_public_identifier": "sample-person",
      "email": "sample.person@example.com",
      "email_source": "manual",
      "email_found_at": "2026-08-20T11:04:02.881Z",
      "phone": "+1 415 555 0100",
      "phone_source": "apollo",
      "phone_found_at": "2026-08-21T09:12:40.105Z",
      "current_company": {
        "id": "comp_00000000000000000000000000",
        "name": "Example Co",
        "domain": "example.com",
        "linkedin_url": "https://www.linkedin.com/company/example",
        "industry": "Software Development",
        "employee_count": 240,
        "location": "Madrid, Spain"
      },
      "positions": [
        {
          "title": "Head of Revenue Operations",
          "company_name": "Example Co",
          "company_linkedin_id": "1234567",
          "start": "2024-03",
          "end": null,
          "is_current": true,
          "description": "Owns the go-to-market stack and pipeline reporting."
        },
        {
          "title": "Revenue Operations Manager",
          "company_name": "Former Co",
          "company_linkedin_id": null,
          "start": "2021-01",
          "end": "2024-02",
          "is_current": false,
          "description": null
        }
      ],
      "education": [
        {
          "school": "Example University",
          "degree": "BSc",
          "field_of_study": "Business Administration",
          "start": "2008",
          "end": "2012"
        }
      ],
      "owner": {
        "id": "user_00000000000000000000000000",
        "name": "Sample Owner",
        "email": "owner@example.com"
      },
      "tags": [
        {
          "id": "tag_00000000000000000000000000",
          "name": "Sample tag"
        }
      ],
      "qualification": {
        "status": "qualified",
        "decided_by": "kairon",
        "decided_at": "2026-08-15T07:31:19.002Z",
        "icp_version_id": "icpv_00000000000000000000000000",
        "rationale": "A sample rationale. Kairon explains every verdict it reaches."
      }
    },
    "previous_attributes": {
      "stage": "interested"
    }
  }
}

import { createHmac, timingSafeEqual } from 'node:crypto';

/**
 * True when a request really came from Kairon.
 * rawBody: the body exactly as it arrived (a Buffer or string), never re-serialized JSON.
 */
export function verifyKaironSignature(rawBody, header, secret, toleranceSeconds = 300) {
  const parts = {};
  for (const piece of String(header ?? '').split(',')) {
    const at = piece.indexOf('=');
    if (at > 0) parts[piece.slice(0, at).trim()] ??= piece.slice(at + 1).trim();
  }
  const t = Number(parts.t);
  if (!Number.isInteger(t) || !parts.v1) return false;
  if (Math.abs(Date.now() / 1000 - t) > toleranceSeconds) return false;

  const expected = createHmac('sha256', secret).update(`${t}.${rawBody}`).digest('hex');
  const got = Buffer.from(parts.v1);
  const want = Buffer.from(expected);
  return got.length === want.length && timingSafeEqual(got, want);
}

import hashlib
import hmac
import time


def verify_kairon_signature(raw_body: bytes, header: str, secret: str, tolerance: int = 300) -> bool:
    """True when a request really came from Kairon. raw_body: the bytes exactly as they arrived."""
    parts = {}
    for piece in (header or "").split(","):
        key, sep, value = piece.partition("=")
        if sep and key.strip() and key.strip() not in parts:
            parts[key.strip()] = value.strip()
    try:
        t = int(parts["t"])
    except (KeyError, ValueError):
        return False
    if "v1" not in parts or abs(time.time() - t) > tolerance:
        return False
    signed = f"{t}.".encode() + raw_body
    expected = hmac.new(secret.encode(), signed, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected.encode(), parts["v1"].encode())

import express from 'express';
import { verifyKaironSignature } from './verify-kairon-signature.js';

const app = express();

// express.raw, NOT express.json: the signature is over the exact bytes we sent.
app.post('/kairon/webhook', express.raw({ type: 'application/json' }), async (req, res) => {
  const ok = verifyKaironSignature(req.body, req.get('Kairon-Signature'), process.env.KAIRON_WEBHOOK_SECRET);
  if (!ok) return res.status(400).send('bad signature');

  const event = JSON.parse(req.body.toString('utf8'));

  // More event types will come: answer 200 to one you do not handle, so it is not retried.
  if (event.type !== 'lead.stage.changed') return res.sendStatus(200);

  // A test event ("Send test event", webhook_test) carries the invented sample lead: accept it, store nothing.
  if (event.data.object.id === 'lead_00000000000000000000000000') return res.sendStatus(200);

  // Same id on every retry and every "Send again": skip what you already have.
  if (await alreadyStored(event.id)) return res.sendStatus(200);

  // Store it, answer, and do the slow work afterwards (a queue, a job, a background task).
  await storeEvent(event);
  res.sendStatus(200);
});

import json
import os

from flask import Flask, request

from verify_kairon_signature import verify_kairon_signature

app = Flask(__name__)


@app.post("/kairon/webhook")
def kairon_webhook():
    # get_data(), NOT get_json(): the signature is over the exact bytes we sent.
    raw = request.get_data()
    if not verify_kairon_signature(raw, request.headers.get("Kairon-Signature"), os.environ["KAIRON_WEBHOOK_SECRET"]):
        return "bad signature", 400

    event = json.loads(raw)

    # More event types will come: answer 200 to one you do not handle, so it is not retried.
    if event["type"] != "lead.stage.changed":
        return "", 200

    # A test event ("Send test event", webhook_test) carries the invented sample lead: accept it, store nothing.
    if event["data"]["object"]["id"] == "lead_00000000000000000000000000":
        return "", 200

    # Same id on every retry and every "Send again": skip what you already have.
    if already_stored(event["id"]):
        return "", 200

    # Store it, answer, and do the slow work afterwards.
    store_event(event)
    return "", 200