Python SDK

response = client.fusion.complete(
    prompt="Analyze this system architecture for single points of failure.",
    panel=["moonshotai/Kimi-K3", "moonshotai/Kimi-K2.6"],
    judge="claude-opus-4-8",
    max_tier=True
)
print(response["choices"][0]["message"]["content"])

from stimulir.realtime import AudioDelta, RealtimeClient, ResponseDone

client = RealtimeClient(instructions="You are a concise voice assistant.")

async with client.connect() as conn:
    await conn.setup()
    await conn.send_audio(pcm16_chunk)
    await conn.commit()
    async for event in conn.events():
        if isinstance(event, AudioDelta):
            play(event.pcm16)
        elif isinstance(event, ResponseDone):
            break

from stimulir import StimulirClient

client = StimulirClient()

prompt = client.prompts.get("customer.chat.agent", label="prod")

session_metadata = {
    **prompt.lineage,
    "model": "Qwen/Qwen2.5-VL-72B-Instruct",
}

# Use prompt.content in your runtime, then store session_metadata with the result.

client.prompts.create_version(
    "customer.chat.agent",
    content=open("prompts/customer_chat_agent.md").read(),
    label="prod",
    prompt_type="system",
    change_notes="Seed current Customer chat agent prompt",
)

client.prompts.patch_metadata(
    "customer.chat.agent",
    3,
    change_notes="Tightened interview scoring rubric",
)

client.prompts.label("customer.chat.agent", 3, "prod")
client.prompts.archive("customer.chat.agent", 2)

# checkout-service
checkout_client = StimulirClient(tags=["checkout-service", "prod"])

# returns-service
returns_client = StimulirClient(tags=["returns-service", "prod"])

checkout_client.data_assets.from_trace(
    "<trace-id>",
    source="agent",
    tags=["retried", "low-confidence"],
)

checkout_client = StimulirClient(tags=["checkout-service", "prod"])

checkout_client.data_assets.create(
    name="checkout-interaction-42",
    kind="trace",
    tags=["low-confidence"],
)
# -> stored tags: ["checkout-service", "prod", "low-confidence"]

asset = client.data_assets.upload(
    "datasets/customer_agent_eval.jsonl",
    name="Customer Agent eval seed",
    target="eval",
)

client.data_assets.patch(asset["id"], description="Held-out interview eval cases")
client.data_assets.stage(asset["id"], "lab", status="ready")

trace_asset = client.data_assets.from_trace(
    "<trace-id>",
    source="agent",
    target="sft",
    name="Cleaned prompt trace",
)

snapshot = client.data_assets.snapshot(trace_asset["id"])

client.data_assets.bulk_stage(
    ["<asset-id-1>", "<asset-id-2>"],
    stage="lab",
    target="eval",
    lineage_event={"job": "redact-normalize-label"},
)

client.data_assets.unstage("<asset-id>")

run = client.lab_evals.create_run(
    name="Customer Agent prompt regression",
    data_asset_id=snapshot["id"],
    prompt_ref="customer.chat.agent:prod",
    model="hybrie-runtime-default",
    execute=True,
)

report = client.lab_evals.get_run(run["id"])

client.lab_evals.score_niah(family="qwen3-4b", checkpoint_dir="/ckpt/d2l")
client.lab_evals.score_adapter(family="qwen3-4b", adapter_dir="/ckpt/adapter")
client.lab_evals.score_rl(family="qwen3-4b", environment="niah", policy="peft-lora", checkpoint_dir="/ckpt/rl")

client = StimulirClient()  # or StimulirClient(runtime_endpoint="http://localhost:8011")

# Route 1, PEFT LoRA (the #1 standard route: explicit rank/alpha)
sft = client.lab_train.sft(family="qwen3-4b", lora_rank=16, examples=200, epochs=3)
rl = client.lab_train.rl(family="qwen3-4b", policy="peft-lora", lora_rank=16, prompts=64)

# Route 2, Doc-to-LoRA (hypernetwork context internalisation)
d2l = client.lab_train.d2l(family="qwen3-4b", examples=500, epochs=3)

# Track training jobs
client.lab_jobs.list()
job = client.lab_jobs.get(sft["job_id"])
client.lab_jobs.cancel(rl["job_id"])

# Hot-swap LoRA adapters into the runtime for inference
client.lab_adapters.list()
client.lab_adapters.load("<adapter-id>")
client.lab_adapters.unload("<adapter-id>")

# Route 3, multimodal projector tuning (frozen encoder -> projector -> frozen LLM).
# stage1 trains the projector only; stage1+2 also jointly tunes a Stage-2 LoRA.
run = client.lab_projectors.create(
    modality="image",
    stages="stage1+2",
    lora_rank=8,
    lora_alpha=16,
    data_asset_id="<paired-multimodal-asset-id>",
    name="vision-projector",
)
client.lab_projectors.get(run["run"]["id"])
client.lab_projectors.list(modality="image")

result = client.capabilities.execute(
    "engineering_usage_summary",
    scope={
        "mode": "chat",
        "modeContext": "workspace",
        "opsWorkspace": "engineering_ops",
    },
    args={"window": "month"},
    confirmed=True,
)

storage_key = client.storage.upload(
    open("clip.mp4", "rb"),
    workspace_id="<workspace-id>",
    filename="clip.mp4",
)

objects = client.storage.list(workspace_id="<workspace-id>")
objects = client.storage.list(workspace_id="<workspace-id>", task_id="<task-id>")
folders = client.storage.folders(workspace_id="<workspace-id>")

link = client.storage.access_url("<object-id>", expires_in_seconds=900, download=True)
print(link["access_url"])

client.storage.delete("<object-id>")

# Close the loop: fetch an object by its storage key (what run/pipe return),
# not its object id. workspace_id defaults from your key's config.
info = client.storage.get_by_key(output_key, workspace_id="<workspace-id>")
data = client.storage.read_by_key(output_key)  # -> bytes

candidates = client.skills.discover("tosi-n", "agentic-frame", workspace_id="<workspace-id>", ref="main")

imported = client.skills.import_(
    "tosi-n", "agentic-frame", ["skills/evidence-clip"],
    workspace_id="<workspace-id>", ref="main",
)

installed = client.skills.list(workspace_id="<workspace-id>")

result = client.sandbox.run(
    "evidence-clip",
    workspace_id="<workspace-id>",
    input_key=storage_key,
    params={"start": 1, "end": 4},
)
print(result["status"], result["output_key"])

# Pipe several skills, threads each output_key into the next input_key.
# Short-circuits on the first step that does not complete (SandboxPipeError).
# Steps: a bare slug, a {"skill", "params"} dict, or a (skill, params) tuple.
results = client.sandbox.pipe(
    ["privacy-layer", {"skill": "capture-traces", "params": {"window": "7d"}}],
    input_key=storage_key,
    workspace_id="<workspace-id>",   # optional, defaults from your key's config
)
final_key = results[-1]["output_key"]

runs = client.sandbox.list(workspace_id="<workspace-id>")

from stimulir import CodexProviderConfig, render_codex_config_snippet

snippet = render_codex_config_snippet(
    CodexProviderConfig(
        model="moonshotai/Kimi-K2.7-Code",
        base_url="https://api.staging.stimulir.com/api/v1/inference",
        project_id="<project-id>",
        auth_command="stimulir",
        auth_args=("codex", "token", "--api-base", "https://api.staging.stimulir.com"),
    )
)
print(snippet)