ZERPY Help

English

Eesti

English

Eesti

Appearance

Code samples ​

Runnable examples for the most common tasks. Every sample reads the key from the environment variable ZERPY_API_KEY and the webhook secret from ZERPY_WEBHOOK_SECRET.

bash
export ZERPY_API_KEY=sk_live_…
export ZERPY_WEBHOOK_SECRET=whsec_…

Helpers ​

A small client per language, used by the samples below. Each one sends the key, parses JSON and turns API errors into exceptions with the error code.

Node.js ​

Node.js 18 or newer (built-in fetch).

js
// zerpy.mjs
const BASE = 'https://zerpy.app/api/v1'

export async function zerpy(path, init = {}) {
  const res = await fetch(BASE + path, {
    ...init,
    headers: {
      Authorization: `Bearer ${process.env.ZERPY_API_KEY}`,
      'Content-Type': 'application/json',
      ...init.headers
    }
  })
  if (res.status === 429) {
    const wait = Number(res.headers.get('Retry-After') ?? 1)
    await new Promise(r => setTimeout(r, wait * 1000))
    return zerpy(path, init)
  }
  const body = await res.json()
  if (!res.ok) {
    const err = new Error(`${body.error?.code}: ${body.error?.message}`)
    err.code = body.error?.code
    err.details = body.error?.details
    err.status = res.status
    throw err
  }
  return body
}

PHP ​

PHP 7.4+ with the curl extension.

php
<?php
// zerpy.php
function zerpy(string $method, string $path, ?array $body = null): array
{
    $ch = curl_init('https://zerpy.app/api/v1' . $path);
    curl_setopt_array($ch, [
        CURLOPT_CUSTOMREQUEST  => $method,
        CURLOPT_RETURNTRANSFER => true,
        CURLOPT_TIMEOUT        => 30,
        CURLOPT_HTTPHEADER     => [
            'Authorization: Bearer ' . getenv('ZERPY_API_KEY'),
            'Content-Type: application/json',
        ],
    ]);
    if ($body !== null) {
        curl_setopt($ch, CURLOPT_POSTFIELDS, json_encode($body));
    }
    $raw    = curl_exec($ch);
    $status = curl_getinfo($ch, CURLINFO_RESPONSE_CODE);
    curl_close($ch);
    if ($raw === false) {
        throw new RuntimeException('Network error');
    }
    $data = json_decode($raw, true);
    if ($status >= 400) {
        throw new RuntimeException(($data['error']['code'] ?? 'error') . ': ' . ($data['error']['message'] ?? $raw), $status);
    }
    return $data;
}

Python ​

Python 3.8+ with requests (pip install requests).

python
# zerpy.py
import os
import time
import requests

BASE = "https://zerpy.app/api/v1"
session = requests.Session()
session.headers["Authorization"] = f"Bearer {os.environ['ZERPY_API_KEY']}"


class ZerpyError(Exception):
    def __init__(self, status, error):
        super().__init__(f"{error.get('code')}: {error.get('message')}")
        self.status = status
        self.code = error.get("code")
        self.details = error.get("details")


def zerpy(method, path, **kwargs):
    res = session.request(method, BASE + path, timeout=30, **kwargs)
    if res.status_code == 429:
        time.sleep(int(res.headers.get("Retry-After", "1")))
        return zerpy(method, path, **kwargs)
    data = res.json()
    if not res.ok:
        raise ZerpyError(res.status_code, data.get("error", {}))
    return data

List products ​

Fetch the whole catalog, 200 products per page.

bash
curl "https://zerpy.app/api/v1/products?page=1&limit=200" \
  -H "Authorization: Bearer $ZERPY_API_KEY"
js
import { zerpy } from './zerpy.mjs'

const products = []
for (let page = 1; ; page++) {
  const res = await zerpy(`/products?page=${page}&limit=200`)
  products.push(...res.data)
  if (!res.has_more) break
}
for (const p of products) {
  console.log(p.sku, p.name, p.price?.gross, p.stock?.available ?? '-')
}
php
<?php
require 'zerpy.php';

$products = [];
$page = 1;
do {
    $res = zerpy('GET', "/products?page={$page}&limit=200");
    $products = array_merge($products, $res['data']);
    $page++;
} while ($res['has_more']);

foreach ($products as $p) {
    echo $p['sku'], ' ', $p['name'], ' ', $p['price']['gross'] ?? '-', ' ', $p['stock']['available'] ?? '-', PHP_EOL;
}
python
from zerpy import zerpy

products, page = [], 1
while True:
    res = zerpy("GET", "/products", params={"page": page, "limit": 200})
    products += res["data"]
    if not res["has_more"]:
        break
    page += 1

for p in products:
    price = (p["price"] or {}).get("gross", "-")
    stock = (p["stock"] or {}).get("available", "-")
    print(p["sku"], p["name"], price, stock)

Incremental sync: add updated_since=2026-11-07T10:00:00Z (URL-encode the + of a time-zone offset as %2B).

Read stock ​

bash
curl "https://zerpy.app/api/v1/stock?skus=P1,P2" \
  -H "Authorization: Bearer $ZERPY_API_KEY"
js
import { zerpy } from './zerpy.mjs'

const { data } = await zerpy('/stock?skus=' + encodeURIComponent('P1,P2'))
for (const s of data) console.log(s.sku, s.available)
php
<?php
require 'zerpy.php';

$res = zerpy('GET', '/stock?skus=' . urlencode('P1,P2'));
foreach ($res['data'] as $s) {
    echo $s['sku'], ' ', $s['available'], PHP_EOL;
}
python
from zerpy import zerpy

for s in zerpy("GET", "/stock", params={"skus": "P1,P2"})["data"]:
    print(s["sku"], s["available"])

Create an order ​

Creates (or, on retry, returns) the order with external_id 1001.

bash
curl -X POST https://zerpy.app/api/v1/orders \
  -H "Authorization: Bearer $ZERPY_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "external_id": "1001",
    "customer": { "email": "mari@example.ee", "name": "Mari Maasikas" },
    "shipping_address": { "line1": "Pärnu mnt 10", "city": "Tallinn", "postal_code": "10148", "country": "EE" },
    "lines": [{ "sku": "P1", "quantity": 2, "unit_price": 12.40 }],
    "shipping": { "description": "Courier", "price": 4.99 }
  }'
js
import { zerpy } from './zerpy.mjs'

const order = {
  external_id: '1001',
  customer: { email: 'mari@example.ee', name: 'Mari Maasikas' },
  shipping_address: { line1: 'Pärnu mnt 10', city: 'Tallinn', postal_code: '10148', country: 'EE' },
  lines: [{ sku: 'P1', quantity: 2, unit_price: 12.40 }],
  shipping: { description: 'Courier', price: 4.99 }
}

// safe to retry: external_id makes the call idempotent
async function sendOrder(body, attempts = 3) {
  try {
    return await zerpy('/orders', { method: 'POST', body: JSON.stringify(body) })
  } catch (err) {
    if (attempts > 1 && (!err.status || err.status >= 500)) return sendOrder(body, attempts - 1)
    throw err
  }
}

const { data, warnings } = await sendOrder(order)
console.log(data.number, data.status, data.totals.gross)
for (const w of warnings) console.warn(`short: ${w.sku} needed ${w.needed}, available ${w.available}`)
php
<?php
require 'zerpy.php';

$order = [
    'external_id' => '1001',
    'customer' => ['email' => 'mari@example.ee', 'name' => 'Mari Maasikas'],
    'shipping_address' => ['line1' => 'Pärnu mnt 10', 'city' => 'Tallinn', 'postal_code' => '10148', 'country' => 'EE'],
    'lines' => [['sku' => 'P1', 'quantity' => 2, 'unit_price' => 12.40]],
    'shipping' => ['description' => 'Courier', 'price' => 4.99],
];

$res = zerpy('POST', '/orders', $order);
echo $res['data']['number'], ' ', $res['data']['status'], ' ', $res['data']['totals']['gross'], PHP_EOL;
foreach ($res['warnings'] as $w) {
    echo "short: {$w['sku']} needed {$w['needed']}, available {$w['available']}", PHP_EOL;
}
python
from zerpy import zerpy, ZerpyError

order = {
    "external_id": "1001",
    "customer": {"email": "mari@example.ee", "name": "Mari Maasikas"},
    "shipping_address": {"line1": "Pärnu mnt 10", "city": "Tallinn", "postal_code": "10148", "country": "EE"},
    "lines": [{"sku": "P1", "quantity": 2, "unit_price": 12.40}],
    "shipping": {"description": "Courier", "price": 4.99},
}

try:
    res = zerpy("POST", "/orders", json=order)
except ZerpyError as e:
    if e.code == "unknown_sku":
        print("Unknown SKUs:", e.details["skus"])
    raise

print(res["data"]["number"], res["data"]["status"], res["data"]["totals"]["gross"])
for w in res["warnings"]:
    print(f"short: {w['sku']} needed {w['needed']}, available {w['available']}")

Needs Montonio connected in ZERPY.

bash
curl -X POST https://zerpy.app/api/v1/orders/SO-00042/payment-link \
  -H "Authorization: Bearer $ZERPY_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{ "return_url": "https://shop.example.com/thank-you" }'
js
const { data } = await zerpy('/orders/SO-00042/payment-link', {
  method: 'POST',
  body: JSON.stringify({ return_url: 'https://shop.example.com/thank-you' })
})
// redirect the customer to data.url
php
<?php
$res = zerpy('POST', '/orders/SO-00042/payment-link', ['return_url' => 'https://shop.example.com/thank-you']);
header('Location: ' . $res['data']['url'], true, 303);
python
res = zerpy("POST", "/orders/SO-00042/payment-link", json={"return_url": "https://shop.example.com/thank-you"})
print(res["data"]["url"])

Verify a webhook ​

Each receiver checks the Zerpy-Signature header against the raw body, rejects timestamps older than 5 minutes and answers quickly with 2xx. See Webhooks for the format.

js
import express from 'express'
import { createHmac, timingSafeEqual } from 'node:crypto'

const app = express()

function verify(raw, header) {
  const m = /t=(\d+),v1=([0-9a-f]+)/.exec(header ?? '')
  if (!m) return false
  const [, t, v1] = m
  if (Math.abs(Date.now() / 1000 - Number(t)) > 300) return false
  const expected = createHmac('sha256', process.env.ZERPY_WEBHOOK_SECRET).update(`${t}.${raw}`).digest('hex')
  return v1.length === expected.length && timingSafeEqual(Buffer.from(v1), Buffer.from(expected))
}

const seen = new Set() // use your database in production

app.post('/zerpy-webhook', express.raw({ type: 'application/json' }), (req, res) => {
  const raw = req.body.toString('utf8')
  if (!verify(raw, req.get('Zerpy-Signature'))) return res.sendStatus(401)
  const event = JSON.parse(raw)
  res.json({ ok: true }) // answer first, work after

  if (seen.has(event.id)) return
  seen.add(event.id)
  if (event.type === 'stock.changed') {
    for (const item of event.data.items) console.log('stock', item.sku, item.available)
  }
  if (event.type === 'order.shipped') {
    console.log('shipped', event.data.order.external_id, event.data.shipment?.tracking_number)
  }
})

app.listen(3000)
ts
// server/api/zerpy-webhook.post.ts
import { createHmac, timingSafeEqual } from 'node:crypto'

export default defineEventHandler(async (event) => {
  const raw = (await readRawBody(event)) ?? ''
  const header = getHeader(event, 'zerpy-signature') ?? ''
  const [, t, v1] = header.match(/t=(\d+),v1=([0-9a-f]+)/) ?? []
  const expected = createHmac('sha256', process.env.ZERPY_WEBHOOK_SECRET!).update(`${t}.${raw}`).digest('hex')
  const valid = !!v1 && v1.length === expected.length && timingSafeEqual(Buffer.from(v1), Buffer.from(expected))
  if (!valid || Math.abs(Date.now() / 1000 - Number(t)) > 300) throw createError({ statusCode: 401 })

  const { type, data } = JSON.parse(raw)
  if (type === 'stock.changed') {
    for (const item of data.items) await updateStock(item.sku, Number(item.available)) // your shop's code
  }
  if (type === 'order.shipped') {
    await markShipped(data.order.external_id, data.shipment?.carrier, data.shipment?.tracking_number)
  }
  return { ok: true }
})
php
<?php
$secret = getenv('ZERPY_WEBHOOK_SECRET');
$raw    = file_get_contents('php://input');
$header = $_SERVER['HTTP_ZERPY_SIGNATURE'] ?? '';

if (!preg_match('/t=(\d+),v1=([0-9a-f]+)/', $header, $m)) {
    http_response_code(401); exit;
}
[$all, $t, $v1] = $m;
$expected = hash_hmac('sha256', $t . '.' . $raw, $secret);
if (!hash_equals($expected, $v1) || abs(time() - (int) $t) > 300) {
    http_response_code(401); exit;
}

$event = json_decode($raw, true);
switch ($event['type']) {
    case 'stock.changed':
        foreach ($event['data']['items'] as $item) {
            // update_stock($item['sku'], (float) $item['available']);
        }
        break;
    case 'order.shipped':
        // mark_shipped($event['data']['order']['external_id'], $event['data']['shipment']['tracking_number'] ?? null);
        break;
}
header('Content-Type: application/json');
echo json_encode(['ok' => true]);
python
import hashlib
import hmac
import json
import os
import re
import time

from flask import Flask, request, abort

app = Flask(__name__)
SECRET = os.environ["ZERPY_WEBHOOK_SECRET"].encode()


def verify(raw: bytes, header: str) -> bool:
    m = re.search(r"t=(\d+),v1=([0-9a-f]+)", header or "")
    if not m:
        return False
    t, v1 = m.group(1), m.group(2)
    if abs(time.time() - int(t)) > 300:
        return False
    expected = hmac.new(SECRET, t.encode() + b"." + raw, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, v1)


@app.post("/zerpy-webhook")
def zerpy_webhook():
    raw = request.get_data()  # raw bytes, before any JSON parsing
    if not verify(raw, request.headers.get("Zerpy-Signature", "")):
        abort(401)
    event = json.loads(raw)
    if event["type"] == "stock.changed":
        for item in event["data"]["items"]:
            print("stock", item["sku"], item["available"])
    elif event["type"] == "order.shipped":
        shipment = event["data"]["shipment"] or {}
        print("shipped", event["data"]["order"]["external_id"], shipment.get("tracking_number"))
    return {"ok": True}