pip install usfbios

pip install usfbios[async]

from usfbios import USFBios

client = USFBios(api_key="sk_live_...")

# Or let the SDK read from USFBIOS_API_KEY env var:
client = USFBios()

info = client.introspect()
print(f"Org: {info.org.name}")
print(f"Workspace: {info.workspace.name}")
print(f"Scopes: {', '.join(info.scopes)}")

# Search for models
results = client.models.search(query="llama", type="llm", limit=10)
for m in results.models:
    print(f"{m.id} - {m.total_params}B params")

# Get model config for GPU planning
config = client.models.get_config("meta-llama/Llama-3.1-8B")
print(f"{config.total_params}B params, MoE: {config.is_moe}")

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

# Upload a local dataset file
ds = client.datasets.upload(
    file_path="./training_data.jsonl",
    name="Customer Support SFT",
)
print(f"Uploaded: {ds.id}")

# List all datasets
datasets = client.datasets.list()
print(f"{len(datasets)} datasets")

# Preview rows before training
preview = client.datasets.preview("ds_abc123", page=1, page_size=5)
print(preview.columns)

# Validate a file before uploading
result = client.datasets.validate("./data.jsonl")
if result.valid:
    print(f"Valid {result.format} - {result.row_count} rows")
else:
    print("Errors:", result.errors)

# Import from HuggingFace
imported = client.datasets.import_from_huggingface(
    repo_id="databricks/dolly-15k",
    integration_id="int_abc123",
    name="Dolly 15k",
)

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

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

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

# Monitor training
metrics = client.training.get_metrics(job.id)
print(f"Loss: {metrics.current_loss}")

# View logs
logs = client.training.get_logs(job.id)
for line in logs.logs:
    print(line)

# List checkpoints
checkpoints = client.training.get_checkpoints(job.id)
for cp in checkpoints:
    print(f"Step {cp.step}: loss {cp.loss}")

# Stop and resume
client.training.stop(job.id)
client.training.resume(job.id, idempotency_key="training-resume-20260711-0001")

# Check wallet balance
balance = client.wallet.get_balance()
print(f"Balance: {balance.balance_cents / 100:.2f{'}'}")

# View transaction history
txns = client.wallet.get_transactions(limit=20)
for t in txns:
    print(f"{t.type}: {t.amount_cents / 100:.2f{'}'} - {t.description}")

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

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

# Import from a connected HuggingFace account
imported = client.datasets.import_from_huggingface(
    repo_id="your-org/private-dataset",
    integration_id="int_abc123",
    name="My Private Dataset",
    split="train",
    max_samples=10000,
)
print(f"Imported: {imported.id} - {imported.row_count} rows")

from usfbios import USFBios, ApiError

client = USFBios()

try:
    job = client.training.get("job_nonexistent")
except ApiError as e:
    print(f"HTTP {e.status}: {e.message}")
    print(f"Code: {e.code}")             # e.g. "NOT_FOUND"
    print(f"Request ID: {e.request_id}")  # for support

# Common error status codes:
# 401 - Invalid or expired API key
# 403 - Key missing required scope
# 404 - Resource not found
# 422 - Validation error
# 429 - Rate limit exceeded (auto-retried)

import asyncio
from usfbios import AsyncUSFBios

async def main():
    client = AsyncUSFBios(api_key="sk_live_...")

    # All methods are awaitable, same signatures as sync client
    models = await client.models.search(query="llama 8B")
    print(f"Found {len(models.models)} models")

    ds = await client.datasets.upload(
        file_path="./data.jsonl",
        name="Async Upload",
    )

    job = await client.training.create(
        model="meta-llama/Llama-3.1-8B-Instruct",
        dataset_id=ds.id,
        method="sft",
        adapter="lora",
    )

    # Monitor in a loop
    while True:
        status = await client.training.get(job.id)
        if status.status in ("completed", "failed"):
            break
        metrics = await client.training.get_metrics(job.id)
        print(f"Step {metrics.global_step}: loss {metrics.current_loss:.4f}")
        await asyncio.sleep(30)

    await client.close()

asyncio.run(main())

from usfbios.types import (
    ModelSearchParams,
    ModelSearchResponse,
    ModelConfig,
    Dataset,
    DatasetUploadParams,
    TrainingJob,
    TrainingCreateParams,
    TrainingMetrics,
    TrainingCheckpoint,
    WalletBalance,
    Transaction,
    GPUPricingResponse,
    ApiKeyIntrospection,
)

# All responses are typed dataclasses with IDE autocompletion
job: TrainingJob = client.training.get("job_abc123")
print(job.model)        # str
print(job.status)       # Literal["pending", "running", "completed", "failed"]
print(job.created_at)   # datetime