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). Keys start with bios- (legacy usf- keys stay 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).

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")

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()

# 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. Pass the same mixing dictionary to preflight and create: {"mode": "interleave", "weights": [3, 1], "seed": 42}. Weights correspond to the ordered 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"],
    "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 metrics
metrics = client.training.get_metrics("job_abc123")
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