Python SDK

What ships in v0.1.0

The first-party Python package is staked-quic-sdk v0.1.0 (module staked_quic_sdk, MIT). It requires Python 3.11 or newer, and the PyO3 extension is built with abi3-py311 so one wheel covers 3.11, 3.12, and 3.13. Runtime dependencies are httpx, pydantic v2, grpcio, and protobuf.

Three clients ship today:

Use LeaderClient unless you only need one piece (a JWT-only flow, a custom QUIC stack, or a signing-oracle smoke test).

Install (from source)

git clone <repo>
cd packages/python/staked-quic-sdk
python -m venv .venv && source .venv/bin/activate
pip install maturin
maturin develop --release   # builds the PyO3 extension + installs the package editable

Once a wheel is published, pip install staked-quic-sdk will be all consumers need. That has not happened yet.

Prerequisites: where your credentials come from

You need an active lease that covers the current epoch. Book one via the Book button on a validator card or detail page (it lands on /book/{validatorId}) or via POST /v1/leases (see the REST section below), then pay the Solana Pay request. Once the lease is active, reveal the trader credentials exactly once with POST /v1/leases/{id}/credentials. The response carries api_token (cleartext, only returned here), tee_endpoint_url, and tls_ca_cert. A second call returns 404 credentials-already-shown. Rotate a lost key with POST /v1/leases/{id}/api-keys.

Each LeaderClientOptions field maps to a concrete source:

Quickstart: LeaderClient

LeaderClient is created via the async factory LeaderClient.create. Construction performs three network calls in order: mint JWT against the marketplace API, gRPC handshake with the host-proxy, RPC query for cluster nodes. Plan a few hundred milliseconds for a cold start, then send_transaction reuses the connection pool.

import asyncio
import os

from staked_quic_sdk import LeaderClient, LeaderClientOptions


async def main() -> None:
    client = await LeaderClient.create(
        LeaderClientOptions(
            marketplace_api_base="https://api.swqos.dev",
            api_key=os.environ["SWQOS_API_KEY"],
            lease_id=os.environ["SWQOS_LEASE_ID"],
            tee_endpoint="https://tee.example.com:443",  # host-proxy, WITH scheme
            rpc_url="https://api.testnet.solana.com",
            tee_tls_ca_pem=open("tee-cert.pem").read(),  # host-proxy CA PEM, sourced out of band today
        )
    )
    try:
        # 32-byte Ed25519 key the TEE presents in its X.509 cert.
        # Compare it with the leased validator's identity before sending value.
        print("staked identity:", client.cert_pubkey().hex())

        # tx_bytes is the bincode wire form of a signed Solana transaction
        # (what solders produces when you serialize a signed Transaction).
        # signature = await client.send_transaction(tx_bytes)  # -> base58 str
    finally:
        await client.close()


asyncio.run(main())

send_transaction(tx_bytes) returns the base58 transaction signature on success. Sending is fire-and-forget at the QUIC layer, so confirm the transaction yourself via your RPC. Always await client.close() when done to free the QUIC and gRPC sockets (it is idempotent).

Failures surface as a single RuntimeError whose message starts with a [REASON_CODE] tag so callers can branch deterministically:

MarketplaceClient: JWT minting only

MarketplaceClient mints the short-lived ES256 lease JWT against POST /v1/leases/{lease_id}/token, sending Authorization: Bearer <api_key>. The JWT's default TTL is 600 seconds. fetch_jwt() makes up to 4 attempts with 200 / 400 / 800 ms backoff on transport errors and 5xx responses, and returns a JwtBundle with jwt (compact ES256 JWT) and expires_at (UNIX seconds, a copy of the exp claim).

import os

from staked_quic_sdk import (
    MarketplaceAuthError,
    MarketplaceClient,
    MarketplaceForbiddenError,
    MarketplaceNetworkError,
    MarketplaceServerError,
)

client = MarketplaceClient(
    api_base="https://api.swqos.dev",
    api_key=os.environ["SWQOS_API_KEY"],
    lease_id=os.environ["SWQOS_LEASE_ID"],
)

try:
    bundle = client.fetch_jwt()
    print(bundle.jwt, bundle.expires_at)
except MarketplaceAuthError:
    # 401: API key is missing, invalid, or revoked.
    raise
except MarketplaceForbiddenError:
    # 403: key is valid but cannot mint for this lease
    # (lease-not-active, lease-epoch-out-of-range).
    raise
except (MarketplaceServerError, MarketplaceNetworkError):
    # Persistent 5xx or transport failure after retries.
    raise

To keep a fresh JWT in the background, use the auto-refresher. It re-mints 60 seconds before exp (Design Doc D7) and swallows transient refresh failures so the timer survives outages:

client.start_auto_refresh()
try:
    bundle = client.get_cached_jwt()  # latest JwtBundle, or None before first fetch
    # ... pass bundle.jwt as the Bearer JWT to the host-proxy ...
finally:
    client.close()  # cancels the refresher and closes the httpx client

MarketplaceClient is also a context manager: with MarketplaceClient(...) as client: closes cleanly on exit.

TpuClient: the signing oracle only

TpuClient talks gRPC to the host-proxy directly. The endpoint is a plain host:port with no scheme. Without a tls argument it opens an insecure channel (local dev only). In production, pass TlsConfig(root_certs=...) with the host-proxy's cert PEM. The per-call deadline defaults to 5 seconds (deadline_s).

import hashlib

from staked_quic_sdk import (
    MarketplaceClient,
    TlsConfig,
    TpuClient,
    TpuRpcError,
)

marketplace = MarketplaceClient(api_base=..., api_key=..., lease_id=...)

with open("tee-cert.pem", "rb") as f:
    root_certs = f.read()

with TpuClient(endpoint="tee.example.com:443", tls=TlsConfig(root_certs=root_certs)) as tpu:
    try:
        bundle = marketplace.fetch_jwt()

        health = tpu.health()           # HealthStatus(provisioned, pubkey_sha256)
        assert health.provisioned, "enclave not provisioned yet"

        cert = tpu.get_certificate()    # 249-byte Firedancer mock X.509 + 32-byte pubkey

        transcript_hash = hashlib.sha256(b"hello swqos").digest()  # 32 bytes
        signature = tpu.sign_certificate_verify(transcript_hash, bundle.jwt)
        assert len(signature) == 64     # Ed25519
    except TpuRpcError as err:
        # err.code is a grpc.StatusCode: UNAUTHENTICATED, RESOURCE_EXHAUSTED,
        # UNAVAILABLE, ...
        raise

Details worth knowing:

• sign_certificate_verify(transcript_hash, jwt) builds the 130-byte TLS 1.3 CertificateVerify payload around the 32-byte SHA-256 transcript hash and forwards the JWT in the gRPC authorization metadata header. Layout: 64 bytes of 0x20, the ASCII label "TLS 1.3, client CertificateVerify\0", then the hash. The helper build_certificate_verify_payload(transcript_hash) is exported if you need the payload yourself.
• get_certificate() and health() are unauthenticated, no JWT required.
• Typed errors: TpuInvalidPayloadError (transcript hash not 32 bytes), TpuRpcError (non-OK gRPC status, carries .code and .details), TpuInvalidSignatureError (response signature not 64 bytes).
• Instances hold an open gRPC channel. Call close() or use the context manager. Sharing one instance across threads is safe.

Runnable example

examples/hello_world.py in the package exercises both paths (LeaderClient, then MarketplaceClient + TpuClient). It reads:

python examples/hello_world.py

What stays outside the SDK: the REST surface

Browsing validators, booking, paying, and extending leases are plain REST calls, not SDK methods. Use httpx (already a dependency) directly. Two things trip people up: the list envelope is nested (data.data), and the Idempotency-Key header must be a ULID. A UUID4 is rejected with 400.

import os

import httpx
from ulid import ULID  # pip install python-ulid

API_BASE = os.environ.get("SWQOS_API_BASE", "https://api.swqos.dev")

# 1. Browse validators (public, no auth).
r = httpx.get(f"{API_BASE}/v1/validators", params={"limit": 20})
r.raise_for_status()
page = r.json()["data"]            # {"data": [...], "next_cursor": ...}
for v in page["data"]:
    print(v["validator_id"], v["display_name"], v["uptime_30d_pct"])

# 2. Book a lease (trader auth + ULID idempotency key).
headers = {
    "Authorization": f"Bearer {os.environ['SWQOS_API_KEY']}",
    "Idempotency-Key": str(ULID()),  # must parse as a ULID, UUIDs get a 400
}
body = {
    "validator_id": "01HXYVAL00000000000000000A",
    "epoch_start": 901,
    "epoch_count": 3,            # 1 to 30
    "mode": "exclusive",         # optional: "exclusive" (default) or "shared"
}
r = httpx.post(f"{API_BASE}/v1/leases", json=body, headers=headers)
r.raise_for_status()
lease = r.json()["data"]
print(lease["lease_id"], lease["payment"]["solana_pay_url"])

The booking response carries payment with solana_pay_url, amount_lamports, and memo. Pay within the 15-minute pending-payment window or the slot is released. The full booking and payment flow, including filters like region, price_min_lamports, price_max_lamports, and available_now on the validator list, is covered in Booking a lease and the API reference. Error bodies are RFC 7807 Problem Details, see Errors.

If you want generated models, the OpenAPI spec is committed at docs/openapi.json in the repo and served live at GET /v1/openapi.json.

Development

From packages/python/staked-quic-sdk:

python3.11 -m venv .venv && source .venv/bin/activate
pip install maturin
pip install ".[dev]"          # BEFORE maturin develop, or the editable install is replaced
maturin develop --release     # rebuild after any change to crate/

pytest                        # 23 tests: marketplace + tpu + leader-client
ruff check src tests examples
cd crate && cargo clippy --all-targets -- -D warnings

The pytest suite covers error paths only. The canonical success-path test lives at sdk-manual-test/rust/ in the repo, driving the upstream staked-quic-tpu-client against the docker-compose dev stack. The gRPC stubs under src/staked_quic_sdk/_proto/ are pre-generated from proto/staked_quic.proto, regenerate them with scripts/generate_proto.py when the sibling proto changes.

See also