Skip to main content
The Python SDK wraps the REST API and adds:
  • a resumable chunked uploader for large WSIs,
  • typed dataclasses for requests and responses,
  • SSE streaming for live job status,
  • and an optional AnnData round-trip for results.
See the Quickstart for the end-to-end happy path.

Install

pip install strand-sdk
# with bioinformatics extras (AnnData / zarr) for downloading results:
pip install "strand-sdk[anndata]"
Requires Python 3.10+.

Configuration

SourceVariable / argumentDefault
EnvSTRAND_API_KEYrequired
EnvSTRAND_BASE_URLhttps://app.strandai.com
ArgClient(api_key=..., base_url=..., timeout=..., max_retries=...)n/a
from strand import Client

client = Client()                      # picks up STRAND_API_KEY from env
client = Client(api_key="sk-strand-…") # explicit

Uploads

upload = client.uploads.upload_file(
    "biopsy.svs",
    progress=lambda done, total: print(f"{done}/{total}"),
)
print(upload.id, upload.width_px, upload.height_px)
upload_file runs the three-step resumable flow under the hood (initiate → chunked PUT to GCS → complete). Chunks are 8 MiB by default and can be tuned via chunk_size.

Predict

# Optional cost estimate before submitting.
est = client.predict.estimate(upload.id, markers=["CD8", "PanCK", "Ki67"])
print(est.estimated_credits, est.org_balance)

# Submit the job. Reserves credits atomically.
job = client.predict.submit(upload.id, markers=["CD8", "PanCK", "Ki67"])
MethodWhat it does
client.predict.estimate(upload_id, markers)Compute the credit cost without reserving credits.
client.predict.submit(upload_id, markers, model=...)Reserve credits and enqueue. Returns a Job.

Selecting a model

predict.submit and predict(...) accept an optional model=. Live Lattice versions:
  • "v0.4": 192-marker panel, original training.
  • "v0.5": 192-marker panel, retrained (current default).
Both share the same GenePT embeddings, so the marker vocabulary is identical between them. Picking a version swaps the model weights while keeping the same vocabulary. Omit model to let the platform pick the current default ("v0.5").
result = client.predict(
    "slide.svs",
    markers=["CD8", "Ki67", "PanCK"],
    model="v0.5",
)
The earlier "v10", "v10-fullpanel", and "v10-fullpanel-v2" names are still accepted as deprecated aliases with a DeprecationWarning; pin the canonical v0.X id to silence the warning and forward-protect against the 2026-12-01 alias sunset. The "v10" alias resolved to the now-sunset v0.3 and is rejected with 400 unknown_model.

Submit without waiting

A full pipeline run blocks for 15+ minutes. To kick a job off and continue with other work, pass wait=False, and predict(...) returns a Job handle as soon as the upload + submit complete:
job = client.predict("slide.svs", markers=["CD3", "CD8"], wait=False)
print(f"submitted {job.id}")

# ...do other work, or shut down the process. The job runs server-side.

# Later (same process or a fresh one via `client.jobs.get(job_id)`):
job.wait()
adata = job.download_results()

# Or stream events:
for event in job.stream_events():
    print(event.status, event.progress)
Static typing follows the wait flag: wait=True returns PredictResult and wait=False returns Job, so IDE completions stay correct without a runtime check.

Jobs

# Block until terminal (uses SSE under the hood, falls back to polling):
status = job.wait(timeout=1800)

# Or stream events yourself:
for event in job.stream_events():
    print(event.status, event.progress)

# Download results as AnnData (requires the `anndata` extra):
adata = job.download_results()

# Or write the raw OME-Zarr store to disk:
path = job.download_results(path="./results-zarr")
MethodWhat it does
job.refresh()Refetch the latest JobStatus snapshot.
job.wait(timeout=...)Block until completed or failed.
job.stream_events()Iterate over SSE JobEvents.
job.download_results()Returns an AnnData (default) or writes to a path.

Errors

All errors inherit from StrandError. Typed subclasses map HTTP codes 1:1 so you can except precisely:
from strand import (
    InsufficientCreditsError, RateLimitError,
    JobFailedError, JobTimeoutError,
)

try:
    job = client.predict.submit(upload.id, markers=["CD8"])
    job.wait(timeout=600)
except InsufficientCreditsError as e:
    print(f"need {e.required} credits; top up first")
except RateLimitError as e:
    print(f"hold off {e.retry_after}s before retrying")
except JobFailedError as e:
    print(f"job failed: {e}")
ExceptionHTTP
AuthError401
BadRequestError400
InsufficientCreditsError402 (carries .required)
NotFoundError404
RateLimitError429 (carries .retry_after)
JobFailedErrorn/a (terminal status failed)
JobTimeoutErrorn/a (wait() timeout)
UploadErrorn/a (GCS resumable session failed)

Public surface

from strand import (
    Client, Job, JobEvent, JobStatus, JobResults,
    Upload, Estimate,
    AuthError, BadRequestError, InsufficientCreditsError,
    JobFailedError, JobTimeoutError, NotFoundError,
    RateLimitError, StrandError, UploadError,
)

See also