cloudmesh-ai-vm API Reference

This page provides the API reference for the cloudmesh-ai-vm library.

API Documentation

Attributes

COMMANDS_DIR `module-attribute`

COMMANDS_DIR = os.path.dirname(__file__)

console `module-attribute`

console = Console()

Classes

VMContext `dataclass`

Source code in src/cloudmesh/ai/command/_shared/context.py

@dataclass
class VMContext:
    cloud_override: str | None = None
    interactive: bool = False

Functions:

cloud_callback

cloud_callback(ctx, param, value)

Source code in src/cloudmesh/ai/command/_shared/context.py

def cloud_callback(ctx: click.Context, param, value):
    if value:
        if ctx.obj is None:
            ctx.obj = VMContext()
        ctx.obj.cloud_override = value
    return value

cmx

cmx()

Cloudmesh AI CLI

Source code in src/cloudmesh/ai/command/__init__.py

@click.group()
def cmx() -> None:
    """Cloudmesh AI CLI"""

debug_callback

debug_callback(ctx, param, value)

Source code in src/cloudmesh/ai/command/_shared/context.py

def debug_callback(ctx: click.Context, param, value):
    if value:
        logging.getLogger("cloudmesh.ai.vm").setLevel(logging.DEBUG)
    return value

load_commands_recursively

load_commands_recursively(group, current_dir, package_path)

Recursively scans the directory for .py files and folders, adding them as commands or subgroups to the provided group.

Source code in src/cloudmesh/ai/command/_shared/utils.py

def load_commands_recursively(group: click.Group, current_dir: str, package_path: str):
    """
    Recursively scans the directory for .py files and folders,
    adding them as commands or subgroups to the provided group.
    """
    for entry in os.listdir(current_dir):
        # Ignore internal folders, files, and __init__.py
        if entry.startswith("_") or entry == "__init__.py":
            continue

        full_path = os.path.join(current_dir, entry)

        if os.path.isdir(full_path):
            # 1. Handle Sub-groups (folders)
            group_name = entry
            module_path = f"{package_path}.{group_name}"
            try:
                mod = importlib.import_module(module_path)
                group_obj = getattr(mod, "cmd", None)

                if not group_obj or not isinstance(group_obj, click.Group):
                    # Fallback: Create a generic group if 'cmd' is missing or not a Group
                    group_obj = click.Group(name=group_name)

                group.add_command(group_obj, name=group_name.replace("_", "-"))

                # Recurse into the folder
                load_commands_recursively(group_obj, full_path, module_path)

            except Exception as e:
                logging.error(
                    f"Failed to load group {group_name} from {module_path}: {e}"
                )

        elif entry.endswith(".py"):
            # 2. Handle Commands (files)
            cmd_name = entry[:-3]
            module_path = f"{package_path}.{cmd_name}"
            try:
                mod = importlib.import_module(module_path)
                cmd_obj = getattr(mod, "cmd", None)

                if cmd_obj:
                    group.add_command(cmd_obj, name=cmd_name.replace("_", "-"))
                else:
                    logging.warning(
                        f"Module {module_path} does not export a 'cmd' object."
                    )

            except Exception as e:
                logging.error(
                    f"Failed to load command {cmd_name} from {module_path}: {e}"
                )

register_providers

register_providers()

Registers all VM providers using entry points. This ensures that the provider map is populated before commands run.

Source code in src/cloudmesh/ai/command/_shared/providers_utils.py

def register_providers():
    """
    Registers all VM providers using entry points.
    This ensures that the provider map is populated before commands run.
    """
    try:
        from cloudmesh.ai.vm.providers import register_providers as _reg

        _reg()
    except Exception as e:
        logging.error(f"Failed to register providers: {e}")

verbose_callback

verbose_callback(ctx, param, value)

Source code in src/cloudmesh/ai/command/_shared/context.py

def verbose_callback(ctx: click.Context, param, value):
    if value:
        pass
    return value

vm_group

vm_group(ctx, interactive=False)

VM management commands

Source code in src/cloudmesh/ai/command/vm/__init__.py

@click.group()
@click.option(
    "--cloud",
    callback=cloud_callback,
    expose_value=False,
    help="Override the default cloud provider",
)
@click.option(
    "--debug",
    is_flag=True,
    callback=debug_callback,
    expose_value=False,
    help="Enable debug logging",
)
@click.option(
    "--verbose",
    is_flag=True,
    callback=verbose_callback,
    expose_value=False,
    help="Print raw subprocess/SSH commands",
)
@click.option("-i", "--interactive", is_flag=True, help="Enter interactive mode")
@click.pass_context
def vm_group(ctx: click.Context, interactive: bool = False) -> None:
    """VM management commands"""
    ctx.ensure_object(VMContext)

    os_cloud = os.environ.get("OS_CLOUD")
    active_cloud = ctx.obj.cloud_override or os_cloud

    # We only enforce OS_CLOUD for commands that actually need a target cloud.
    # Commands like 'provider list' should be allowed without it.
    # Since we are in the group callback, we'll check if the command being called requires it.
    # For now, we'll allow it to be None and let the specific command handle the error if needed.
    if active_cloud:
        console.print(f"Running on cloud: [bold blue]{active_cloud}[/bold blue]")

    if interactive:
        console.print(
            "[bold green]Entering interactive VM shell. Type 'exit' or 'quit' to leave.[/bold green]"
        )

        while True:
            try:
                prompt = f"[bold blue]({active_cloud}) vm>[bold blue] "

                user_input = console.input(prompt)
                if not user_input.strip():
                    continue
                if user_input.strip().lower() in ("exit", "quit"):
                    break

                args = shlex.split(user_input)
                cmd_name = args[0]
                cmd_args = args[1:]

                # Use the group's get_command to find the command
                cmd = vm_group.get_command(ctx, cmd_name)
                if cmd:
                    ctx.invoke(cmd, *cmd_args)
                else:
                    console.print(
                        f"[red]Unknown command: {cmd_name}[/red]. Type 'help' for available commands."
                    )
            except KeyboardInterrupt:
                console.print("\\n[yellow]Use 'exit' to leave the shell.[/yellow]")
            except Exception as e:
                console.print(f"[bold red]Shell Error:[/bold red] {e!s}")
    else:
        pass

Modules

account

Functions:

cmd

cmd(ctx)

List account information and resource limits for the current VM provider.

Source code in src/cloudmesh/ai/command/vm/account.py

@click.command()
@click.pass_context
def cmd(ctx: click.Context):
    """
    List account information and resource limits for the current VM provider.
    """
    # Get the active provider using the helper function
    try:
        provider = get_active_provider(ctx)
    except click.ClickException as e:
        console.print(f"[red]{e!s}[/red]")
        return

    console.print(
        f"[bold blue]Fetching account information for provider: {provider.cloud_name}...[/bold blue]"
    )

    # Call the provider's account info method
    info = provider.get_account_info()

    if not info:
        console.print(
            "[yellow]No account information available for this provider.[/yellow]"
        )
        return

    if "error" in info:
        console.print(f"[red]Error: {info['error']}[/red]")
        return

    # 1. Print Basic Account Information
    console.print("\\n[bold magenta]Account Details:[/bold magenta]")

    # Extract basic fields common to most providers
    basic_fields = {
        "Project ID": info.get("project_id"),
        "Project Name": info.get("project_name"),
        "Site/Domain": info.get("site") or info.get("domain_id"),
        "User ID": info.get("user_id"),
        "Allocation": info.get("allocation"),
    }

    for label, value in basic_fields.items():
        if value:
            console.print(f"[bold]{label}:[/bold] {value}")

    # 2. Print Quotas/Limits if available (common in OpenStack)
    quotas = info.get("quotas")
    if quotas:
        console.print("\\n[bold magenta]Resource Quotas/Limits:[/bold magenta]")

        # Prepare data for render_table
        # Quotas can be a dict of {resource: value} or a more complex structure
        rows = []
        if isinstance(quotas, dict):
            for resource, value in quotas.items():
                # Some providers might return a dict for value (e.g., {'limit': 10, 'used': 2})
                if isinstance(value, dict):
                    # Try to find 'limit' or 'used' in the value dict
                    limit = value.get("limit", "N/A")
                    used = value.get("used", "N/A")
                    rows.append([resource, limit, used])
                else:
                    rows.append([resource, value, "N/A"])

        if rows:
            render_table(
                title="VM Resource Limits",
                columns=["Resource", "Limit", "Used"],
                rows=rows,
                alignments=["left", "right", "right"],
            )
        else:
            console.print("[yellow]No detailed quota information available.[/yellow]")
    else:
        # If no quotas, but we have basic info, just finish
        pass

    console.print("\\n")

cloud

config

Functions:

config_group

config_group()

Manage VM CLI configuration.

Source code in src/cloudmesh/ai/command/vm/config/__init__.py

@click.group()
def config_group():
    """Manage VM CLI configuration."""

Modules

get

Functions:

get_config

get_config(ctx, key)

Get a specific configuration value.

Source code in src/cloudmesh/ai/command/vm/config/get.py

@click.command()
@click.argument("key")
@handle_errors
def get_config(ctx: click.Context, key: str):
    """Get a specific configuration value."""
    val = state.config.get(key)
    if val is not None:
        console.print(f"{key}: [bold green]{val}[/bold green]")
    else:
        console.print(f"Configuration key [red]{key}[/red] not found.")

init

Functions:

init_config

init_config(ctx)

Initialize the VM CLI configuration.

Source code in src/cloudmesh/ai/command/vm/config/init.py

@click.command()
@handle_errors
def init_config(ctx: click.Context):
    """Initialize the VM CLI configuration."""
    # Implementation would go here to create the config file
    console.print("VM configuration initialized.")

list

Functions:

list_config

list_config(ctx)

List all VM configuration settings.

Source code in src/cloudmesh/ai/command/vm/config/list.py

@click.command()
@handle_errors
def list_config(ctx: click.Context):
    """List all VM configuration settings."""
    config = state.config.to_dict()
    render_table(
        "VM Configuration", ["Key", "Value"], [[k, v] for k, v in config.items()]
    )

set

Functions:

parse_value

parse_value(value)

Attempt to parse value as boolean, integer, or float; fallback to string.

Source code in src/cloudmesh/ai/command/vm/config/set.py

def parse_value(value: str):
    """Attempt to parse value as boolean, integer, or float; fallback to string."""
    try:
        return ast.literal_eval(value)
    except (ValueError, SyntaxError):
        return value

set_config

set_config(ctx, key, value)

Set a configuration value.

Source code in src/cloudmesh/ai/command/vm/config/set.py

@click.command()
@click.argument("key")
@click.argument("value")
@handle_errors
def set_config(ctx: click.Context, key: str, value: str):
    """Set a configuration value."""
    parsed_val = parse_value(value)
    state.config.set(key, parsed_val)
    console.print(
        f"Configured [bold green]{key}[/bold green] = {parsed_val} ({type(parsed_val).__name__})."
    )

delete

Classes

Functions:

delete

delete(ctx, name=None, count=None, vm_range=None, **kwargs)

Deletes one or more VMs.

If a name is provided, it can be a single name or a hostlist specification (e.g. 'node[1-3]'). If --count is provided, it deletes the last N VMs created. If --range is provided, it deletes VMs in the specified index range (e.g. '1-5' -> username-1...username-5).

Example

cmx vm delete my-vm cmx vm delete "node[1-3]" cmx vm delete --count 3 cmx vm delete --range 1-5 cmx vm delete

Source code in src/cloudmesh/ai/command/vm/delete.py

@click.command()
@click.pass_context
@click.argument("name", required=False)
@click.option(
    "--count", type=int, help="Delete the last N VMs (sorted by timestamp if available)"
)
@click.option("--range", "vm_range", help="Delete VMs in a range (e.g. 1-5)")
@vm_options
@handle_errors
def delete(
    ctx: click.Context,
    name: str | None = None,
    count: int | None = None,
    vm_range: str | None = None,
    **kwargs,
) -> None:
    """
    Deletes one or more VMs.

    If a name is provided, it can be a single name or a hostlist specification (e.g. 'node[1-3]').
    If --count is provided, it deletes the last N VMs created.
    If --range is provided, it deletes VMs in the specified index range (e.g. '1-5' -> username-1...username-5).

    Example:
        cmx vm delete my-vm
        cmx vm delete "node[1-3]"
        cmx vm delete --count 3
        cmx vm delete --range 1-5
        cmx vm delete
    """
    provider = get_active_provider(ctx)

    # Resolve the list of VM names to delete
    vms_to_delete = resolve_vms(ctx, name=name, count=count, vm_range=vm_range)

    # Execute deletion
    deleted_count = 0
    for vm_name in vms_to_delete:
        if provider.delete(vm_name):
            console.print(
                f"Successfully deleted VM [bold green]{vm_name}[/bold green]."
            )
            deleted_count += 1
        else:
            console.print(f"[red]Failed to delete VM {vm_name}.[/red]")

    if deleted_count == 0:
        raise VMCommandError("Failed to delete any of the requested VMs.")

doctor

Functions:

doctor

doctor(ctx, strict=False, config=False, **kwargs)

Diagnose the active VM provider without modifying any VM resources.

Checks provider resolution, local requirements, configuration, and provider information.

Source code in src/cloudmesh/ai/command/vm/doctor.py

@click.command()
@click.option(
    "--strict",
    is_flag=True,
    help="Exit with a non-zero status when a doctor check fails.",
)
@click.option(
    "--config",
    is_flag=True,
    help="Run exhaustive configuration validation (replicates smoke test).",
)
@click.pass_context
@vm_options
def doctor(
    ctx: click.Context,
    strict: bool = False,
    config: bool = False,
    **kwargs,
) -> None:
    """
    Diagnose the active VM provider without modifying any VM resources.

    Checks provider resolution, local requirements, configuration, and
    provider information.
    """
    rows = []
    failures = 0
    warnings = 0

    # 1. Resolve provider
    provider = get_active_provider(ctx)
    cloud_name = getattr(provider, "cloud_name", "unknown")

    rows.append(
        [
            "Provider",
            _status_label("PASS"),
            f"Resolved provider: {cloud_name}",
        ]
    )

    # 2. Check local provider requirements
    requirements_ok = None

    if hasattr(provider, "check_requirements"):
        try:
            requirements_ok = provider.check_requirements()

            if requirements_ok:
                rows.append(
                    [
                        "Requirements",
                        _status_label("PASS"),
                        "Required provider tools are available.",
                    ]
                )
            else:
                failures += 1
                rows.append(
                    [
                        "Requirements",
                        _status_label("FAIL"),
                        "One or more required provider tools are unavailable.",
                    ]
                )
        except Exception as exc:
            failures += 1
            rows.append(
                [
                    "Requirements",
                    _status_label("FAIL"),
                    f"{type(exc).__name__}: {exc}",
                ]
            )
    else:
        warnings += 1
        rows.append(
            [
                "Requirements",
                _status_label("SKIP"),
                "Provider does not implement check_requirements().",
            ]
        )

    # 3. Validate provider configuration
    if hasattr(provider, "validate_config"):
        try:
            validation = provider.validate_config()
            status, details = _format_validation_result(validation)

            if status == "FAIL":
                failures += 1
            elif status == "WARN":
                warnings += 1

            rows.append(
                [
                    "Configuration",
                    _status_label(status),
                    details,
                ]
            )
        except Exception as exc:
            failures += 1
            rows.append(
                [
                    "Configuration",
                    _status_label("FAIL"),
                    f"{type(exc).__name__}: {exc}",
                ]
            )
    else:
        warnings += 1
        rows.append(
            [
                "Configuration",
                _status_label("SKIP"),
                "Provider does not implement validate_config().",
            ]
        )

    # 4. Retrieve provider information.
    # Skip runtime provider queries if required local tools are unavailable.
    if requirements_ok is False:
        warnings += 1
        rows.append(
            [
                "Provider Info",
                _status_label("SKIP"),
                "Skipped because provider requirements are unavailable.",
            ]
        )

    elif hasattr(provider, "get_provider_info"):
        try:
            info = provider.get_provider_info()

            if info:
                rows.append(
                    [
                        "Provider Info",
                        _status_label("PASS"),
                        "Provider information retrieved successfully.",
                    ]
                )
            else:
                warnings += 1
                rows.append(
                    [
                        "Provider Info",
                        _status_label("WARN"),
                        "Provider returned no diagnostic information.",
                    ]
                )

        except Exception as exc:
            failures += 1
            rows.append(
                [
                    "Provider Info",
                    _status_label("FAIL"),
                    f"{type(exc).__name__}: {exc}",
                ]
            )

    else:
        warnings += 1
        rows.append(
            [
                "Provider Info",
                _status_label("SKIP"),
                "Provider does not implement get_provider_info().",
            ]
        )

    # -------------------------------------------------------------------
    # Optional exhaustive configuration validation (full_config flag).
    # -------------------------------------------------------------------
    if config:
        all_checks = run_all_config_checks()
        for section, checks in all_checks.items():
            for check_name, (status, details) in checks.items():
                # Row label combines the section and the individual check name
                row_label = f"{section}: {check_name}"
                rows.append([row_label, _status_label(status), details])
                if status == "FAIL":
                    failures += 1
                elif status == "WARN":
                    warnings += 1

    render_table(
        f"VM Doctor: {cloud_name}",
        ["Check", "Status", "Details"],
        rows,
    )

    if failures:
        console.print(
            f"\n[bold red]Doctor result: NEEDS ATTENTION[/bold red] "
            f"({failures} failed, {warnings} warning(s))"
        )
    elif warnings:
        console.print(
            f"\n[bold yellow]Doctor result: HEALTHY WITH WARNINGS[/bold yellow] "
            f"({warnings} warning(s))"
        )
    else:
        console.print(
            "\n[bold green]Doctor result: HEALTHY[/bold green]"
        )

    if strict and failures:
        ctx.exit(1)

flavor

Functions:

flavor

flavor(ctx, **kwargs)

List available VM flavor.

Source code in src/cloudmesh/ai/command/vm/flavor.py

@click.command()
@click.pass_context
@vm_options
@handle_errors
def flavor(ctx: click.Context, **kwargs):
    """List available VM flavor."""
    provider = get_active_provider(ctx)
    flv = provider.get_flavors()  # Adjusted to use get_flavors()
    if flv:
        render_table(
            "Flavors",
            ["Name", "CPU", "RAM"],
            [
                [
                    f.get("name", "Unknown"),
                    f.get("cpu", "Unknown"),
                    f.get("ram", "Unknown"),
                ]
                for f in flv
            ],
        )
    else:
        console.print("No flavors found.")

horizon

Functions:

cmd

cmd(ctx, site=None, cloud=None, debug=False, verbose=False, **kwargs)

Open the Horizon dashboard for the active cloud provider.

Source code in src/cloudmesh/ai/command/vm/horizon.py

@click.command()
@vm_options
@click.option(
    "--site",
    type=click.Choice(list(URL_MAP.keys()), case_sensitive=False),
    help="Specify the cloud site",
)
@click.pass_context
def cmd(
    ctx: click.Context,
    site: str = None,
    cloud: str = None,
    debug: bool = False,
    verbose: bool = False,
    **kwargs,
) -> None:
    """Open the Horizon dashboard for the active cloud provider."""
    from cloudmesh.ai.command._shared.context import state

    # 1. Determine which site to use
    target_site = site

    if not target_site:
        # Try to get the active provider from context
        active_provider = (
            ctx.obj.cloud_override
            or getattr(state.config, "default_cloud", "multipass")
            or "multipass"
        )

        if active_provider == "chameleon":
            # Look up the site in the chameleon config
            chameleon_config = getattr(state.config, "chameleon", {})
            target_site = getattr(chameleon_config, "site", None)
        elif active_provider == "jetstream":
            target_site = "Jetstream"

    # 2. Resolve the URL
    if target_site:
        # Match the input (potentially lowercased by click.Choice) to the keys in URL_MAP
        lookup_key = next(
            (k for k in URL_MAP if k.lower() == target_site.lower()), None
        )
        if lookup_key:
            url = URL_MAP[lookup_key]
            click.echo(f"Opening Horizon for {lookup_key}...")
            webbrowser.open(url)
            return

    click.echo("Could not determine the target cloud site.")
    click.echo("Available sites:")
    for s in URL_MAP:
        click.echo(f"  - {s}")
    click.echo("\nUse 'cmx vm horizon --site <SITE>' to open a specific one.")

image

Functions:

image

image(ctx, **kwargs)

List available VM image.

Source code in src/cloudmesh/ai/command/vm/image.py

@click.command()
@click.pass_context
@vm_options
@handle_errors
def image(ctx: click.Context, **kwargs):
    """List available VM image."""
    provider = get_active_provider(ctx)
    imgs = provider.get_images()  # Adjusted to use get_images() from MultipassManager
    if imgs:
        columns = ["Name", "ID"]
        rows = [[img.get("name", "Unknown"), img.get("id", "Unknown")] for img in imgs]
        render_table("Available VM Images", columns, rows)
    else:
        console.print("No images found.")

info

Classes

Functions:

info

info(ctx, name=None, as_json=False, kind='vm', **kwargs)

Get detailed information about a specific VM.

Source code in src/cloudmesh/ai/command/vm/info.py

@click.command()
@click.pass_context
@click.argument("name", required=False, default=None)
@click.option(
    "--json",
    "as_json",
    is_flag=True,
    default=False,
    help="Print the VM information as JSON (YAML is the default)",
)
@click.option(
    "--kind",
    default="vm",
    help="Type of information to return (e.g., 'vm', 'all', 'list')",
)
@vm_options
@handle_errors
def info(
    ctx: click.Context,
    name: str | None = None,
    as_json: bool = False,  # ← new argument
    kind: str = "vm",
    **kwargs,
) -> None:
    """
    Get detailed information about a specific VM.
    """
    provider = get_active_provider(ctx)

    # 1. Fetch VM information
    if name is None:
        from cloudmesh.ai.command._shared.context import resolve_vm_name

        name = resolve_vm_name(ctx, name)

    vm_info = provider.info(name, kind=kind)

    if not vm_info:
        raise VMCommandError(f"No information found for VM '{name}'.")

    if "error" in vm_info:
        raise VMCommandError(
            f"Could not retrieve info for VM '{name}': {vm_info['error']}"
        )

    # 2. Display the information
    console.print(f"\n[bold blue]Information for VM: {name}[/bold blue]")
    console.print("-" * 30)

    # 2. Choose the output format
    if as_json:
        # JSON – compact but still pretty‑printed
        console.print(json.dumps(vm_info, indent=2, sort_keys=True))
    else:
        # YAML – this is now the *default* format
        # safe_dump keeps the output deterministic and human‑readable
        yaml_str = yaml.safe_dump(vm_info, sort_keys=False, default_flow_style=False)
        # Replace yellow (default for some YAML values) with blue for IDs/UUIDs
        import re

        # Regex to match UUIDs: 8-4-4-4-12 hex chars
        uuid_pattern = r"([a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12})"
        yaml_str = re.sub(uuid_pattern, r"[blue]\1[/blue]", yaml_str)
        console.print(yaml_str)

    # Iterate through the returned dictionary and print key-value pairs
    # for key, value in vm_info.items():
    #    # Format keys to be more human-readable (e.g., "public_ip" -> "Public IP")
    #    display_key = key.replace("_", " ").title()
    #    console.print(f"[bold]{display_key}:[/bold] {value}")

    console.print("-" * 30)

ip

Functions:

ip_group

ip_group()

Manage VM floating IPs.

Source code in src/cloudmesh/ai/command/vm/ip/__init__.py

@click.group()
def ip_group():
    """Manage VM floating IPs."""

Modules

assign

Functions:

assign_ip

assign_ip(ctx, name)

Assign a floating IP to a VM.

Source code in src/cloudmesh/ai/command/vm/ip/assign.py

@click.command()
@click.pass_context
@click.argument("name")
@handle_errors
def assign_ip(ctx: click.Context, name: str):
    """Assign a floating IP to a VM."""
    provider = get_active_provider(ctx)
    ip = provider.assign_floating_ip(name)
    if ip:
        console.print(
            f"Assigned floating IP [bold green]{ip}[/bold green] to VM {name}."
        )
    else:
        raise VMCommandError(f"Failed to assign a floating IP to VM {name}.")

release

Functions:

release_ip

release_ip(ctx, name)

Release the floating IP of a VM.

Source code in src/cloudmesh/ai/command/vm/ip/release.py

@click.command()
@click.pass_context
@click.argument("name")
@handle_errors
def release_ip(ctx: click.Context, name: str):
    """Release the floating IP of a VM."""
    provider = get_active_provider(ctx)
    if provider.release_floating_ip(name):
        console.print(
            f"Released the floating IP of VM [bold green]{name}[/bold green]."
        )
    else:
        raise VMCommandError(f"Failed to release the floating IP of VM {name}.")

key

Functions:

key_group

key_group()

SSH key management commands for the current VM provider.

Source code in src/cloudmesh/ai/command/vm/key/__init__.py

@click.group(name="key")
def key_group():
    """SSH key management commands for the current VM provider."""

Modules

upload

Compatibility shim for the cloudmesh.ai.command.vm.key.upload import.

Historically the vm key upload command lived in cloudmesh.ai.command.key and the test suite expects it to be importable via the legacy path cloudmesh.ai.command.vm.key.upload. The original module defines the Click command upload_key and a cmd alias. To satisfy the test without duplicating logic we re‑implement the command here by copying the implementation from cloudmesh.ai.command.key.upload. This ensures that patching get_active_provider on this module (as performed in the unit tests) works as intended.

Classes

Functions:

upload_key

upload_key(ctx, key_path=None, name=None)

Upload a public SSH key to the cloud provider.

This implementation mirrors the original upload_key command located in cloudmesh.ai.command.key.upload. The logic is kept in sync to avoid divergence between the legacy import path and the current command group.

Source code in src/cloudmesh/ai/command/vm/key/upload.py

@click.command()
@click.pass_context
@click.argument("key_path", required=False)
@click.option("--name", help="Custom name for the key")
@handle_errors
def upload_key(ctx: click.Context, key_path: str = None, name: str = None):
    """Upload a public SSH key to the cloud provider.

    This implementation mirrors the original ``upload_key`` command located in
    ``cloudmesh.ai.command.key.upload``. The logic is kept in sync to avoid
    divergence between the legacy import path and the current command group.
    """
    provider = get_active_provider(ctx)
    config = provider.get_cloud_config(provider.cloud_name) or {}

    # Resolve key_path
    if not key_path:
        # 1. Try config
        key_path = config.get("key_path")

        if not key_path:
            # 2. Try default fallbacks in ~/.ssh/
            ssh_dir = os.path.expanduser("~/.ssh")
            fallbacks = ["id_ed25519.pub", "id_rsa.pub"]
            for fb in fallbacks:
                path = os.path.join(ssh_dir, fb)
                if os.path.exists(path):
                    key_path = path
                    break

            if not key_path:
                # 3. Try any *.pub file
                pub_files = glob.glob(os.path.join(ssh_dir, "*.pub"))
                if len(pub_files) == 1:
                    key_path = pub_files[0]
                elif len(pub_files) > 1:
                    raise VMCommandError(
                        f"Multiple public keys found in {ssh_dir} ({len(pub_files)} files). "
                        "The tool cannot decide which key to use. Please specify 'key_path' "
                        "as an argument or configure it in clouds.yaml."
                    )
                else:
                    raise VMCommandError(
                        f"No public keys found in {ssh_dir}. Please provide a key_path."
                    )

    key_path = os.path.expanduser(key_path)

    # Resolve name
    if not name:
        name = config.get("key_name")

    if provider.upload_key(key_path, name):
        console.print(f"Successfully uploaded key [bold green]{key_path}[/bold green].")
    else:
        raise VMCommandError(f"Failed to upload key {key_path}.")

list

Functions:

list_group

list_group(ctx, all)

List VM resources.

Source code in src/cloudmesh/ai/command/vm/list.py

@click.group(invoke_without_command=True)
@click.pass_context
@click.option("--all", is_flag=True, help="List VMs from all enabled providers")
def list_group(ctx: click.Context, all: bool):
    """List VM resources."""
    if ctx.obj is None:
        from cloudmesh.ai.command._shared.context import VMContext

        ctx.obj = VMContext()
    ctx.obj.list_all = all
    if ctx.invoked_subcommand is None:
        ctx.invoke(list_vms)

list_regions

list_regions(ctx, **kwargs)

List regions.

Source code in src/cloudmesh/ai/command/vm/list.py

@list_group.command(name="regions")
@click.pass_context
@vm_options
@handle_errors
def list_regions(ctx: click.Context, **kwargs):
    """List regions."""
    provider = get_active_provider(ctx)
    # Check if provider has list_regions, otherwise fallback to a generic message
    if hasattr(provider, "list_regions"):
        regions = provider.list_regions()
        if regions:
            render_table("Regions", ["Region Name"], [[r] for r in regions])
        else:
            console.print("No regions found.")
    else:
        console.print("Region listing not supported by this provider.")

list_vms

list_vms(ctx, json=False, **kwargs)

List VMs.

Source code in src/cloudmesh/ai/command/vm/list.py

@list_group.command(name="vms")
@click.pass_context
@click.option("--json", is_flag=True, help="Output in JSON format")
@vm_options
@handle_errors
def list_vms(ctx: click.Context, json: bool = False, **kwargs):
    """List VMs."""
    all_flag = getattr(ctx.obj, "list_all", False)
    if all_flag:
        from cloudmesh.ai.command._shared.context import state
        from cloudmesh.ai.vm.providers import PROVIDER_MAP, get_provider

        config = state.config
        configured_clouds = config.clouds if config else {}

        any_vms = False
        all_vms_data = {}
        for cloud_name in sorted(PROVIDER_MAP.keys()):
            cloud_cfg = configured_clouds.get(cloud_name)
            is_enabled = False
            if cloud_cfg:
                if hasattr(cloud_cfg, "enabled"):
                    is_enabled = cloud_cfg.enabled
                elif isinstance(cloud_cfg, dict):
                    is_enabled = cloud_cfg.get("enabled", False)

            if not is_enabled:
                continue

            try:
                provider = get_provider(cloud_name)
                vms = provider.list()
                if vms:
                    if json:
                        all_vms_data[cloud_name] = vms
                    if _render_vms(cloud_name, vms):
                        any_vms = True
            except Exception as e:
                console.print(f"Could not list VMs for provider {cloud_name}: {e}")

        if json:
            console.print(json.dumps(all_vms_data, indent=4))
            return

        if not any_vms:
            console.print("No VMs found on any enabled providers.")
        return

    provider = get_active_provider(ctx)
    cloud_name = provider.cloud_name
    vms = provider.list()
    if json:
        console.print(json.dumps({cloud_name: vms}, indent=4))
        return

    if not _render_vms(cloud_name, vms):
        console.print(f"No VMs found on {cloud_name}.")

login(ctx, name=None, **kwargs)

Log into a VM or get connection details. If name is omitted, the last started VM is used.

Source code in src/cloudmesh/ai/command/vm/login.py

@click.command()
@click.argument("name", required=False)
@vm_options
@handle_errors
@click.pass_context
def login(ctx: click.Context, name: str | None = None, **kwargs) -> None:
    """
    Log into a VM or get connection details.
    If name is omitted, the last started VM is used.
    """
    provider = get_active_provider(ctx)
    vm_name = resolve_vm_name(ctx, name)

    if not vm_name:
        raise VMCommandError("No VM specified and no last-used VM found in context.")

    if not provider.login(vm_name):
        raise VMCommandError(f"Failed to login to VM {vm_name}.")

provider

Functions:

provider_group

provider_group()

Manage VM cloud provider.

Source code in src/cloudmesh/ai/command/vm/provider/__init__.py

@click.group()
def provider_group():
    """Manage VM cloud provider."""

Modules

get

Functions:

get_provider

get_provider(ctx)

Get the current default cloud provider.

Source code in src/cloudmesh/ai/command/vm/provider/get.py

@click.command(name="get")
@click.pass_context
@handle_errors
def get_provider(ctx: click.Context):
    """Get the current default cloud provider."""
    provider = state.config.db.get("default_cloud", "multipass") or "multipass"
    console.print(f"Current provider: [bold green]{provider}[/bold green].")

info

Functions:

provider_info

provider_info(ctx, **kwargs)

Display detailed information about the active VM provider.

Source code in src/cloudmesh/ai/command/vm/provider/info.py

@click.command()
@click.pass_context
@vm_options
@handle_errors
def provider_info(ctx: click.Context, **kwargs) -> None:
    """Display detailed information about the active VM provider."""
    provider = get_active_provider(ctx)

    try:
        # Fetch provider-specific info
        info = provider.get_provider_info()

        if not info:
            console.print(
                f"No detailed information available for provider: {provider.cloud_name}"
            )
            return

        # Prepare data for table
        rows = [[k, v] for k, v in info.items()]

        table_title = f"Provider Info: {provider.cloud_name}"
        render_table(table_title, ["Key", "Value"], rows)

    except Exception as e:
        # handle_errors usually catches this, but we can be explicit
        raise click.ClickException(str(e))

list

Functions:

list_providers

list_providers(ctx)

List all supported VM providers and their configuration status.

Source code in src/cloudmesh/ai/command/vm/provider/list.py

@click.command(name="list")
@click.pass_context
@handle_errors
def list_providers(ctx: click.Context):
    """List all supported VM providers and their configuration status."""
    from cloudmesh.ai.vm.providers import PROVIDER_METADATA

    # Access the current configuration from the state proxy
    config = state.config
    configured_clouds = config.db.get("clouds", {}) if config else {}
    default_cloud = config.db.get("default_cloud", "multipass") if config else None

    # Prepare rows for the provider support matrix
    rows = []
    # Iterate over the supported providers defined in metadata
    sorted_providers = sorted(PROVIDER_METADATA.keys())

    for p_name in sorted_providers:
        meta = PROVIDER_METADATA.get(p_name, {})

        # Format the provider name for display
        display_name = p_name.capitalize()
        if p_name == "aws":
            display_name = "AWS"
        elif p_name == "azure":
            display_name = "Azure"
        elif p_name == "google":
            display_name = "Google"
        elif p_name == "wsl2":
            display_name = "WSL2"
        elif p_name == "vbox":
            display_name = "VirtualBox"
        elif p_name == "chameleon":
            display_name = "Chameleon Cloud"
        elif p_name == "jetstream":
            display_name = "Jetstream"

        # Check if the provider is the default
        is_default = "⭐" if p_name == default_cloud else ""

        # Check if the provider is defined in clouds.yaml AND has enabled=True
        cloud_cfg = configured_clouds.get(p_name)
        is_enabled_val = False
        if cloud_cfg:
            if hasattr(cloud_cfg, "enabled"):
                is_enabled_val = cloud_cfg.enabled
            elif isinstance(cloud_cfg, dict):
                is_enabled_val = cloud_cfg.get("enabled", False)

        is_enabled = "🟢" if is_enabled_val else "🔴"

        rows.append(
            [
                is_default,
                display_name,
                is_enabled,
                meta.get("lifecycle", "🔴"),
                meta.get("remote_exec", "🔴"),
                meta.get("status", "Unknown"),
            ]
        )

    # Define alignments for the columns:
    # ["", "Provider", "Enabled", "Lifecycle", "Remote Exec", "Status"]
    # We center the status icons but leave descriptions left-aligned.
    alignments = ["center", "left", "center", "center", "left", "left"]

    render_table(
        "Provider Support & Configuration Matrix",
        ["", "Provider", "Enabled", "Lifecycle", "Remote Exec", "Status"],
        rows,
        alignments=alignments,
    )

    if default_cloud:
        console.print(
            f"\n[bold]Default cloud provider:[/bold] [bold green]{default_cloud}[/bold green]"
        )

    console.print(
        "[bold]Cloudmesh Config file path:[/bold] [dim]~/.config/cloudmesh/clouds.yaml[/dim]"
    )
    console.print(
        "[bold]Openstack Config file path:[/bold] [dim]~/.config/openstack/clouds.yaml[/dim]"
    )

set

Functions:

set_provider

set_provider(ctx, provider)

Set the default cloud provider.

Source code in src/cloudmesh/ai/command/vm/provider/set.py

@click.command(name="set")
@click.pass_context
@click.argument("provider")
@handle_errors
def set_provider(ctx: click.Context, provider: str):
    """Set the default cloud provider."""
    state.db.set("default_cloud", provider)
    state.save()
    console.print(f"Default provider set to [bold green]{provider}[/bold green].")

reservation

Functions:

reservation

reservation()

Manage VM reservations (leases) on the cloud provider.

Source code in src/cloudmesh/ai/command/vm/reservation/__init__.py

@click.group()
def reservation():
    """Manage VM reservations (leases) on the cloud provider."""

Modules

duration

Classes

Functions:

duration

duration(ctx, name, node_type, count, start, days)

Creates a reservation with a specified duration and relative start time.

Example

cmx vm reservation duration "my-lease" --node-type "compute_haswell" --count 1 --start +10m

Source code in src/cloudmesh/ai/command/vm/reservation/duration.py

@click.command()
@click.pass_context
@click.argument("name")
@click.option(
    "--node-type",
    required=True,
    help="Type of node to reserve (e.g., 'compute_haswell')",
)
@click.option("--count", type=int, default=1, help="Number of nodes to reserve")
@click.option(
    "--start",
    help="Relative start time offset (e.g., '+10m', '+1h', '+1d'). Defaults to now.",
)
@click.option("--days", type=int, default=1, help="Duration of the reservation in days")
@handle_errors
def duration(
    ctx: click.Context,
    name: str,
    node_type: str,
    count: int,
    start: str | None,
    days: int,
) -> None:
    """
    Creates a reservation with a specified duration and relative start time.

    Example:
        cmx vm reservation duration "my-lease" --node-type "compute_haswell" --count 1 --start +10m
    """
    provider = get_active_provider(ctx)

    # Verify provider supports reservations
    if not hasattr(provider, "create_reservation"):
        raise VMCommandError(
            f"The active provider '{provider.cloud_name}' does not support reservations."
        )

    # Calculate start and end dates
    now = datetime.now()
    if start:
        delta = parse_relative_offset(start)
        start_dt = now + delta
    else:
        start_dt = now

    end_dt = start_dt + timedelta(days=days)

    # Format dates as "YYYY-MM-DD HH:MM" (required by chi/OpenStack)
    start_date_str = start_dt.strftime("%Y-%m-%d %H:%M")
    end_date_str = end_dt.strftime("%Y-%m-%d %H:%M")

    console.print(f"Creating reservation [bold blue]{name}[/bold blue]...")
    console.print(f"Nodes: {count} x {node_type}")
    console.print(f"Start: {start_date_str}")
    console.print(f"End: {end_date_str}")

    if provider.create_reservation(
        name=name,
        node_type=node_type,
        count=count,
        start_date=start_date_str,
        end_date=end_date_str,
    ):
        console.print(
            f"Successfully created reservation [bold green]{name}[/bold green]."
        )
    else:
        raise VMCommandError(f"Failed to create reservation '{name}'.")

parse_relative_offset

parse_relative_offset(offset_str)

Parses a relative time offset string like '+10m', '+2h', '+1d' and returns a timedelta object.

Source code in src/cloudmesh/ai/command/vm/reservation/duration.py

def parse_relative_offset(offset_str: str) -> timedelta:
    """
    Parses a relative time offset string like '+10m', '+2h', '+1d'
    and returns a timedelta object.
    """
    match = re.match(r"^\+(\d+)([mhd])$", offset_str)
    if not match:
        raise click.BadParameter(
            f"Invalid offset format '{offset_str}'. Expected '+<number>[m|h|d]' (e.g., '+10m', '+2h', '+1d')."
        )

    value = int(match.group(1))
    unit = match.group(2)

    if unit == "m":
        return timedelta(minutes=value)
    elif unit == "h":
        return timedelta(hours=value)
    elif unit == "d":
        return timedelta(days=value)

    raise click.BadParameter(
        f"Unsupported time unit '{unit}'. Use 'm' for minutes, 'h' for hours, or 'd' for days."
    )

reset

Classes

Functions:

reset

reset(ctx, name=None, **kwargs)

Resets a VM or the provider service.

Example

cmx vm reset multipass

Source code in src/cloudmesh/ai/command/vm/reset.py

@click.command()
@click.pass_context
@click.argument("name", required=False)
@vm_options
@handle_errors
def reset(ctx: click.Context, name: str | None = None, **kwargs) -> None:
    """
    Resets a VM or the provider service.

    Example:
        cmx vm reset multipass
    """
    vm_name = resolve_vm_name(ctx, name)
    provider = get_active_provider(ctx)
    if provider.reset(vm_name):
        state.set_last_vm(provider.cloud_name, vm_name)
        console.print(f"Successfully reset [bold green]{vm_name}[/bold green].")
    else:
        raise VMCommandError(f"Failed to reset {vm_name}.")

restart

Classes

Functions:

restart

restart(ctx, name=None, count=None, vm_range=None, **kwargs)

Restarts one or more VMs.

If a name is provided, it can be a single name or a hostlist specification (e.g. 'node[1-3]'). If --count is provided, it restarts the last N VMs created. If --range is provided, it restarts VMs in the specified index range (e.g. '1-5' -> username-1...username-5).

Example

cmx vm restart my-vm cmx vm restart "node[1-3]" cmx vm restart --count 3 cmx vm restart --range 1-5 cmx vm restart

Source code in src/cloudmesh/ai/command/vm/restart.py

@click.command()
@click.pass_context
@click.argument("name", required=False)
@click.option("--count", type=int, help="Restart the last N VMs")
@click.option("--range", "vm_range", help="Restart VMs in a range (e.g. 1-5)")
@vm_options
@handle_errors
def restart(
    ctx: click.Context,
    name: str | None = None,
    count: int | None = None,
    vm_range: str | None = None,
    **kwargs,
) -> None:
    """
    Restarts one or more VMs.

    If a name is provided, it can be a single name or a hostlist specification (e.g. 'node[1-3]').
    If --count is provided, it restarts the last N VMs created.
    If --range is provided, it restarts VMs in the specified index range (e.g. '1-5' -> username-1...username-5).

    Example:
        cmx vm restart my-vm
        cmx vm restart "node[1-3]"
        cmx vm restart --count 3
        cmx vm restart --range 1-5
        cmx vm restart
    """
    provider = get_active_provider(ctx)
    vms_to_restart = resolve_vms(ctx, name=name, count=count, vm_range=vm_range)

    restarted_count = 0
    for vm_name in vms_to_restart:
        if provider.restart(vm_name):
            console.print(
                f"Successfully restarted VM [bold green]{vm_name}[/bold green]."
            )
            restarted_count += 1
        else:
            console.print(f"[red]Failed to restart VM {vm_name}.[/red]")

    if restarted_count == 0:
        raise VMCommandError("Failed to restart any of the requested VMs.")

    # Update last VM to the last one successfully restarted
    last_restarted = vms_to_restart[-1]
    state.set_last_vm(provider.cloud_name, last_restarted)

run

Classes

Functions:

run

run(ctx, name=None, command=None, **kwargs)

Executes a command on a VM.

Example

cmx vm run my-vm "ls -la /home" cmx vm run "ls -la /home"

Source code in src/cloudmesh/ai/command/vm/run.py

@click.command()
@click.pass_context
@click.argument("name", required=False)
@click.argument("command", required=False)
@vm_options
@handle_errors
def run(
    ctx: click.Context,
    name: str | None = None,
    command: str | None = None,
    **kwargs,
) -> None:
    """
    Executes a command on a VM.

    Example:
        cmx vm run my-vm "ls -la /home"
        cmx vm run "ls -la /home"
    """
    if command is None:
        actual_command = name
        vm_name = resolve_vm_name(ctx, None)
    else:
        actual_command = command
        vm_name = resolve_vm_name(ctx, name)

    if not actual_command:
        raise VMCommandError("No command provided to run.")

    provider = get_active_provider(ctx)
    result = provider.run_command(vm_name, actual_command)
    if result:
        state.set_last_vm(provider.cloud_name, vm_name)
        console.print(result)
    else:
        raise VMCommandError(f"Failed to execute command on VM {vm_name}.")

security_group

Classes