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`.
Contract reference: https://api.genomicintelligence.ai/redoc
"""
from __future__ import annotations
import time
from typing import Any, Dict, Optional
import requests
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": ...}}
"""
def __init__(self, status: int, body: Dict[str, Any]):
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", "")
self.request_id = err.get("request_id")
self.details = err.get("details")
super().__init__(
f"[{status} {self.code}] {self.message} (request_id={self.request_id})"
)
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:
body = {"error": {"code": "non_json", "message": resp.text[:200]}}
if not resp.ok:
raise GIError(resp.status_code, body)
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]:
r = self._session.get(
f"{self.base_url}/v1/tasks/{task}/models", timeout=self.timeout
)
return self._check(r)
def predict(
self,
task: str,
sequence: str,
sequence_name: str = "sequence",
model: Optional[str] = None,
options: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
"""Synchronous prediction. Returns the full ``{data, meta}`` body."""
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
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,
) -> str:
"""Submit a task in async mode. Returns the ``job_id``."""
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
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"]
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:
body = {"error": {"code": "non_json", "message": r.text[:200]}}
raise GIError(r.status_code, body)
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
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, 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
_section("Expression (TSS-centered 9,198 bp window — HBB in K562)")
# The expression model expects a fixed 9,198 bp window centered on the
# TSS. The bundled fixture is 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).
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", {})
print(
f"expression: {pred.get('expression_log_tpm')} log(TPM+1) "
f"({pred.get('expression_tpm')} TPM)"
)
except GIError as exc:
print(f"expression failed: {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
_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: