Webhooks

curl https://api.aisearchapi.dev/v1/search \
  -H "Authorization: Bearer $AISEARCH_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "query": "best crm for startups",
    "surfaces": ["chatgpt", "perplexity"],
    "regions": [{ "country": "US" }],
    "webhook": {
      "url": "https://example.com/hooks/aisearch",
      "secret": "whsec_your_signing_secret"
    }
  }'

const res = await fetch('https://api.aisearchapi.dev/v1/search', {
  method: 'POST',
  headers: {
    Authorization: `Bearer ${process.env.AISEARCH_API_KEY}`,
    'Content-Type': 'application/json',
  },
  body: JSON.stringify({
    query: 'best crm for startups',
    surfaces: ['chatgpt', 'perplexity'],
    regions: [{ country: 'US' }],
    webhook: {
      url: 'https://example.com/hooks/aisearch',
      secret: 'whsec_your_signing_secret',
    },
  }),
});

const job = await res.json();

import os, requests

res = requests.post(
    "https://api.aisearchapi.dev/v1/search",
    headers={
        "Authorization": f"Bearer {os.environ['AISEARCH_API_KEY']}",
        "Content-Type": "application/json",
    },
    json={
        "query": "best crm for startups",
        "surfaces": ["chatgpt", "perplexity"],
        "regions": [{"country": "US"}],
        "webhook": {
            "url": "https://example.com/hooks/aisearch",
            "secret": "whsec_your_signing_secret",
        },
    },
)

job = res.json()

import express from "express";
import { createHmac, timingSafeEqual } from "node:crypto";

const app = express();
const SECRET = process.env.AISEARCH_WEBHOOK_SECRET;
const MAX_SKEW_SECONDS = 300; // reject deliveries older than ~5 minutes

function verify(rawBody, timestamp, signature, secret) {
const ts = Number(timestamp);
if (!Number.isFinite(ts)) return false;

// Bound the clock skew to defeat replays.
if (Math.abs(Date.now() / 1000 - ts) > MAX_SKEW_SECONDS) return false;

// Sign `${timestamp}.${body}` over the exact received bytes.
const signed = Buffer.concat([Buffer.from(`${timestamp}.`), rawBody]);
const expected = createHmac("sha256", secret).update(signed).digest("hex");
const a = Buffer.from(expected);
const b = Buffer.from(signature ?? "");
return a.length === b.length && timingSafeEqual(a, b);
}

// Capture the raw bytes — do NOT use express.json() on this route.
app.post(
"/hooks/aisearch",
express.raw({ type: "application/json" }),
(req, res) => {
const timestamp = req.get("X-AISearch-Timestamp");
const signature = req.get("X-AISearch-Signature");

    if (!verify(req.body, timestamp, signature, SECRET)) {
      return res.status(401).send("invalid signature");
    }

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

    // Acknowledge fast, then process out of band. Dedupe on event.id.
    res.status(200).send("ok");
    queueForProcessing(event); // e.g. push to a job queue

}
);

app.listen(3000);

import hmac
import hashlib
import os
import time
from flask import Flask, request, abort

app = Flask(__name__)
SECRET = os.environ["AISEARCH_WEBHOOK_SECRET"].encode()
MAX_SKEW_SECONDS = 300  # reject deliveries older than ~5 minutes


def verify(raw_body: bytes, timestamp: str | None, signature: str | None) -> bool:
    if timestamp is None or signature is None:
        return False
    try:
        ts = int(timestamp)
    except ValueError:
        return False

    # Bound the clock skew to defeat replays.
    if abs(time.time() - ts) > MAX_SKEW_SECONDS:
        return False

    # Sign `${timestamp}.${body}` over the exact received bytes.
    signed = timestamp.encode() + b"." + raw_body
    expected = hmac.new(SECRET, signed, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, signature)


@app.post("/hooks/aisearch")
def aisearch_webhook():
    # request.get_data() returns the exact raw bytes.
    raw = request.get_data()
    timestamp = request.headers.get("X-AISearch-Timestamp")
    signature = request.headers.get("X-AISearch-Signature")

    if not verify(raw, timestamp, signature):
        abort(401)

    event = request.get_json()

    # Acknowledge fast, then process out of band. Dedupe on event["id"].
    queue_for_processing(event)
    return "ok", 200