← Developers

End-to-end examples

Spec matching api v1

Every snippet below hits the real endpoints — no mocks. The edge flows target the node runtime (http://<edge>:8291, the manifests/LayoutManifest.production-line.example.json vertical), the cloud flows target the Control Plane (https://cp.axiom-trace.eu). If you haven’t read The Contract yet, read it first.

1 — Operator session (edge)

# PIN login → session token (TTL in expiresInHours, revocable, audited)
TOKEN=$(curl -s http://localhost:8291/api/v1/auth/login \
  -H 'content-type: application/json' \
  -d '{"tenantId":"<tenant-uuid>","deviceCode":"IPC-01","pin":"1234"}' \
  | jq -r .token)

# PIN shared between operators → 409 PIN_AMBIGUOUS + candidate list;
# re-send with "operatorId" to disambiguate.

2 — Lot lifecycle: receive → work → inspect

# VERIFY_LOT (role MAGAZZINIERE) — FSM IN_ATTESA → VERIFICATO
curl -X POST http://localhost:8291/api/v1/actions/VERIFY_LOT \
  -H "Authorization: Bearer $TOKEN" -H 'content-type: application/json' \
  -H "X-Idempotency-Key: $(uuidgen)" \
  -d '{"payload":{"lot_id":"LOT-2026-091","peso_dichiarato":100.0,
        "peso_misurato":100.3}}'

# ACCEPT_LOT — formula `weight_delta_pct` runs, constraint `peso_coerente`
# fails closed beyond ±0.5%: 422-style rejection, zero commit.
curl -X POST http://localhost:8291/api/v1/actions/ACCEPT_LOT \
  -H "Authorization: Bearer $TOKEN" -H "X-Idempotency-Key: $(uuidgen)" \
  -d '{"payload":{"lot_id":"LOT-2026-091","peso_dichiarato":100.0,
        "peso_misurato":100.3}}'

# Inspection (role QUALITA) — START_MEASURE then QC_PASS/QC_FAIL
curl -X POST http://localhost:8291/api/v1/actions/QC_PASS \
  -H "Authorization: Bearer $TOKEN" -H "X-Idempotency-Key: $(uuidgen)" \
  -d '{"payload":{"piece_id":"PZ-001","misura":0.018}}'
# misura > 0.02 → CONSTRAINT_VIOLATED (tolleranza_collaudo).
# A trigger your role doesn't own → 403. Out-of-state trigger → 409.
// Same flow in Node — one async function, zero dependencies
async function closeOperation(base, token, order) {
  const r = await fetch(`${base}/api/v1/actions/CLOSE_OPERATION`, {
    method: 'POST',
    headers: { 'content-type': 'application/json',
               authorization: `Bearer ${token}`,
               'x-idempotency-key': crypto.randomUUID() },
    body: JSON.stringify({ payload: {
      order_id: order, qta: 500, qta_ordinata: 500 } })
  });
  if (r.status === 401) throw new Error('sessione scaduta — rifare login');
  return r.json();
}
// qta > qta_ordinata → the rule `qta_entro_ordine` rejects: you can never
// commit more than the order. The formula `tempo_std` runs for the audit row.
# Read queries are declared in the manifest and addressed as triggers
curl -X POST http://localhost:8291/api/v1/actions/LIST_LOTTI \
  -H "Authorization: Bearer $TOKEN" -H "X-Idempotency-Key: $(uuidgen)" \
  -d '{"payload":{"lot_id":"LOT-2026-091"}}' | jq '.rows'

4 — Cloud side: portal session → tenant surface

# Portal login (cloud) → Bearer session
PS=$(curl -s https://cp.axiom-trace.eu/api/v1/portal/login \
  -H 'content-type: application/json' \
  -d '{"email":"qa@customer.com","password":"…","tenant_id":"<uuid>"}' \
  | jq -r .token)

# One snapshot: contract, fleet heartbeat, monthly metering
curl -s https://cp.axiom-trace.eu/api/v1/tenant/usage \
  -H "Authorization: Bearer $PS" | jq '{plan: .subscription.plan, edges: .edges|length}'

# Re-verify the tenant hash-chained audit trail — self-service integrity
curl -s https://cp.axiom-trace.eu/api/v1/tenant/audit/verify \
  -H "Authorization: Bearer $PS" | jq '{ok, checked, head}'

# SIEM export — NDJSON cursor-paginated, same facts the chain verifies
curl -s "https://cp.axiom-trace.eu/api/v1/tenant/audit?limit=1000" \
  -H "Authorization: Bearer $PS" | head -1 | jq '{seq, action, integrity}'
# Same in Python — incremental audit export for a SIEM collector
import httpx
S = httpx.Client(base_url="https://cp.axiom-trace.eu",
                 headers={"authorization": f"Bearer {token}"})
since = 0
while True:
    rows = S.get("/api/v1/tenant/audit",
                 params={"since_seq": since, "limit": 1000}).json()
    events = rows["events"] if isinstance(rows, dict) else rows
    if not events:
        break
    since = int(events[-1]["seq"])   # seq is a 64-bit string — coerce
    ship_to_siem(events)

5 — Ingest (edge → cloud channel)

# The edge daemon's outbound drain — idempotent on event id, ≤1000/batch
curl -X POST https://cp.axiom-trace.eu/api/v1/sync/ingest \
  -H "Authorization: Bearer $EDGE_API_KEY" \
  -H "x-edge-node: edge-01" \
  -H "x-edge-signature: ed25519.<b64-of-raw-body>" \
  -H 'content-type: application/json' \
  -d '{"events":[{"id":"<uuid>","aggregate_type":"BIZ_LOTTI",
        "aggregate_id":"<uuid>","event_type":"INSERT_BIZ_LOTTI",
        "payload":{"lotto":"LOT-2026-091","peso_misurato":100.3},
        "created_at":"2026-09-26T14:01:03Z"}]}'
# → 201 {processed_count, duplicate_count, latest_manifest_version}
#   resending the same batch is not an error, it's the designed retry path.

Where next