runbios-sdk

Official Python SDK for the Run BiOS fine-tuning platform API.

Installation

The distribution is runbios-sdk; the import name is bios (runbios is provided as an alias). (bios on PyPI is an unrelated project — installing it will not give you this SDK.)

pip install runbios-sdk

Quick Start

from bios import RunBiOS

client = RunBiOS(api_key="bios-...")

# Search the hosted catalog. Rows come straight from the model registry, so
# they are snake_case, and the HANDLE you pass everywhere else is repo_id
# (`id` is the registry UUID).
result = client.models.search(query="llama", type="llm", limit=5)
for model in result["models"]:
    print(f"{model['repo_id']} -- {model['params_total_b']}B params")

# Create a training job
job = client.training.create(
    idempotency_key="training-create-20260711-0001",
    model="meta-llama/Llama-3.1-8B-Instruct",
    dataset_ids=["ds_abc123", "ds_def456"],
    method="sft",
    adapter="lora",
    epochs=3,
    learning_rate=2e-4,
    lora_rank=16,
)
print(f"Job {job['id']} created -- status: {job['status']}")

Authentication

The SDK supports two authentication methods:

API Key (recommended). Platform keys default to bios-, but provider-shaped and custom prefixes also work. A workspace-bound key with the serverless scope can be used immediately for serverless inference. Legacy usf- keys remain valid:

client = RunBiOS(api_key="bios-...")

Environment variables. When api_key, base_url, or inference_key is omitted, the SDK reads RUNBIOS_API_KEY, RUNBIOS_BASE_URL, and RUNBIOS_INFERENCE_KEY from the environment (the legacy BIOS_API_KEY, BIOS_BASE_URL, and BIOS_INFERENCE_KEY names are still accepted as fallbacks):

export RUNBIOS_API_KEY=bios-...
export RUNBIOS_BASE_URL=https://api.runbios.ai   # optional; this is the default
export RUNBIOS_INFERENCE_KEY=sk-bios-...        # optional; defaults to RUNBIOS_API_KEY

client = RunBiOS()  # uses RUNBIOS_API_KEY / RUNBIOS_BASE_URL / RUNBIOS_INFERENCE_KEY

JWT Access Token:

client = RunBiOS(
    access_token="eyJhbG...",
    org_id="org_abc123",
)

Configuration

Resources

Inference

Inference keys are separate from control-plane API keys, but you never pass one twice: inference_key falls back to RUNBIOS_INFERENCE_KEY and then to the control-plane api_key, so a platform key carrying the serverless scope calls /v1 directly. Pass an explicit inference_key for a per-deployment sk-bios-... key. Streaming yields each OpenAI SSE chunk as a dictionary and closes the upstream response when the iterator is closed. Calls are never retried automatically; if your application chooses to retry, reuse the same idempotency_key. The header is propagated, but the SDK does not claim server-side replay/deduplication unless the endpoint returns an explicit replay acknowledgement.

client = RunBiOS(
    api_key="bios-control-plane-key",
    inference_key="sk-bios-deployment-key",  # omit to reuse api_key
    # Use https://api-dev.runbios.ai explicitly during dev.
)

tools = [{
    "type": "function",
    "function": {
        "name": "lookup",
        "parameters": {"type": "object", "properties": {"id": {"type": "integer"}}},
    },
}]

stream = client.inference.stream_chat_completions(
    messages=[{"role": "user", "content": "Look up record 42"}],
    tools=tools,
    idempotency_key="chat-42-attempt-1",
)
try:
    for chunk in stream:
        print(chunk)
finally:
    stream.close()  # cancels/disconnects an unfinished generation

Serverless catalog models

Call any catalog model by id on the unified /v1 endpoint with a workspace platform key that carries the serverless scope — no per-deployment key. The gateway routes by model; dedicated deployments and serverless models share the same endpoint. reasoning_effort is forwarded, and streaming surfaces content and reasoning_content deltas incrementally.

# A platform key with the serverless scope is enough — `api_key` doubles as the
# inference key, so RunBiOS(api_key=K).inference.chat_completions(...) just works.
client = RunBiOS(api_key="bios-platform-key-with-serverless-scope")

stream = client.inference.stream_chat_completions(
    model="meta-llama/Llama-3.1-8B-Instruct",  # serverless catalog id
    messages=[{"role": "user", "content": "Explain tensor parallelism briefly."}],
    reasoning_effort="low",
)
try:
    for chunk in stream:
        delta = (chunk.get("choices") or [{}])[0].get("delta", {})
        if delta.get("reasoning_content"):
            print(delta["reasoning_content"], end="", flush=True)
        if delta.get("content"):
            print(delta["content"], end="", flush=True)
finally:
    stream.close()

# Non-streaming; the final response carries "usage" when the model reports it.
completion = client.inference.chat_completions(
    model="meta-llama/Llama-3.1-8B-Instruct",
    messages=[{"role": "user", "content": "One sentence on GPUs."}],
)
choice = completion["choices"][0]
answer = choice["message"].get("content")
if answer:
    print(answer)
else:
    # A 200 is NOT proof the model answered. Reasoning models can spend the
    # whole budget inside reasoning_content and return content: null with
    # finish_reason "length" — an empty answer that looks like success. Always
    # read choices[0].message.content, and raise max_tokens (or lower
    # reasoning_effort) when finish_reason is "length".
    print("no answer:", choice.get("finish_reason"),
          "reasoning tokens only:", bool(choice["message"].get("reasoning_content")))

Streaming billing is charged server-side on completed usage; the SDK only needs to request usage where the endpoint exposes it (no client change).

Other supported /v1 inference tasks

Serverless catalogs serve OpenAI chat and, when the model supports that dialect, Anthropic Messages (inference.messages / stream_messages). Serverless does not serve /v1/completions, /v1/embeddings, or /v1/rerank. Those three routes require a dedicated deployment that advertises the matching task and a credential with deployments:read or deployments:write; a chat-only deployment cannot embed or rerank. Read the server's preflight/status capability rather than guessing from a model name. The SDK forwards a configured inference_key, otherwise the workspace platform api_key.

from bios.inference import Inference

reply = client.inference.messages(
    model="catalog-model-with-messages-support", max_tokens=64,
    messages=[{"role": "user", "content": "Hello."}],
)

dedicated = Inference()
text = dedicated.completions(model="completion-deployment", prompt="Continue")
vectors = dedicated.embeddings(model="embedding-deployment", input=["First", "Second"])
ranking = dedicated.rerank(model="rerank-deployment", query="Question", documents=["A", "B"])
stream = dedicated.stream_completions(model="completion-deployment", prompt="Continue")
try:
    for event in stream:
        print(event)
finally:
    stream.close()

stream_messages similarly yields the Anthropic SSE events in order. Set RUNBIOS_INFERENCE_KEY for the dedicated example. None of these POSTs is retried automatically; idempotency_key is sent only if supplied.

Workspace serverless usage and limits (read-only)

A workspace-bound platform key or hosted OAuth grant with analytics:read can read saved workspace RPM/spend settings and usage. serverless alone allows inference, not analytics. Set RUNBIOS_API_KEY to a workspace key with that scope (or pass the key in the client config); a dedicated deployment serving key cannot read the control plane. No workspace override is accepted by these methods. Key creation/listing, per-key usage, org-wide totals and every RPM or spend-cap write stay behind the authenticated console session.

reporting = RunBiOS()
limits = reporting.inference.serverless_limits()
overview = reporting.inference.serverless_usage_overview("24h")
by_model = reporting.inference.serverless_usage_by_model("7d")
requests = reporting.inference.serverless_usage_requests(window="7d", outcome="failed", limit=25)
buckets = reporting.inference.serverless_usage_timeseries("spend", "7d")
daily = reporting.inference.serverless_usage_daily(30)
savings = reporting.inference.serverless_usage_savings("30d")

Windows are 1h, 24h, 7d, 30d or 90d. Overview, model detail, requests and timeseries come from the request ledger (about seven days retained). daily comes from full-history workspace counters over 1–92 UTC days and includes failed admitted requests. Saved workspace RPM is not the effective tier clamp or a remaining-quota counter. An incomplete 200 is an error, not zero usage or an unlimited workspace.

Models

Search the model catalog, fetch training configs, and check adapter compatibility. The catalog lists only models hosted on Run BiOS (the platform's own verified registry, mirrored in Run BiOS storage) — every result can be trained and deployed; it is never a live Hugging Face search.

Search results are the registry's own rows: snake_case fields, repo_id as the model handle (id is the registry UUID), plus maxContext — the native context window that caps a deployment's context_length — and weightBytes, the on-disk weight size. query becomes the registry's q filter, which is the only search parameter it reads.

# Search models
result = client.models.search(query="llama", type="llm", limit=10)
for m in result["models"]:
    print(m["repo_id"], m["params_total_b"], m["surface_type"],
          m.get("maxContext"), m.get("weightBytes"))

# One model, by its author/name handle
detail = client.models.get("meta-llama/Llama-3.1-8B-Instruct")
print(detail["model"]["architecture"], detail["model"]["maxContext"])

# The context ceiling on its own — None when the registry does not record it
print(client.models.native_max_context("meta-llama/Llama-3.1-8B-Instruct"))

# Get model config
config = client.models.get_config("meta-llama/Llama-3.1-8B-Instruct")
print(f"{config['totalParams']}B params, MoE: {config['isMoE']}")

# Check adapter compatibility
compat = client.models.get_adapter_compatibility(
    model_type="llama",
    training_method="rlhf",
    rlhf_algorithm="dpo",
)
usable = [a for a in compat["adapters"] if a["compatible"]]
print(f"{len(usable)} compatible adapters")

Models this credential can invoke

client.models.list() reads GET /v1/models with the SDK's platform API key, returning serverless pool models and this workspace's dedicated deployments according to its scopes. client.models.retrieve(model_id) uses the same boundary. These are different from the trainable-model registry above: models.search() / models.get() do not say whether this key can invoke a model. A separate dedicated inference key selects the corresponding roster via client.inference.list_models() and retrieve_model(model_id), using the same key and base URL as chat.

available = client.inference.list_models()
for model in available["data"]:
    print(model["id"])
if available.get("usf_unreachable_sources"):
    print("Some model sources were unreachable; retry before concluding they are empty.")
if available["data"]:
    detail = client.inference.retrieve_model(available["data"][0]["id"])
    print(detail["id"])

A partial response keeps usf_unreachable_sources; it never collapses an unreadable source to an empty list. These reads do not create API keys or change account limits.

Datasets

Upload, import, preview, and manage training datasets.

# List datasets
datasets = client.datasets.list()

# Upload a dataset
uploaded = client.datasets.upload(
    file_path="./training_data.jsonl",
    name="My SFT Dataset",
)
print(f"Uploaded: {uploaded['id']}")

# Import from HuggingFace
imported = client.datasets.import_from_huggingface(
    repo_id="HuggingFaceH4/ultrachat_200k",
    integration_id="int_abc123",
    name="Ultrachat SFT",
    subset="default",
    split="train_sft",
)

# Preview dataset rows
preview = client.datasets.preview("ds_abc123", page=1, page_size=5)
print(preview["samples"][0].keys() if preview["samples"] else [])

# Validate before uploading
result = client.datasets.validate("./data.jsonl")
if result["format_valid"]:
    print(f"Valid {result['detected_format']} with {result['num_samples']} rows")
else:
    print("Errors:", result["validation_errors"])

# Search HuggingFace Hub
hub_results = client.datasets.search_hub(query="code instruct")

# Preview a Hub dataset
hub_preview = client.datasets.preview_hub(
    dataset_id="HuggingFaceH4/ultrachat_200k",
    subset="default",
    split="train_sft",
    limit=5,
)

# Get format specs
specs = client.datasets.get_format_specs()

# Get storage usage
usage = client.datasets.get_storage_usage()

# Maintain an existing dataset explicitly
client.datasets.update_column_mapping("ds_abc123", {"conversation": "messages"})
client.datasets.revalidate("ds_abc123")
client.datasets.update_source("ds_abc123")  # reference-mode Hub datasets only
client.datasets.set_integration("ds_abc123", "int_replacement")  # None detaches

# Delete a dataset
client.datasets.delete("ds_abc123")

Connected Hub search and exact source selection

Use client.integrations.list() to discover connected accounts, then select the returned ID explicitly. client.integrations.browse(integration_id, query=..., page=..., limit=...) searches with that account's access. Stored credentials are never returned by these discovery methods.

Pass the same integration_id to datasets.preview_hub and datasets.import_from_huggingface. Preview returns available_configs and available_splits; choose the actual subset/split rather than assuming train. A missing or unauthorized integration is an error, not anonymous fallback. Public imports without an integration use datasets.register_huggingface and require workspace_id on the client or method call.

Both import methods accept revision, max_samples, sample_strategy and import_mode. The service records immutable source metadata for resume. A Hub preview is not a guarantee of readiness or a preview of an arbitrary older revision: inspect the registered dataset when importing an explicit revision. Poll datasets.get_status(id) until ready before selecting it for training; a client polling timeout does not cancel the background import.

Training

Create, monitor, stop, and resume fine-tuning jobs.

dataset_ids preserves source order. eval_dataset_ids selects separate held-out datasets and is mutually exclusive with eval_split; evaluation data is composed sequentially and is never used for optimization. Both sets are frozen at job creation—launch a new job to change them; an active run cannot accept more data. Pass the same mixing dictionary to preflight and create: {"mode": "interleave", "weights": [3, 1], "seed": 42}. Weights correspond to the ordered training IDs, not alphabetic order. Modes are sequential, shuffle, interleave, and phased; phased plans contain ordered phases with name, portion, optional weights, and shuffle. The default composition seed is 42. Source pins, order, plan and checksum are retained for metadata-only reconstruction on resume.

request = {
    "idempotency_key": "training-create-20260711-0001",
    "model": "meta-llama/Llama-3.1-8B-Instruct",
    "dataset_ids": ["ds_abc123", "ds_def456"],
    "eval_dataset_ids": ["ds_holdout"],
    "method": "sft",
    "adapter": "lora",
    "epochs": 3,
    "learning_rate": 2e-4,
    "lora_rank": 16,
    "lora_alpha": 32,
    "gpu_type": "A100_80GB",
    "gpu_count": 1,
    "gpu_priorities": [
        {"gpu_type": "A100_80GB", "gpu_count": 1},
        {"gpu_type": "H100_80GB", "gpu_count": 1},
        {"gpu_type": "L40S_48GB", "gpu_count": 1},
    ],
    "queue_if_unavailable": True,
    "queue_deadline": "2026-07-18T00:00:00Z",
    "max_price_hour_cents": 500,
}

# Side-effect-free validation, canonical sizing, live stock and alternatives
check = client.training.preflight(request)
print(check["request_hash"], check.get("recommended"), check["queue_eligible"])

# Create the paid job only after reviewing preflight
job = client.training.create(**request)

# List jobs
jobs = client.training.list(status="running")
page = client.training.list_page(limit=50, offset=0)

# Get job details
job = client.training.get("job_abc123")
print(f"Status: {job['status']}, Progress: {job.get('progress', 0)}%")

# Get a compact, model-readable report (metric averages/ranges/best steps and GPU averages/peaks)
report = client.training.get_metrics("job_abc123", view="summary")
for metric in report.get("metric_summaries", []):
    print(metric["label"], metric["latest"], metric["average"], metric.get("best"))

# Request a bounded curve only when you need its shape
metrics = client.training.get_metrics("job_abc123", view="series", max_points=200)
for point in metrics["metrics"]:
    print(point["step"], point.get("loss"))

# Get checkpoints
checkpoints = client.training.get_checkpoints("job_abc123")
for cp in checkpoints:
    print(f"{cp['name']}: {cp['size_bytes']} bytes")

# Get logs
logs = client.training.get_logs("job_abc123")
for entry in logs["logs"]:
    print(entry["level"], entry["message"])

# Stop a job
client.training.stop("job_abc123", keep_data=True)

# Resume a stopped job
client.training.resume("job_abc123", idempotency_key="training-resume-20260711-0001")

# Delete a checkpoint
client.training.delete_checkpoint("job_abc123", "cp_xyz789")

Wallet

View wallet balance and transaction history.

balance_cents is the deposited balance; available_balance_cents is what can actually be spent right now (balance minus active_holds_cents and accruing_cents). Spend decisions read the second one. Auto top-up is flat (auto_topup_enabled / auto_topup_threshold / auto_topup_amount), and the transaction list is a wrapped page.

# Get balance
balance = client.wallet.get_balance()
print(f"Balance:   ${balance['balance_cents'] / 100:.2f}")
print(f"Spendable: ${balance['available_balance_cents'] / 100:.2f}")
print(f"On hold:   ${balance['active_holds_cents'] / 100:.2f}")
print(f"Accruing:  ${balance['accruing_cents'] / 100:.2f}")

# List transactions -- the rows are WRAPPED, so iterate ["transactions"]
page = client.wallet.get_transactions(limit=20)
print(f"{page['total']} transactions")
for t in page["transactions"]:
    print(f"{t['type']}/{t['category']}: ${t['amount_cents'] / 100:.2f} -- {t.get('description')}")

# Get pricing
pricing = client.wallet.get_pricing()

Inference deployment management

The same client.inference resource that makes OpenAI-compatible chat calls also manages model-serving deployments (create, monitor, stop, and delete).

allow_capacity_queue=True is explicit consent to wait for stock, so it requires 3 to 5 ranked gpu_priorities — the SDK rejects the combination locally before any request. Leave the queue off to book exactly one placement.

serving_mode, model_task, and supports_images are all server-derived and immutable: the platform reads them off the resolved model and rejects a conflicting assertion, so omit them and read the result back from the deployment. Which model_task values are accepted depends on the model, so the SDK does not judge one locally — it forwards whatever you pass and the server rules on it. preflight() echoes the derived task at canonical_request["model_task"] if you want to see it before creating.

context_length is optional and follows one policy: omit it and the server pre-fills min(native_max, 262144); the window is adjustable only when the model's native max exceeds the 32,768 floor, and it can never exceed the model's own native max. It is also a sizing input — a bigger window means a bigger KV cache, which can raise the minimum GPU count. Read the ceiling with client.models.native_max_context(model_id) first, and read it back from native_max_context on the deployment detail. Pass it as an int: the local capacity pre-check only runs on a window it can prove fits, so anything else (a numeric string included) is forwarded for the server to answer.

request = {
    "name": "llama-api",
    "source_type": "hf_model",
    "hf_model_id": "meta-llama/Llama-3.1-8B-Instruct",
    "gpu_type": "H100_80GB",
    "gpu_count": 1,
    # Queueing needs 3-5 ranked placements; the first is the one booked now.
    "allow_capacity_queue": True,
    "gpu_priorities": [
        {"gpu_type": "H100_80GB", "gpu_count": 1},
        {"gpu_type": "A100_80GB", "gpu_count": 1},
        {"gpu_type": "L40S_48GB", "gpu_count": 2},
    ],
    "max_price_hour_cents": 500,
}

# Never ask for more context than the model has.
native_max = client.models.native_max_context(request["hf_model_id"])
if native_max:
    request["context_length"] = min(65536, native_max)

# No wallet mutation and no GPU allocation.
check = client.inference.preflight(request)
print(check["selected_gpu"], check["alternatives"], check["billing"])

deployment = client.inference.create(request, idempotency_key="deploy-create-20260711-0001")
print(deployment["inference_key"])  # returned once; store securely

status = client.inference.status(deployment["id"])
print(status["status"], status.get("status_reason"), status.get("queue_expires_at"), status.get("wallet_authorization_status"))
# The detail resolves the per-deployment serving settings a list row omits.
print(status["context_length"], status["native_max_context"], status["model_task"])
# status stays "failed" for existing filters when status_reason is "queue_expired".
notification_history = client.inference.notifications(deployment["id"])
print([(row["event_type"], row["state"], row["attempt_count"]) for row in notification_history])

# Bounded newest-first listing; keep filters unchanged while using next_cursor.
# LIST rows are a lean grid projection: the model handle is `model_ref` (there
# is no `model` key) and the serving settings above are absent -- call
# client.inference.get(id) for those.
page = client.inference.list_page(limit=100, status="running", search="llama")
if page["has_more"]:
    older = client.inference.list_page(
        limit=100, status="running", search="llama", cursor=page["next_cursor"]
    )
    print(older["deployments"])
# Lazy traversal avoids one unbounded response.
for item in client.inference.iter_all(status="running"):
    print(item["name"], item["model_ref"], item["requests_total"])

client.inference.stop(deployment["id"])
client.inference.delete(deployment["id"])

GPU

View GPU pricing and get hardware recommendations.

# Get all GPU pricing
pricing = client.gpu.get_pricing()
for gpu in pricing["gpus"]:
    print(f"{gpu['display_name']}: {gpu['price_display']} -- {gpu['vram_gb']}GB VRAM")

# Authoritative model-aware choices, live stock, total pricing, and alternatives
options = client.gpu.get_options(
    "meta-llama/Llama-3.1-8B-Instruct",
    train_type="qlora",
    method="sft",
)

# Get recommended GPU for a model
rec = client.gpu.get_recommended("meta-llama/Llama-3.1-8B-Instruct")
if rec:
    print(f"Recommended: {rec['display_name']} x{rec['recommended_count']}")
    print(f"Total VRAM: {rec['total_vram_gb']}GB")
    print(f"Cost: ${rec['estimated_cost_per_hour_cents'] / 100:.2f}/hr")
    print(f"Reason: {rec['reason']}")

Introspect

Discover the permissions and scope of your API key.

info = client.introspect()
print(f"Org: {info['org']['name']}")
print(f"Scopes: {', '.join(info['scopes'])}")
print(f"Allowed tools: {len(info['allowed_mcp_tools'])}")

Error Handling

All API errors raise ApiError with status, code, request_id, and message attributes.

from bios import RunBiOS, ApiError

client = RunBiOS(api_key="bios-...")

try:
    job = client.training.get("bad_id")
except ApiError as e:
    if e.status == 404:
        print("Job not found")
    elif e.status == 401:
        print("Invalid API key")
    elif e.status == 403:
        print("Insufficient permissions")
    else:
        print(f"API error {e.status}: {e.message}")
        if e.request_id:
            print(f"Request ID: {e.request_id}")

GPU rejections

Every rejected GPU selection raises CapacityUnavailableError (a subclass of ApiError), for training and for inference, so one except reaches all of them and every one carries the same recovery data: alternatives (the canonical available_gpus list), minimum_requirement, selected, reason, and checked_at.

The rejections split into two families. Read permanent (or queue_offered), never the status or the code, to tell them apart:

CAPACITY_UNAVAILABLE means the GPU is not in stock right now, so waiting or joining the capacity queue can still get the request booked. The other four are fixed facts about the request: waiting does not help and the capacity queue cannot help either, however long the request waits. Only a different GPU type or a different GPU count can make it run.

from bios import RunBiOS, CapacityUnavailableError

try:
    deployment = client.inference.create(request, idempotency_key="deploy-create-20260725-0001")
except CapacityUnavailableError as e:
    print(e.message)
    print(e.reason, e.code, e.minimum_requirement)
    if e.permanent:
        # Never offer to wait here. Pick from the alternatives instead.
        for gpu in e.alternatives:
            print(gpu["gpu_type"], gpu["min_gpus"], gpu["valid_counts"])
    elif e.queue_offered:
        print("Out of stock right now. Retry later, or submit with the capacity queue.")
    if e.gpu_priorities_entry:
        print(f"gpu_priorities entry {e.gpu_priorities_entry} is the one to change")

inference.create runs the same check before it submits anything, so a selection that can never run this model raises the identical error (same code, same 400, queue_offered False) without a round trip. The service stays the authority and re-checks on create.

Python Version Support

• Python 3.10+

License

MIT