API Reference

This page provides the technical API reference for the cloudmesh-ai-cmc library, automatically generated from the source code.

Modules

context

CMC Context Initialization.

This module initializes the global state for the Cloudmesh AI CMC application, including configuration loading, logging setup, and telemetry initialization.

Global Objects

config (Config): The application configuration object. logger (Logger): The main application logger. telemetry (Telemetry): The telemetry collection object.

Classes

Functions:

Modules

main

Classes

DelegatingCommand

Bases: Group

A command that delegates execution to another click object.

This is used to handle cases where a command is loaded from a different Click version or instance, preventing version mismatch errors.

Source code in src/cloudmesh/ai/cmc/main.py

class DelegatingCommand(click.Group):
    """A command that delegates execution to another click object.

    This is used to handle cases where a command is loaded from a different 
    Click version or instance, preventing version mismatch errors.
    """
    def __init__(self, name, delegate, **kwargs):
        """Initializes the DelegatingCommand.

        Args:
            name (str): The name of the command.
            delegate (Any): The actual click object or factory to delegate to.
            **kwargs: Additional arguments passed to the click.Group constructor.
        """
        # For a Group, ignore_unknown_options is a valid argument in many click versions
        # If it still fails, we can remove it and try another way.
        super().__init__(name, **kwargs)
        self.delegate = delegate

    def get_command(self, ctx, name):
        """Overrides get_command to delegate subcommand retrieval to the delegate.

        Args:
            ctx (click.Context): The Click context.
            name (str): The name of the command to retrieve.

        Returns:
            click.Command: The retrieved command from the delegate, or None.
        """
        if hasattr(self.delegate, 'get_command'):
            return self.delegate.get_command(ctx, name)
        return None

    def invoke(self, ctx):
        """Invokes the delegate object with the provided context.

        Args:
            ctx (click.Context): The Click context containing the arguments.

        Raises:
            RuntimeError: If the delegate cannot be resolved or is not callable.
        """
        # We manually call the delegate's main method with the remaining arguments
        # This bypasses the need for the delegate to be a 'core' click object
        try:
            # If the delegate is a function (factory), call it first.
            # click.Group is also callable, so we check if it already has 'main' or 'commands'.
            if callable(self.delegate) and not hasattr(self.delegate, 'main') and not hasattr(self.delegate, 'commands'):
                logger.debug(f"Calling delegate factory {self.delegate}")
                actual_delegate = self.delegate()
                logger.debug(f"Factory returned {actual_delegate}")
            else:
                actual_delegate = self.delegate

            if actual_delegate is None:
                raise RuntimeError(f"Delegate for {self.name} resolved to None")

            # If the delegate is a click object (Group or Command), it doesn't have a .main() method.
            # We check for .main() first (for wrapper objects), otherwise we call it directly.
            if hasattr(actual_delegate, 'main') and callable(actual_delegate.main):
                actual_delegate.main(args=ctx.args, standalone_mode=True)
            elif callable(actual_delegate):
                # For click.Group/Command, we can't easily call them with args from here
                # without creating a new context. The safest way is to use a subprocess
                # or to use the click internal API.
                # However, since we are in a DelegatingCommand, we can try to invoke it.
                try:
                    # Try to use the click object's own entry point logic if it's a group
                    if hasattr(actual_delegate, 'cli'):
                        actual_delegate.cli(args=ctx.args, standalone_mode=True)
                    else:
                        # Fallback: call it as a function if it's a simple wrapper
                        actual_delegate(args=ctx.args, standalone_mode=True)
                except TypeError:
                    # If it doesn't accept args, it might be a standard click object
                    # In that case, we might need to use a different approach.
                    # For now, let's try to call it.
                    actual_delegate()
            else:
                raise RuntimeError(f"Delegate for {self.name} is not callable and has no .main() method")
        except Exception as e:
            logger.error(f"Delegation failed for {self.name}: {e}")
            sys.exit(1)

Methods:

__init__

__init__(name, delegate, **kwargs)

Initializes the DelegatingCommand.

Parameters:

Name	Type	Description	Default
`name`	`str`	The name of the command.	required
`delegate`	`Any`	The actual click object or factory to delegate to.	required
`**kwargs`		Additional arguments passed to the click.Group constructor.	`{}`

Source code in src/cloudmesh/ai/cmc/main.py

def __init__(self, name, delegate, **kwargs):
    """Initializes the DelegatingCommand.

    Args:
        name (str): The name of the command.
        delegate (Any): The actual click object or factory to delegate to.
        **kwargs: Additional arguments passed to the click.Group constructor.
    """
    # For a Group, ignore_unknown_options is a valid argument in many click versions
    # If it still fails, we can remove it and try another way.
    super().__init__(name, **kwargs)
    self.delegate = delegate

get_command

get_command(ctx, name)

Overrides get_command to delegate subcommand retrieval to the delegate.

Parameters:

Name	Type	Description	Default
`ctx`	`Context`	The Click context.	required
`name`	`str`	The name of the command to retrieve.	required

Returns:

Type	Description
	click.Command: The retrieved command from the delegate, or None.

Source code in src/cloudmesh/ai/cmc/main.py

def get_command(self, ctx, name):
    """Overrides get_command to delegate subcommand retrieval to the delegate.

    Args:
        ctx (click.Context): The Click context.
        name (str): The name of the command to retrieve.

    Returns:
        click.Command: The retrieved command from the delegate, or None.
    """
    if hasattr(self.delegate, 'get_command'):
        return self.delegate.get_command(ctx, name)
    return None

invoke

invoke(ctx)

Invokes the delegate object with the provided context.

Parameters:

Name	Type	Description	Default
`ctx`	`Context`	The Click context containing the arguments.	required

Raises:

Type	Description
`RuntimeError`	If the delegate cannot be resolved or is not callable.

Source code in src/cloudmesh/ai/cmc/main.py

def invoke(self, ctx):
    """Invokes the delegate object with the provided context.

    Args:
        ctx (click.Context): The Click context containing the arguments.

    Raises:
        RuntimeError: If the delegate cannot be resolved or is not callable.
    """
    # We manually call the delegate's main method with the remaining arguments
    # This bypasses the need for the delegate to be a 'core' click object
    try:
        # If the delegate is a function (factory), call it first.
        # click.Group is also callable, so we check if it already has 'main' or 'commands'.
        if callable(self.delegate) and not hasattr(self.delegate, 'main') and not hasattr(self.delegate, 'commands'):
            logger.debug(f"Calling delegate factory {self.delegate}")
            actual_delegate = self.delegate()
            logger.debug(f"Factory returned {actual_delegate}")
        else:
            actual_delegate = self.delegate

        if actual_delegate is None:
            raise RuntimeError(f"Delegate for {self.name} resolved to None")

        # If the delegate is a click object (Group or Command), it doesn't have a .main() method.
        # We check for .main() first (for wrapper objects), otherwise we call it directly.
        if hasattr(actual_delegate, 'main') and callable(actual_delegate.main):
            actual_delegate.main(args=ctx.args, standalone_mode=True)
        elif callable(actual_delegate):
            # For click.Group/Command, we can't easily call them with args from here
            # without creating a new context. The safest way is to use a subprocess
            # or to use the click internal API.
            # However, since we are in a DelegatingCommand, we can try to invoke it.
            try:
                # Try to use the click object's own entry point logic if it's a group
                if hasattr(actual_delegate, 'cli'):
                    actual_delegate.cli(args=ctx.args, standalone_mode=True)
                else:
                    # Fallback: call it as a function if it's a simple wrapper
                    actual_delegate(args=ctx.args, standalone_mode=True)
            except TypeError:
                # If it doesn't accept args, it might be a standard click object
                # In that case, we might need to use a different approach.
                # For now, let's try to call it.
                actual_delegate()
        else:
            raise RuntimeError(f"Delegate for {self.name} is not callable and has no .main() method")
    except Exception as e:
        logger.error(f"Delegation failed for {self.name}: {e}")
        sys.exit(1)

LazyCommand `dataclass`

Represents a command that is loaded lazily from a module.

Attributes:

Name	Type	Description
`name`	`str`	The name of the command.
`module_name`	`str`	The name of the module containing the command.
`entry_point_name`	`str`	The name of the entry point function/object in the module.
`hidden`	`bool`	Whether the command should be hidden from help output. Defaults to False.

Source code in src/cloudmesh/ai/cmc/main.py

@dataclass
class LazyCommand:
    """Represents a command that is loaded lazily from a module.

    Attributes:
        name (str): The name of the command.
        module_name (str): The name of the module containing the command.
        entry_point_name (str): The name of the entry point function/object in the module.
        hidden (bool): Whether the command should be hidden from help output. Defaults to False.
    """
    name: str
    module_name: str
    entry_point_name: str
    hidden: bool = False

    def get_short_help_str(self, limit=None):
        """Returns a short help string for the lazy-loaded command.

        Args:
            limit (int, optional): Maximum length of the help string. Defaults to None.

        Returns:
            str: A short description of the command.
        """
        return "Lazy-loaded extension"

Methods:

get_short_help_str

get_short_help_str(limit=None)

Returns a short help string for the lazy-loaded command.

Parameters:

Name	Type	Description	Default
`limit`	`int`	Maximum length of the help string. Defaults to None.	`None`

Returns:

Name	Type	Description
`str`		A short description of the command.

Source code in src/cloudmesh/ai/cmc/main.py

def get_short_help_str(self, limit=None):
    """Returns a short help string for the lazy-loaded command.

    Args:
        limit (int, optional): Maximum length of the help string. Defaults to None.

    Returns:
        str: A short description of the command.
    """
    return "Lazy-loaded extension"

SubcommandHelpGroup

Bases: Group

Custom Click Group to show subcommands in help output with dynamic width.

This group overrides the default command retrieval and formatting to support lazy loading of extensions and to ensure help text utilizes the full terminal width.

Source code in src/cloudmesh/ai/cmc/main.py

class SubcommandHelpGroup(click.Group):
    """Custom Click Group to show subcommands in help output with dynamic width.

    This group overrides the default command retrieval and formatting to support
    lazy loading of extensions and to ensure help text utilizes the full terminal width.
    """

    def get_command(self, ctx, name):
        """Retrieves a command, supporting lazy loading of extensions.

        Args:
            ctx (click.Context): The Click context.
            name (str): The name of the command to retrieve.

        Returns:
            click.Command: The loaded command, or None if not found.
        """
        logger.debug(f"get_command called for {name}")
        cmd = super().get_command(ctx, name)
        logger.debug(f"initial cmd type for {name}: {type(cmd)}")

        # Use attribute check instead of isinstance to avoid Click version/instance mismatches
        if hasattr(cmd, 'module_name') and hasattr(cmd, 'entry_point_name'):
            # To prevent Click version mismatches between core and extensions,
            # we ensure the extension uses the same click module as the core.
            import click as core_click
            sys.modules['click'] = core_click

            # Load the actual command
            if getattr(cmd, 'module_name', None):
                # Core or Pip extension
                try:
                    module = importlib.import_module(cmd.module_name)
                    cmd = getattr(module, cmd.entry_point_name, None)
                except Exception as e:
                    logger.error(
                        f"Failed to lazy-load extension {cmd.module_name}: {e}"
                    )
                    cmd = None

            if cmd:
                # If the entry point is 'main', it's often just a wrapper that calls the real CLI.
                # We try to find the actual click object in the module.
                if getattr(cmd, '__name__', None) == 'main' and module:
                    for attr_name in ['cli', 'entry_point', 'group']:
                        attr = getattr(module, attr_name, None)
                        if attr and callable(attr):
                            logger.debug(f"Found actual click object or factory {attr_name} instead of main wrapper")
                            cmd = attr
                            break

                logger.debug(f"loaded cmd type for {name}: {type(cmd)}")
                # If the loaded command is seen as a function but is actually a click.Group from a different instance,
                # we wrap it in a DelegatingCommand to avoid Click version mismatches.
                if callable(cmd) and not hasattr(cmd, 'make_context'):
                    logger.debug(f"Wrapping {name} in DelegatingCommand due to Click version mismatch")
                    cmd = DelegatingCommand(name=name, delegate=cmd)

                # Cache the loaded command in the group
                self.commands[name] = cmd
        return cmd

    def format_commands(self, ctx, formatter):
        """Overwrites the default command formatting for full terminal width.

        Ensures that help text is not prematurely truncated by syncing the 
        formatter width with the actual terminal size.

        Args:
            ctx (click.Context): The Click context.
            formatter (click.helpers.HelpFormatter): The formatter to use.
        """
        # 1. Sync formatter width with actual terminal width
        term_width, _ = shutil.get_terminal_size()
        formatter.width = term_width

        commands = []
        for subcommand in self.list_commands(ctx):
            cmd = self.get_command(ctx, subcommand)
            if cmd is None or cmd.hidden:
                continue
            commands.append((subcommand, cmd))

        if not commands:
            return

        # 2. Calculate the column width for the command names
        # We find the longest command name to align the help text column
        max_name_len = max([len(name) for name, _ in commands]) if commands else 0

        # 3. Calculate available width for the help text
        # Margin (2) + Name + Padding (2)
        available_help_width = term_width - max_name_len - 6

        with formatter.section("Commands"):
            rows = []
            for name, cmd in commands:
                # We use a larger limit for the help string to ensure
                # the docstrings we wrote in speedtest.py and tree.py are visible
                help_text = cmd.get_short_help_str(limit=available_help_width)
                rows.append((name, help_text))

            formatter.write_dl(rows)

Methods:

format_commands

format_commands(ctx, formatter)

Overwrites the default command formatting for full terminal width.

Ensures that help text is not prematurely truncated by syncing the formatter width with the actual terminal size.

Parameters:

Name	Type	Description	Default
`ctx`	`Context`	The Click context.	required
`formatter`	`HelpFormatter`	The formatter to use.	required

Source code in src/cloudmesh/ai/cmc/main.py

def format_commands(self, ctx, formatter):
    """Overwrites the default command formatting for full terminal width.

    Ensures that help text is not prematurely truncated by syncing the 
    formatter width with the actual terminal size.

    Args:
        ctx (click.Context): The Click context.
        formatter (click.helpers.HelpFormatter): The formatter to use.
    """
    # 1. Sync formatter width with actual terminal width
    term_width, _ = shutil.get_terminal_size()
    formatter.width = term_width

    commands = []
    for subcommand in self.list_commands(ctx):
        cmd = self.get_command(ctx, subcommand)
        if cmd is None or cmd.hidden:
            continue
        commands.append((subcommand, cmd))

    if not commands:
        return

    # 2. Calculate the column width for the command names
    # We find the longest command name to align the help text column
    max_name_len = max([len(name) for name, _ in commands]) if commands else 0

    # 3. Calculate available width for the help text
    # Margin (2) + Name + Padding (2)
    available_help_width = term_width - max_name_len - 6

    with formatter.section("Commands"):
        rows = []
        for name, cmd in commands:
            # We use a larger limit for the help string to ensure
            # the docstrings we wrote in speedtest.py and tree.py are visible
            help_text = cmd.get_short_help_str(limit=available_help_width)
            rows.append((name, help_text))

        formatter.write_dl(rows)

get_command

get_command(ctx, name)

Retrieves a command, supporting lazy loading of extensions.

Parameters:

Name	Type	Description	Default
`ctx`	`Context`	The Click context.	required
`name`	`str`	The name of the command to retrieve.	required

Returns:

Type	Description
	click.Command: The loaded command, or None if not found.

Source code in src/cloudmesh/ai/cmc/main.py

def get_command(self, ctx, name):
    """Retrieves a command, supporting lazy loading of extensions.

    Args:
        ctx (click.Context): The Click context.
        name (str): The name of the command to retrieve.

    Returns:
        click.Command: The loaded command, or None if not found.
    """
    logger.debug(f"get_command called for {name}")
    cmd = super().get_command(ctx, name)
    logger.debug(f"initial cmd type for {name}: {type(cmd)}")

    # Use attribute check instead of isinstance to avoid Click version/instance mismatches
    if hasattr(cmd, 'module_name') and hasattr(cmd, 'entry_point_name'):
        # To prevent Click version mismatches between core and extensions,
        # we ensure the extension uses the same click module as the core.
        import click as core_click
        sys.modules['click'] = core_click

        # Load the actual command
        if getattr(cmd, 'module_name', None):
            # Core or Pip extension
            try:
                module = importlib.import_module(cmd.module_name)
                cmd = getattr(module, cmd.entry_point_name, None)
            except Exception as e:
                logger.error(
                    f"Failed to lazy-load extension {cmd.module_name}: {e}"
                )
                cmd = None

        if cmd:
            # If the entry point is 'main', it's often just a wrapper that calls the real CLI.
            # We try to find the actual click object in the module.
            if getattr(cmd, '__name__', None) == 'main' and module:
                for attr_name in ['cli', 'entry_point', 'group']:
                    attr = getattr(module, attr_name, None)
                    if attr and callable(attr):
                        logger.debug(f"Found actual click object or factory {attr_name} instead of main wrapper")
                        cmd = attr
                        break

            logger.debug(f"loaded cmd type for {name}: {type(cmd)}")
            # If the loaded command is seen as a function but is actually a click.Group from a different instance,
            # we wrap it in a DelegatingCommand to avoid Click version mismatches.
            if callable(cmd) and not hasattr(cmd, 'make_context'):
                logger.debug(f"Wrapping {name} in DelegatingCommand due to Click version mismatch")
                cmd = DelegatingCommand(name=name, delegate=cmd)

            # Cache the loaded command in the group
            self.commands[name] = cmd
    return cmd

Functions:

alias_plugins_check

alias_plugins_check(ctx)

Alias for plugins check.

Parameters:

Name	Type	Description	Default
`ctx`	`Context`	The Click context.	required

Source code in src/cloudmesh/ai/cmc/main.py

@cli.command(name="pch", hidden=True)
@click.pass_context
def alias_plugins_check(ctx):
    """Alias for plugins check.

    Args:
        ctx (click.Context): The Click context.
    """
    plugins_group = cli.get_command("plugins")
    if plugins_group:
        ctx.invoke(plugins_group.get_command("check"))

alias_plugins_list

alias_plugins_list(ctx)

Alias for plugins list.

Parameters:

Name	Type	Description	Default
`ctx`	`Context`	The Click context.	required

Source code in src/cloudmesh/ai/cmc/main.py

@cli.command(name="pl", hidden=True)
@click.pass_context
def alias_plugins_list(ctx):
    """Alias for plugins list.

    Args:
        ctx (click.Context): The Click context.
    """
    plugins_group = cli.get_command("plugins")
    if plugins_group:
        ctx.invoke(plugins_group.get_command("list"))

alias_telemetry

alias_telemetry(ctx)

Alias for telemetry.

Parameters:

Name	Type	Description	Default
`ctx`	`Context`	The Click context.	required

Source code in src/cloudmesh/ai/cmc/main.py

@cli.command(name="tel", hidden=True)
@click.pass_context
def alias_telemetry(ctx):
    """Alias for telemetry.

    Args:
        ctx (click.Context): The Click context.
    """
    ctx.invoke(cli.get_command("telemetry"))

alias_version

alias_version(ctx)

Alias for version.

Parameters:

Name	Type	Description	Default
`ctx`	`Context`	The Click context.	required

Source code in src/cloudmesh/ai/cmc/main.py

@cli.command(name="v", hidden=True)
@click.pass_context
def alias_version(ctx):
    """Alias for version.

    Args:
        ctx (click.Context): The Click context.
    """
    ctx.invoke(cli.get_command("version"))

cli

cli(ctx, debug)

cmc: Cloudmesh Commands.

The main entry point for the Cloudmesh AI CMC tool.

Parameters:

Name	Type	Description	Default
`ctx`	`Context`	The Click context.	required
`debug`	`bool`	Whether to enable debug logging.	required

Source code in src/cloudmesh/ai/cmc/main.py

@click.group(cls=SubcommandHelpGroup)
@click.option("-v", "--debug", is_flag=True, help="Enable debug logging.")
@click.pass_context
def cli(ctx, debug):
    """cmc: Cloudmesh Commands.

    The main entry point for the Cloudmesh AI CMC tool.

    Args:
        ctx (click.Context): The Click context.
        debug (bool): Whether to enable debug logging.
    """
    if debug:
        logger.setLevel(logging.DEBUG)
        logger.debug("Debug logging enabled")

generate_completion_script

generate_completion_script(shell_type)

Prints the shell completion script for the specified shell type.

Parameters:

Name	Type	Description	Default
`shell_type`	`str`	The type of shell (e.g., 'bash_source', 'zsh_source', 'fish_source').	required

Source code in src/cloudmesh/ai/cmc/main.py

def generate_completion_script(shell_type):
    """Prints the shell completion script for the specified shell type.

    Args:
        shell_type (str): The type of shell (e.g., 'bash_source', 'zsh_source', 'fish_source').
    """
    scripts = {
        "bash_source": (
            "_cmc() {\n"
            "    local cur=\"${COMP_WORDS[COMP_CWORD]}\"\n"
            "    local suggestions=$(CLICOMPLETE=1 cmc \"${COMP_WORDS[@]:1}\")\n"
            "    COMPREPLY=( $(compgen -W \"$suggestions\" -- \"$cur\") )\n"
            "}\n"
            "complete -F _cmc cmc"
        ),
        "zsh_source": (
            "autoload -Uz compinit && compinit\n"
            "_cmc() {\n"
            "    local -a opts\n"
            "    # In Zsh completion functions, $words contains the current command line\n"
            "    # We skip the first word (the command itself) and pass the rest to cmc\n"
            "    # We use \"${words[2,CURRENT]}\" to get the arguments up to the cursor\n"
            "    opts=($(CLICOMPLETE=1 cmc \"${words[2,CURRENT]}\"))\n"
            "    compadd -a opts\n"
            "}\n"
            "compdef _cmc cmc"
        ),
        "fish_source": (
            "complete -c cmc -f\n"
            "complete -c cmc -a \"(CLICOMPLETE=1 cmc)\""
        ),
    }
    script = scripts.get(shell_type, "")
    print(script)

handle_completion

handle_completion()

Manually handles shell completion requests.

This function is called when the environment variable CLICOMPLETE=1 is set. It determines the current word being typed and suggests matching commands or subcommands.

Source code in src/cloudmesh/ai/cmc/main.py

def handle_completion():
    """Manually handles shell completion requests.

    This function is called when the environment variable CLICOMPLETE=1 is set.
    It determines the current word being typed and suggests matching commands 
    or subcommands.
    """
    # Suppress all logs except WARNING/ERROR during completion to avoid 
    # polluting the shell completion list.
    logging.getLogger().setLevel(logging.WARNING)
    logger.setLevel(logging.WARNING)

    try:
        # Ensure extensions are loaded before completing
        load_core_extensions(cli)
        load_pip_extensions(cli)
    except Exception as e:
        # We print to stderr so it doesn't pollute the completion list
        print(f"Error loading extensions during completion: {e}", file=sys.stderr)

    args = sys.argv[1:]

    # Support reading completion arguments from stdin (e.g., cat FILE | cmc)
    # if no arguments are provided via CLI.
    if not args and not sys.stdin.isatty():
        stdin_content = sys.stdin.read().strip()
        if stdin_content:
            # Split by whitespace to simulate argv
            args = stdin_content.split()

    # Determine the current word being typed and the preceding context
    if not args:
        current_word = ""
        context_args = []
    elif args[-1] == "":
        current_word = ""
        context_args = args[:-1]
    else:
        current_word = args[-1]
        context_args = args[:-1]

    if not context_args:
        # Root level completion: suggest top-level commands
        all_commands = list(cli.commands.keys())

        # If the user has already typed a full command name (possibly with trailing space) and hit TAB,
        # check if it's a group and suggest its subcommands instead.
        stripped_word = current_word.rstrip()
        if stripped_word in all_commands:
            cmd = cli.commands.get(stripped_word)
            # Use hasattr to avoid Click version mismatch issues with isinstance
            if cmd and hasattr(cmd, 'commands'):
                for sub_cmd in cmd.commands.keys():
                    print(sub_cmd)
                return

        # Otherwise, suggest commands that start with the current word
        # Strip trailing whitespace to allow matching (e.g., "ban " matches "banner")
        search_word = current_word.rstrip()
        for cmd_name in all_commands:
            if cmd_name.startswith(search_word):
                print(cmd_name)
    else:
        # Subcommand completion: suggest subcommands for the first argument
        group_name = context_args[0]
        group = cli.commands.get(group_name)

        # Use hasattr to avoid Click version mismatch issues with isinstance
        if group and hasattr(group, 'commands'):
            for sub_cmd in group.commands.keys():
                if sub_cmd.startswith(current_word):
                    print(sub_cmd)
        else:
            # If the first argument isn't a group, we can't suggest subcommands.
            # We print nothing, which allows the shell to potentially fall back
            # to other completion types, but we've tried our best.
            pass

load_core_extensions

load_core_extensions(cli)

Recursively loads all core extensions from the command directory.

Iterates through the cloudmesh.ai.command package and adds any found Click commands or groups to the provided CLI object.

Parameters:

Name	Type	Description	Default
`cli`	`Group`	The main CLI group to which extensions are added.	required

Source code in src/cloudmesh/ai/cmc/main.py

def load_core_extensions(cli):
    """Recursively loads all core extensions from the command directory.

    Iterates through the `cloudmesh.ai.command` package and adds any found 
    Click commands or groups to the provided CLI object.

    Args:
        cli (click.Group): The main CLI group to which extensions are added.
    """
    found_any = False

    # Handle namespace packages where __path__ is a list
    paths = getattr(extensions, '__path__', [])
    if isinstance(paths, str):
        paths = [paths]

    def register_recursive(current_cli, path, package_prefix, visited=None):
        nonlocal found_any
        if visited is None:
            visited = set()

        if package_prefix in visited:
            return
        visited.add(package_prefix)

        # Ensure path is a list for pkgutil.iter_modules
        search_path = path if isinstance(path, (list, tuple)) else [path]

        try:
            for _, module_name, is_pkg in pkgutil.iter_modules(search_path):
                full_module_name = f"{package_prefix}.{module_name}"
                logger.debug(f"Found module: {full_module_name} (pkg={is_pkg})")
                # Try to use the module/package's own register function if it exists
                try:
                    mod = importlib.import_module(full_module_name)
                    if hasattr(mod, 'register') and callable(mod.register):
                        mod.register(current_cli)
                        visited.add(full_module_name)
                        found_any = True
                        continue 
                except Exception as e:
                    logger.debug(f"Could not use register function for {full_module_name}: {e}")

                if is_pkg:
                    # Fallback: Create a group for the sub-package and recurse
                    group = click.Group(name=module_name)
                    current_cli.add_command(group)
                    try:
                        pkg = importlib.import_module(full_module_name)
                        register_recursive(group, pkg.__path__, full_module_name, visited)
                    except Exception as e:
                        logger.error(f"Failed to recurse into {full_module_name}: {e}")
                else:
                    try:
                        module = importlib.import_module(full_module_name)
                        # Find any click Command or Group in the module and add it
                        for attr in vars(module).values():
                            if isinstance(attr, (click.Command, click.Group)):
                                # Use the command's own name if it's explicitly set, otherwise use module_name
                                cmd_name = getattr(attr, 'name', module_name) or module_name
                                # Prevent adding the group to itself
                                if attr is not current_cli:
                                    logger.debug(f"Adding command {cmd_name} to {current_cli}")
                                    current_cli.add_command(attr, name=cmd_name)
                                    found_any = True
                                break
                    except Exception as e:
                        logger.error(f"Could not load {full_module_name}: {e}")
        except Exception as e:
            logger.error(f"Failed to iterate {package_prefix}: {e}")

    for path in paths:
        register_recursive(cli, path, "cloudmesh.ai.command")

    if not found_any:
        logger.warning("No core extensions found in cloudmesh.ai.command")

load_pip_extensions

load_pip_extensions(cli)

Loads extensions installed via pip using entry points lazily.

Searches for entry points in the 'cloudmesh.ai.command' group and adds them as LazyCommand objects to the CLI.

Parameters:

Name	Type	Description	Default
`cli`	`Group`	The main CLI group to which extensions are added.	required

Source code in src/cloudmesh/ai/cmc/main.py

def load_pip_extensions(cli):
    """Loads extensions installed via pip using entry points lazily.

    Searches for entry points in the 'cloudmesh.ai.command' group and adds 
    them as LazyCommand objects to the CLI.

    Args:
        cli (click.Group): The main CLI group to which extensions are added.
    """
    eps = entry_points().select(group="cloudmesh.ai.command")
    for entry_point in eps:
        # Avoid overwriting extensions already loaded from the registry
        if entry_point.name in cli.commands:
            logger.debug(f"Skipping pip extension {entry_point.name} as it is already registered")
            continue

        # We use the entry point's value as the module name
        # entry_point.value is usually 'module.function'
        module_path, func_name = entry_point.value.rsplit(":", 1)
        cli.add_command(
            LazyCommand(
                name=entry_point.name,
                module_name=module_path,
                entry_point_name=func_name,
            ),
            name=entry_point.name,
        )

main

main()

Main entry point for the CMC application.

Initializes pip extensions and invokes the Click CLI.

Source code in src/cloudmesh/ai/cmc/main.py

def main():
    """Main entry point for the CMC application.

    Initializes pip extensions and invokes the Click CLI.
    """
    # Use telemetry to track the entire execution of the cmc tool
    with telemetry.track(message="Executing CMC command"):
        logger.debug("CMC main() is executing")

        # 1. Core extensions are now loaded at module level

        # 2. Load extensions installed via pip (entry points)
        load_pip_extensions(cli)
        cli()

Modules

utils

Classes

CMCError

Bases: Exception

Base class for CMC exceptions.

Source code in src/cloudmesh/ai/cmc/utils.py

class CMCError(Exception):
    """Base class for CMC exceptions."""
    pass

Config

Handles configuration for CMC from a YAML file.

Source code in src/cloudmesh/ai/cmc/utils.py

class Config:
    """Handles configuration for CMC from a YAML file."""

    DEFAULT_CONFIG_PATH = Path(path_expand("~/.config/cloudmesh/cmc.yaml"))

    DEFAULTS = {
        "telemetry": {
            "enabled": True,
            "path": "~/cmc_telemetry.jsonl",
            "backend": "json"
        },
        "logging": {
            "level": "WARNING"
        }
    }

    SCHEMA = {
        "telemetry.enabled": {"type": bool, "desc": "Enable or disable telemetry collection"},
        "telemetry.path": {"type": str, "desc": "Path to the telemetry log file"},
        "telemetry.backend": {"type": str, "desc": "Telemetry backend (e.g., 'json', 'text')"},
        "logging.level": {"type": str, "desc": "Logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL)"},
    }

    def __init__(self, config_path: Optional[Path] = None):
        """Initializes the Config object.

        Args:
            config_path (Optional[Path]): Path to the configuration file. 
                Defaults to DEFAULT_CONFIG_PATH if not provided.
        """
        self.path = config_path or self.DEFAULT_CONFIG_PATH
        self.data = copy.deepcopy(self.DEFAULTS)
        self._load_config()

    def _load_config(self):
        """Loads configuration from the YAML file on disk and updates defaults."""
        user_config = load_yaml(str(self.path))
        if user_config:
            self._deep_update(self.data, user_config)

    def _deep_update(self, base: Dict, update: Dict):
        """Recursively updates a dictionary.

        Args:
            base (Dict): The dictionary to be updated.
            update (Dict): The dictionary containing updates to apply.
        """
        for k, v in update.items():
            if isinstance(v, dict) and k in base and isinstance(base[k], dict):
                self._deep_update(base[k], v)
            else:
                base[k] = v

    def get(self, key_path: str, default: Any = None) -> Any:
        """Gets a value from the config using a dot-separated path.

        Environment variables can override config values. For example, 
        'telemetry.path' can be overridden by 'CMC_TELEMETRY_PATH'.

        Args:
            key_path (str): Dot-separated path to the configuration value.
            default (Any, optional): Value to return if the key is not found. Defaults to None.

        Returns:
            Any: The configuration value or the default value.
        """
        # 1. Check for environment variable override
        env_var = f"CMC_{key_path.replace('.', '_').upper()}"
        env_val = os.environ.get(env_var)
        if env_val is not None:
            # Try to cast to the type specified in SCHEMA
            if key_path in self.SCHEMA:
                expected_type = self.SCHEMA[key_path]["type"]
                try:
                    if expected_type is bool:
                        return env_val.lower() in ("true", "1", "yes")
                    return expected_type(env_val)
                except (ValueError, TypeError):
                    logger.warning(f"Environment variable {env_var} has invalid value '{env_val}' for type {expected_type.__name__}. Using config value.")
            else:
                return env_val

        # 2. Fallback to config data
        keys = key_path.split(".")
        val = self.data
        try:
            for k in keys:
                val = val[k]
            return val
        except (KeyError, TypeError):
            return default

    def validate(self, key_path: str, value: Any):
        """Validates a configuration value against the schema if it exists.

        Args:
            key_path (str): Dot-separated path to the configuration value.
            value (Any): The value to validate.

        Raises:
            TypeError: If the value does not match the expected type in the schema.
        """
        if key_path in self.SCHEMA:
            expected_type = self.SCHEMA[key_path]["type"]
            if not isinstance(value, expected_type):
                raise TypeError(f"Invalid type for '{key_path}'. Expected {expected_type.__name__}, got {type(value).__name__}.")
        # If not in SCHEMA, we allow it (dynamic plugin configuration)

    def set(self, key_path: str, value: Any):
        """Sets a value in the config using a dot-separated path.

        Args:
            key_path (str): Dot-separated path to the configuration value.
            value (Any): The value to set.
        """
        self.validate(key_path, value)
        keys = key_path.split(".")
        val = self.data
        for k in keys[:-1]:
            if k not in val or not isinstance(val[k], dict):
                val[k] = {}
            val = val[k]
        val[keys[-1]] = value

    def save(self):
        """Saves the current configuration to the YAML file.

        Raises:
            Exception: If the configuration file cannot be saved.
        """
        dump_yaml(str(self.path), self.data)

Methods:

__init__

__init__(config_path=None)

Initializes the Config object.

Parameters:

Name	Type	Description	Default
`config_path`	`Optional[Path]`	Path to the configuration file. Defaults to DEFAULT_CONFIG_PATH if not provided.	`None`

Source code in src/cloudmesh/ai/cmc/utils.py

def __init__(self, config_path: Optional[Path] = None):
    """Initializes the Config object.

    Args:
        config_path (Optional[Path]): Path to the configuration file. 
            Defaults to DEFAULT_CONFIG_PATH if not provided.
    """
    self.path = config_path or self.DEFAULT_CONFIG_PATH
    self.data = copy.deepcopy(self.DEFAULTS)
    self._load_config()

get

get(key_path, default=None)

Gets a value from the config using a dot-separated path.

Environment variables can override config values. For example, 'telemetry.path' can be overridden by 'CMC_TELEMETRY_PATH'.

Parameters:

Name	Type	Description	Default
`key_path`	`str`	Dot-separated path to the configuration value.	required
`default`	`Any`	Value to return if the key is not found. Defaults to None.	`None`

Returns:

Name	Type	Description
`Any`	`Any`	The configuration value or the default value.

Source code in src/cloudmesh/ai/cmc/utils.py

def get(self, key_path: str, default: Any = None) -> Any:
    """Gets a value from the config using a dot-separated path.

    Environment variables can override config values. For example, 
    'telemetry.path' can be overridden by 'CMC_TELEMETRY_PATH'.

    Args:
        key_path (str): Dot-separated path to the configuration value.
        default (Any, optional): Value to return if the key is not found. Defaults to None.

    Returns:
        Any: The configuration value or the default value.
    """
    # 1. Check for environment variable override
    env_var = f"CMC_{key_path.replace('.', '_').upper()}"
    env_val = os.environ.get(env_var)
    if env_val is not None:
        # Try to cast to the type specified in SCHEMA
        if key_path in self.SCHEMA:
            expected_type = self.SCHEMA[key_path]["type"]
            try:
                if expected_type is bool:
                    return env_val.lower() in ("true", "1", "yes")
                return expected_type(env_val)
            except (ValueError, TypeError):
                logger.warning(f"Environment variable {env_var} has invalid value '{env_val}' for type {expected_type.__name__}. Using config value.")
        else:
            return env_val

    # 2. Fallback to config data
    keys = key_path.split(".")
    val = self.data
    try:
        for k in keys:
            val = val[k]
        return val
    except (KeyError, TypeError):
        return default

save

save()

Saves the current configuration to the YAML file.

Raises:

Type	Description
`Exception`	If the configuration file cannot be saved.

Source code in src/cloudmesh/ai/cmc/utils.py

def save(self):
    """Saves the current configuration to the YAML file.

    Raises:
        Exception: If the configuration file cannot be saved.
    """
    dump_yaml(str(self.path), self.data)

set

set(key_path, value)

Sets a value in the config using a dot-separated path.

Parameters:

Name	Type	Description	Default
`key_path`	`str`	Dot-separated path to the configuration value.	required
`value`	`Any`	The value to set.	required

Source code in src/cloudmesh/ai/cmc/utils.py

def set(self, key_path: str, value: Any):
    """Sets a value in the config using a dot-separated path.

    Args:
        key_path (str): Dot-separated path to the configuration value.
        value (Any): The value to set.
    """
    self.validate(key_path, value)
    keys = key_path.split(".")
    val = self.data
    for k in keys[:-1]:
        if k not in val or not isinstance(val[k], dict):
            val[k] = {}
        val = val[k]
    val[keys[-1]] = value

validate

validate(key_path, value)

Validates a configuration value against the schema if it exists.

Parameters:

Name	Type	Description	Default
`key_path`	`str`	Dot-separated path to the configuration value.	required
`value`	`Any`	The value to validate.	required

Raises:

Type	Description
`TypeError`	If the value does not match the expected type in the schema.

Source code in src/cloudmesh/ai/cmc/utils.py

def validate(self, key_path: str, value: Any):
    """Validates a configuration value against the schema if it exists.

    Args:
        key_path (str): Dot-separated path to the configuration value.
        value (Any): The value to validate.

    Raises:
        TypeError: If the value does not match the expected type in the schema.
    """
    if key_path in self.SCHEMA:
        expected_type = self.SCHEMA[key_path]["type"]
        if not isinstance(value, expected_type):
            raise TypeError(f"Invalid type for '{key_path}'. Expected {expected_type.__name__}, got {type(value).__name__}.")

PluginDependencyError

Bases: CMCError

Raised when a plugin dependency is missing or inactive.

Source code in src/cloudmesh/ai/cmc/utils.py

class PluginDependencyError(CMCError):
    """Raised when a plugin dependency is missing or inactive."""
    pass

PluginVersionError

Bases: CMCError

Raised when a plugin version is incompatible.

Source code in src/cloudmesh/ai/cmc/utils.py

class PluginVersionError(CMCError):
    """Raised when a plugin version is incompatible."""
    pass

Registry

Manages the registration and status of AI extensions.

Persists extension metadata within the main CMC configuration file.

Source code in src/cloudmesh/ai/cmc/utils.py

class Registry:
    """Manages the registration and status of AI extensions.

    Persists extension metadata within the main CMC configuration file.
    """

    def __init__(self):
        # Initialize the main config to store registry data
        from cloudmesh.ai.cmc.utils import Config
        self.config = Config()
        self.extensions = self.config.get("registry", {})

    def _save_registry(self):
        """Saves the current registry state to the main config file."""
        try:
            self.config.set("registry", self.extensions)
            self.config.save()
        except Exception as e:
            logger.error(f"Failed to save registry to config: {e}")

    def register(self, name: str, directory: str):
        """Registers a new extension path.

        Args:
            name (str): The name of the extension.
            directory (str): Path to the extension directory.
        """
        # Try to find version in pyproject.toml
        version = "0.0.0"
        pyproject_path = Path(directory) / "pyproject.toml"
        if pyproject_path.exists():
            try:
                import tomllib # Python 3.11+
                with open(pyproject_path, "rb") as f:
                    data = tomllib.load(f)
                    version = data.get("project", {}).get("version", "0.0.0")
            except Exception:
                pass

        self.extensions[name] = {
            "path": str(Path(directory).resolve()),
            "version": version,
            "active": True
        }
        self._save_registry()

    def set_status(self, name: str, active: bool) -> bool:
        """Updates the active status of an extension.

        Returns:
            bool: True if updated, False if extension not found.
        """
        if name in self.extensions:
            self.extensions[name]["active"] = active
            self._save_registry()
            return True
        return False

    def unregister(self, name: str):
        """Removes an extension from the registry."""
        if name in self.extensions:
            del self.extensions[name]
            self._save_registry()

    def list_all_details(self):
        """Returns a list of all registered extensions with their details."""
        return [
            {"name": name, **details}
            for name, details in self.extensions.items()
        ]

Methods:

list_all_details

list_all_details()

Returns a list of all registered extensions with their details.

Source code in src/cloudmesh/ai/cmc/utils.py

def list_all_details(self):
    """Returns a list of all registered extensions with their details."""
    return [
        {"name": name, **details}
        for name, details in self.extensions.items()
    ]

register

register(name, directory)

Registers a new extension path.

Parameters:

Name	Type	Description	Default
`name`	`str`	The name of the extension.	required
`directory`	`str`	Path to the extension directory.	required

Source code in src/cloudmesh/ai/cmc/utils.py

def register(self, name: str, directory: str):
    """Registers a new extension path.

    Args:
        name (str): The name of the extension.
        directory (str): Path to the extension directory.
    """
    # Try to find version in pyproject.toml
    version = "0.0.0"
    pyproject_path = Path(directory) / "pyproject.toml"
    if pyproject_path.exists():
        try:
            import tomllib # Python 3.11+
            with open(pyproject_path, "rb") as f:
                data = tomllib.load(f)
                version = data.get("project", {}).get("version", "0.0.0")
        except Exception:
            pass

    self.extensions[name] = {
        "path": str(Path(directory).resolve()),
        "version": version,
        "active": True
    }
    self._save_registry()

set_status

set_status(name, active)

Updates the active status of an extension.

Returns:

Name	Type	Description
`bool`	`bool`	True if updated, False if extension not found.

Source code in src/cloudmesh/ai/cmc/utils.py

def set_status(self, name: str, active: bool) -> bool:
    """Updates the active status of an extension.

    Returns:
        bool: True if updated, False if extension not found.
    """
    if name in self.extensions:
        self.extensions[name]["active"] = active
        self._save_registry()
        return True
    return False

unregister

unregister(name)

Removes an extension from the registry.

Source code in src/cloudmesh/ai/cmc/utils.py

def unregister(self, name: str):
    """Removes an extension from the registry."""
    if name in self.extensions:
        del self.extensions[name]
        self._save_registry()

Functions:

handle_errors

handle_errors(func)

Decorator to provide standardized error handling for CMC commands.

Logs the full traceback for developers, emits a telemetry failure if available, and displays a clean, user-friendly error message.

Parameters:

Name	Type	Description	Default
`func`	`Callable`	The function to be wrapped.	required

Returns:

Name	Type	Description
`Callable`		The wrapped function.

Source code in src/cloudmesh/ai/cmc/utils.py

def handle_errors(func):
    """Decorator to provide standardized error handling for CMC commands.

    Logs the full traceback for developers, emits a telemetry failure if 
    available, and displays a clean, user-friendly error message.

    Args:
        func (Callable): The function to be wrapped.

    Returns:
        Callable: The wrapped function.
    """
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except Exception as e:
            # 1. Log the full traceback for developers
            logger.exception(f"Unhandled exception in {func.__name__}: {str(e)}")

            # 2. Try to emit telemetry failure if telemetry is available in the global scope
            try:
                from cloudmesh.ai.cmc.main import telemetry
                telemetry.fail(error=str(e), function=func.__name__)
            except ImportError:
                pass

            # 3. Show clean error to user
            if isinstance(e, PluginDependencyError):
                console.print(f"\n[bold red]Dependency Error:[/bold red] {str(e)}")
                console.print("[yellow]Action: Ensure all required plugins are installed and active using 'cmc plugins list'.[/yellow]")
            elif isinstance(e, PluginVersionError):
                console.print(f"\n[bold red]Version Error:[/bold red] {str(e)}")
                console.print("[yellow]Action: Please update your CMC installation to the required version.[/yellow]")
            else:
                console.print(f"\n[bold red]Error:[/bold red] {str(e)}")
                console.print("[dim]Check logs for full traceback.[/dim]")

            sys.exit(1)
    return wrapper

register_group_extensions

register_group_extensions(parent_group, group_package, child_target=None)

Iteratively loads all modules in a package and calls their register() function.

Parameters:

Name	Type	Description	Default
`parent_group`	`Any`	The parent group to register extensions to.	required
`group_package`	`Module`	The package containing the extension modules.	required
`child_target`	`Any`	An alternative target for registration. Defaults to parent_group if not provided.	`None`

Source code in src/cloudmesh/ai/cmc/utils.py

def register_group_extensions(parent_group, group_package, child_target=None):
    """Iteratively loads all modules in a package and calls their register() function.

    Args:
        parent_group (Any): The parent group to register extensions to.
        group_package (Module): The package containing the extension modules.
        child_target (Any, optional): An alternative target for registration. 
            Defaults to parent_group if not provided.
    """
    target = child_target or parent_group

    for loader, module_name, is_pkg in pkgutil.iter_modules(group_package.__path__):
        full_name = f"{group_package.__name__}.{module_name}"
        try:
            module = importlib.import_module(full_name)
            if hasattr(module, "register"):
                module.register(target)
        except Exception as e:
            logger.error(f"Error loading extension {full_name}: {e}")

API Reference

Modules

context

Classes

Functions:

Modules

main

Classes

DelegatingCommand

Methods:

LazyCommand dataclass

Methods:

SubcommandHelpGroup

Methods:

Functions:

alias_plugins_check

alias_plugins_list

alias_telemetry

alias_version

cli

generate_completion_script

handle_completion

load_core_extensions

load_pip_extensions

main

Modules

utils

Classes

CMCError

Config

Methods:

PluginDependencyError

PluginVersionError

Registry

Methods:

Functions:

handle_errors

register_group_extensions

Modules

LazyCommand `dataclass`