3torus — Python SDK

Typed Python client for the 3Torus GPU compute marketplace API — the Python sibling of the 3torus CLI. It exposes the platform's public gRPC surface as a typed client with Pydantic models, a typed exception hierarchy and built-in retry.

Install

pip install 3torus          # distribution name

import threetorus           # import name

Python ≥ 3.11.

Quickstart — submit a GPU job in 5 lines

from threetorus import Client

with Client(api_key="3t_...") as c:        # public gateway, TLS — the default
    job = c.jobs.submit(image="busybox:1.36", gpu_count=1, command=["sh", "-c"], args=["echo hello"])
    print(job.id, job.status.name)          # gpu-job-1a2b3c4d5e6f SUBMITTED
    result = c.jobs.get(job.id)             # poll until terminal
    print(result.status.name, result.result and result.result.exit_code)

Since 0.3.1 Client() defaults to the public gateway, api.dev.3torus.com:443, over TLS, so the address is optional. Callers inside the platform pass their own address; plaintext is then chosen automatically for local and in-cluster hosts, and secure=True / secure=False override the choice. (On 0.3.0, pass "api.dev.3torus.com:443" and secure=True explicitly.)

The API key rides on every RPC as an authorization: Bearer <key> gRPC metadata header — it is never logged, never placed in the address, and never echoed in an exception. The SDK reads no environment variable for it: load it yourself from an environment variable or secret store and pass it explicitly; never hard-code it. Each method also needs a scope on the key — a key without it gets PermissionDeniedError.

The GitBook SDK Quickstart is the published copy of this page — keep the two in sync.

Verb surface

Each sub-client fronts one gRPC service:

Sub-client	Service	Methods
client.clusters	user.v1 ComputeService (your own clusters)	list, get
client.jobs	user.v1 JobsService	submit, get, result, list, cancel
client.billing	user.v1 BillingService	usage, quota, invoices, balance
client.providers	core.v1 ProvidersService	list, get, gpu_skus
client.health	core.v1 HealthService	check
client.auth	core.v1 AuthService (validate), user.v1 IdentityService (whoami → Identity)	validate, whoami

from threetorus import models as m

clusters, page = c.clusters.list(phase=m.ClusterPhase.READY)   # your tenant only
usage = c.billing.usage()                                       # the caller's current period
quota = c.billing.quota()                                       # limits + consumption
skus, _ = c.providers.gpu_skus(tier="datacenter")
status = c.health.check()
me = c.auth.whoami()

jobs.submit parameters

Keyword-only: image (required), gpu_count (0 means 1; the upper bound is the largest GPU node in the fleet), name, command / args (at most 64 entries together), env (at most 64 entries, non-empty keys), gpu_model and priority (not supported yet — a non-empty / non-zero value is rejected with UnimplementedError). submit and cancel are never retried automatically.

Typed models

Responses are Pydantic models with native Python types — datetime for timestamps, Decimal for money, IntEnum for enums — instead of raw protobuf accessors:

job: m.Job = c.jobs.get("job-id")
job.status            # m.JobStatus.COMPLETED  (IntEnum, .name == "COMPLETED")
job.completed_at      # datetime | None  (UTC; None when unset)
usage.estimated_cost  # m.Money(amount=Decimal("4.20"), currency="USD")

Errors

Every gRPC failure is mapped to a typed exception carrying the original grpc.StatusCode and the server's message verbatim:

from threetorus import NotFoundError

try:
    c.clusters.get("does-not-exist")
except NotFoundError as e:
    print(e.code, e.message)        # StatusCode.NOT_FOUND ...

UNAVAILABLE / DEADLINE_EXCEEDED are retried automatically with capped exponential backoff before surfacing; permanent codes are never retried. Override the policy with Client(..., retry=RetryPolicy(max_attempts=5)).

Known limits

• Resubmitting an identical job returns the existing one. The SDK sends no idempotency key, so a job is identified by its spec (image, name, GPU count, command, args, env); submitting the same spec again returns the job you already have. Change name to run it again.
• jobs.submit() does not yet expose a result output path, a maximum run time, or a start tier (the CLI's jobs submit does: --output, --max-run-seconds, --start-tier).
• providers.gpu_skus(provider=…) is rejected with UnimplementedError; filter by tier= instead.
• Paging is offset-based (page= / page_size=); cursor pagination is not supported.
• With a machine key, auth.whoami() returns an empty user_id and may return an empty tenant_id.

Version 0.3.0 was checked against the live gateway after release: health.check() returned SERVING, and a clusters.list() without a key was refused with UnauthenticatedError.

Changelog

Pre-1.0: a minor bump may break the API; a patch bump never does. The version lives in TWO places that must match — pyproject.toml [project].version and threetorus.__version__ (the latter rides every RPC in the x-3torus-client header) — and the release tag must equal them.

0.3.1

Changed:

• Client() now defaults to the public gateway, api.dev.3torus.com:443, over TLS. The previous default address was reachable only from inside the platform.
• secure now defaults to None = choose from the address, with the same rule as the CLI: TLS for port 443 or any non-local host; plaintext for localhost, loopback IPs, Kubernetes service names (*.svc, *.svc.cluster.local) and bare host names. An explicit secure=True / secure=False is honoured as before.

Not breaking for callers that pass their address: those hosts keep the transport they had. A caller that relied on the old default address now reaches the public gateway.

Docs: JobArtifact.download_url is signed for the platform's public object-store endpoint, so it can be fetched from outside the platform until it expires (the docstring used to say otherwise).

0.3.0 (2026-10-11)

Breaking — client.clusters now speaks the tenant-scoped torus.user.v1.ComputeService:

• clusters.list() returns only YOUR tenant's clusters; clusters.get() of another tenant's cluster raises NotFoundError.
• clusters.list(role=…) removed — passing it raises TypeError.
• clusters.list_nodes() and clusters.capacity() removed (never implemented server-side; no user-plane equivalent). models.Node, models.Capacity, models.ClusterRole removed with them.
• models.Cluster drops role, gains gpu_model.

Fixed: jobs.list(page=…/page_size=…) and billing.invoices(page=…/ page_size=…) raised TypeError before sending anything.

0.2.0 (never released)

Jobs, billing and auth.whoami() moved to torus.user.v1. Superseded by 0.3.0 before any release.

Development (contributors)

pip install -e ".[dev]"     # from sdk/python/ in a checkout of the platform repo
ruff check src tests
pytest

The generated protobuf stubs are not vendored here — the wheel force-includes the platform repo's canonical gen/core and gen/user trees (produced by buf generate), so the SDK never drifts from the proto contracts.

Releasing to PyPI

Releases are tag-driven and publish through PyPI trusted publishing (GitHub OIDC → short-lived upload token); there is no long-lived PyPI token. The workflow is .github/workflows/release-sdk.yml.

1. Bump [project].version in pyproject.toml and threetorus.__version__ on a branch, merge to main.
2. Tag that commit on main and push the tag (git tag v0.3.1 && git push origin v0.3.1 — the tag MUST equal the declared version).
3. The workflow asserts the tag equals the declared version, asserts the tagged commit is reachable from main, builds the wheel with the stubs inside, runs twine check, and publishes the wheel to https://pypi.org/project/3torus/.
4. Verify from a clean environment: pip install 3torus==<version>.

Dry run (build + twine check + version assert, no publish): gh workflow run release-sdk.yml --ref main — dry_run defaults to true.

Wheel only. The sdist omits the generated stubs (the force-include is wheel-scoped), so it cannot be built into a working install and is not published.