Python SDK

Use Python to submit containerized Jobs, follow their progress, and run work inside persistent Workers. Client handles API configuration and authentication for Job and Worker operations. Other operations, including Services and bucket management, are available through the generated API clients.

Install and configure

Install anycloud-sdk with Python 3.10 or newer. The distribution includes both anycloud and anycloud_workflows:

pip install anycloud-sdk

You need a running AnyCloud API, a login token, and a saved compute credential for remote VM Jobs. See Getting Started for setup. Use the SDK version that matches your API version.

from anycloud import Client

with Client() as client:
    print(client.sdk_version)

The client discovers configuration in this order:

Setting	Precedence, highest first
API URL	`api_url=` argument, `API_URL`, `$ANYCLOUD_DIR/api-url`, `http://localhost:8080`
Login token	`token=` argument, `$ANYCLOUD_DIR/.token`, `ANYCLOUD_TOKEN`, `GITHUB_TOKEN`

ANYCLOUD_DIR defaults to ~/.anycloud. anycloud login saves the token, and anycloud api use saves the API target. A saved nonempty token takes precedence over token environment variables. To override discovery, pass Client(api_url="https://api.example.com", token="github-access-token"). The client accepts an API URL with or without /v1. Empty explicit values, invalid URLs, and unreadable configuration files raise ValueError.

API authentication does not select cloud compute credentials. Pass credential_name on each remote image submission, or supply inline credentials through cloud_config. A Job targeting an existing Worker uses that Worker's Cluster capacity and needs no per-Job compute credential.

Keep the with Client() context open while using its Job handles. The context closes its connections when it exits; applications managing the lifetime explicitly can call client.close().

Submit and inspect a Job

A dedicated Job runs a published container image on its own VM. Select a saved credential and GPU, then inspect the returned handle:

from anycloud import Client

with Client() as client:
    job = client.submit(
        "ghcr.io/acme/trainer:latest",
        credential_name="aws-prod",
        gpu="h100:8",
        cloud_config={"spot": True},
        command=["python", "train.py"],
    )
    print(job.id)
    status = job.status()
    print(status.deployment.state)

Build your application and its dependencies into the image. Use an immutable tag or digest when reproducibility matters; command overrides the image's default command. See Container Images.

submit() returns after admission; provisioning and execution continue in the background. status() fetches a fresh observation with the Job state and events. States are strings such as "queued", "running", and "completed". See Job lifecycle for their meaning.

Use cloud_config for spot, region, VM type, disk, or bucket settings, env for environment variables, and secrets for saved Secret names. The submission reference lists the arguments, and Configuration explains their constraints.

Jobs on Workers

A targeted Job runs inside an existing Worker's application. Pass its name or ID as a string, with an optional Job ID:

import httpx

from anycloud import Client
from anycloud.exceptions import ApiException

with Client() as client:
    worker_name = "model-workers"
    job = client.submit(worker=worker_name, deployment_id="request-123")
    status = job.status()
    state = status.deployment.state if status.deployment is not None else None
    terminal_states = {"completed", "errored", "failed", "invalid", "terminated"}

    if state in terminal_states:
        message = f"Job {state}"
    elif state == "running":
        message = "Picked up"
    elif state is None:
        message = "Job status unknown"
    elif state != "queued":
        message = f"Job {state}"
    else:
        ahead = status.jobs_ahead
        if ahead is None:
            position = "Queue position unknown"
        elif ahead == 0:
            position = "First in the queue"
        else:
            position = f"{ahead} Jobs ahead"

        try:
            worker = client.get_worker(worker_name)
        except (httpx.HTTPError, ApiException):
            availability = "Worker observation unavailable"
        else:
            workload = worker.workload
            if worker.deletion_requested_at is not None or worker.deleted_at is not None:
                availability = "Worker is draining or deleted"
            elif workload is None or workload.observed_generation is None:
                availability = "Worker readiness unknown"
            elif workload.observed_generation != worker.generation:
                availability = "Waiting for the current Worker generation"
            elif not workload.container_ready:
                availability = "Worker container is not ready"
            elif worker.unused_capacity == 0:
                availability = "All Worker concurrency is currently held"
            elif ahead is not None and worker.unused_capacity > ahead:
                availability = "Capacity currently available for this queue position"
            else:
                availability = "Waiting for pickup"
        message = f"{position}; {availability}"

    print(message)

Supply exactly one of image or worker. Targeted submissions accept only worker and optional deployment_id. Image, compute, credential, and container options are rejected even if you pass an empty mapping or list. Configure those settings on the Worker instead. Pass worker.id when you have a WorkerSummary, rather than passing the summary itself.

job.status().jobs_ahead counts earlier queued Jobs on the same Worker:

0 means first in the queue; positive values count Jobs ahead.
None applies outside the queued Worker-targeted case or when the response omits the field.
The count covers the shared Worker queue, including other callers' Jobs. Ordering uses original submission time, then Job ID. Retries and resubmissions retain that time, so older Jobs returning to the queue can move ahead.

Job state, events, and queue count come from one API database snapshot. Worker lookup is a separate observation: worker.jobs.queued counts the whole queue, and worker.unused_capacity accounts for executions still holding concurrency, including cleanup. Readiness requires the current desired generation and a ready container. Draining or deleted Workers cannot accept work.

Only "running" confirms pickup. Capacity and queue position are advisory; zero Jobs ahead does not guarantee immediate start or an ETA. Missing or failed observations mean unknown rather than a confirmed capacity blocker. Refresh status at a cadence appropriate to your application, checking terminal states before displaying queue or Worker messages.

Wait and terminate

Use job.wait() to wait for successful completion. Save the Job ID to observe it from another process with client.get():

from anycloud import Client
from anycloud_workflows import DeploymentWaitTimeout, JobFailedError

with Client() as client:
    job = client.get("request-123")  # An existing Job ID.
    try:
        job.wait(timeout_seconds=3600, poll_interval_seconds=2)
        print(f"Completed {job.id}")
    except JobFailedError as error:
        print(f"Job {error.deployment_id}: {error.state}")
        print(error.status.events)
    except DeploymentWaitTimeout:
        print(f"Stopped waiting for {job.id}; check its status before deciding what to do")

wait() returns the same Job handle on completed. It raises JobFailedError for errored, failed, invalid, or terminated; error.status contains the terminal status response. Omitting timeout_seconds waits without a deadline. A DeploymentWaitTimeout stops polling without terminating work.

To stop a Job, request termination explicitly:

from anycloud import Client

with Client() as client:
    job = client.get("request-123")
    response = job.terminate()
    print(response.message)

Termination records the terminal state and starts cleanup. It does not wait for resources to finish releasing. API and transport errors propagate from status, wait, and termination calls; see Error handling.

Reattach to a Job

client.get(deployment_id) fetches status immediately, checks that the resource is a Job, and returns a handle. Missing or inaccessible IDs raise a generated API exception; a Service or VM ID raises ValueError. The handle starts with last_status populated and submission=None.

For an existing Job in a resubmittable terminal state, after cleanup completes:

from anycloud import Client

with Client() as client:
    job = client.get("request-123")
    job.resubmit()
    job.wait()

Resubmission keeps the Job ID and returns a generated DeploymentResubmitResponse. An early resubmission can fail while cleanup is pending. Success clears last_status; a failed request leaves it intact. Any admission response known to the handle stays available as submission.

VM-backed Jobs reuse the container configuration accepted at submission. Worker Jobs retain their Worker target and use its current configuration when selected for execution; resubmission can therefore run newer configuration. Per-Job logs and exec are unavailable for Worker Jobs.

Handle Jobs inside a Worker

Inside a Worker Pod, current_worker() discovers the injected API URL, Worker ID, and rotating workload token. Its next_job(timeout=20) waits up to 20 seconds to discover a candidate and returns a CurrentJob handle or None. An attempted claim may resolve later within the separate request budget. Omitting timeout uses the 20-second wait; pass timeout=0 for one immediate lookup. A handler needs only that handle:

import time

import httpx

from anycloud.exceptions import ServiceException
from anycloud_workflows import CurrentJob, current_worker


def handle(job: CurrentJob) -> None:
    # Replace these steps with your application work, using job.id to load inputs.
    for step in range(3):
        if job.cancellation_requested():
            # Stop local work and release its resources before acknowledging cleanup.
            job.cleanup()
            return
        print(f"Job {job.id}: step {step}", flush=True)
        time.sleep(1)
    job.complete()


with current_worker() as worker:
    while True:
        try:
            job = worker.next_job()
        except (httpx.HTTPError, ServiceException, TimeoutError):
            time.sleep(1)
            continue
        if job is None:
            time.sleep(1)
            continue
        handle(job)

The example handles one Job at a time and delays empty polls by one second. Applications own capacity, concurrency, and cooperative cancellation; a handle does not interrupt work or clean up application resources. Keep the context open until every handler has finished. current_worker(timeout_seconds=30) sets a separate elapsed request budget, capped at 30 seconds. Local caller serialization, discovery, claim, response decoding, and cancellation share that budget. Handler outcome and cleanup calls remain usable while discovery waits.

None means no unresolved claim from this call can assign work later. Discovery does not reserve work. If a claim response is lost, keep the same Worker context and retry next_job(); it retains the exact Job, submission revision, previous execution, and new execution UUID. A later call may use a different timeout. Temporary capacity, readiness, authentication, and network errors preserve this pending claim. A request timeout raises an exception and never means empty. The SDK checks both returned IDs before accepting a claim.

Member	Purpose	Return value
`job.id`	Identify the Job and load application inputs.	`str`
`job.complete()`	Publish successful completion.	`None`
`job.error(message)`	Publish an application failure.	`None`
`job.invalid(message)`	Reject invalid input without retrying.	`None`
`job.retry(message)`	Report a retryable failure for the API to reschedule.	`None`
`job.cancellation_requested()`	Check whether local work should stop.	`bool`
`job.cleanup()`	Confirm that terminated local work has stopped and released its resources.	`None`

Handle methods refresh the projected token before each operation and preserve generated API exceptions. job.selected is the original generated SelectedJob model, including its Job and execution IDs. Context methods such as worker.outcome(job, "completed") accept either handles or generated selections. Ordinary handlers do not need to manipulate execution IDs.

The lower-level WorkerJobPoller accepts an application-owned generated ApiClient and returns SelectedJob from next_job(). Keep one poller alive across retries: it retains the complete claim identity after an uncertain acquisition failure so the next request can safely replay it. Install the matching API/SDK release in Worker images.

Manage Workers

Create a Worker on a Ready NVIDIA GPU Cluster with an application like the handler above:

from anycloud import Client

with Client() as client:
    worker = client.create_worker(
        "model-workers",
        "ghcr.io/acme/model-worker:latest",
        cluster="training",
        docker_options={"gpus": "all"},
        command=["python", "worker.py"],
        max_concurrent_jobs_per_replica=4,
    )
    print(worker.id, worker.generation)

Creation returns a WorkerSummary before the application is ready. Use client.get_worker(worker.id) to fetch current observations. The Cluster determines the accelerator model and per-VM supply. GPU options use the same docker_options representation as VM Jobs (dockerOptions in the API): {"gpus": "all"} or a positive integer string such as {"gpus": "1"}. Omitting GPU options on verified NVIDIA capacity uses all GPUs on one capacity VM; omitting them on CPU capacity requests none. Zero is invalid, and explicit GPU access requires sufficient verified NVIDIA capacity. Image names and CUDA metadata do not infer a GPU request.

GPU Workers select the NVIDIA runtime automatically. An explicit {"runtime": "nvidia"} is supported on NVIDIA capacity. CPU {"runtime": "runc"} selects the default managed runtime; combining runc with GPU access is rejected by the current managed adapter because its device injection requires the NVIDIA runtime. See Cluster GPU rules. Worker named Secrets and per-Job buckets are not supported.

Update an existing Worker by name or ID:

from anycloud import Client

with Client() as client:
    worker = client.update_worker(
        "model-workers",
        image="ghcr.io/acme/model-worker:v2",
        env={"MODEL_REVISION": "v2"},
    )
    print(worker.generation)

Updates preserve omitted fields, including submitted GPU intent. A supplied docker_options dictionary replaces the entire option map; omitting gpus from that replacement reapplies the capacity default. None clears command, env, or docker_options; clearing Docker options also reapplies capacity defaults. Empty lists and maps remain explicit values. image and max_concurrent_jobs_per_replica cannot be None. An empty update raises ValueError. The Worker's name and Cluster cannot change.

client.delete_worker("model-workers") starts draining/deletion and returns a summary without waiting for active Jobs or Pods to finish.

List all Workers or filter by a Cluster name or ID:

from anycloud import Client

with Client(request_timeout_seconds=10) as client:
    all_workers = client.list_workers()
    for worker in client.list_workers(cluster="training"):
        print(worker.name, worker.jobs.queued, worker.unused_capacity)

list_workers() returns a list of generated WorkerSummary models, or an empty list when no Workers match. Omitting cluster or passing None lists Workers across Clusters. The optional request timeout is configured once on Client and defaults to 30 seconds.

Chain Jobs

Submit the next Job after the previous one completes. Use an output bucket as the next Job's input to pass data between their containers:

from anycloud import Client

with Client() as client:
    prepared = client.submit(
        "ghcr.io/acme/prepare:latest",
        credential_name="aws-prod",
        gpu="h100",
        cloud_config={"output_bucket": "prepared-training-data"},
    )
    prepared.wait()
    trained = client.submit(
        "ghcr.io/acme/trainer:latest",
        credential_name="aws-prod",
        gpu="h100",
        cloud_config={"input_bucket": "prepared-training-data"},
    )
    trained.wait()

The first application writes to /mnt/output; the next reads /mnt/input. Use distinct buckets or application prefixes for concurrent runs. A failed wait raises before the next submission. For parallel independent Jobs, submit them in a Python loop before waiting; the application owns partial-failure handling.

Client reference

Client(api_url=None, token=None, request_timeout_seconds=30) discovers any omitted connection settings. The request timeout applies to submission, status, Worker management, termination, and resubmission. job.wait() has its own polling deadline and interval.

Submit options

client.submit(image=None, *, worker=None, ...) requires exactly one target. For image submissions:

Argument	Meaning
`credential_name`	Saved compute credential name.
`cloud_config`	Cloud, region, VM, spot, disk, and bucket settings.
`gpu`	GPU model/count string, or an ordered list such as `["h100:8", "a100:8"]`.
`env`	Mapping of environment names to string values.
`secrets`	Sequence of saved Secret names.
`docker_options`	Container settings such as `{"shm_size": "8g"}`.
`command`	Sequence of command arguments, such as `["python", "train.py"]`.
`deployment_id`	Optional custom Job ID.

cloud_config and docker_options accept generated models or mappings with Python field names (vm_type) or wire aliases (vmType). Pass command and secrets as sequences rather than a single string. For a Worker target, only worker, deployment_id, timeout, and wait are accepted. Worker submission allocates a valid Job ID locally when omitted, so a lost admission response can still retain a reference.

timeout is an optional nonnegative number of seconds from admission to first claim. wait=True blocks until that claim is durable, independently of handler completion. wait=False returns after admission; job.wait_for_claim() can observe the same persisted deadline later. Omitting timeout sets no claim deadline, including with wait=True. A zero timeout expires an unclaimed submission immediately. job.resubmit(timeout=..., wait=True) keeps the ID and records a new submission; omitting its timeout clears the old deadline. Automatic retries keep their submission revision and first-claim evidence.

JobSubmissionExpired confirms that this submission never claimed work and cannot start later. JobSubmissionTerminated identifies other termination before claim, and JobSubmissionChanged identifies an explicit resubmission that superseded the handle's receipt. Each retains .job, .deployment_id, and the handle's submission .revision, so a failed blocking submission still provides its recovery handle. These exceptions are exported from anycloud_workflows. Worker Client.submit() may raise JobSubmissionUncertain with .job and the original .cause when admission is uncertain or its internal claim wait fails. Definitive admission rejections, such as HTTP 401, retain their generated exception types. Inspect or retry with the retained handle; uncertainty does not confirm expiry. Request errors from existing-handle resubmit() and wait_for_claim() retain their original types. Every Job resubmission uses the client's configured request timeout as its elapsed request budget. Management API and SDK releases must match exactly.

See Configuration for field mappings, defaults, hardware constraints, storage credentials, and environment precedence.

Handle results

Member	Result
`job.id`	Job ID string.
`job.submission`	Generated `DeploymentAdmissionResponse`, or `None` after reattachment.
`job.status()`	Fresh generated `DeploymentStatusResponse`, also saved in `last_status`.
`job.last_status`	Last cached status, or `None` before observation/after successful resubmission.
`job.wait(...)`	The same handle on success; raises on unsuccessful terminal state or timeout.
`job.wait_for_claim()`	The same handle after this Worker submission's first durable claim.
`job.terminate()`	Generated `MessageResponse`; cleanup continues asynchronously.
`job.resubmit()`	Generated `DeploymentResubmitResponse`, including the Worker submission receipt; Job ID stays the same.

Worker creation accepts name, image, required keyword cluster, and optional command, env, docker_options, and max_concurrent_jobs_per_replica (default 1). Updates accept those mutable settings except name and cluster. get_worker(identifier), update_worker(worker, ...), and delete_worker(worker) accept name or ID strings. list_workers(*, cluster=None) accepts an optional Cluster name or ID and returns a list of WorkerSummary models.

Deployment operations

Use generated API families for Services, deployment listing, cost, and SSH metadata. They can share a Client's authentication and connections:

from anycloud import Client
from anycloud.api.deployment_admission_api import DeploymentAdmissionApi
from anycloud.api.deployments_api import DeploymentsApi
from anycloud.models.deployment_admission_input import DeploymentAdmissionInput

with Client() as client:
    service = DeploymentAdmissionApi(client.api_client).submit_deployment_sync(
        client.sdk_version,
        DeploymentAdmissionInput.from_dict(
            {
                "deploymentType": "service",
                "image": "ghcr.io/acme/inference:latest",
                "credentialName": "aws-prod",
                "cloudConfig": {},
                "gpuType": "l40s",
                "command": ["python", "serve.py"],
            }
        ),
    )
    print(service.id, service.url)
    deployments = DeploymentsApi(client.api_client)
    for deployment in deployments.list_deployments_sync(client.sdk_version, limit="20"):
        print(deployment.id, deployment.state)

Service admission does not wait for readiness. Use get_deployment_status_sync(id, client.sdk_version) and terminate_deployment_sync(id, client.sdk_version) on the same DeploymentsApi instance while the client is open. See Services for a full example; Client.get() supports Job handles only.

list_deployments_sync() returns generated deployment models. Worker Jobs have worker_id set and image, image_digest, command, and docker_options set to None; VM-backed deployments require an image. List results keep env and github_token as None and ssh_private_key as "". Inspect the Worker's configuration with client.get_worker(deployment.worker_id) while the client is open.

get_deployment_status_sync().deployment exposes identity, lifecycle, and non-sensitive container metadata, including worker_id. Worker Jobs have the same null container fields as list results. Status responses omit environment, registry-token, and SSH private-key fields.

Wait for any deployment

The lower-level wait helper returns a generated status response for any terminal state, including failure, without raising JobFailedError:

from anycloud import Client
from anycloud_workflows import wait_for_terminal_deployment

with Client() as client:
    status = wait_for_terminal_deployment(
        client.api_client,
        "request-123",
        client.sdk_version,
        timeout_seconds=900,
    )
    print(status.deployment.state)

Use it when you want to handle every terminal state yourself. It raises DeploymentWaitTimeout when the polling deadline expires and preserves API exceptions. Services normally run until termination, so waiting for a terminal state does not check Service readiness.

Other API families

Use generated API families for operations beyond the workflow helpers. They share a Client's authentication and connections through client.api_client; pass client.sdk_version to management methods.

API family	Operations
`APIAccessApi`	API access policy and admitted users.
`BucketsApi`	Bucket creation, deletion, existence, listing, and object listing/deletion.
`ClustersApi`	Cluster management.
`CredentialsApi`	Saved compute and storage credentials.
`DatabaseApi`	Database inspection.
`DeploymentAdmissionApi`, `DeploymentsApi`	Submit, list, inspect, terminate, resubmit, and query cost/SSH metadata.
`NotificationsApi`, `QuotaApi`, `SpendControlsApi`	Notification, quota, and spending operations.
`SecretsApi`	Create, list, and delete named Secrets.
`ServiceUpgradesApi`	Service upgrades.
`WorkersApi`, `WorkerWorkloadApi`	Worker management and in-Pod Job lifecycle operations.

See Buckets, Secrets, and Configuration for examples. Generated responses are models with attributes; method arguments and fields follow the API contract.

Generated client setup

Applications can construct an ApiClient directly. Generated clients require explicit configuration and the installed SDK version as the any_cloud_client_version argument to management methods; they do not discover CLI settings themselves. The argument supplies the AnyCloud-Client-Version header and must match the API version.

from importlib.metadata import version

from anycloud.api.deployments_api import DeploymentsApi
from anycloud.api_client import ApiClient
from anycloud.configuration import Configuration
from anycloud.sync_helper import run_sync

api_client = ApiClient(Configuration(
    host="http://localhost:8080/v1",
    access_token="github-access-token",
))
try:
    deployments = DeploymentsApi(api_client).list_deployments_sync(
        version("anycloud-sdk"), limit="20"
    )
    print([deployment.id for deployment in deployments])
finally:
    run_sync(api_client.close())

Set host to your selected API URL including /v1. access_token supplies Bearer authentication. The application owns this client's lifetime.

Async operations

Use generated methods without the _sync suffix in an async context:

import asyncio
from importlib.metadata import version

from anycloud.api.secrets_api import SecretsApi
from anycloud.api_client import ApiClient
from anycloud.configuration import Configuration


async def main() -> None:
    configuration = Configuration(
        host="http://localhost:8080/v1",
        access_token="github-access-token",
    )
    async with ApiClient(configuration) as api_client:
        secrets = await SecretsApi(api_client).list_secrets(version("anycloud-sdk"))
        print([secret.name for secret in secrets])


asyncio.run(main())

Error handling

JobFailedError and DeploymentWaitTimeout describe waiting outcomes. Generated API exceptions propagate unchanged from both Client and current_worker(); their HTTP status is in ApiException.status. Local argument/configuration checks may raise ValueError, TypeError, or model validation errors before a request.

For declared API errors, error.data.actual_instance contains the generated error envelope. For example, deleting a Secret used by an active deployment can raise ConflictException:

from anycloud import Client
from anycloud.api.secrets_api import SecretsApi
from anycloud.exceptions import ConflictException
from anycloud.models.secret_in_use_api_error import SecretInUseApiError

with Client() as client:
    try:
        SecretsApi(client.api_client).delete_secret_sync("training", client.sdk_version)
    except ConflictException as error:
        envelope = error.data.actual_instance
        if isinstance(envelope, SecretInUseApiError):
            print(envelope.code, envelope.data.deployment_ids)
        else:
            raise

Unexpected envelopes are retained as UndefinedApiError. An API failure is separate from a Job's terminal state: inspect status/events for workload failures, and handle httpx.HTTPError for transport failures where appropriate.

Migrating existing code

from anycloud import Client is the public Job and Worker client. anycloud_workflows.Client remains the same class. Import current_worker, CurrentJob, JobFailedError, and DeploymentWaitTimeout from anycloud_workflows; API families, models, and HTTP exceptions live in anycloud.

When updating older SDK code:

Pass compute selection as submit(credential_name=..., cloud_config=...). Constructor-level credentials and deployment configuration defaults are unavailable. Token discovery does not automatically choose a compute account.
Use job.wait(timeout_seconds=...) and handle JobFailedError or DeploymentWaitTimeout. Read state from job.status().deployment.state; the old DeploymentFailedError, ConflictError, and NotFoundError types are replaced by workflow errors or generated API exceptions.
Replace current_deployment() with current_worker() and submit(deployment=...) with submit(worker="name-or-id"). Target a Worker by its name or ID rather than a handle. Worker summaries are generated data; use Client methods for management.
Replace old anycloud.types and Bucket configuration objects with supported cloud_config mappings or generated models. Attaching a bucket to a Job is supported independently of SDK byte transfers.

The SDK has no Service handles, JobGroup, submit_many(), or get_or_submit(). Use generated operations for Services and ordinary Python for orchestration. SDK log streaming, exec, and bucket-byte upload/download helpers are also unavailable; use the CLI for those tasks. Generated bucket metadata operations and Job bucket attachments remain available.

Install and configure​

Submit and inspect a Job​

Jobs on Workers​

Wait and terminate​

Reattach to a Job​

Handle Jobs inside a Worker​

Manage Workers​

Chain Jobs​

Client reference​

Submit options​

Handle results​

Deployment operations​

Wait for any deployment​

Other API families​

Generated client setup​

Async operations​

Error handling​

Migrating existing code​