Continuous capture via the Streams API

Not every verification starts in a phone. Drive-through cameras at a hub gate, a kiosk camera in a service bay, or a Raspberry Pi watching a parking bay all need to push frames at the API on a schedule and act on the result. The Streams API is that path.

A stream is a registered camera with:

• A stream_id that the client posts frames to.
• A bcrypt-hashed stream_token for authentication on each frame.
• A policy_id — every frame is verified against the same policy.
• An owning customer_id, the same one billed for the inferences.

Registering a stream

Streams are created from the dashboard (Settings → Streams → New). The dashboard call inserts a row into verify_ai_streams (see 20260622_streams.sql) with:

| Column | Notes | | --------------------- | ----------------------------------------------------------------------------- | | id | UUID, used in the frame URL. | | customer_id | Owning customer. | | site_id | Optional — link the stream to a site row. | | name | Human label (e.g. "Hub-East gate camera"). | | location | Free-text location string. | | asset_type | Optional — bias dispatch / billing tags. | | region | Optional — same. | | expected_throughput | Optional integer hint for capacity planning. | | policy_id | Policy applied to every frame (FK to verify_ai_policies). | | stream_token_hash | bcrypt hash of the secret token — never stored plain. Null until paired. | | status | active, paused, archived. | | group_id | Cameras with the same group_id are correlated in the v2 multi-camera pass. | | last_frame_at | Updated best-effort on every successful frame post. |

The plaintext token is minted by POST /api/v1/streams/pair and returned once in that response. After that we only have the hash; if you lose it, re-pair.

Pairing flow

For physical kiosks we expose a pairing exchange so an operator never has to type a token. The dashboard provisions the stream + a short-lived pairing code (rendered as a QR on screen), the operator shows the QR to the kiosk camera, the kiosk POSTs the scanned code to /api/v1/streams/pair, and the server mints + returns the long-lived stream token:

# 1. Operator opens the dashboard:
#    Settings -> Streams -> New. Picks a policy + name. The server
#    INSERTs a row into verify_ai_streams (status='active',
#    stream_token_hash=null) and a row into verify_ai_stream_pairing_codes
#    (expires_at = now + 10 minutes), and shows the operator a QR
#    containing the code.
 
# 2. Kiosk reads the QR and exchanges the code for a stream token:
curl -X POST https://verify.switchlabs.dev/api/v1/streams/pair \
  -H "Content-Type: application/json" \
  -d '{"code":"<scanned-code>"}'
# → {
#     "stream_id":    "f6c1e7e2-...",
#     "stream_token": "vai_strm_...",
#     "policy_id":    "pol_forest1"
#   }

The token is returned exactly once and only its bcrypt hash is stored on the stream row. The kiosk persists the plaintext locally and uses it on every frame. Codes expire 10 minutes after creation (expires_at) and are single-use (the row's consumed_at flips on successful exchange — re-using a code returns 410 Gone).

Posting frames

Once paired, the camera posts JPEGs to:

POST /api/v1/streams/{stream_id}/frame

Multipart body, single frame field (image/jpeg), X-Stream-Token header. We target <800ms p95 end-to-end (network + inference + policy) so a gate camera can decide whether to open a barrier without the driver noticing.

import requests, time
from pathlib import Path
 
STREAM_ID    = "f6c1e7e2-..."           # UUID returned by /streams/pair
STREAM_TOKEN = "vai_strm_..."           # plaintext token, kept on-device only
 
def post_frame(jpeg_bytes: bytes) -> dict:
    r = requests.post(
        f"https://verify.switchlabs.dev/api/v1/streams/{STREAM_ID}/frame",
        headers={"X-Stream-Token": STREAM_TOKEN},
        files={"frame": ("frame.jpg", jpeg_bytes, "image/jpeg")},
        timeout=2.0,
    )
    r.raise_for_status()
    return r.json()
 
# Pi 5 + Arducam capture loop — 1 fps is plenty for parking / gate use.
while True:
    jpeg = capture_jpeg()        # whatever your camera lib produces
    result = post_frame(jpeg)
    if result["is_compliant"]:
        open_barrier()
    time.sleep(1)

Same shape from the shell, when scripting an integration test:

curl -X POST \
  -H "X-Stream-Token: $STREAM_TOKEN" \
  -F "frame=@./gate.jpg" \
  https://verify.switchlabs.dev/api/v1/streams/$STREAM_ID/frame

The response body is the standard verification object — same shape as POST /v1/verify — so existing branching logic ports straight over.

Hardware patterns

We're deliberately not opinionated about hardware. Three shapes we expect to see in the wild:

• Reference design (deferred) — Pi 5 + Arducam 64MP autofocus + PoE hat. The Python loop above runs on it directly. We are publishing a ready-to-flash image later in the cycle; the capture-and-post script is the same one in this guide.
• Axis ACAP — Axis network cameras can run a small ACAP app that pushes frames to our endpoint on motion. Planned partner integration, not yet certified.
• Verkada — Verkada Command webhooks can fire on configured triggers; we forward those frames into the Streams API. Also planned.

Anything that can produce a JPEG and POST a multipart request works today.

Preview: vehicle-presence gating and multi-camera correlation

For continuous capture we don't actually want to send 1 frame/sec when nothing is happening. The intended flow is:

1. A small on-edge motion / presence model decides "vehicle in frame".
2. Only then do we POST to /streams/{id}/frame.
3. For multi-camera setups (front and rear of the same gate), the server correlates frames inside a short window so one vehicle produces one decision, not two.

Both of those are v2 — see docs/streams/ROADMAP.md. Today the endpoint will happily verify every frame you send, and you'll get one result per POST. Plan billing accordingly while the gating layer is still building.

What's next

• Verifying parked vehicles — same policy engine, but in a handheld scanner.
• Model dispatch and targeting — pinning a specific model to a stream via metadata.asset_type.