SDK recipes

These recipes keep one project client open, pass canonical values between doors, and persist every durable operation reference before awaiting it.

Capture and freeze a training snapshot

import sx


async def capture_snapshot(project_id, task_ref, embodiment_name, dataset_name):
    async with sx.connect(project_id) as client:
        task = await client.experience.task(task_ref)
        embodiment = await client.embodiment.get(embodiment_name)

        capture = await client.experience.capture(
            task,
            with_embodiment=embodiment,
            episodes=100,
        )
        save_operation_ref(str(capture.ref))
        receipt = await capture

        snapshot = await (
            client.experience
            .dataset(dataset_name)
            .where(task=task)
            .snapshot()
        )
        return receipt, snapshot

The capture receipt proves the bounded collection work. The snapshot is the immutable value a training run should consume; the live dataset name alone is not reproducible input.

Evaluate an exact application

async def evaluate(project_id, task_ref, application_ref):
    async with sx.connect(project_id) as client:
        task = await client.experience.task(task_ref)
        application = await client.autonomy.application(application_ref)

        operation = await client.worlds.evaluate(
            application,
            for_task=task,
            seed=7,
        )
        save_operation_ref(str(operation.ref))
        return await operation

Resolve the application and task in the same project context. Keep the returned evaluation with its operation reference and evidence; do not reduce it to a detached global score.

Train without deploying

from pathlib import Path

import yaml
from sx_autonomy import application_manifest_from_dict


def load_manifest(path: str):
    return application_manifest_from_dict(
        yaml.safe_load(Path(path).read_text())
    )


async def train(project_id, manifest_path):
    manifest = load_manifest(manifest_path)
    async with sx.connect(project_id) as client:
        operation = await client.train(manifest)
        save_operation_ref(str(operation.ref))
        return await operation

Use this path for a governed build whose result will be evaluated or compared separately.

Run the complete deployment program

async def deploy(project_id, manifest_path):
    manifest = load_manifest(manifest_path)
    async with sx.connect(project_id) as client:
        operation = await client.deploy(manifest)
        save_operation_ref(str(operation.ref))
        return await operation

This is the default promotion path: one operation owns build or reuse, Worlds evidence, qualification, task approvals, and rollout. Use direct client.fleet.deploy(...) only when your own promotion controller already holds every exact evidence value.

Resume after process loss

async def resume(project_id, operation_ref):
    async with sx.connect(project_id) as client:
        operation = client.operation(operation_ref)
        status = await operation.status()
        print(status.state, status.cursor)
        return await operation

Never resubmit solely because a notebook, worker, or agent disconnected. Reattach first.

Run in CI

import os

async with sx.connect(
    project_id,
    api_key=os.environ["SX_API_KEY"],
) as client:
    operation = await client.worlds.evaluate(application, for_task=task)
    publish_ci_output("operation_ref", str(operation.ref))
    evaluation = await operation

Use a revocable key scoped to the project and minimum required grants. Inject it with the CI secret store; never print it or write it to the workspace. Publishing the operation reference is safe and lets a human or agent inspect the same durable work later.

Recipe invariant

Resolve exact inputs, submit once, persist the reference, await the owner, and return the typed result. Every recipe in your codebase should preserve that shape.