Sample code (Python)

Complete, runnable Python examples for the main flows. Configuration comes from environment variables: VAST_ACCESS_TOKEN, VAST_PRINTER_ID, VAST_LABEL_TEMPLATE_ID, and VAST_WEBHOOK_SECRET.

Client setup & token verification

vast_client.py

# vast_client.py  -  pip install requests
import os
import requests

BASE_URL = "https://api.vastvision.io/third-party-access"
TOKEN = os.environ["VAST_ACCESS_TOKEN"]          # <YOUR_ACCESS_TOKEN>, server-side only

session = requests.Session()
session.headers.update({
    "Authorization": f"Bearer {TOKEN}",
    "Content-Type": "application/json",
})


class VastAPIError(Exception):
    def __init__(self, status: int, detail):
        super().__init__(f"{status}: {detail}")
        self.status, self.detail = status, detail


def request(method: str, path: str, **kwargs):
    resp = session.request(method, f"{BASE_URL}{path}", timeout=60, **kwargs)
    if resp.status_code != 200:
        try:
            detail = resp.json().get("detail")
        except ValueError:
            detail = resp.text
        raise VastAPIError(resp.status_code, detail)
    return resp.json()


def ping() -> dict:
    return request("GET", "/ping")       # {"status": "ok", "organization_id": 42}


if __name__ == "__main__":
    print(ping())

Printing labels for three new assets

print_labels.py

# print_labels.py
import os
from vast_client import request, VastAPIError

PRINTER_ID = int(os.environ["VAST_PRINTER_ID"])           # <PRINTER_ID>
TEMPLATE_ID = int(os.environ["VAST_LABEL_TEMPLATE_ID"])   # <TEMPLATE_ID>


def print_labels(assets: list[dict]) -> dict:
    payload = {
        "printer_id": PRINTER_ID,
        "rfid_label_template_id": TEMPLATE_ID,
        "asset_list": assets,
    }
    try:
        job = request("POST", "/print-rfid-labels-and-create-assets/", json=payload)
    except VastAPIError as err:
        if err.status == 403:
            raise RuntimeError("Printer is not in this organization; check PRINTER_ID") from err
        if err.status == 404:
            raise RuntimeError("Printer, its location, or the template was not found") from err
        raise                                   # 401 / 422 / 5xx

    sent_epcs = {a["epc_hex"] for a in assets if a.get("epc_hex")}
    created_epcs = {a["epc_hex"] for a in job["assets"]}
    skipped = sent_epcs - created_epcs
    if skipped:
        print(f"Skipped (EPC already exists): {sorted(skipped)}")

    if job["status"] == "printer_received":
        print(f"Print job {job['id']}: {len(job['assets'])} label(s) sent to printer")
    elif job["status"] == "printer_failed":
        # Assets WERE created; only printing failed. Surface this to the user.
        print(f"Print job {job['id']} failed at the printer. Created EPCs: {sorted(created_epcs)}")
    return job


if __name__ == "__main__":
    small_polo = {"name": "Peter Millar Polo", "status": "available", "classification_ids": [101, 102]}
    print_labels([dict(small_polo) for _ in range(3)])

Webhook receiver with signature verification

Receives the two automation recipes from How automation recipes work, verifies each signature against its path, ignores test sends, and records the asset’s new state from the last step.

webhook_receiver.py

# webhook_receiver.py  -  pip install flask
import base64, hashlib, hmac, os, time
from flask import Flask, request

app = Flask(__name__)
SECRET = os.environ["VAST_WEBHOOK_SECRET"]
MAX_AGE = 300

# Paths exactly as entered in each recipe's Webhook URL field -> state they signal
ROUTES = {
    "/webhooks/vast/out-of-inventory": "out_of_inventory",
    "/webhooks/vast/in-inventory": "in_inventory",
}


def verified(path: str) -> bool:
    ts = request.headers.get("X-API-Timestamp", "")
    sig = request.headers.get("X-API-Signature", "")
    if not ts.isdigit() or not sig:
        return False
    now = int(time.time())
    if int(ts) > now or now - int(ts) > MAX_AGE:
        return False
    msg = f"POST{path}{ts}".encode("utf-8")
    expected = base64.b64encode(
        hmac.new(SECRET.encode("utf-8"), msg, hashlib.sha256).digest()).decode("ascii")
    return hmac.compare_digest(expected, sig)


def record_state(epc: str, state: str, zone: dict, at: str) -> None:
    # Hand off to a queue in production. Skip if `at` is older than what you stored.
    print(f"{epc} -> {state} ({zone['name']}, {at})")


def make_handler(path: str, state: str):
    def handler():
        if not verified(path):
            return "invalid signature", 401
        body = request.get_json(silent=True) or {}
        steps = body.get("data") or []
        if body.get("is_test") or not steps:
            return "", 200                  # acknowledge, but change nothing
        last = steps[-1]                    # newest position is the last step
        record_state(last["asset"]["epc_hex"], state, last["zone"], last["timestamp"])
        return "", 200                      # exactly 200
    return handler


for path, state in ROUTES.items():
    app.add_url_rule(path, state, make_handler(path, state), methods=["POST"])