Webhooks

curl -X POST https://api.boxline.dev/v1/webhooks \  -H "x-api-key: $BOXLINE_API_KEY" \  -H "content-type: application/json" \  -d '{"url": "https://example.com/webhooks/boxline", "events": ["*"]}'

import { Boxline } from "@boxline/sdk";const bx = new Boxline(); // reads BOXLINE_API_KEY and BOXLINE_API_URLconst endpoint = await bx.webhooks.create({ url: "https://example.com/webhooks/boxline", events: ["*"] });// endpoint.secret is shown only now: store it in your secret manager.

from boxline import Boxlinebx = Boxline()  # reads BOXLINE_API_KEY and BOXLINE_API_URLendpoint = bx.webhooks.create("https://example.com/webhooks/boxline", ["*"])# endpoint["secret"] is shown only now: store it in your secret manager.

curl https://api.boxline.dev/v1/webhooks/event-types \  -H "x-api-key: $BOXLINE_API_KEY"

import { Boxline } from "@boxline/sdk";const bx = new Boxline(); // reads BOXLINE_API_KEY and BOXLINE_API_URLconst { data, all } = await bx.webhooks.eventTypes();for (const e of data) console.log(e.group, e.type, e.description);console.log("every type, also later ones:", all); // "*"

from boxline import Boxlinebx = Boxline()  # reads BOXLINE_API_KEY and BOXLINE_API_URLtypes = bx.webhooks.event_types()for e in types["data"]:    print(e["group"], e["type"], e["description"])print("every type, also later ones:", types["all"])  # "*"

curl -X POST https://api.boxline.dev/v1/webhooks \  -H "x-api-key: $BOXLINE_API_KEY" \  -H "content-type: application/json" \  -d '{"url": "https://example.com/webhooks/boxline", "events": ["session.stopped", "agent_run.finished"]}'# The answer holds "secret": "whsec_…", shown only this once.

import { Boxline } from "@boxline/sdk";const bx = new Boxline(); // reads BOXLINE_API_KEY and BOXLINE_API_URLconst endpoint = await bx.webhooks.create({  url: "https://example.com/webhooks/boxline",  events: ["session.stopped", "agent_run.finished"],});// endpoint.secret ("whsec_…") is shown only now: put it in your secret manager, never in code or logs.console.log(endpoint.id, endpoint.events);

from boxline import Boxlinebx = Boxline()  # reads BOXLINE_API_KEY and BOXLINE_API_URLendpoint = bx.webhooks.create("https://example.com/webhooks/boxline", ["session.stopped", "agent_run.finished"])# endpoint["secret"] ("whsec_…") is shown only now: put it in your secret manager, never in code or logs.print(endpoint["id"], endpoint["events"])

import { verifyWebhook, WebhookSignatureError, type WebhookEvent } from "@boxline/sdk";// body: the raw request body (a Buffer or string) exactly as it arrived; header: its Boxline-Signature.export function receive(body: Buffer, header: string | undefined): { status: number; event?: WebhookEvent } {  try {    const event = verifyWebhook(body, header, process.env.BOXLINE_WEBHOOK_SECRET!); // within 5 minutes by default    return { status: 200, event };  } catch (err) {    if (err instanceof WebhookSignatureError) return { status: 400 }; // err.reason: malformed, timestamp_out_of_range, no_matching_signature    throw err;  }}

import osfrom boxline import WebhookSignatureError, verify_webhookdef receive(body: bytes, header):    """body: the raw request body exactly as it arrived; header: its Boxline-Signature."""    try:        event = verify_webhook(body, header, os.environ["BOXLINE_WEBHOOK_SECRET"])  # within 5 minutes by default        return 200, event    except WebhookSignatureError as e:  # e.reason: malformed, timestamp_out_of_range, no_matching_signature        return 400, None

import { createHmac, timingSafeEqual } from "node:crypto";export function verifyWebhook(body: Buffer | string, header: string, secret: string, toleranceSeconds = 300) {  const pairs = String(header).split(",").map((p) => [p.slice(0, p.indexOf("=")), p.slice(p.indexOf("=") + 1)]);  const t = Number(pairs.find(([k, v]) => k === "t" && /^\d+$/.test(v))?.[1]);  if (!t || Math.abs(Date.now() / 1000 - t) > toleranceSeconds) throw new Error("timestamp out of range");  const want = createHmac("sha256", secret).update(`${t}.`).update(body).digest();  const ok = pairs.some(([k, v]) => {    const got = Buffer.from(k === "v1" ? v : "", "hex");    return got.length === want.length && timingSafeEqual(got, want);  });  if (!ok) throw new Error("no matching signature");  return JSON.parse(body.toString());}

import hashlibimport hmacimport jsonimport timedef verify_webhook(body: bytes, header: str, secret: str, tolerance: int = 300):    pairs = [p.split("=", 1) for p in header.split(",") if "=" in p]    t = next((int(v) for k, v in pairs if k == "t" and v.isdigit()), None)    if t is None or abs(time.time() - t) > tolerance:        raise ValueError("timestamp out of range")    want = hmac.new(secret.encode(), f"{t}.".encode() + body, hashlib.sha256).digest()    got = [bytes.fromhex(v) for k, v in pairs if k == "v1" and len(v) == 64 and all(c in "0123456789abcdef" for c in v)]    if not any(hmac.compare_digest(g, want) for g in got):        raise ValueError("no matching signature")    return json.loads(body)

import { verifyWebhook, WebhookSignatureError, type WebhookEventPayload } from "@boxline/sdk";const handled = new Set<string>(); // keep handled event ids in your database// body: the raw request body exactly as it arrived; header: its Boxline-Signature. Returns the status to answer.export function receive(body: Buffer, header: string | undefined): number {  let event: WebhookEventPayload;  try {    event = verifyWebhook(body, header, process.env.BOXLINE_WEBHOOK_SECRET!) as WebhookEventPayload;  } catch (err) {    if (err instanceof WebhookSignatureError) return 400;    throw err;  }  if (handled.has(event.id)) return 200; // delivered twice: already handled  handled.add(event.id);  switch (event.type) {    case "agent_run.finished": {      const run = event.data; // typed: status, errorCode, resumable, variableNames, usage…      if (run.resumable) console.log(`run ${run.id} stopped at ${run.errorCode}: resume it before ${run.resumable.until}`);      else console.log(`run ${run.id} ${run.status}`, run.resultText);      break;    }    case "session.expiring":      console.log(`session ${event.data.sessionId} stops in ${event.data.secondsLeft} s: give it a longer timeout now if it should go on`);      break;    case "usage.limit_reached":      console.log(`limit ${event.data.kind} reached (${event.data.used} of ${event.data.limit}), resets ${event.data.resetsAt}`);      break;    case "credential.changed":      console.log(`${event.data.type} ${event.data.name} ${event.data.action} by ${event.data.by}`); // never a value      break;    case "credential.code_needed":      console.log(`a run waits for a ${event.data.type} for ${event.data.credential}: send it now`); // never the code or link      break;    default:      break; // a type this receiver does not handle, or one added later: ignore it, and still answer 200  }  return 200;}

import osfrom typing import castfrom boxline import WebhookSignatureError, verify_webhookfrom boxline.types import WebhookEventPayloadhandled: set = set()  # keep handled event ids in your databasedef receive(body: bytes, header) -> int:    """body: the raw request body exactly as it arrived; header: its Boxline-Signature. Returns the status to answer."""    try:        event = cast(WebhookEventPayload, verify_webhook(body, header, os.environ["BOXLINE_WEBHOOK_SECRET"]))    except WebhookSignatureError:        return 400    if event["id"] in handled:        return 200  # delivered twice: already handled    handled.add(event["id"])    if event["type"] == "agent_run.finished":        run = event["data"]  # status, errorCode, resumable, variableNames, usage…        if run["resumable"]:            print(f"run {run['id']} stopped at {run['errorCode']}: resume it before {run['resumable']['until']}")        else:            print(f"run {run['id']} {run['status']}", run["resultText"])    elif event["type"] == "session.expiring":        print(f"session {event['data']['sessionId']} stops in {event['data']['secondsLeft']} s: give it a longer timeout now if it should go on")    elif event["type"] == "usage.limit_reached":        d = event["data"]        print(f"limit {d['kind']} reached ({d['used']} of {d['limit']}), resets {d['resetsAt']}")    elif event["type"] == "credential.changed":        d = event["data"]        print(f"{d['type']} {d['name']} {d['action']} by {d['by']}")  # never a value    elif event["type"] == "credential.code_needed":        d = event["data"]        print(f"a run waits for a {d['type']} for {d['credential']}: send it now")  # never the code or link    # Any other type, or one added later: ignore it, and still answer 200.    return 200

curl -X PATCH https://api.boxline.dev/v1/webhooks/$WEBHOOK_ID \  -H "x-api-key: $BOXLINE_API_KEY" \  -H "content-type: application/json" \  -d '{"rotateSecret": true}'# The answer holds the new "secret" (shown only now) and "previousSecretExpiresAt".

import { Boxline } from "@boxline/sdk";const bx = new Boxline(); // reads BOXLINE_API_KEY and BOXLINE_API_URLconst rotated = await bx.webhooks.update("WEBHOOK_ID", { rotateSecret: true });// Store rotated.secret. Until rotated.previousSecretExpiresAt, deliveries carry both signatures, so your// receiver can check both while you roll out the new secret: verifyWebhook(body, header, [newSecret, oldSecret]).console.log("the old secret signs until", rotated.previousSecretExpiresAt);

from boxline import Boxlinebx = Boxline()  # reads BOXLINE_API_KEY and BOXLINE_API_URLrotated = bx.webhooks.update("WEBHOOK_ID", rotate_secret=True)# Store rotated["secret"]. Until previousSecretExpiresAt, deliveries carry both signatures, so your# receiver can check both while you roll out the new secret: verify_webhook(body, header, [new_secret, old_secret]).print("the old secret signs until", rotated["previousSecretExpiresAt"])

curl -X PATCH https://api.boxline.dev/v1/webhooks/$WEBHOOK_ID \  -H "x-api-key: $BOXLINE_API_KEY" \  -H "content-type: application/json" \  -d '{"enabled": true}'

import { Boxline } from "@boxline/sdk";const bx = new Boxline(); // reads BOXLINE_API_KEY and BOXLINE_API_URLconst w = await bx.webhooks.get("WEBHOOK_ID");if (!w.enabled) console.log("switched off:", w.disabledReason, "since", w.disabledAt);await bx.webhooks.update("WEBHOOK_ID", { enabled: true }); // on again, with its failures forgotten

from boxline import Boxlinebx = Boxline()  # reads BOXLINE_API_KEY and BOXLINE_API_URLw = bx.webhooks.get("WEBHOOK_ID")if not w["enabled"]:    print("switched off:", w["disabledReason"], "since", w["disabledAt"])bx.webhooks.update("WEBHOOK_ID", enabled=True)  # on again, with its failures forgotten

# Failed deliveries, newest first, with each attempt.curl "https://api.boxline.dev/v1/webhooks/$WEBHOOK_ID/deliveries?status=failed" \  -H "x-api-key: $BOXLINE_API_KEY"# Send one again now, with the same event id and body.curl -X POST https://api.boxline.dev/v1/webhooks/$WEBHOOK_ID/deliveries/$DELIVERY_ID/retry \  -H "x-api-key: $BOXLINE_API_KEY"

import { Boxline } from "@boxline/sdk";const bx = new Boxline(); // reads BOXLINE_API_KEY and BOXLINE_API_URLfor await (const d of bx.webhooks.deliveries("WEBHOOK_ID", { status: "failed" })) {  console.log(d.id, d.eventType, d.attempts, d.errorCode, d.responseStatus);  for (const a of d.history) console.log("  ", a.at, a.status ?? a.errorCode, a.durationMs, "ms");}const again = await bx.webhooks.retryDelivery("WEBHOOK_ID", "DELIVERY_ID");console.log(again.status, again.responseStatus);

from boxline import Boxlinebx = Boxline()  # reads BOXLINE_API_KEY and BOXLINE_API_URLfor d in bx.webhooks.deliveries("WEBHOOK_ID", status="failed"):    print(d["id"], d["eventType"], d["attempts"], d["errorCode"], d["responseStatus"])    for a in d["history"]:        print("  ", a["at"], a.get("status") or a.get("errorCode"), a["durationMs"], "ms")again = bx.webhooks.retry_delivery("WEBHOOK_ID", "DELIVERY_ID")print(again["status"], again["responseStatus"])

# webhook.test, or with "type" a made-up sample of that event (its body has "test": true):curl -X POST https://api.boxline.dev/v1/webhooks/$WEBHOOK_ID/test \  -H "x-api-key: $BOXLINE_API_KEY" \  -H "content-type: application/json" \  -d '{"type": "agent_run.finished"}'

import { Boxline } from "@boxline/sdk";const bx = new Boxline(); // reads BOXLINE_API_KEY and BOXLINE_API_URL// webhook.test, or with type a made-up sample of that event (its body has test: true):const delivery = await bx.webhooks.test("WEBHOOK_ID", { type: "agent_run.finished" }); // waits up to 12 s for your answerconsole.log(delivery.status, delivery.responseStatus, delivery.errorCode, delivery.responseBody);

from boxline import Boxlinebx = Boxline()  # reads BOXLINE_API_KEY and BOXLINE_API_URL# webhook.test, or with type a made-up sample of that event (its body has test: True):delivery = bx.webhooks.test("WEBHOOK_ID", type="agent_run.finished")  # waits up to 12 s for your answerprint(delivery["status"], delivery["responseStatus"], delivery["errorCode"], delivery["responseBody"])

import { createServer } from "node:http";import { Boxline, WebhookSignatureError, verifyWebhook } from "@boxline/sdk";const port = 4900;const url = `http://127.0.0.1:${port}/webhooks`;// The endpoint's secret; after a rotation, both: "whsec_new,whsec_old".let secrets = (process.env.BOXLINE_WEBHOOK_SECRET ?? "").split(",").filter(Boolean);const bx = new Boxline();let registered: string | undefined;if (process.argv.includes("--register")) {  // Local development only: your local API must run with WEBHOOKS_ALLOW_LOCAL=1. Real endpoints are public HTTPS.  const endpoint = await bx.webhooks.create({ url, events: ["session.stopped", "agent_run.finished", "crawl.finished"] });  secrets = [endpoint.secret]; // kept in memory, never printed  registered = endpoint.id;}if (!secrets.length) throw new Error("set BOXLINE_WEBHOOK_SECRET, or pass --register");const handled = new Set<string>(); // event ids already handled; a real receiver keeps them in its databaseconst server = createServer((req, res) => {  if (req.method !== "POST" || req.url !== "/webhooks") return void res.writeHead(404).end();  const chunks: Buffer[] = [];  req.on("data", (c: Buffer) => chunks.push(c));  req.on("end", () => {    const body = Buffer.concat(chunks); // the raw bytes: re-serialised JSON would not match the signature    try {      const event = verifyWebhook(body, req.headers["boxline-signature"] as string | undefined, secrets);      if (!handled.has(event.id)) {        handled.add(event.id); // a delivery can arrive more than once: drop the ones you have handled        console.log(event.type, event.id, JSON.stringify(event.data).slice(0, 120));      }      res.writeHead(200).end("ok"); // answer fast; do slow work afterwards    } catch (err) {      if (!(err instanceof WebhookSignatureError)) throw err;      res.writeHead(400).end(err.reason);    }  });});server.listen(port, "127.0.0.1", () => console.log(`listening on ${url}`));process.on("SIGINT", async () => {  if (registered) await bx.webhooks.delete(registered); // clean up the local endpoint  process.exit(0);});

import osimport sysfrom http.server import BaseHTTPRequestHandler, ThreadingHTTPServerfrom boxline import Boxline, WebhookSignatureError, verify_webhookPORT = 4900URL = f"http://127.0.0.1:{PORT}/webhooks"# The endpoint's secret; after a rotation, both: "whsec_new,whsec_old".secrets = [s for s in os.environ.get("BOXLINE_WEBHOOK_SECRET", "").split(",") if s]bx = Boxline()registered = Noneif "--register" in sys.argv:    # Local development only: your local API must run with WEBHOOKS_ALLOW_LOCAL=1. Real endpoints are public HTTPS.    endpoint = bx.webhooks.create(URL, ["session.stopped", "agent_run.finished", "crawl.finished"])    secrets = [endpoint["secret"]]  # kept in memory, never printed    registered = endpoint["id"]if not secrets:    raise SystemExit("set BOXLINE_WEBHOOK_SECRET, or pass --register")handled = set()  # event ids already handled; a real receiver keeps them in its databaseclass Receiver(BaseHTTPRequestHandler):    def answer(self, status: int, text: str) -> None:        self.send_response(status)        self.send_header("content-length", str(len(text)))        self.end_headers()        self.wfile.write(text.encode())    def do_POST(self) -> None:        if self.path != "/webhooks":            return self.answer(404, "")        # The raw bytes: re-serialised JSON would not match the signature.        body = self.rfile.read(int(self.headers.get("content-length") or 0))        try:            event = verify_webhook(body, self.headers.get("Boxline-Signature"), secrets)        except WebhookSignatureError as e:            return self.answer(400, e.reason)        if event["id"] not in handled:            handled.add(event["id"])  # a delivery can arrive more than once: drop the ones you have handled            print(event["type"], event["id"], str(event["data"])[:120])        self.answer(200, "ok")  # answer fast; do slow work afterwardsserver = ThreadingHTTPServer(("127.0.0.1", PORT), Receiver)print(f"listening on {URL}")try:    server.serve_forever()except KeyboardInterrupt:    if registered:        bx.webhooks.delete(registered)  # clean up the local endpoint