Python client
The client is one dependency-light file you drop into your project — no SDK to install, no build step. Read it here, or download the whole kit:
Download the integration kit (.tar.gz) — every file below plus the docs, bundled.
client/gi_client.py
Single-file client. Depends only on requests. Wraps sync predict, async submit + poll, the composite annotation→expression workflow, job lifecycle, and model listing; returns the unified {data, meta} envelope and raises a typed GIError on any non-2xx.
client/gi_client.py
"""Tiny client for the Genomic Intelligence API.
Wraps `requests` with bearer auth, the unified `{data, meta}` /
`{error}` envelope, and a polling helper for async jobs. Drop this file
into your project and `from gi_client import Client`.
Endpoints covered:
- the six per-task predictors, each its own published operation:
``POST /v1/tasks/{promoter,splice,enhancer,chromatin,annotation,
expression}/predict``
- the composite ``POST /v1/workflows/find-genes-and-predict-expression``
- the async job endpoints ``GET /v1/tasks/jobs[/{job_id}]``
- model discovery ``GET /v1/tasks/{task}/models``
Contract reference: https://api.genomicintelligence.ai/redoc
"""
from __future__ import annotations
import time
from typing import Any, Dict, Mapping, Optional
import requests
#: Minimum ``sequence`` length per task, in bp, measured AFTER whitespace is
#: stripped. This is a MIRROR for local pre-flight only — the authority is the
#: ``minLength`` published on each task's request schema in the live OpenAPI
#: document (``GET /v1/openapi.json``, public and unauthenticated), generated
#: from the service's ``api/core/limits.py``.
#:
#: The two can no longer disagree quietly. ``scripts/check_contract.py`` reads
#: this table straight out of this file and asserts it against
#: ``scripts/contract.json``, a pin of the live document; it runs on every PR
#: and **blocks the deploy**. ``scripts/sync_contract.py --check`` re-verifies
#: the pin against the live API daily. If you need to change a number here,
#: re-pin first and reconcile this table to the contract — never the reverse.
TASK_MIN_BP: Dict[str, int] = {
"promoter": 300,
"splice": 100,
"enhancer": 50,
"chromatin": 200,
"annotation": 1_000,
"expression": 9_198,
"find-genes-and-predict-expression": 1_000,
}
#: Maximum ``sequence`` length, in bp, for every task (same mirror caveat).
#: Also published per model as ``bio_spec.request_max_bp``.
MAX_BP = 500_000
#: Above this, the composite workflow refuses synchronous delivery with
#: ``413 sync_too_large`` — resend the same body with ``Prefer: respond-async``.
FIND_GENES_SYNC_LIMIT_BP = 50_000
class GIError(RuntimeError):
"""Raised on any non-2xx response from the API.
Attributes mirror the unified error envelope so callers can switch
on ``code`` rather than HTTP status alone:
{"error": {"code": "...", "message": "...",
"request_id": "...", "details": ...}}
``details`` is carried through verbatim and never parsed. Its shape is
keyed on the sibling ``code`` and has varied across releases — for
``validation_failed`` you may see either a bare ``[{loc, msg, type}, ...]``
array or ``{"errors": [...]}`` — and splice's response-size cap emits a
``validation_failed`` with neither (``{task, record_count,
maximum_records, sequence_length, threshold}``). Branch on ``code`` and
treat ``details`` as display-only, and you are correct against all of them.
"""
def __init__(
self,
status: int,
body: Dict[str, Any],
headers: Optional[Mapping[str, str]] = None,
):
err = (body or {}).get("error", {}) if isinstance(body, dict) else {}
self.status = status
self.code = err.get("code", "http_error")
self.message = err.get("message", "")
# The X-Request-Id header is set on every response, including any
# error whose body omits error.request_id. Fall back to it — support
# tickets always need a correlation id.
self.request_id = err.get("request_id") or (headers or {}).get("X-Request-Id")
self.details = err.get("details")
rid = self.request_id or "unset"
super().__init__(f"[{status} {self.code}] {self.message} (request_id={rid})")
class Client:
"""Thin synchronous client.
>>> c = Client(api_key="gi_…")
>>> r = c.predict("promoter", sequence="ACGT" * 500, sequence_name="demo")
>>> r["meta"]["inference_time_ms"]
"""
def __init__(
self,
api_key: str,
base_url: str = "https://api.genomicintelligence.ai",
timeout: float = 120.0,
) -> None:
self.base_url = base_url.rstrip("/")
self.timeout = timeout
self._session = requests.Session()
self._session.headers.update(
{
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
"Accept": "application/json",
}
)
# ------------------------------------------------------------------ helpers
def _check(self, resp: requests.Response) -> Dict[str, Any]:
try:
body = resp.json()
except ValueError:
# A response arrived but is not JSON — usually an edge proxy. Use
# http_error: error.code is a closed enum published in the schema,
# so a synthesized value outside it is one no caller can match on.
# Client-origin errors carry no request_id (the server never saw
# this one), which is how you tell them apart from server errors.
body = {"error": {"code": "http_error", "message": resp.text[:200]}}
if not resp.ok:
raise GIError(resp.status_code, body, resp.headers)
return body
# ----------------------------------------------------------------- requests
def health(self) -> Dict[str, Any]:
r = self._session.get(f"{self.base_url}/health", timeout=self.timeout)
return self._check(r)
def list_models(self, task: str) -> Dict[str, Any]:
"""``GET /v1/tasks/{task}/models`` — requires the same bearer key.
Returns a FLAT ``{task, default_model, models: [...]}`` object, not
the ``{data, meta}`` envelope. Each model carries ``bio_spec`` with
``request_max_bp`` (the enforced ceiling), ``context_window_bp``
(the model's own window — compare your sequence length against it to
know whether the model saw real sequence or padding; null for
annotation and expression) and ``trained_window_bp`` (fixed
receptive field, 9198 for ``g0-expression``; null for sliding-window
models). ``request_max_bp`` is the only one of the three that is a
cap; the window fields describe what the model scores, not what the
route accepts.
An unknown ``task`` is ``404 not_found``.
"""
r = self._session.get(
f"{self.base_url}/v1/tasks/{task}/models", timeout=self.timeout
)
return self._check(r)
@staticmethod
def _body(
sequence: str,
sequence_name: str,
model: Optional[str],
options: Optional[Dict[str, Any]],
tss_index: Optional[int] = None,
) -> Dict[str, Any]:
body: Dict[str, Any] = {"sequence": sequence, "sequence_name": sequence_name}
if model is not None:
body["model"] = model
if options is not None:
body["options"] = options
if tss_index is not None:
body["tss_index"] = tss_index
return body
def predict(
self,
task: str,
sequence: str,
sequence_name: str = "sequence",
model: Optional[str] = None,
options: Optional[Dict[str, Any]] = None,
tss_index: Optional[int] = None,
) -> Dict[str, Any]:
"""Synchronous prediction. Returns the full ``{data, meta}`` body.
``task`` is one of ``promoter``, ``splice``, ``enhancer``,
``chromatin``, ``annotation``, ``expression`` — each a separately
published operation at the same URL this method builds. An
unrecognised task is ``404 not_found``, not a validation error.
``model`` is a TOP-LEVEL field, not ``options["model"]``: every
task's ``options`` object is closed (``additionalProperties:
false``), so an unknown key — ``model`` included — is rejected with
``422 validation_failed`` at ``loc ["body","options","<key>"]``.
The accepted option keys differ per task; see the request schema in
ReDoc (``PromoterOptions``, ``SpliceOptions``, …). ``expression``
additionally *requires* ``options["description"]``.
``sequence`` must be within this task's bounds (see
:data:`TASK_MIN_BP` / :data:`MAX_BP`, measured after whitespace is
stripped); outside them is ``422 validation_failed``, never ``413``.
A sequence above the floor but shorter than the selected model's
``bio_spec.context_window_bp`` (from :meth:`list_models`) is still
accepted and scored — against a window padded out to that size.
``tss_index`` applies to the ``expression`` task only: the 0-based
offset of the TSS into the whitespace-stripped ``sequence``. It is
required there unless ``sequence`` is exactly 9,198 bp, and must
satisfy ``4599 <= tss_index <= len(sequence) - 4599``. The client
does not validate it — the API returns ``422 validation_failed``
(reported at ``loc ["body"]``, so switch on ``error.code``).
"""
body = self._body(sequence, sequence_name, model, options, tss_index)
r = self._session.post(
f"{self.base_url}/v1/tasks/{task}/predict",
json=body,
timeout=self.timeout,
)
return self._check(r)
def submit_async(
self,
task: str,
sequence: str,
sequence_name: str = "sequence",
model: Optional[str] = None,
options: Optional[Dict[str, Any]] = None,
tss_index: Optional[int] = None,
) -> str:
"""Submit a task in async mode. Returns the ``job_id``.
``Prefer: respond-async`` is a declared header parameter on all six
predict operations and on the composite workflow. Async delivery is
JSON-only: combining it with a text ``?format=`` is a ``400``.
``tss_index`` is the ``expression``-only TSS offset; see :meth:`predict`.
"""
body = self._body(sequence, sequence_name, model, options, tss_index)
r = self._session.post(
f"{self.base_url}/v1/tasks/{task}/predict",
headers={"Prefer": "respond-async"},
json=body,
timeout=self.timeout,
)
body = self._check(r)
# 202 envelope is {data: {job_id, status, links}, meta: {...}} —
# uniform with the {data, meta} shape every other successful
# response on the inference URL produces.
return body["data"]["job_id"]
# --------------------------------------------------- composite workflow
def find_genes_and_predict_expression(
self,
sequence: str,
description: str,
sequence_name: str = "sequence",
options: Optional[Dict[str, Any]] = None,
async_: bool = False,
) -> Dict[str, Any]:
"""``POST /v1/workflows/find-genes-and-predict-expression``.
Runs annotation over the region, centres a 9,198 bp window on each
found gene's TSS, and scores expression for each — one round trip.
``sequence`` is 1,000–500,000 bp (the annotation stage's floor).
``description`` is the experimental context applied to every gene;
it is required (missing or empty is ``422 validation_failed``) and
is passed as ``options["description"]``. Extra ``options`` may add
``annotation_model``, ``expression_model``, ``batch_size`` (1–128,
default 8) and ``shift_coordinates``; the object is closed, so any
other key is a ``422``.
Above :data:`FIND_GENES_SYNC_LIMIT_BP` (50,000 bp) synchronous
delivery is refused with ``413 sync_too_large`` carrying
``details = {sequence_length, threshold}`` — resend the identical
body with ``async_=True``.
Returns the ``{data, meta}`` body; with ``async_=True`` that body is
the ``202`` acceptance ``{data: {job_id, status, links}, meta}``,
so poll :meth:`wait_for_job` on ``body["data"]["job_id"]``.
``meta.task_specific_counts`` is
``{genes_found, genes_predicted, genes_skipped}`` with
``genes_predicted + genes_skipped == genes_found``; per-gene causes
are in ``data.expression_predictions[].skip_reason``.
"""
opts: Dict[str, Any] = dict(options or {})
opts["description"] = description
body = self._body(sequence, sequence_name, None, opts)
headers = {"Prefer": "respond-async"} if async_ else None
r = self._session.post(
f"{self.base_url}/v1/workflows/find-genes-and-predict-expression",
headers=headers,
json=body,
timeout=self.timeout,
)
return self._check(r)
def get_job(self, job_id: str) -> requests.Response:
"""Single poll. The caller inspects ``status_code`` to discriminate."""
return self._session.get(
f"{self.base_url}/v1/tasks/jobs/{job_id}", timeout=self.timeout
)
def wait_for_job(
self,
job_id: str,
poll_interval: float = 2.0,
max_wait: float = 30 * 60,
on_progress=None,
) -> Dict[str, Any]:
"""Poll until terminal. Returns ``{data, meta}`` on success, raises ``GIError``."""
deadline = time.monotonic() + max_wait
while True:
r = self.get_job(job_id)
if r.status_code == 200:
return r.json()
if r.status_code == 202:
if on_progress is not None:
try:
# 202 poll body is also {data: {progress: {...}}, meta}.
on_progress((r.json().get("data") or {}).get("progress") or {})
except Exception:
pass
if time.monotonic() > deadline:
raise TimeoutError(f"job {job_id} did not finish within {max_wait}s")
time.sleep(poll_interval)
continue
# Terminal error
try:
body = r.json()
except ValueError:
# http_error for the same reason as _check above: the code must
# come from the published enum.
body = {"error": {"code": "http_error", "message": r.text[:200]}}
raise GIError(r.status_code, body, r.headers)
def list_jobs(self, limit: int = 50) -> Dict[str, Any]:
r = self._session.get(
f"{self.base_url}/v1/tasks/jobs",
params={"limit": limit},
timeout=self.timeout,
)
return self._check(r)
client/quickstart.py
Runs every task end to end against the bundled real sequences. The fastest way to confirm your key works before you write any code.
client/quickstart.py
"""Genomic Intelligence API — partner quickstart.
Hits every public task endpoint with a real, biologically meaningful
sequence drawn from the bundled ``sequences/`` directory (the same
fixtures the service uses for its golden numeric-regression tests).
Sync calls are sized so each task completes in a few seconds on a warm
GPU; ``annotation`` is the one outlier and is intentionally exercised
both sync and async.
pip install -r requirements.txt
export GI_API_KEY=gi_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
export GI_BASE_URL=https://api.genomicintelligence.ai # optional
python quickstart.py
"""
from __future__ import annotations
import os
import pathlib
import sys
import requests
from gi_client import Client, GIError
SEQ_DIR = pathlib.Path(__file__).parent / "sequences"
def load_fasta(filename: str) -> tuple[str, str]:
"""Return (sequence_name, sequence) from a single-record FASTA file."""
text = (SEQ_DIR / filename).read_text()
lines = text.splitlines()
header = lines[0].lstrip(">").strip()
sequence = "".join(line.strip() for line in lines[1:] if line.strip())
return header, sequence.upper()
def _section(title: str) -> None:
print(f"\n=== {title} ===")
def _summary(label: str, body: dict) -> None:
meta = body.get("meta", {})
counts = meta.get("task_specific_counts", {})
print(
f"{label:<22} model={meta.get('model'):<32} "
f"{meta.get('inference_time_ms', '?')} ms counts={counts}"
)
def main() -> int:
api_key = os.environ.get("GI_API_KEY")
if not api_key:
print("ERROR: set GI_API_KEY (your gi_… bearer key)", file=sys.stderr)
return 2
base_url = os.environ.get("GI_BASE_URL", "https://api.genomicintelligence.ai")
client = Client(api_key=api_key, base_url=base_url)
_section("Health")
print(client.health())
# All sequences are real biological inputs from the service's golden
# fixtures — see client/sequences/. Each FASTA carries
# provenance in its header (gene symbol, coordinates, assembly).
tp53_name, tp53_seq = load_fasta("promoter_tp53.fa") # human, ~26 kb, gene-sense (TP53 is - strand)
hbb_name, hbb_seq = load_fasta("splice_hbb.fa") # human, ~4 kb
eve_name, eve_seq = load_fasta("enhancer_eve.fa") # drosophila, ~10 kb, eve developmental enhancer (+ strand)
chr19_name, chr19_seq = load_fasta("chromatin_active_promoter_chr19.fa") # human, ~40 kb
hbb_tss_name, hbb_tss_seq = load_fasta("expression_hbb_k562.fa") # human, exactly 9198 bp, TSS-centered
_section("Sync inference (real biological sequences)")
try:
_summary("promoter (TP53)",
client.predict("promoter", tp53_seq, tp53_name))
_summary("splice (HBB)",
client.predict("splice", hbb_seq, hbb_name))
# Enhancer model is Drosophila-trained — use a fly enhancer.
_summary("enhancer (eve)",
client.predict("enhancer", eve_seq, eve_name))
_summary("chromatin (chr19)",
client.predict("chromatin", chr19_seq, chr19_name))
except GIError as exc:
print(f"sync task failed: {exc}", file=sys.stderr)
return 1
except requests.RequestException as exc:
# Transport-level: DNS, TLS, connection refused, read timeout.
# Routine for a hosted service and not a request problem.
print(f"sync task failed: network error: {type(exc).__name__}: {exc}", file=sys.stderr)
return 1
_section("Expression (TSS-centered 9,198 bp window — HBB in K562)")
# The expression model scores exactly one 9,198 bp window (TSS +/- 4599).
# Anything shorter is a hard 422 -- no padding, no truncation. The bundled
# fixture is exactly that window for HBB, centered on its canonical TSS;
# with the K562 cell-type description this should report HIGH expression
# (HBB is highly expressed in K562 erythroleukemia cells).
#
# At exactly 9198 bp, tss_index is optional and defaults to 4599. To submit
# a longer locus (up to 500,000 bp) and let the server cut the window, pass
# tss_index=<0-based TSS offset into the whitespace-stripped sequence>; it
# is REQUIRED at any other length and must satisfy
# 4599 <= tss_index <= len(sequence) - 4599. Note "whitespace-stripped":
# an index counted over line-wrapped FASTA characters is off by the newline
# count and can silently score the wrong window, so assert on the
# scored_window echoed back below.
try:
body = client.predict(
"expression",
sequence=hbb_tss_seq,
sequence_name=hbb_tss_name,
options={
"description": (
"assay term name is polyA plus RNA-seq. "
"biosample summary is Homo sapiens K562."
),
},
)
pred = body.get("data", {}).get("prediction", {})
counts = body.get("meta", {}).get("task_specific_counts", {})
print(
f"expression: {pred.get('expression_log_tpm')} log(TPM+1) "
f"({pred.get('expression_tpm')} TPM)"
)
# Windowing provenance: confirm the server scored the window you meant.
print(
f" tss_index={counts.get('tss_index')} "
f"scored_window={counts.get('scored_window')}"
)
except GIError as exc:
print(f"expression failed: {exc}", file=sys.stderr)
except requests.RequestException as exc:
print(f"expression failed: network error: {type(exc).__name__}: {exc}", file=sys.stderr)
_section("Async job — annotation on TP53 (~26 kb)")
# Annotation is the slowest atomic task and the one where async
# actually matters. TP53 is well-annotated; expect at least one
# transcript in the response.
try:
job_id = client.submit_async(
"annotation",
sequence=tp53_seq,
sequence_name=tp53_name,
options={"batch_size": 8},
)
print(f"submitted job_id={job_id}")
def progress(p):
pct = p.get("current_percent")
msg = p.get("message", "")
print(f" {pct:>3}% {msg}")
body = client.wait_for_job(job_id, poll_interval=2.0, on_progress=progress)
meta = body.get("meta", {})
counts = meta.get("task_specific_counts", {})
transcripts = body.get("data", {}).get("transcripts", []) or []
print(
f"done — counts={counts} "
f"transcripts={len(transcripts)} "
f"total_time_ms={meta.get('inference_time_ms')}"
)
except GIError as exc:
print(f"annotation failed: {exc}", file=sys.stderr)
return 1
except requests.RequestException as exc:
print(f"annotation failed: network error: {type(exc).__name__}: {exc}", file=sys.stderr)
return 1
except TimeoutError as exc:
# wait_for_job gives up rather than polling forever.
print(f"annotation timed out: {exc}", file=sys.stderr)
return 1
_section("Recent jobs")
print(client.list_jobs(limit=5))
return 0
if __name__ == "__main__":
raise SystemExit(main())
client/requirements.txt
The client's only runtime dependency.
client/requirements.txt
requests>=2.31
Example sequences
Real sequences the quickstart and recipes run against — download or curl them directly: