Helix Connect Python SDK

Official Python SDK for the Helix Connect data marketplace platform.

Overview

The Helix Connect SDK gives data producers and consumers programmatic access to the Helix Connect data marketplace. Producers upload and price datasets, manage partner access, and track earnings; consumers browse, subscribe to, and download the datasets they have access to. Every dataset is encrypted in transit and at rest, with encryption, compression, and decryption handled automatically by the SDK.

Installation

pip install helix-connect

Requires Python 3.10 or later.

Authentication & Credentials

Every SDK call is authenticated with three values: HELIX_CUSTOMER_ID, AWS_ACCESS_KEY_ID, and AWS_SECRET_ACCESS_KEY. You get these from the Helix Connect portal (https://portal.helix.tools) — sign in and open the Credentials page, where they're revealed only once you're authenticated.

The SDK does not read these from the environment for you; your application reads them (e.g. from env vars or a secrets manager) and passes them into the constructor, as shown throughout this README. The one variable the SDK does resolve automatically is HELIX_API_ENDPOINT, used as a fallback for api_endpoint when it's omitted; it otherwise defaults to https://api-go.helix.tools.

import os
from helix_connect import HelixProducer

producer = HelixProducer(
    aws_access_key_id=os.environ["AWS_ACCESS_KEY_ID"],
    aws_secret_access_key=os.environ["AWS_SECRET_ACCESS_KEY"],
    customer_id=os.environ["HELIX_CUSTOMER_ID"],
)

STS session credentials (opt-in)

By default, the SDK signs every request with the long-lived AWS key you provide (credential_mode="static") — unchanged since 1.0.0. Opt into short-lived, auto-refreshing AWS STS session credentials with one constructor argument: credential_mode="sts". That key is then used only as a bootstrap credential — the SDK mints a 15-minute session credential from the Helix credential broker and refreshes it automatically before it expires. Everything else (uploads, downloads, notification polling) works identically in both modes.

from helix_connect import HelixConsumer

consumer = HelixConsumer(
    aws_access_key_id=os.environ["AWS_ACCESS_KEY_ID"],
    aws_secret_access_key=os.environ["AWS_SECRET_ACCESS_KEY"],
    customer_id=os.environ["HELIX_CUSTOMER_ID"],
    credential_mode="sts",  # opt-in; default is "static"
)

# Force an immediate re-mint (test/e2e hook; no-op in static mode)
consumer.force_refresh()

credential_mode="sts" also accepts broker_endpoint (override the credential-broker URL; defaults to api_endpoint), credential_scope (optional dict merged into the mint request), and auto_refresh (default True). Available on both HelixConsumer and HelixProducer. A failed mint raises CredentialRefreshError carrying the broker's error code (e.g. subscription_expired).

Quickstart — Producer

from helix_connect import HelixProducer

producer = HelixProducer(
    aws_access_key_id=os.environ["AWS_ACCESS_KEY_ID"],
    aws_secret_access_key=os.environ["AWS_SECRET_ACCESS_KEY"],
    customer_id=os.environ["HELIX_CUSTOMER_ID"],
)

# Upload a dataset. Encryption and compression are handled automatically.
dataset = producer.upload_dataset(
    file_path="./data/customers.json",
    dataset_name="customer-records",
    description="Monthly customer export",
    category="general",
    data_freshness="daily",
)
print(f"Uploaded {dataset['name']} ({dataset['id']})")

# List everything this producer has uploaded
my_datasets = producer.list_my_datasets()
print(f"{len(my_datasets)} dataset(s) on this account")

# Update metadata without re-uploading the underlying data
producer.update_dataset(
    dataset_id=dataset["id"],
    updates={"description": "Updated description with more details"},
)

# Replace a dataset's data by re-uploading a new file (metadata unchanged)
producer.update_dataset_data(
    dataset_id=dataset["id"],
    file_path="./data/customers-v2.json",
)

Quickstart — Consumer

from helix_connect import HelixConsumer

consumer = HelixConsumer(
    aws_access_key_id=os.environ["AWS_ACCESS_KEY_ID"],
    aws_secret_access_key=os.environ["AWS_SECRET_ACCESS_KEY"],
    customer_id=os.environ["HELIX_CUSTOMER_ID"],
)

# See what this customer is subscribed to as a consumer
subscriptions = consumer.list_subscriptions(role="consumer")

# List datasets this customer has access to (first page, unpaginated call --
# unchanged pre-existing behavior when page/limit are omitted)
datasets = consumer.list_datasets()

# Pass page/limit to walk every page: the return value switches to the full
# envelope (datasets + total_count + page + limit + total_pages) so you know
# when to stop.
page = 1
all_datasets = []
while True:
    envelope = consumer.list_datasets(page=page, limit=100)
    all_datasets.extend(envelope["datasets"])
    if page >= envelope["total_pages"]:
        break
    page += 1

# Poll the per-consumer notification queue for DATA_PUBLISHED events.
# Messages are auto-acknowledged (deleted) by default once returned.
notifications = consumer.poll_notifications(
    max_messages=10,
    wait_time_seconds=20,
)

for notif in notifications:
    output_path = f"./downloads/{notif['dataset_id']}.json"
    # auto_decrypt / auto_decompress both default to True.
    consumer.download_dataset(
        dataset_id=notif["dataset_id"],
        output_path=output_path,
        auto_decrypt=True,
        auto_decompress=True,
    )
    print(f"Downloaded {notif.get('dataset_name', notif['dataset_id'])} -> {output_path}")

Marketplace

The marketplace surface lets consumers browse and pay for listed datasets, and lets producers price their own datasets and track earnings. Both marketplace surfaces raise DatasetNotFoundError (server 404) while the marketplace_payments feature flag is off. Listed prices are the producer's own; platform terms are at https://helix.tools/#pricing.

Browsing and subscribing (consumer)

# Search + paginate the public listing (all filters optional)
results = consumer.browse_marketplace(search="phone", category="phone-numbers", page=1)
for ds in results["datasets"]:
    print(f"{ds['name']}: {ds.get('marketplace')}")

first = results["datasets"][0]

# A composite view: dataset + reviews + related datasets + the caller's
# own subscription info
details = consumer.get_dataset_details(dataset_id=first["id"])
print(details["dataset"]["name"], details["subscription_info"])

# Pay for a priced/listed dataset via Stripe Checkout. Pass exactly one of
# dataset_id (subscribe directly) or request_id (pay for a request that was
# already approved). Returns the Checkout URL only -- the SDK never opens it.
checkout_url = consumer.create_subscription_checkout(dataset_id=first["id"])
print(f"Complete payment at: {checkout_url}")

Pricing and earnings (producer)

producer.set_dataset_marketplace(
    dataset_id=dataset["id"],
    price_monthly_cents=4900,  # $49.00/mo, in USD cents; 0 = free
    listed=True,
)

earnings = producer.get_earnings()  # optionally get_earnings(period="2026-07")
print(earnings)

Approving requests at a per-consumer price

A producer can pin a different price for each consumer when approving their subscription request — comp one consumer while charging another for the same dataset. price_monthly_cents is optional and three-valued:

• omitted — the dataset's own marketplace price applies.
• 0 — a free grant, provisioned immediately even on a currently-paid dataset (comps this consumer).
• a positive value — the consumer must complete checkout at that price, even if the dataset is currently free.

requests_ = producer.list_subscription_requests(status="pending")
request_id = requests_["requests"][0]["request_id"]

# Comp this specific consumer regardless of the dataset's own price.
producer.approve_subscription_request(request_id, price_monthly_cents=0)

# Charge a different consumer a custom price for the same dataset (your own
# price, in USD cents -- platform terms are at https://helix.tools/#pricing).
producer.approve_subscription_request(other_request_id, price_monthly_cents=1500)

# Omit the argument to fall back to the dataset's own marketplace price.
producer.approve_subscription_request(third_request_id)

Partner Invites

A producer can self-serve inviting a consumer partner directly — no platform-admin step required. Requires the partner_invite feature flag on the producer's account; without it these calls raise PermissionDeniedError.

# Contract violations (bad email, empty datasets, etc.) raise ValueError
# before any network traffic.
result = producer.invite_consumer(
    company_name="Acme Analytics",
    business_email="data@acme.example",
    datasets=[dataset["id"]],  # 1-50 dataset IDs auto-granted at invite time
    tier="free",               # currently the only supported tier
)
print(result["consumer_id"], result["status"])

consumers = producer.list_consumers()
if consumers:
    producer.deactivate_consumer(consumer_id=consumers[0]["consumer_id"])

datasets also accepts a list of {"dataset_id": ..., "tier": "free"|"paid"} objects (typed as helix_connect.InviteConsumerDatasetGrant, the same shape name used by the TypeScript and Go SDKs) instead of plain ID strings, to comp this consumer on one dataset while granting paid access to another in the same invite. Per-dataset tier overrides the invite-wide tier for that entry; the two forms cannot be mixed in one call.

result = producer.invite_consumer(
    company_name="Acme Analytics",
    business_email="data@acme.example",
    datasets=[
        {"dataset_id": dataset["id"], "tier": "free"},       # comp this one
        {"dataset_id": other_dataset["id"], "tier": "paid"}, # charge for this one
    ],
)

Payouts (Stripe Connect)

Producers accept payouts through a hosted Stripe Connect Express flow. The SDK never opens or redirects to any of the returned URLs itself — send the producer to them.

# One-time: connect a Stripe Express account to receive payouts
onboard = producer.connect_onboard()
print(f"Open this to finish onboarding: {onboard['url']}")

# Check payout account status any time
status = producer.get_connect_status()
if status.get("can_price_datasets"):
    print("Payouts are enabled — datasets can be priced above $0")
else:
    print(f"Still due: {status.get('requirements_due')}")

# Once onboarding is complete, get a one-time link to the Stripe Express
# dashboard (raises PermissionDeniedError if onboarding isn't complete yet)
login = producer.create_connect_login_link()
print(f"Manage your payout account: {login['url']}")

Versioning & Changelog

This SDK follows semantic versioning. See CHANGELOG.md for the full release history.

Support

• Documentation: https://dev.helix.tools (sign in at https://portal.helix.tools and open SDK Docs)
• Issues: https://github.com/helix-tools/sdk-python/issues
• Email: support@helix.tools

License

See LICENSE for details.