When to use it: You want to hear when a price or stock level changes, without polling for it.

Uses	`tasks.create` with `output` and `schedule`, `tasks.run` and `waitForRun` (`wait_for_run`), `webhooks.create`, `verifyWebhook` (`verify_webhook`)
Needs	a plan with task schedules, and a model on the API; the receiver must be reachable by the API (public HTTPS; a local API with `WEBHOOKS_ALLOW_LOCAL=1` also delivers to this computer)
Site	a books.toscrape.com product page (a demo shop)
Output	`output/result.json` (the schedule, each run, each webhook and what the receiver made of it) and `output/last-price.json` (the receiver's memory)
Languages	Node and Python

receiver.ts / receiver.py is the receiver. The example starts it in its own process and runs the task right away (RUNS times) instead of waiting for the schedule; since that receiver stops with the example, it switches the schedule off at the end unless KEEP_SCHEDULE=1. To keep watching, run the receiver on a public HTTPS address (WEBHOOK_SECRET=whsec_… PORT=8787 npx tsx node/receiver.ts) and point a webhook endpoint at it.

The code

The example as it is in its folder. Both versions do the same job and write the same output/, so one result check covers both.

/** * Watch a price: a task reads a product's price and stock into a fixed shape (an output schema), a schedule runs it * every hour, and each finished run goes to your webhook receiver (task_run.finished), which compares the price with * the last one it saw. * *   npx tsx node/index.ts      (BOXLINE_API_KEY; PRODUCT_URL, SCHEDULE, RUNS; KEEP_SCHEDULE=1 to leave it running) * * The example starts the receiver (receiver.ts) in this process and runs the task right away RUNS times instead of * waiting for the schedule. Because this receiver stops with the example, it switches the schedule off at the end * unless KEEP_SCHEDULE=1 (run receiver.ts on a public HTTPS address for that). Writes output/result.json. */import { mkdirSync, writeFileSync } from "node:fs";import { join } from "node:path";import { Boxline, type TaskCreateParams } from "@boxline/sdk";import { startReceiver, type Price } from "./receiver.js";const out = process.env.OUTPUT_DIR ?? "output";const productUrl = process.env.PRODUCT_URL ?? "https://books.toscrape.com/catalogue/sapiens-a-brief-history-of-humankind_996/index.html";const cron = process.env.SCHEDULE ?? "0 * * * *"; // every hour, on the hour (at most every 5 minutes)const runs = Number(process.env.RUNS ?? 1);const keep = process.env.KEEP_SCHEDULE === "1";const bx = new Boxline();mkdirSync(out, { recursive: true });// 1. The receiver, and a webhook endpoint for it. Its secret is shown only now: the receiver keeps it in memory.const receiver = await startReceiver({ stateFile: join(out, "last-price.json") });const endpoint = await bx.webhooks.create({ url: receiver.url, events: ["task_run.finished"], description: "Watch a price (example)" });receiver.setSecret(endpoint.secret);console.log(`Webhook endpoint ${endpoint.id} → ${receiver.url}`);// 2. The task and its schedule; running the example again updates the task of the same name.const name = `Watch a price: ${new URL(productUrl).pathname.split("/").filter(Boolean).slice(-2, -1)[0] ?? productUrl}`.slice(0, 100);const spec: TaskCreateParams = {  name,  instruction: "Open %url% and read the product on it: its title, its price as a number, the currency, and whether it is in stock.",  variables: [{ name: "url", default: productUrl, description: "The product page" }],  output: {    type: "object",    properties: {      title: { type: "string", minLength: 1 },      price: { type: "number", exclusiveMinimum: 0 },      currency: { type: "string", description: "ISO 4217 code, e.g. GBP" },      inStock: { type: "boolean" },    },    required: ["title", "price", "currency", "inStock"],    additionalProperties: false,  },  maxSteps: 10,  schedule: { cron, timezone: "UTC", enabled: true },};const existing = (await bx.tasks.list({ limit: 100 })).data.find((t) => t.name === name);const task = existing ? await bx.tasks.update(existing.id, spec) : await bx.tasks.create(spec);const schedule = { cron: task.schedule!.cron, timezone: task.schedule!.timezone, nextRunAt: task.schedule!.nextRunAt, setAt: new Date().toISOString() };console.log(`Task ${task.id} "${name}": runs "${cron}" (UTC), next at ${schedule.nextRunAt}`);// 3. Run it now instead of waiting for the schedule; each finished run's webhook reaches the receiver.const done = [];try {  for (let i = 0; i < runs; i++) {    const started = await bx.tasks.run<Price>(task.id);    console.log(`Run ${started.id} (agent run ${started.runId}) · Session: ${started.sessionId}`);    const run = await bx.tasks.waitForRun<Price>(started, { timeoutMs: 5 * 60_000 });    console.log(`  ${run.status}: ${JSON.stringify(run.result)}`);    const delivery = await receiver.waitFor(run.id, 90_000);    done.push({ taskRunId: run.id, runId: run.runId, status: run.status, result: run.result, costUsd: run.usage.costUsd, webhook: delivery.message });  }} finally {  await receiver.close();  await bx.webhooks.delete(endpoint.id); // this receiver is gone  if (!keep) await bx.tasks.update(task.id, { schedule: { enabled: false } });}const after = await bx.tasks.get(task.id);console.log(keep ? `The schedule stays on: next run at ${after.schedule?.nextRunAt}` : "The schedule is switched off (this receiver stopped); KEEP_SCHEDULE=1 leaves it on.");writeFileSync(  join(out, "result.json"),  JSON.stringify(    {      productUrl,      task: { id: task.id, name, schedule, enabledAtEnd: after.schedule?.enabled ?? false },      runs: done,      deliveries: receiver.deliveries,      agentRuns: done.flatMap((r) => (r.runId ? [{ id: r.runId }] : [])),      made: { tasks: [task.id], webhooks: [endpoint.id] },    },    null,    2,  ),);

"""Watch a price: a task reads a product's price and stock into a fixed shape (an output schema), a schedule runs itevery hour, and each finished run goes to your webhook receiver (task_run.finished), which compares the price with thelast one it saw.    python python/main.py      (BOXLINE_API_KEY; PRODUCT_URL, SCHEDULE, RUNS; KEEP_SCHEDULE=1 to leave it running)The example starts the receiver (receiver.py) in this process and runs the task right away RUNS times instead ofwaiting for the schedule. Because this receiver stops with the example, it switches the schedule off at the end unlessKEEP_SCHEDULE=1 (run receiver.py on a public HTTPS address for that). Writes output/result.json."""import jsonimport osimport sysfrom datetime import datetime, timezonefrom pathlib import Pathfrom urllib.parse import urlparsesys.path.insert(0, str(Path(__file__).parent))from receiver import Receiver  # noqa: E402from boxline import Boxline  # noqa: E402out = Path(os.environ.get("OUTPUT_DIR", "output"))product_url = os.environ.get("PRODUCT_URL", "https://books.toscrape.com/catalogue/sapiens-a-brief-history-of-humankind_996/index.html")cron = os.environ.get("SCHEDULE", "0 * * * *")  # every hour, on the hour (at most every 5 minutes)runs = int(os.environ.get("RUNS", "1"))keep = os.environ.get("KEEP_SCHEDULE") == "1"bx = Boxline()out.mkdir(parents=True, exist_ok=True)# 1. The receiver, and a webhook endpoint for it. Its secret is shown only now: the receiver keeps it in memory.receiver = Receiver(out / "last-price.json")endpoint = bx.webhooks.create(receiver.url, ["task_run.finished"], description="Watch a price (example)")receiver.secret = endpoint["secret"]print(f"Webhook endpoint {endpoint['id']} -> {receiver.url}")# 2. The task and its schedule; running the example again updates the task of the same name.parts = [p for p in urlparse(product_url).path.split("/") if p]name = f"Watch a price: {parts[-2] if len(parts) >= 2 else product_url}"[:100]spec = dict(    instruction="Open %url% and read the product on it: its title, its price as a number, the currency, and whether it is in stock.",    variables=[{"name": "url", "default": product_url, "description": "The product page"}],    output={        "type": "object",        "properties": {            "title": {"type": "string", "minLength": 1},            "price": {"type": "number", "exclusiveMinimum": 0},            "currency": {"type": "string", "description": "ISO 4217 code, e.g. GBP"},            "inStock": {"type": "boolean"},        },        "required": ["title", "price", "currency", "inStock"],        "additionalProperties": False,    },    max_steps=10,    schedule={"cron": cron, "timezone": "UTC", "enabled": True},)existing = next((t for t in bx.tasks.list(limit=100).data if t["name"] == name), None)task = bx.tasks.update(existing["id"], name=name, **spec) if existing else bx.tasks.create(name, **spec)schedule = {**{k: task["schedule"][k] for k in ("cron", "timezone", "nextRunAt")}, "setAt": datetime.now(timezone.utc).isoformat()}print(f'Task {task["id"]} "{name}": runs "{cron}" (UTC), next at {schedule["nextRunAt"]}')# 3. Run it now instead of waiting for the schedule; each finished run's webhook reaches the receiver.done = []try:    for _ in range(runs):        started = bx.tasks.run(task["id"])        print(f"Run {started['id']} (agent run {started['runId']}) · Session: {started['sessionId']}", flush=True)        run = bx.tasks.wait_for_run(started, timeout=300)        print(f"  {run['status']}: {json.dumps(run['result'])}")        delivery = receiver.wait_for(run["id"], 90)        done.append({"taskRunId": run["id"], "runId": run["runId"], "status": run["status"], "result": run["result"],                     "costUsd": run["usage"]["costUsd"], "webhook": delivery["message"]})finally:    receiver.close()    bx.webhooks.delete(endpoint["id"])  # this receiver is gone    if not keep:        bx.tasks.update(task["id"], schedule={"enabled": False})after = bx.tasks.get(task["id"])if keep:    print(f"The schedule stays on: next run at {after['schedule']['nextRunAt']}")else:    print("The schedule is switched off (this receiver stopped); KEEP_SCHEDULE=1 leaves it on.")result = {    "productUrl": product_url,    "task": {"id": task["id"], "name": name, "schedule": schedule, "enabledAtEnd": bool(after["schedule"] and after["schedule"]["enabled"])},    "runs": done,    "deliveries": receiver.deliveries,    "agentRuns": [{"id": r["runId"]} for r in done if r["runId"]],    "made": {"tasks": [task["id"]], "webhooks": [endpoint["id"]]},}(out / "result.json").write_text(json.dumps(result, indent=2, ensure_ascii=False))

receiver, which the example starts in its own process:

/** * The webhook receiver of "Watch a price": it verifies each task_run.finished delivery against the raw body, drops * repeats (a delivery can arrive more than once), and compares the run's price with the last one it saw for that task * (kept in a JSON file). index.ts starts it in the same process; to keep watching, run it on a public HTTPS address: * *   WEBHOOK_SECRET=whsec_… PORT=8787 STATE_FILE=last-price.json npx tsx node/receiver.ts */import { existsSync, readFileSync, writeFileSync } from "node:fs";import { createServer } from "node:http";import type { AddressInfo } from "node:net";import { pathToFileURL } from "node:url";import { verifyWebhook, WebhookSignatureError } from "@boxline/sdk";export interface Price {  title: string;  price: number;  currency: string;  inStock: boolean;}export interface Delivery {  eventId: string;  taskRunId: string;  status: string;  price: number | null;  previous: number | null;  changed: boolean | null;  message: string;}export async function startReceiver(opts: { port?: number; stateFile: string; secret?: string }) {  let secret = opts.secret ?? "";  const seen = new Set<string>();  const deliveries: Delivery[] = [];  const waiting = new Map<string, (d: Delivery) => void>();  const server = createServer((req, res) => {    const chunks: Buffer[] = [];    req.on("data", (c: Buffer) => chunks.push(c));    req.on("end", () => {      let event;      try {        event = verifyWebhook<{ taskId: string; taskRunId: string; status: string; result: Price | null }>(Buffer.concat(chunks), req.headers["boxline-signature"] as string, secret);      } catch (err) {        res.writeHead(400).end(err instanceof WebhookSignatureError ? err.reason : "bad request");        return;      }      res.writeHead(200).end("ok");      if (event.type !== "task_run.finished" || seen.has(event.id)) return;      seen.add(event.id);      const { taskId, taskRunId, status, result } = event.data;      const state: Record<string, Price & { at: string }> = existsSync(opts.stateFile) ? JSON.parse(readFileSync(opts.stateFile, "utf8")) : {};      const last = state[taskId];      let delivery: Delivery;      if (status !== "completed" || !result) {        delivery = { eventId: event.id, taskRunId, status, price: null, previous: last?.price ?? null, changed: null, message: `the run ${status}: nothing to compare` };      } else {        const changed = last ? last.price !== result.price || last.inStock !== result.inStock : null;        const message =          changed === null            ? `first price seen: ${result.price} ${result.currency}`            : changed              ? `CHANGED: ${last!.price} → ${result.price} ${result.currency}${last!.inStock !== result.inStock ? `, now ${result.inStock ? "in stock" : "out of stock"}` : ""}`              : `no change: still ${result.price} ${result.currency}`;        state[taskId] = { ...result, at: event.createdAt };        writeFileSync(opts.stateFile, JSON.stringify(state, null, 2));        delivery = { eventId: event.id, taskRunId, status, price: result.price, previous: last?.price ?? null, changed, message };      }      deliveries.push(delivery);      console.log(`webhook ${event.id}: ${delivery.message}`);      waiting.get(taskRunId)?.(delivery);    });  });  await new Promise<void>((ok) => server.listen(opts.port ?? 0, "127.0.0.1", ok));  const port = (server.address() as AddressInfo).port;  return {    url: `http://127.0.0.1:${port}/boxline`,    deliveries,    setSecret: (s: string) => (secret = s),    /** The delivery for this task run (already here, or the next one to arrive). */    waitFor: (taskRunId: string, timeoutMs = 60_000) =>      new Promise<Delivery>((ok, fail) => {        const done = deliveries.find((d) => d.taskRunId === taskRunId);        if (done) return ok(done);        const timer = setTimeout(() => fail(new Error(`no webhook for task run ${taskRunId} within ${timeoutMs / 1000} s`)), timeoutMs);        waiting.set(taskRunId, (d) => (clearTimeout(timer), ok(d)));      }),    close: () => new Promise<void>((ok) => server.close(() => ok())),  };}// Run on its own: a long-running receiver.if (process.argv[1] && import.meta.url === pathToFileURL(process.argv[1]).href) {  if (!process.env.WEBHOOK_SECRET) throw new Error("Set WEBHOOK_SECRET to the endpoint's secret (whsec_…)");  const r = await startReceiver({ port: Number(process.env.PORT ?? 8787), stateFile: process.env.STATE_FILE ?? "last-price.json", secret: process.env.WEBHOOK_SECRET });  console.log(`Listening on ${r.url} (put it behind HTTPS on a public address)`);}

"""The webhook receiver of "Watch a price": it verifies each task_run.finished delivery against the raw body, dropsrepeats (a delivery can arrive more than once), and compares the run's price with the last one it saw for that task(kept in a JSON file). main.py starts it in a thread; to keep watching, run it on a public HTTPS address:    WEBHOOK_SECRET=whsec_… PORT=8787 STATE_FILE=last-price.json python python/receiver.py"""import jsonimport osimport threadingfrom http.server import BaseHTTPRequestHandler, ThreadingHTTPServerfrom pathlib import Pathfrom boxline import WebhookSignatureError, verify_webhookclass Receiver:    def __init__(self, state_file: Path, port: int = 0, secret: str = "") -> None:        self.state_file = Path(state_file)        self.secret = secret        self.deliveries: list = []        self._seen: set = set()        self._arrived = threading.Condition()        receiver = self        class Handler(BaseHTTPRequestHandler):            def do_POST(self) -> None:  # noqa: N802 (http.server's naming)                body = self.rfile.read(int(self.headers.get("Content-Length") or 0))                try:                    event = verify_webhook(body, self.headers.get("Boxline-Signature"), receiver.secret)                except WebhookSignatureError as e:                    self.send_response(400)                    self.end_headers()                    self.wfile.write(e.reason.encode())                    return                self.send_response(200)                self.end_headers()                self.wfile.write(b"ok")                receiver._handle(event)            def log_message(self, *args) -> None:  # quiet                pass        self._server = ThreadingHTTPServer(("127.0.0.1", port), Handler)        self.url = f"http://127.0.0.1:{self._server.server_address[1]}/boxline"        threading.Thread(target=self._server.serve_forever, daemon=True).start()    def _handle(self, event: dict) -> None:        if event["type"] != "task_run.finished" or event["id"] in self._seen:            return        self._seen.add(event["id"])        data = event["data"]        state = json.loads(self.state_file.read_text()) if self.state_file.exists() else {}        last = state.get(data["taskId"])        result = data.get("result")        if data["status"] != "completed" or not result:            delivery = {"eventId": event["id"], "taskRunId": data["taskRunId"], "status": data["status"], "price": None,                        "previous": last["price"] if last else None, "changed": None, "message": f"the run {data['status']}: nothing to compare"}        else:            changed = None if last is None else (last["price"] != result["price"] or last["inStock"] != result["inStock"])            if changed is None:                message = f"first price seen: {result['price']} {result['currency']}"            elif changed:                message = f"CHANGED: {last['price']} -> {result['price']} {result['currency']}"            else:                message = f"no change: still {result['price']} {result['currency']}"            state[data["taskId"]] = {**result, "at": event["createdAt"]}            self.state_file.write_text(json.dumps(state, indent=2))            delivery = {"eventId": event["id"], "taskRunId": data["taskRunId"], "status": data["status"], "price": result["price"],                        "previous": last["price"] if last else None, "changed": changed, "message": message}        print(f"webhook {event['id']}: {delivery['message']}", flush=True)        with self._arrived:            self.deliveries.append(delivery)            self._arrived.notify_all()    def wait_for(self, task_run_id: str, timeout: float = 60) -> dict:        """The delivery for this task run (already here, or the next one to arrive)."""        with self._arrived:            found = self._arrived.wait_for(lambda: next((d for d in self.deliveries if d["taskRunId"] == task_run_id), None), timeout)        if not found:            raise TimeoutError(f"no webhook for task run {task_run_id} within {timeout:g} s")        return found    def close(self) -> None:        self._server.shutdown()if __name__ == "__main__":    if not os.environ.get("WEBHOOK_SECRET"):        raise SystemExit("Set WEBHOOK_SECRET to the endpoint's secret (whsec_…)")    r = Receiver(Path(os.environ.get("STATE_FILE", "last-price.json")), int(os.environ.get("PORT", "8787")), os.environ["WEBHOOK_SECRET"])    print(f"Listening on {r.url} (put it behind HTTPS on a public address)")    threading.Event().wait()

Inputs

Environment variables the example reads:

PRODUCT_URL: the product page (default https://books.toscrape.com/catalogue/sapiens-a-brief-history-of-humankind_996/index.html).
SCHEDULE: the cron (at most every 5 minutes; UTC) (default 0 * * * *).
RUNS: how many runs to start right away (default 1).
KEEP_SCHEDULE: leave the schedule on at the end (for example 1).

Run it

Get the examples with git clone https://github.com/Boxline-dev/examples.git and open this one: cd examples/watch-a-price. Put your API key in BOXLINE_API_KEY (and BOXLINE_API_URL for another API), then run either version from the example’s folder. Both write to output/ and stop their sessions when they finish.

# Node 18 or newer, in the example's folderexport BOXLINE_API_KEY=…   # your API keynpm install @boxline/sdk tsxnpx tsx node/index.ts

# Python 3.9 or newer, in the example's folderexport BOXLINE_API_KEY=…   # your API keypip install boxline-sdkpython python/main.py

Then check the result, from the same folder: npx tsx check.ts output. It exits with 0 only when the result is right.

What its check proves

check.ts reads what the example wrote and passes only when the result is right, not merely when the example ran.

The schedule's next run is the next full hour, every run completed with an answer in the schema and the site's price, each sent a signed task_run.finished webhook with the same result, the receiver found the first price new and the second one unchanged, and the schedule was switched off at the end.

Source

github.com/Boxline-dev/examples/watch-a-price on GitHub, with its README.md, check.ts and .env.example. The examples are open source. See all examples.

These docs describe a beta API. Endpoints, fields and SDK methods may change before general availability; breaking changes will be listed in the changelog.