Skip to main content

Lium

from lium.sdk import Lium

Defined in lium.sdk.client.

Clean Unix-style SDK for Lium.

Lium(config: Optional[lium.sdk.config.Config] = None, source: str = 'sdk')

Attributes​

NameTypeDescription
config
headers

Methods​

MethodDescription
list_ssh_keysReturn SSH keys registered for the current user.
register_ssh_keyRegister a new SSH public key under the current user.
default_ssh_key_namecli-<user>@<host> sanitised to A-Za-z0-9.@-.
upStart a new pod on a specific node.
podRetrieve detailed information about a specific pod.
logsStream logs from a pod.
editEdit a pod's template configuration.
lsList available nodes.
psList active pods.
downStop a pod.
rmRemove pod (alias for down()).
rebootReboot a pod.
get_default_imagesGet default images for GPU type and driver version.
default_docker_templateResolve the best default template for a node ID.
templatesList available templates.
get_executorResolve a node by ID.
gpu_typesGet list of available GPU types.
get_templateFetch a template by ID/HUID/name.
get_template_by_image_nameFetch a template by its Docker image + tag.
ssh_connectionSSH connection context manager.
execExecute a shell command on a pod over SSH.
stream_execExecute a shell command and stream incremental output.
exec_allExecute a shell command on multiple pods in parallel.
wait_readyPoll until a pod reports RUNNING + SSH metadata.
scpUpload a local file to a pod via SFTP.
downloadDownload a file from a pod via SFTP.
uploadUpload a file to a pod via SFTP.
sshGet SSH command string for connecting to a pod.
rsyncSync directories with rsync.
switch_templateSwitch the template of a running pod.
create_templateCreate a new template.
wait_template_readyWait for template verification to complete.
get_my_user_idGet the current user's ID.
update_templateUpdate an existing template owned by the caller.
walletsGet the caller's configured funding wallets.
add_walletLink a Bittensor wallet with the user account.
backup_createCreate or replace a backup configuration for a pod.
backup_nowTrigger an immediate backup for a pod.
backup_configReturn the backup configuration for a pod if one exists.
backup_listList all backup configurations across all pods.
backup_logsGet recent backup logs for a pod.
backup_deleteDelete a backup configuration by ID.
restoreRestore a backup to a pod.
restore_logsGet recent restore logs for a pod.
get_deployment_estimateEstimate deployment time for a template on a node.
balanceGet current account balance.
volumesList all volumes for the current user.
volumeGet a specific volume by ID.
volume_createCreate a new volume.
volume_updateUpdate a volume's metadata.
volume_deleteDelete a volume.
schedule_terminationSchedule a pod for automatic termination at a future date and time.
cancel_scheduled_terminationCancel a scheduled termination for a pod.
install_jupyterInstall Jupyter Notebook on a pod.

list_ssh_keys​

def list_ssh_keys() -> List[lium.sdk.models.SSHKey]:

Return SSH keys registered for the current user.

register_ssh_key​

def register_ssh_key(*, name: str, public_key: str) -> lium.sdk.models.SSHKey:

Register a new SSH public key under the current user.

default_ssh_key_name​

def default_ssh_key_name() -> str:

cli-<user>@<host> sanitised to [A-Za-z0-9._@-].

up​

def up(
    *,
    executor_id: str,
    name: str = 'Your Pod',
    template_id: Optional[str] = None,
    volume_id: Optional[str] = None,
    ports: Optional[int] = None,
    ssh_keys: Optional[List[str]] = None,
    ssh_name: Optional[str] = None
) -> Dict[str, Any]:

Start a new pod on a specific node.

Arguments:

  • executor_id: Target node ID string.
  • name: Human-friendly pod name (defaults to "Your Pod").
  • template_id: Template ID. Defaults to the node's default template.
  • volume_id: Optional volume ID to attach on spawn.
  • ports: Number of exposed ports to request.
  • ssh_keys: SSH public keys to authorize. Defaults to the keys discovered by the Config.
  • ssh_name: Optional name to use when registering a new SSH key with the backend. Defaults to cli-<user>@<hostname>. Only applied to keys that are not already registered server-side.

Returns: Pod metadata as returned by the rent API (id, name, status, ssh command, etc.).

pod​

def pod(pod_id: str) -> Dict[str, Any]:

Retrieve detailed information about a specific pod.

Arguments:

  • pod_id: The unique identifier of the pod to retrieve.

Returns: Raw pod data dictionary including template, node, status, and connection info.

logs​

def logs(
    pod_id: str,
    *,
    tail: int = 100,
    follow: bool = False
) -> Generator[bytes, NoneType, NoneType]:

Stream logs from a pod.

Arguments:

  • pod_id: The unique identifier of the pod.
  • tail: Number of lines to retrieve from the end of the logs (default: 100).
  • follow: If True, stream logs continuously (default: False).

Yields: Log lines as bytes.

edit​

def edit(pod_id: str, **kwargs) -> Dict[str, Any]:

Edit a pod's template configuration.

Updates the template associated with a pod by merging the provided keyword arguments with the existing template settings.

Arguments:

  • pod_id: The unique identifier of the pod whose template to edit.
  • **kwargs: Template fields to update. Common fields include:
    • docker_image (str): Docker image repository.
    • docker_image_tag (str): Docker image tag.
    • startup_commands (str): Commands to run on container start.
    • internal_ports (List[int]): Ports to expose.
    • environment (Dict[str, str]): Environment variables.
    • volumes (List[str]): Volume mount paths.

Returns: Updated template data dictionary from the API.

Example:

lium.edit(pod_id, startup_commands="python main.py", environment={"DEBUG": "1"})

ls​

def ls(
    *,
    gpu_type: Optional[str] = None,
    gpu_count: Optional[int] = None,
    lat: Optional[float] = None,
    lon: Optional[float] = None,
    max_distance_miles: Optional[int] = None,
    min_cuda_version: Optional[float] = None
) -> List[lium.sdk.models.ExecutorInfo]:

List available nodes.

Arguments:

  • gpu_type: Optional GPU filter such as "A100" or "H200".
  • gpu_count: Exact GPU count to match (defaults to 8, pass None to disable).
  • lat: Optional latitude for geospatial filtering. Must be used together with lon and max_distance_miles.
  • lon: Optional longitude for geospatial filtering. Must be used together with lat and max_distance_miles.
  • max_distance_miles: Optional radius (in miles) for geospatial filtering. Must be used together with lat and lon.
  • min_cuda_version: Optional minimum CUDA version to require (e.g. 12.4). Nodes whose max_cuda_version is None or below this threshold are excluded. NVIDIA drivers are backward compatible, so a node with a higher driver CUDA version satisfies the requirement.

Returns: A list of ExecutorInfo objects that satisfy the filters.

ps​

def ps() -> List[lium.sdk.models.PodInfo]:

List active pods.

Returns: List of PodInfo objects representing the caller's running pods.

down​

def down(pod: lium.sdk.models.PodInfo) -> Dict[str, Any]:

Stop a pod.

Arguments:

  • pod: Pod to terminate.

Returns: API response payload from the delete call.

rm​

def rm(pod: lium.sdk.models.PodInfo) -> Dict[str, Any]:

Remove pod (alias for down()).

Arguments:

  • pod: Pod to terminate.

Returns: API response payload from the delete call.

reboot​

def reboot(
    pod: lium.sdk.models.PodInfo,
    volume_id: Optional[str] = None
) -> Dict[str, Any]:

Reboot a pod.

Arguments:

  • pod: Pod to reboot.
  • volume_id: Optional volume ID to attach for the reboot request.

Returns: Pod data from the API response after issuing the reboot.

get_default_images​

def get_default_images(gpu_model: Optional[str], driver_version: Optional[str]) -> list[dict]:

Get default images for GPU type and driver version.

default_docker_template​

def default_docker_template(executor_id: str) -> lium.sdk.models.Template:

Resolve the best default template for a node ID.

Arguments:

  • executor_id: Node identifier returned by ls().

Returns: Template best suited for the node.

Raises:

  • ValueError: If no matching node or template exists.

templates​

def templates(
    filter: Optional[str] = None,
    only_my: bool = False
) -> List[lium.sdk.models.Template]:

List available templates.

Arguments:

  • filter: Optional substring to filter by image or name.
  • only_my: When True return only templates owned by the caller.

Returns: List of Template.

get_executor​

def get_executor(executor: str) -> Optional[lium.sdk.models.ExecutorInfo]:

Resolve a node by ID.

Arguments:

  • executor: Node ID string.

Returns: Matching ExecutorInfo or None if not found.

gpu_types​

def gpu_types() -> set[str]:

Get list of available GPU types.

Returns: Set of GPU type strings advertised by the API.

get_template​

def get_template(template_id: str) -> Optional[lium.sdk.models.Template]:

Fetch a template by ID/HUID/name.

Arguments:

  • template_id: Template ID, HUID, or name to match.

Returns: Matching Template or None if not found.

get_template_by_image_name​

def get_template_by_image_name(
    image_name: Optional[str] = None,
    image_tag: Optional[str] = None
) -> Optional[lium.sdk.models.Template]:

Fetch a template by its Docker image + tag.

Arguments:

  • image_name: Repository/image name.
  • image_tag: Tag to match.

Returns: Matching Template or None if not found.

ssh_connection​

def ssh_connection(pod: lium.sdk.models.PodInfo, timeout: int = 30):

SSH connection context manager.

Arguments:

  • pod: Pod whose SSH metadata is used.
  • timeout: Connection timeout in seconds.

Yields: An active paramiko.SSHClient.

exec​

def exec(
    pod: lium.sdk.models.PodInfo,
    *,
    command: str,
    env: Optional[Dict[str, str]] = None
) -> Dict[str, Any]:

Execute a shell command on a pod over SSH.

Arguments:

  • pod: Pod to target.
  • command: Shell command to run remotely.
  • env: Optional environment variables exported before the command runs.

Returns: Dict containing stdout, stderr, exit_code, and success flag.

stream_exec​

def stream_exec(
    pod: lium.sdk.models.PodInfo,
    *,
    command: str,
    env: Optional[Dict[str, str]] = None
) -> Generator[Dict[str, str], NoneType, NoneType]:

Execute a shell command and stream incremental output.

Arguments:

  • pod: Pod to target.
  • command: Shell command to run remotely.
  • env: Optional environment variables exported before the command runs.

Yields: Streaming output chunks as \{"type": "stdout"|"stderr", "data": str\}.

exec_all​

def exec_all(
    pods: List[lium.sdk.models.PodInfo],
    *,
    command: str,
    env: Optional[Dict[str, str]] = None,
    max_workers: int = 10
) -> List[Dict]:

Execute a shell command on multiple pods in parallel.

Arguments:

  • pods: List of pods to target.
  • command: Shell command to run on each pod.
  • env: Optional environment variables exported before each command.
  • max_workers: Maximum number of SSH workers to spawn.

Returns: List of result dictionaries mirroring exec().

wait_ready​

def wait_ready(
    pod: Union[str, lium.sdk.models.PodInfo, Dict],
    *,
    timeout: int = 300,
    poll_interval: int = 10
) -> Optional[lium.sdk.models.PodInfo]:

Poll until a pod reports RUNNING + SSH metadata.

Arguments:

  • pod: Pod identifier, PodInfo, or dict with an id field.
  • timeout: Maximum number of seconds to wait.
  • poll_interval: Interval between successive ps calls.

Returns: PodInfo when the pod is ready, otherwise None if timeout expires.

scp​

def scp(pod: lium.sdk.models.PodInfo, *, local: str, remote: str) -> None:

Upload a local file to a pod via SFTP.

download​

def download(pod: lium.sdk.models.PodInfo, *, remote: str, local: str) -> None:

Download a file from a pod via SFTP.

Arguments:

  • pod: The pod to download from.
  • remote: Remote file path on the pod.
  • local: Local destination path.

Raises:

  • ValueError: If SSH is not configured for the pod.

upload​

def upload(pod: lium.sdk.models.PodInfo, *, local: str, remote: str) -> None:

Upload a file to a pod via SFTP.

This is an alias for scp() for parity with the CLI.

Arguments:

  • pod: The pod to upload to.
  • local: Local file path to upload.
  • remote: Remote destination path on the pod.

Raises:

  • ValueError: If SSH is not configured for the pod.

ssh​

def ssh(pod: lium.sdk.models.PodInfo) -> str:

Get SSH command string for connecting to a pod.

Arguments:

  • pod: The pod to generate SSH command for.

Returns: SSH command string with the configured SSH key path.

Raises:

  • ValueError: If SSH is not configured for the pod or no SSH key path is set.

rsync​

def rsync(pod: lium.sdk.models.PodInfo, *, local: str, remote: str) -> None:

Sync directories with rsync.

Arguments:

  • pod: Pod to sync.
  • local: Local path or directory (rsync source).
  • remote: Remote path on the pod.

Raises:

  • RuntimeError: If the rsync command fails.

switch_template​

def switch_template(
    pod: lium.sdk.models.PodInfo,
    *,
    template_id: str
) -> lium.sdk.models.PodInfo:

Switch the template of a running pod.

Arguments:

  • pod: Pod to update.
  • template_id: ID of the template to switch to.

Returns: PodInfo object with updated pod information.

create_template​

def create_template(
    name: str,
    docker_image: str,
    docker_image_digest: str = '',
    docker_image_tag: str = 'latest',
    ports: Optional[List[int]] = None,
    start_command: Optional[str] = None,
    **kwargs
) -> lium.sdk.models.Template:

Create a new template.

Arguments:

  • name: Friendly template name.
  • docker_image: Image repository (e.g., "daturaai/pytorch").
  • docker_image_digest: Digest string for pinning (defaults to empty string).
  • docker_image_tag: Image tag (defaults to "latest").
  • ports: Internal ports to expose (defaults to [22, 8000]).
  • start_command: Optional command executed on container start.
  • **kwargs: Additional template fields:
    • category (str): Template category (defaults to "UBUNTU").
    • is_private (bool): Whether template is private (defaults to True).
    • volumes (List[str]): Volume mount paths (defaults to ["/workspace"]).
    • description (str): Template description.
    • environment (Dict[str, str]): Environment variables.
    • entrypoint (str): Container entrypoint.

Returns: Newly created Template.

wait_template_ready​

def wait_template_ready(
    template_id: str,
    timeout: int = 300
) -> Optional[lium.sdk.models.Template]:

Wait for template verification to complete.

Arguments:

  • template_id: Template identifier.
  • timeout: Maximum seconds to wait.

Returns: Template when verification succeeds, otherwise None if the timeout expires.

Raises:

  • LiumError: If template verification fails.

get_my_user_id​

def get_my_user_id() -> str:

Get the current user's ID.

Returns: The ID returned by /users/me.

update_template​

def update_template(
    template_id: str,
    name: str,
    docker_image: str,
    docker_image_digest: str,
    docker_image_tag: str = 'latest',
    ports: Optional[List[int]] = None,
    start_command: Optional[str] = None,
    **kwargs
) -> lium.sdk.models.Template:

Update an existing template owned by the caller.

Arguments:

  • template_id: Template identifier.
  • name: Friendly name.
  • docker_image: Image repository.
  • docker_image_digest: Optional digest.
  • docker_image_tag: Image tag.
  • ports: Internal ports to expose.
  • start_command: Startup command.
  • **kwargs: Additional override fields.

Returns: Updated Template.

Raises:

  • ValueError: If the template is missing or not owned by the caller.

wallets​

def wallets() -> List[Dict[str, Any]]:

Get the caller's configured funding wallets.

Returns: Raw wallet records returned by the pay API.

add_wallet​

def add_wallet(bt_wallet: Any) -> None:

Link a Bittensor wallet with the user account.

Arguments:

  • bt_wallet: Wallet object exposing coldkey/coldkeypub for signing.

Raises:

  • LiumError: If verification or wallet polling fails.

backup_create​

def backup_create(
    pod: lium.sdk.models.PodInfo,
    *,
    path: str = '/home',
    frequency_hours: int = 6,
    retention_days: int = 7
) -> lium.sdk.models.BackupConfig:

Create or replace a backup configuration for a pod.

Arguments:

  • pod: Pod to configure.
  • path: Filesystem path to back up.
  • frequency_hours: Backup interval in hours.
  • retention_days: Retention period in days.

Returns: Created BackupConfig.

backup_now​

def backup_now(
    pod: lium.sdk.models.PodInfo,
    *,
    name: str,
    description: str = ''
) -> Dict[str, Any]:

Trigger an immediate backup for a pod.

Arguments:

  • pod: Pod to back up.
  • name: Backup name.
  • description: Optional description.

Returns: API response payload from the run-now endpoint.

backup_config​

def backup_config(pod: lium.sdk.models.PodInfo) -> Optional[lium.sdk.models.BackupConfig]:

Return the backup configuration for a pod if one exists.

Arguments:

  • pod: Pod to inspect.

Returns: BackupConfig if present, otherwise None.

backup_list​

def backup_list() -> List[lium.sdk.models.BackupConfig]:

List all backup configurations across all pods.

Returns: List of BackupConfig.

backup_logs​

def backup_logs(pod: lium.sdk.models.PodInfo) -> List[lium.sdk.models.BackupLog]:

Get recent backup logs for a pod.

Arguments:

  • pod: Pod to inspect.

Returns: List of BackupLog entries (possibly empty).

backup_delete​

def backup_delete(config_id: str) -> Dict[str, Any]:

Delete a backup configuration by ID.

Arguments:

  • config_id: Backup configuration identifier.

Returns: API response payload.

restore​

def restore(
    pod: lium.sdk.models.PodInfo,
    *,
    backup_id: str,
    restore_path: str = '/root'
) -> Dict[str, Any]:

Restore a backup to a pod.

Arguments:

  • pod: Pod to restore to.
  • backup_id: ID of the backup to restore.
  • restore_path: Path where to restore the backup (default: /root).

Returns: Response from the restore API.

restore_logs​

def restore_logs(pod: lium.sdk.models.PodInfo) -> List[lium.sdk.models.RestoreLog]:

Get recent restore logs for a pod.

Arguments:

  • pod: Pod to inspect.

Returns: List of RestoreLog entries (possibly empty).

get_deployment_estimate​

def get_deployment_estimate(executor_id: str, template_id: str) -> dict:

Estimate deployment time for a template on a node.

Arguments:

  • executor_id: Node UUID.
  • template_id: Template UUID.

Returns: Dict with estimated_seconds, is_slow_machine, warning_message, is_cached_template, and docker_image_size (image size in bytes, or None if unknown).

balance​

def balance() -> float:

Get current account balance.

Returns: Floating-point balance value reported by /users/me.

volumes​

def volumes() -> List[lium.sdk.models.VolumeInfo]:

List all volumes for the current user.

Returns: List of VolumeInfo.

volume​

def volume(volume_id: str) -> lium.sdk.models.VolumeInfo:

Get a specific volume by ID.

Arguments:

  • volume_id: Volume identifier.

Returns: VolumeInfo for the requested volume.

volume_create​

def volume_create(name: str, *, description: str = '') -> lium.sdk.models.VolumeInfo:

Create a new volume.

Arguments:

  • name: Volume name.
  • description: Optional description.

Returns: Created VolumeInfo.

volume_update​

def volume_update(
    volume_id: str,
    *,
    name: Optional[str] = None,
    description: Optional[str] = None
) -> lium.sdk.models.VolumeInfo:

Update a volume's metadata.

Arguments:

  • volume_id: Volume identifier.
  • name: Optional new name.
  • description: Optional description.

Returns: Updated VolumeInfo.

Raises:

  • ValueError: If neither name nor description is provided.

volume_delete​

def volume_delete(volume_id: str) -> Dict[str, Any]:

Delete a volume.

Arguments:

  • volume_id: Volume identifier.

Returns: API response payload from the delete request.

schedule_termination​

def schedule_termination(pod: lium.sdk.models.PodInfo, *, termination_time: str) -> Dict[str, Any]:

Schedule a pod for automatic termination at a future date and time.

Arguments:

  • pod: Pod to schedule
  • termination_time: ISO 8601 formatted datetime string (e.g., "2025-10-17T15:30:00Z")

Returns: Response from the schedule termination API

cancel_scheduled_termination​

def cancel_scheduled_termination(pod: lium.sdk.models.PodInfo) -> Dict[str, Any]:

Cancel a scheduled termination for a pod.

Arguments:

  • pod: Pod to cancel the schedule for

Returns: Response from the cancel scheduled termination API

install_jupyter​

def install_jupyter(
    pod: lium.sdk.models.PodInfo,
    *,
    jupyter_internal_port: int
) -> Dict[str, Any]:

Install Jupyter Notebook on a pod.

Arguments:

  • pod: Pod to install Jupyter on
  • jupyter_internal_port: Internal port for Jupyter Notebook

Returns: Response from the install Jupyter API