SDKs¶

Venturi maintains first-party, fully typed SDKs for TypeScript and Python. Both are generated from the same OpenAPI 3.1 document that defines the REST API, so the client surface always matches the contract. Each SDK builds in the platform conventions you would otherwise implement by hand: token refresh, cursor-pagination iterators, idempotency-key generation, Retry-After-aware backoff, and typed problem+json errors.

Package availability

The SDKs are distributed to every onboarded customer in the onboarding artifact bundle, which you download from the authenticated onboarding channel in the product, and through your private package channel. The TypeScript package is @venturi/sdk; the Python distribution is the venturi_sdk wheel, which installs the importable venturi package. Neither is published to a public registry: @venturi/sdk is not on the public npm registry, and the public PyPI project named venturi is unrelated, so do not install either from a public registry. That public venturi project also installs a top-level venturi module, so it cannot share an environment with the SDK: if pip show venturi finds it, run pip uninstall venturi before you install the venturi_sdk wheel, or install the SDK into a separate virtual environment. Your bundle manifest pins each package filename, version, SHA-256 digest, and matching OpenAPI version.

Generated, never hand-written

The SDKs are regenerated from each released OpenAPI version and pinned for reproducible builds. The generated client cannot drift from the published contract (a mismatch fails CI), so the methods, models, and error types you use are exactly the API’s.

Install¶

Run these commands from the directory that holds the files from your onboarding bundle. Set VERSION to the version in the bundle manifest first.

TypeScriptPython

Bash

VERSION="<version from your bundle manifest>"
npm install "./venturi-sdk-${VERSION}.tgz"

Bash

VERSION="<version from your bundle manifest>"
pip install "./venturi_sdk-${VERSION}-py3-none-any.whl"

Configure the client¶

Set the base URL by the base-URL rule: https://api.venturi.systems on the SaaS tier, or your own instance host when self-hosted. Supply OAuth client credentials (or an API key) and the client manages token refresh.

TypeScriptPython

TypeScript

import { Venturi } from "@venturi/sdk";

const venturi = new Venturi({
  baseUrl: "https://<your-venturi-instance>",
  clientId: process.env.VENTURI_CLIENT_ID,
  clientSecret: process.env.VENTURI_CLIENT_SECRET,
  // or: apiKey: process.env.VENTURI_API_KEY
});

Python

import os

from venturi import Venturi

venturi = Venturi(
    base_url="https://<your-venturi-instance>",
    client_id=os.environ["VENTURI_CLIENT_ID"],
    client_secret=os.environ["VENTURI_CLIENT_SECRET"],
    # or: api_key=os.environ["VENTURI_API_KEY"]
)

Reading attribution¶

Every record carries the interpretation evidence card (stage_origin, confidence, confidence_band, evidence_basis, model_version, degradation_state, freshness, and export_eligibility), fully typed.

TypeScriptPython

TypeScript

const record = await venturi.attribution.get("attr_01HF8SNWNW4VMC42M411HGMRWC");
console.log(record.costUsd, record.interpretation.confidenceBand);

if (record.interpretation.exportEligibility === "eligible") {
  // safe to include in chargeback
}

Python

record = venturi.attribution.get("attr_01HF8SNWNW4VMC42M411HGMRWC")
print(record.cost_usd, record.interpretation.confidence_band)

if record.interpretation.export_eligibility == "eligible":
    ...  # safe to include in chargeback

Pagination with transparent iterators¶

Paginated resources expose an async iterator that follows cursors for you. You never construct, sign, or carry a cursor by hand. Filter and sort keys use the REST field names in both SDKs (cost_usd); the TypeScript SDK exposes response fields in camelCase (costUsd), and the Python SDK keeps them in snake_case.

TypeScriptPython

TypeScript

for await (const record of venturi.attribution.list({
  filter: { provider: { eq: "openai" } },
  sort: "-cost_usd",
})) {
  console.log(record.id, record.costUsd);
}

Python

for record in venturi.attribution.list(
    filter={"provider": {"eq": "openai"}},
    sort="-cost_usd",
):
    print(record.id, record.cost_usd)

The iterator walks next_cursor to exhaustion, returning each record exactly once. See pagination for the underlying contract.

Ingestion coverage¶

Both SDKs include typed clients for the InvocationEvent and generic observability ingestion routes (POST /api/v1/invocations and POST /api/v1/events), including caller-supplied idempotency keys. The REST wire format on the Ingestion API page remains authoritative.

Automatic idempotency¶

The SDK attaches a fresh Idempotency-Key to every mutating call by default, so a transient network retry never creates a duplicate. To make a specific operation safe to retry across process restarts, pass your own key. It must be a UUID or ULID, like every Idempotency-Key: generate one for the logical operation, store it with that operation, and reuse it on every retry.

TypeScriptPython

TypeScript

const job = await venturi.exports.create(
  { format: "focus", period: { from: "2026-05-01", to: "2026-05-31" } },
  { idempotencyKey: "b298fbca-c258-450f-8ce7-c050522e9338" }, // one fixed UUID per logical operation; reuse it on every retry
);

Python

job = venturi.exports.create(
    format="focus",
    period={"from": "2026-05-01", "to": "2026-05-31"},
    idempotency_key="b298fbca-c258-450f-8ce7-c050522e9338",  # one fixed UUID per logical operation; reuse it on every retry
)

Automatic rate-limit backoff¶

On a 429, the SDK honors Retry-After and retries with jittered backoff. After exhausting retries it raises a typed RateLimitError rather than a bare HTTP error, so you can branch deliberately.

TypeScriptPython

TypeScript

import { RateLimitError } from "@venturi/sdk";

const period = { from: "2026-05-01", to: "2026-05-31" };

try {
  await venturi.exports.create({ format: "focus", period });
} catch (err) {
  if (err instanceof RateLimitError) {
    // backoff exhausted: slow down or queue
  }
}

Python

from venturi import RateLimitError

period = {"from": "2026-05-01", "to": "2026-05-31"}

try:
    venturi.exports.create(format="focus", period=period)
except RateLimitError:
    ...  # backoff exhausted: slow down or queue

Typed errors¶

Errors are deserialized into a typed union derived from the published error catalog. Branch on the error_code, not on prose.

TypeScriptPython

TypeScript

import { VenturiError } from "@venturi/sdk";

const exportRequest = {
  format: "focus",
  period: { from: "2026-05-01", to: "2026-05-31" },
};

try {
  await venturi.exports.create(exportRequest);
} catch (err) {
  if (err instanceof VenturiError && err.errorCode === "INSUFFICIENT_SCOPE") {
    // this credential cannot create exports
  }
}

Python

from venturi import VenturiError

export_request = {
    "format": "focus",
    "period": {"from": "2026-05-01", "to": "2026-05-31"},
}

try:
    venturi.exports.create(**export_request)
except VenturiError as err:
    if err.error_code == "INSUFFICIENT_SCOPE":
        ...  # this credential cannot create exports

Verifying webhooks¶

Each SDK ships a signature-verification helper so you do not reimplement the HMAC check. The helper throws on a bad signature or timestamp skew, so each example catches that error and rejects the delivery with 400. Each one also answers the webhook.verification challenge, because a new endpoint receives no production events until it passes ownership verification. See webhooks.

TypeScriptPython

TypeScript

import express from "express";
import { verifyWebhook } from "@venturi/sdk";

const app = express();
const secret = process.env.VENTURI_WEBHOOK_SECRET ?? "";

// express.raw keeps the exact bytes Venturi signed.
app.post("/venturi", express.raw({ type: "application/json" }), (req, res) => {
  let event;
  try {
    // throws on a bad signature or timestamp skew
    event = verifyWebhook(req.body, req.get("Venturi-Signature") ?? "", secret);
  } catch {
    res.sendStatus(400);
    return;
  }
  if (event.type === "webhook.verification") {
    res.json({ challenge: event.data.challenge });
    return;
  }
  handleEvent(event); // your processing
  res.sendStatus(200);
});

Python

import os

from flask import Flask, request
from venturi.webhooks import verify_webhook

app = Flask(__name__)
secret = os.environ["VENTURI_WEBHOOK_SECRET"]


@app.post("/venturi")
def handle():
    try:
        # get_data() returns the exact bytes Venturi signed.
        # raises on a bad signature or timestamp skew
        event = verify_webhook(request.get_data(), request.headers.get("Venturi-Signature", ""), secret)
    except Exception:
        return "", 400
    if event["type"] == "webhook.verification":
        return {"challenge": event["data"]["challenge"]}
    process(event)  # your processing
    return "", 200

Sample apps¶

Each SDK bundle includes runnable reference apps you can point at the sandbox:

Chargeback exporter: generates a FOCUS export and downloads the result.
Webhook receiver: verifies signatures and processes events.
Anomaly-alert forwarder: relays anomaly.detected events to your channel.

Versioning¶

SDK major versions track API major versions. A breaking API change ships alongside a coordinated new SDK major version, and the changelog records the mapping. Within a major version, upgrades are additive and safe. See versioning and rate limits.

Give feedback on this page

Sign in to send feedback. Opens in a new tab.

Copy page details

Docs feedback: SDKs
Page: https://docs.venturi.systems/developers/sdks/