cloudmesh-ai-ssh API Reference

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

API Documentation

Attributes

console module-attribute

console = Console()

entry_point module-attribute

entry_point = ssh_group

logger module-attribute

logger = get_contextual_logger('ssh')

telemetry module-attribute

telemetry = Telemetry('ssh')

Classes

SSHConfig

Bases: SSHBase

Managing the SSH config file (usually ~/.ssh/config).

Source code in src/cloudmesh/ai/ssh/ssh_config.py

class SSHConfig(SSHBase):
    """Managing the SSH config file (usually ~/.ssh/config)."""

    def __init__(self, filename: Optional[Union[str, Path]] = None, debug: bool = False):
        super().__init__(debug=debug)
        if filename is not None:
            self.filename = self.resolve_path(str(filename))
        else:
            self.filename = self.resolve_path("~/.ssh/config")

        self.conf: Optional[SshConf] = None
        self._resolved_cache: Dict[str, Dict[str, str]] = {}
        self.load()
    def get_content(self) -> str:
        """Return the raw content of the SSH config file."""
        try:
            with open(self.filename, "r") as f:
                return f.read()
        except Exception as e:
            logger.error(f"Could not read config file {self.filename}: {e}")
            return ""

        self.load()

    def names(self) -> List[str]:
        """The names defined in the SSH config.

        Returns:
            List[str]: the host names.
        """
        return self.list()

    def load(self):
        """Parse the SSH config file using sshconf."""
        try:
            self.conf = SshConf(str(self.filename))
        except Exception as e:
            raise SSHConfigError(f"Could not load ssh config file {self.filename}: {e}")
            self.conf = None

    def _get_resolved_config(self, host: str) -> Dict[str, str]:
        """Use 'ssh -G' to get the fully resolved configuration for a host, with caching."""
        if host in self._resolved_cache:
            return self._resolved_cache[host]

        try:
            # -G: print resolved configuration for this host
            # -F: use specific config file
            result = subprocess.run(
                ["ssh", "-F", str(self.filename), "-G", host],
                capture_output=True,
                text=True,
                check=False
            )
            if result.returncode == 0:
                config = {}
                for line in result.stdout.splitlines():
                    line = line.strip()
                    if not line:
                        continue
                    # ssh -G output is typically 'option value'
                    parts = line.split(None, 1)
                    if len(parts) == 2:
                        k, v = parts
                        config[k.strip()] = v.strip()
                self._resolved_cache[host] = config
                return config
        except Exception as e:
            logger.error(f"Error resolving config for host {host} via ssh -G: {e}")

        return {}


    def _get_explicit_keys(self, host: str) -> List[str]:
        """Identify all keys explicitly defined in the config file for a host.

        This includes keys defined in the specific host block and the 'Host *' block.
        """
        explicit_keys = set()
        try:
            with open(self.filename, "r") as f:
                lines = f.readlines()
        except Exception as e:
            logger.error(f"Could not read config file for key extraction: {e}")
            return []

        current_hosts = []
        for line in lines:
            line = line.strip()
            if not line or line.startswith("#"):
                continue

            if line.lower().startswith("host "):
                # Start of a new block
                current_hosts = line[5:].strip().split()
            elif current_hosts:
                # Inside a host block
                # Check if this block applies to the target host or is a wildcard
                if any(h.lower() == host.lower() or h == "*" for h in current_hosts):
                    # Extract the key from 'Key Value'
                    parts = line.split(None, 1)
                    if len(parts) == 2:
                        explicit_keys.add(parts[0])

        return list(explicit_keys)

    def get_options(self, host: str) -> str:
        """Get configuration options for a host that are explicitly defined in the config file.

        Args:
            host: the host name.

        Returns:
            str: comma-separated string of options.
        """
        explicit_keys = self._get_explicit_keys(host)
        if not explicit_keys:
            return ""

        # Get fully resolved configuration from ssh -G
        resolved_config = self._get_resolved_config(host)

        # Only keep options that were explicit in the config, and filter out HostName/User
        filtered = {}
        for k in explicit_keys:
            k_lower = k.lower()
            if k_lower not in ["hostname", "user"]:
                # Prefer the resolved value from ssh -G
                # ssh -G output keys are lowercase
                if k_lower in resolved_config:
                    filtered[k] = resolved_config[k_lower]
                else:
                    # Fallback to raw value from config if ssh -G didn't return it
                    # (this is rare for valid options)
                    try:
                        # We can't easily get the raw value without a real parser, 
                        # so we'll just use the key name or skip it.
                        # But for now, let's just use the resolved config.
                        pass
                    except Exception:
                        pass

        if not filtered:
            return ""
        return ", ".join([f"{k}={v}" for k, v in filtered.items()])

    def list(self) -> List[str]:
        """List the hosts defined in the config file.

        Returns:
            List[str]: list of host names.
        """
        hosts = []
        try:
            with open(self.filename, "r") as f:
                for line in f:
                    trimmed = line.strip()
                    if trimmed.lower().startswith("host "):
                        # Extract the host part, handling multiple hosts on one line
                        # e.g., "Host host1 host2"
                        parts = trimmed.split()
                        hosts.extend(parts[1:])
        except Exception as e:
            logger.error(f"Error reading hosts from {self.filename}: {e}")

        return hosts

    def __str__(self) -> str:
        """The string representation of the config as JSON."""
        self._ensure_loaded()
        if not self.conf:
            return "{}"

        # Convert sshconf to a dictionary for JSON representation
        hosts_dict = {}
        try:
            for host in self.conf.hosts():
                hosts_dict[host] = self.conf.get_all(host)
        except Exception as e:
            logger.error(f"Error parsing config for JSON representation: {e}")

        return json.dumps(hosts_dict, indent=4)

    def login(self, name: str):
        """Login to the host defined in .ssh/config by name.

        Args:
            name: the name of the host as defined in the config file.
        """
        logger.info(f"Logging into host: {name}")
        try:
            self._execute(["ssh", name], capture_output=False)
        except Exception as e:
            logger.error(f"Failed to login to {name}: {e}")

    def execute(self, name: str, command: str, use_pty: bool = False) -> Union[CommandResult, str]:
        """Execute the command on the named host.

        Args:
            name: the name of the host in config.
            command: the command to be executed.
            use_pty: whether to allocate a pseudo-terminal.

        Returns:
            Union[CommandResult, str]: CommandResult for remote, stdout for local.
        """
        if name == "localhost":
            # Execute locally
            result = self._execute(["sh", "-c", command])
            return result.stdout if result.stdout else result.stderr

        # Execute via Fabric
        user = self.username(name)
        return self._run_remote(name, command, user=user, use_pty=use_pty)

    def sudo_execute(self, name: str, command: str, use_pty: bool = False) -> CommandResult:
        """Execute the command on the named host with sudo.

        Args:
            name: the name of the host in config.
            command: the command to be executed.
            use_pty: whether to allocate a pseudo-terminal.

        Returns:
            CommandResult: structured result of the execution.
        """
        user = self.username(name)
        return self._run_remote(name, command, user=user, use_sudo=True, use_pty=use_pty)

    def execute_parallel(self, hosts: List[str], command: str) -> Dict[str, CommandResult]:
        """Execute the same command on multiple hosts in parallel.

        Args:
            hosts: list of host names.
            command: the command to execute.

        Returns:
            Dict[str, CommandResult]: mapping of host to its result.
        """
        from concurrent.futures import ThreadPoolExecutor

        results = {}
        with ThreadPoolExecutor() as executor:
            future_to_host = {executor.submit(self.execute, host, command): host for host in hosts}
            for future in future_to_host:
                host = future_to_host[future]
                try:
                    results[host] = future.result()
                except Exception as e:
                    logger.error(f"Parallel execution failed for {host}: {e}")

        return results

    def local(self, command: str) -> str:
        """Execute the command on the localhost.

        Args:
            command: the command to execute.

        Returns:
            str: the output of the command.
        """
        return self.execute("localhost", command)

    def username(self, host: str) -> Optional[str]:
        """Returns the username for a given host, falling back to local user."""
        opts = self._get_resolved_config(host)
        user = opts.get("user", opts.get("User", ""))
        if user:
            return user
        return os.environ.get("USER", "user")
    def hostname(self, host: str) -> str:
        """Returns the actual HostName for the given host."""
        opts = self._get_resolved_config(host)
        hostname = opts.get("hostname", opts.get("HostName", ""))
        return hostname if hostname else host

    def yaml(self) -> str:
        """Returns the parsed SSH configuration in YAML format.

        Returns:
            A YAML string representation of the parsed hosts dictionary.
        """
        if not self.conf:
            return "{}"

        hosts_dict = {}
        for host in self.conf.hosts():
            hosts_dict[host] = self.conf.get_all(host)
        return yaml.dump(hosts_dict, default_flow_style=False)

    def get_tunnels(self) -> List[Dict]:
        """Extract all tunnel forwards (LocalForward, RemoteForward) from the config.

        Returns:
            List[Dict]: A list of tunnel definitions.
        """
        tunnels = []
        if not self.conf:
            return tunnels

        try:
            hosts = self.conf.hosts()
        except Exception as e:
            logger.error(f"Error parsing hosts for tunnels from {self.filename}: {e}")
            return tunnels

        for host in hosts:
            try:
                # sshconf stores forwards in the config dict for the host
                all_conf = self.conf.get_all(host)

                # LocalForward and RemoteForward can be lists or single strings
                for key in ['localforward', 'remoteforward']:
                    forward = all_conf.get(key)
                    if not forward:
                        continue

                    forwards = forward if isinstance(forward, list) else [forward]
                    for f in forwards:
                        # Forward format: "local_port remote_host:remote_port"
                        parts = f.split()
                        if len(parts) >= 2:
                            local_port = parts[0]
                            remote_target = parts[1]
                            tunnels.append({
                                "host": host,
                                "type": "Local" if key == 'localforward' else "Remote",
                                "local_port": local_port,
                                "remote_target": remote_target
                            })
            except Exception as e:
                logger.warn(f"Skipping host {host} due to parsing error: {e}")
                continue
        return tunnels

    def delete(self, name: str):
        """Removes a host entry from the SSH config file.

        Args:
            name: the name of the host to remove.
        """
        if not self.conf:
            return

        try:
            self.conf.remove(name)
            self.conf.save()
        except Exception as e:
            raise SSHConfigError(f"Failed to delete host {name} from {self.filename}: {e}")

    def generate(
        self,
        host: str,
        hostname: str,
        identity: Optional[str] = None,
        user: Optional[str] = None,
        verbose: bool = False,
    ):
        """Adds a host to the config file with given parameters.

        Args:
            host: the alias for the host.
            hostname: the actual hostname or IP.
            identity: the path to the identity file.
            user: the username for the host.
            verbose: prints debug messages.
        """
        if not self.conf:
            return

    def _friendly_error(self, e: Exception) -> str:
        """Convert technical exceptions into user-friendly error messages."""
        msg = str(e)
        if "not enough values to unpack" in msg:
            return "Configuration syntax error: a line in this block is missing a required value (expected 'Key Value' format)."
        return msg


    def check(self) -> List[Dict]:
        """Check the SSH config file for malformed entries.

        Returns:
            List[Dict]: A list of errors found. Each error contains 'line' and 'message'.
        """
        errors = []
        if not self.filename.exists():
            return [{"line": 0, "message": f"Config file not found: {self.filename}"}]

        # Deep dive: isolate blocks to find the culprit
        try:
            with open(self.filename, "r") as f:
                lines = f.readlines()
        except Exception as e:
            return [{"line": 0, "message": f"Could not read file: {e}"}]

        current_host = None
        current_block = []
        start_line = 0

        for i, line in enumerate(lines, 1):
            trimmed = line.strip()
            if not trimmed or trimmed.startswith("#"):
                continue

            if trimmed.lower().startswith("host "):
                # Process previous block
                if current_host:
                    block_res = self._validate_block(current_host, current_block)
                    if not block_res["valid"]:
                        errors.append({"line": start_line, "message": f"Host {current_host}: {block_res['error']}"})

                current_host = trimmed[5:].strip()
                current_block = [line]
                start_line = i
            elif current_host:
                current_block.append(line)
            else:
                # Line before any Host block
                if trimmed:
                    errors.append({"line": i, "message": f"Global configuration line (applies to all hosts): {trimmed}"})

        # Process the last block
        if current_host:
            block_res = self._validate_block(current_host, current_block)
            if not block_res["valid"]:
                errors.append({"line": start_line, "message": f"Host {current_host}: {block_res['error']}"})

        return errors

    def _validate_block(self, host: str, block_lines: List[str]) -> Dict:
        """Validate a single host block by using the actual ssh binary.

        This is 100% compatible with OpenSSH syntax as it uses the real parser.
        """
        import tempfile
        import os
        import subprocess

        try:
            with tempfile.NamedTemporaryFile(mode='w', delete=False) as tmp:
                tmp.writelines(block_lines)
                tmp_path = tmp.name

            try:
                # -F: use specific config file
                # -G: print configuration for this host (parses the config)
                # We use 'localhost' as a dummy host because ssh -G will parse the 
                # entire config file and fail if there are syntax errors, regardless 
                # of the host provided.
                result = subprocess.run(
                    ["ssh", "-F", tmp_path, "-G", "localhost"],
                    capture_output=True,
                    text=True,
                    check=False
                )

                if result.returncode == 0:
                    return {"valid": True}
                else:
                    # Use stderr for the error message, fallback to stdout
                    error_msg = result.stderr.strip() or result.stdout.strip() or "Unknown SSH configuration error"
                    return {"valid": False, "error": error_msg}

            finally:
                if os.path.exists(tmp_path):
                    os.remove(tmp_path)
        except Exception as e:
            return {"valid": False, "error": f"Internal validation error: {e}"}

Methods:

__str__

__str__()

The string representation of the config as JSON.

Source code in src/cloudmesh/ai/ssh/ssh_config.py

def __str__(self) -> str:
    """The string representation of the config as JSON."""
    self._ensure_loaded()
    if not self.conf:
        return "{}"

    # Convert sshconf to a dictionary for JSON representation
    hosts_dict = {}
    try:
        for host in self.conf.hosts():
            hosts_dict[host] = self.conf.get_all(host)
    except Exception as e:
        logger.error(f"Error parsing config for JSON representation: {e}")

    return json.dumps(hosts_dict, indent=4)

check

check()

Check the SSH config file for malformed entries.

Returns:

Type	Description
List[Dict]	List[Dict]: A list of errors found. Each error contains 'line' and 'message'.

Source code in src/cloudmesh/ai/ssh/ssh_config.py

def check(self) -> List[Dict]:
    """Check the SSH config file for malformed entries.

    Returns:
        List[Dict]: A list of errors found. Each error contains 'line' and 'message'.
    """
    errors = []
    if not self.filename.exists():
        return [{"line": 0, "message": f"Config file not found: {self.filename}"}]

    # Deep dive: isolate blocks to find the culprit
    try:
        with open(self.filename, "r") as f:
            lines = f.readlines()
    except Exception as e:
        return [{"line": 0, "message": f"Could not read file: {e}"}]

    current_host = None
    current_block = []
    start_line = 0

    for i, line in enumerate(lines, 1):
        trimmed = line.strip()
        if not trimmed or trimmed.startswith("#"):
            continue

        if trimmed.lower().startswith("host "):
            # Process previous block
            if current_host:
                block_res = self._validate_block(current_host, current_block)
                if not block_res["valid"]:
                    errors.append({"line": start_line, "message": f"Host {current_host}: {block_res['error']}"})

            current_host = trimmed[5:].strip()
            current_block = [line]
            start_line = i
        elif current_host:
            current_block.append(line)
        else:
            # Line before any Host block
            if trimmed:
                errors.append({"line": i, "message": f"Global configuration line (applies to all hosts): {trimmed}"})

    # Process the last block
    if current_host:
        block_res = self._validate_block(current_host, current_block)
        if not block_res["valid"]:
            errors.append({"line": start_line, "message": f"Host {current_host}: {block_res['error']}"})

    return errors

delete

delete(name)

Removes a host entry from the SSH config file.

Parameters:

Name	Type	Description	Default
name	str	the name of the host to remove.	required

Source code in src/cloudmesh/ai/ssh/ssh_config.py

def delete(self, name: str):
    """Removes a host entry from the SSH config file.

    Args:
        name: the name of the host to remove.
    """
    if not self.conf:
        return

    try:
        self.conf.remove(name)
        self.conf.save()
    except Exception as e:
        raise SSHConfigError(f"Failed to delete host {name} from {self.filename}: {e}")

execute

execute(name, command, use_pty=False)

Execute the command on the named host.

Parameters:

Name	Type	Description	Default
name	str	the name of the host in config.	required
command	str	the command to be executed.	required
use_pty	bool	whether to allocate a pseudo-terminal.	False

Returns:

Type	Description
Union[CommandResult, str]	Union[CommandResult, str]: CommandResult for remote, stdout for local.

Source code in src/cloudmesh/ai/ssh/ssh_config.py

def execute(self, name: str, command: str, use_pty: bool = False) -> Union[CommandResult, str]:
    """Execute the command on the named host.

    Args:
        name: the name of the host in config.
        command: the command to be executed.
        use_pty: whether to allocate a pseudo-terminal.

    Returns:
        Union[CommandResult, str]: CommandResult for remote, stdout for local.
    """
    if name == "localhost":
        # Execute locally
        result = self._execute(["sh", "-c", command])
        return result.stdout if result.stdout else result.stderr

    # Execute via Fabric
    user = self.username(name)
    return self._run_remote(name, command, user=user, use_pty=use_pty)

execute_parallel

execute_parallel(hosts, command)

Execute the same command on multiple hosts in parallel.

Parameters:

Name	Type	Description	Default
hosts	List[str]	list of host names.	required
command	str	the command to execute.	required

Returns:

Type	Description
Dict[str, CommandResult]	Dict[str, CommandResult]: mapping of host to its result.

Source code in src/cloudmesh/ai/ssh/ssh_config.py

def execute_parallel(self, hosts: List[str], command: str) -> Dict[str, CommandResult]:
    """Execute the same command on multiple hosts in parallel.

    Args:
        hosts: list of host names.
        command: the command to execute.

    Returns:
        Dict[str, CommandResult]: mapping of host to its result.
    """
    from concurrent.futures import ThreadPoolExecutor

    results = {}
    with ThreadPoolExecutor() as executor:
        future_to_host = {executor.submit(self.execute, host, command): host for host in hosts}
        for future in future_to_host:
            host = future_to_host[future]
            try:
                results[host] = future.result()
            except Exception as e:
                logger.error(f"Parallel execution failed for {host}: {e}")

    return results

generate

generate(host, hostname, identity=None, user=None, verbose=False)

Adds a host to the config file with given parameters.

Parameters:

Name	Type	Description	Default
host	str	the alias for the host.	required
hostname	str	the actual hostname or IP.	required
identity	Optional[str]	the path to the identity file.	None
user	Optional[str]	the username for the host.	None
verbose	bool	prints debug messages.	False

Source code in src/cloudmesh/ai/ssh/ssh_config.py

def generate(
    self,
    host: str,
    hostname: str,
    identity: Optional[str] = None,
    user: Optional[str] = None,
    verbose: bool = False,
):
    """Adds a host to the config file with given parameters.

    Args:
        host: the alias for the host.
        hostname: the actual hostname or IP.
        identity: the path to the identity file.
        user: the username for the host.
        verbose: prints debug messages.
    """
    if not self.conf:
        return

get_content

get_content()

Return the raw content of the SSH config file.

Source code in src/cloudmesh/ai/ssh/ssh_config.py

def get_content(self) -> str:
    """Return the raw content of the SSH config file."""
    try:
        with open(self.filename, "r") as f:
            return f.read()
    except Exception as e:
        logger.error(f"Could not read config file {self.filename}: {e}")
        return ""

    self.load()

get_options

get_options(host)

Get configuration options for a host that are explicitly defined in the config file.

Parameters:

Name	Type	Description	Default
host	str	the host name.	required

Returns:

Name	Type	Description
str	str	comma-separated string of options.

Source code in src/cloudmesh/ai/ssh/ssh_config.py

def get_options(self, host: str) -> str:
    """Get configuration options for a host that are explicitly defined in the config file.

    Args:
        host: the host name.

    Returns:
        str: comma-separated string of options.
    """
    explicit_keys = self._get_explicit_keys(host)
    if not explicit_keys:
        return ""

    # Get fully resolved configuration from ssh -G
    resolved_config = self._get_resolved_config(host)

    # Only keep options that were explicit in the config, and filter out HostName/User
    filtered = {}
    for k in explicit_keys:
        k_lower = k.lower()
        if k_lower not in ["hostname", "user"]:
            # Prefer the resolved value from ssh -G
            # ssh -G output keys are lowercase
            if k_lower in resolved_config:
                filtered[k] = resolved_config[k_lower]
            else:
                # Fallback to raw value from config if ssh -G didn't return it
                # (this is rare for valid options)
                try:
                    # We can't easily get the raw value without a real parser, 
                    # so we'll just use the key name or skip it.
                    # But for now, let's just use the resolved config.
                    pass
                except Exception:
                    pass

    if not filtered:
        return ""
    return ", ".join([f"{k}={v}" for k, v in filtered.items()])

get_tunnels

get_tunnels()

Extract all tunnel forwards (LocalForward, RemoteForward) from the config.

Returns:

Type	Description
List[Dict]	List[Dict]: A list of tunnel definitions.

Source code in src/cloudmesh/ai/ssh/ssh_config.py

def get_tunnels(self) -> List[Dict]:
    """Extract all tunnel forwards (LocalForward, RemoteForward) from the config.

    Returns:
        List[Dict]: A list of tunnel definitions.
    """
    tunnels = []
    if not self.conf:
        return tunnels

    try:
        hosts = self.conf.hosts()
    except Exception as e:
        logger.error(f"Error parsing hosts for tunnels from {self.filename}: {e}")
        return tunnels

    for host in hosts:
        try:
            # sshconf stores forwards in the config dict for the host
            all_conf = self.conf.get_all(host)

            # LocalForward and RemoteForward can be lists or single strings
            for key in ['localforward', 'remoteforward']:
                forward = all_conf.get(key)
                if not forward:
                    continue

                forwards = forward if isinstance(forward, list) else [forward]
                for f in forwards:
                    # Forward format: "local_port remote_host:remote_port"
                    parts = f.split()
                    if len(parts) >= 2:
                        local_port = parts[0]
                        remote_target = parts[1]
                        tunnels.append({
                            "host": host,
                            "type": "Local" if key == 'localforward' else "Remote",
                            "local_port": local_port,
                            "remote_target": remote_target
                        })
        except Exception as e:
            logger.warn(f"Skipping host {host} due to parsing error: {e}")
            continue
    return tunnels

hostname

hostname(host)

Returns the actual HostName for the given host.

Source code in src/cloudmesh/ai/ssh/ssh_config.py

def hostname(self, host: str) -> str:
    """Returns the actual HostName for the given host."""
    opts = self._get_resolved_config(host)
    hostname = opts.get("hostname", opts.get("HostName", ""))
    return hostname if hostname else host

list

list()

List the hosts defined in the config file.

Returns:

Type	Description
List[str]	List[str]: list of host names.

Source code in src/cloudmesh/ai/ssh/ssh_config.py

def list(self) -> List[str]:
    """List the hosts defined in the config file.

    Returns:
        List[str]: list of host names.
    """
    hosts = []
    try:
        with open(self.filename, "r") as f:
            for line in f:
                trimmed = line.strip()
                if trimmed.lower().startswith("host "):
                    # Extract the host part, handling multiple hosts on one line
                    # e.g., "Host host1 host2"
                    parts = trimmed.split()
                    hosts.extend(parts[1:])
    except Exception as e:
        logger.error(f"Error reading hosts from {self.filename}: {e}")

    return hosts

load

load()

Parse the SSH config file using sshconf.

Source code in src/cloudmesh/ai/ssh/ssh_config.py

def load(self):
    """Parse the SSH config file using sshconf."""
    try:
        self.conf = SshConf(str(self.filename))
    except Exception as e:
        raise SSHConfigError(f"Could not load ssh config file {self.filename}: {e}")
        self.conf = None

local

local(command)

Execute the command on the localhost.

Parameters:

Name	Type	Description	Default
command	str	the command to execute.	required

Returns:

Name	Type	Description
str	str	the output of the command.

Source code in src/cloudmesh/ai/ssh/ssh_config.py

def local(self, command: str) -> str:
    """Execute the command on the localhost.

    Args:
        command: the command to execute.

    Returns:
        str: the output of the command.
    """
    return self.execute("localhost", command)

login

login(name)

Login to the host defined in .ssh/config by name.

Parameters:

Name	Type	Description	Default
name	str	the name of the host as defined in the config file.	required

Source code in src/cloudmesh/ai/ssh/ssh_config.py

def login(self, name: str):
    """Login to the host defined in .ssh/config by name.

    Args:
        name: the name of the host as defined in the config file.
    """
    logger.info(f"Logging into host: {name}")
    try:
        self._execute(["ssh", name], capture_output=False)
    except Exception as e:
        logger.error(f"Failed to login to {name}: {e}")

names

names()

The names defined in the SSH config.

Returns:

Type	Description
List[str]	List[str]: the host names.

Source code in src/cloudmesh/ai/ssh/ssh_config.py

def names(self) -> List[str]:
    """The names defined in the SSH config.

    Returns:
        List[str]: the host names.
    """
    return self.list()

sudo_execute

sudo_execute(name, command, use_pty=False)

Execute the command on the named host with sudo.

Parameters:

Name	Type	Description	Default
name	str	the name of the host in config.	required
command	str	the command to be executed.	required
use_pty	bool	whether to allocate a pseudo-terminal.	False

Returns:

Name	Type	Description
CommandResult	CommandResult	structured result of the execution.

Source code in src/cloudmesh/ai/ssh/ssh_config.py

def sudo_execute(self, name: str, command: str, use_pty: bool = False) -> CommandResult:
    """Execute the command on the named host with sudo.

    Args:
        name: the name of the host in config.
        command: the command to be executed.
        use_pty: whether to allocate a pseudo-terminal.

    Returns:
        CommandResult: structured result of the execution.
    """
    user = self.username(name)
    return self._run_remote(name, command, user=user, use_sudo=True, use_pty=use_pty)

username

username(host)

Returns the username for a given host, falling back to local user.

Source code in src/cloudmesh/ai/ssh/ssh_config.py

def username(self, host: str) -> Optional[str]:
    """Returns the username for a given host, falling back to local user."""
    opts = self._get_resolved_config(host)
    user = opts.get("user", opts.get("User", ""))
    if user:
        return user
    return os.environ.get("USER", "user")

yaml

yaml()

Returns the parsed SSH configuration in YAML format.

Returns:

Type	Description
str	A YAML string representation of the parsed hosts dictionary.

Source code in src/cloudmesh/ai/ssh/ssh_config.py

def yaml(self) -> str:
    """Returns the parsed SSH configuration in YAML format.

    Returns:
        A YAML string representation of the parsed hosts dictionary.
    """
    if not self.conf:
        return "{}"

    hosts_dict = {}
    for host in self.conf.hosts():
        hosts_dict[host] = self.conf.get_all(host)
    return yaml.dump(hosts_dict, default_flow_style=False)

SSHTunnelError

Bases: SSHError

Raised when an SSH tunnel fails to start, stop, or maintain connectivity.

Source code in src/cloudmesh/ai/ssh/exceptions.py

class SSHTunnelError(SSHError):
    """Raised when an SSH tunnel fails to start, stop, or maintain connectivity."""
    pass

Telemetry

Handles structured telemetry emission for AI commands. Supports multiple backends for flexible data export.

Source code in cloudmesh/ai/common/telemetry.py

class Telemetry:
    """
    Handles structured telemetry emission for AI commands.
    Supports multiple backends for flexible data export.
    """

    def __init__(
        self, 
        command_name: str, 
        telemetry_file: Optional[Union[str, Path]] = None,
        backends: Optional[List[TelemetryBackend]] = None
    ) -> None:
        """Initialize the Telemetry collector.

        Args:
            command_name: Name of the command emitting telemetry.
            telemetry_file: Backward compatibility: path to a JSONL file.
            backends: List of TelemetryBackend implementations to use.

        """
        self.command_name = command_name
        self.logger = ai_log.get_logger(f"{command_name}.telemetry")
        self.backends: List[TelemetryBackend] = backends or []

        # Maintain backward compatibility with telemetry_file
        if telemetry_file:
            self.backends.append(JSONFileBackend(telemetry_file))

    def _get_system_context(self) -> Dict[str, Any]:
        """Gathers basic system context to accompany telemetry metrics.

        Returns:
            A dictionary containing basic system information.
        """
        info = ai_sys.systeminfo()
        return {
            "cpu": info.get("cpu"),
            "gpu_present": info.get("gpu.present"),
            "gpu_model": info.get("gpu.model"),
            "memory_total": info.get("memory.total"),
        }

    def emit(
        self, 
        status: str, 
        metrics: Optional[Dict[str, Any]] = None, 
        message: Optional[str] = None,
        stdout: bool = False,
        **kwargs: Any
    ) -> None:
        """Emits a structured telemetry record to all configured backends.

        Args:
            status: The current status of the command (e.g., 'started', 'completed', 'failed').
            metrics: A dictionary of KPIs and measurements.
            message: An optional human-readable message.
            stdout: If True, prints the JSON record to stdout.
        """
        if os.environ.get("CLOUDMESH_AI_TELEMETRY_DISABLED", "").lower() in ("1", "true", "yes"):
            return
        all_metrics = (metrics or {}).copy()
        all_metrics.update(kwargs)
        record = {
            "timestamp": datetime.now().isoformat(),
            "command": self.command_name,
            "status": status,
            "metrics": all_metrics,
            "system": self._get_system_context(),
        }
        if message:
            record["message"] = message

        json_record = json.dumps(record)

        # 1. Log via the standard logging system
        self.logger.info(f"TELEMETRY: {json_record}")

        # 2. Emit to all configured backends
        for backend in self.backends:
            backend.emit(record)

        # 3. Optional stdout for direct ingestion/piping
        if stdout:
            print(json_record, file=sys.stdout)

    def start(self, message: Optional[str] = None, **kwargs: Any) -> None:
        """Helper to emit a 'started' status.

        Args:
            message: Optional message for the start event.
            **kwargs: Additional metrics to include in the record.

        """
        self.emit("started", message=message, **kwargs)

    def complete(self, metrics: Optional[Dict[str, Any]] = None, message: Optional[str] = "Command completed successfully", **kwargs: Any) -> None:
        """Helper to emit a 'completed' status with final metrics.

        Args:
            metrics: A dictionary of final KPIs and measurements.
            message: An optional human-readable message.
            **kwargs: Additional metrics to include in the record.

        """
        self.emit("completed", metrics=metrics, message=message, **kwargs)

    def fail(self, error: str, metrics: Optional[Dict[str, Any]] = None, **kwargs: Any) -> None:
        """Helper to emit a 'failed' status.

        Args:
            error: The error message to record.
            metrics: A dictionary of KPIs at the time of failure.
            **kwargs: Additional metrics to include in the record.

        """
        self.emit("failed", metrics=metrics, message=error, **kwargs)

    @contextmanager
    def track(self, message: Optional[str] = None, metrics: Optional[Dict[str, Any]] = None):
        """
        Context manager to automatically track the start and completion of a task.

        Args:
            message: Optional message for the start event.
            metrics: Initial metrics to include.
        """
        self.start(message=message, **(metrics or {}))
        start_time = time.perf_counter()
        try:
            yield
            duration = time.perf_counter() - start_time
            final_metrics = (metrics or {}).copy()
            final_metrics["duration_sec"] = duration
            self.complete(metrics=final_metrics)
        except Exception as e:
            duration = time.perf_counter() - start_time
            final_metrics = (metrics or {}).copy()
            final_metrics["duration_sec"] = duration
            self.fail(error=str(e), metrics=final_metrics)
            raise

Methods:

__init__

__init__(command_name, telemetry_file=None, backends=None)

Initialize the Telemetry collector.

Parameters:

Name	Type	Description	Default
command_name	str	Name of the command emitting telemetry.	required
telemetry_file	Optional[Union[str, Path]]	Backward compatibility: path to a JSONL file.	None
backends	Optional[List[TelemetryBackend]]	List of TelemetryBackend implementations to use.	None

Source code in cloudmesh/ai/common/telemetry.py

def __init__(
    self, 
    command_name: str, 
    telemetry_file: Optional[Union[str, Path]] = None,
    backends: Optional[List[TelemetryBackend]] = None
) -> None:
    """Initialize the Telemetry collector.

    Args:
        command_name: Name of the command emitting telemetry.
        telemetry_file: Backward compatibility: path to a JSONL file.
        backends: List of TelemetryBackend implementations to use.

    """
    self.command_name = command_name
    self.logger = ai_log.get_logger(f"{command_name}.telemetry")
    self.backends: List[TelemetryBackend] = backends or []

    # Maintain backward compatibility with telemetry_file
    if telemetry_file:
        self.backends.append(JSONFileBackend(telemetry_file))

complete

complete(metrics=None, message='Command completed successfully', **kwargs)

Helper to emit a 'completed' status with final metrics.

Parameters:

Name	Type	Description	Default
metrics	Optional[Dict[str, Any]]	A dictionary of final KPIs and measurements.	None
message	Optional[str]	An optional human-readable message.	'Command completed successfully'
**kwargs	Any	Additional metrics to include in the record.	{}

Source code in cloudmesh/ai/common/telemetry.py

def complete(self, metrics: Optional[Dict[str, Any]] = None, message: Optional[str] = "Command completed successfully", **kwargs: Any) -> None:
    """Helper to emit a 'completed' status with final metrics.

    Args:
        metrics: A dictionary of final KPIs and measurements.
        message: An optional human-readable message.
        **kwargs: Additional metrics to include in the record.

    """
    self.emit("completed", metrics=metrics, message=message, **kwargs)

emit

emit(status, metrics=None, message=None, stdout=False, **kwargs)

Emits a structured telemetry record to all configured backends.

Parameters:

Name	Type	Description	Default
status	str	The current status of the command (e.g., 'started', 'completed', 'failed').	required
metrics	Optional[Dict[str, Any]]	A dictionary of KPIs and measurements.	None
message	Optional[str]	An optional human-readable message.	None
stdout	bool	If True, prints the JSON record to stdout.	False

Source code in cloudmesh/ai/common/telemetry.py

def emit(
    self, 
    status: str, 
    metrics: Optional[Dict[str, Any]] = None, 
    message: Optional[str] = None,
    stdout: bool = False,
    **kwargs: Any
) -> None:
    """Emits a structured telemetry record to all configured backends.

    Args:
        status: The current status of the command (e.g., 'started', 'completed', 'failed').
        metrics: A dictionary of KPIs and measurements.
        message: An optional human-readable message.
        stdout: If True, prints the JSON record to stdout.
    """
    if os.environ.get("CLOUDMESH_AI_TELEMETRY_DISABLED", "").lower() in ("1", "true", "yes"):
        return
    all_metrics = (metrics or {}).copy()
    all_metrics.update(kwargs)
    record = {
        "timestamp": datetime.now().isoformat(),
        "command": self.command_name,
        "status": status,
        "metrics": all_metrics,
        "system": self._get_system_context(),
    }
    if message:
        record["message"] = message

    json_record = json.dumps(record)

    # 1. Log via the standard logging system
    self.logger.info(f"TELEMETRY: {json_record}")

    # 2. Emit to all configured backends
    for backend in self.backends:
        backend.emit(record)

    # 3. Optional stdout for direct ingestion/piping
    if stdout:
        print(json_record, file=sys.stdout)

fail

fail(error, metrics=None, **kwargs)

Helper to emit a 'failed' status.

Parameters:

Name	Type	Description	Default
error	str	The error message to record.	required
metrics	Optional[Dict[str, Any]]	A dictionary of KPIs at the time of failure.	None
**kwargs	Any	Additional metrics to include in the record.	{}

Source code in cloudmesh/ai/common/telemetry.py

def fail(self, error: str, metrics: Optional[Dict[str, Any]] = None, **kwargs: Any) -> None:
    """Helper to emit a 'failed' status.

    Args:
        error: The error message to record.
        metrics: A dictionary of KPIs at the time of failure.
        **kwargs: Additional metrics to include in the record.

    """
    self.emit("failed", metrics=metrics, message=error, **kwargs)

start

start(message=None, **kwargs)

Helper to emit a 'started' status.

Parameters:

Name	Type	Description	Default
message	Optional[str]	Optional message for the start event.	None
**kwargs	Any	Additional metrics to include in the record.	{}

Source code in cloudmesh/ai/common/telemetry.py

def start(self, message: Optional[str] = None, **kwargs: Any) -> None:
    """Helper to emit a 'started' status.

    Args:
        message: Optional message for the start event.
        **kwargs: Additional metrics to include in the record.

    """
    self.emit("started", message=message, **kwargs)

track

track(message=None, metrics=None)

Context manager to automatically track the start and completion of a task.

Parameters:

Name	Type	Description	Default
message	Optional[str]	Optional message for the start event.	None
metrics	Optional[Dict[str, Any]]	Initial metrics to include.	None

Source code in cloudmesh/ai/common/telemetry.py

@contextmanager
def track(self, message: Optional[str] = None, metrics: Optional[Dict[str, Any]] = None):
    """
    Context manager to automatically track the start and completion of a task.

    Args:
        message: Optional message for the start event.
        metrics: Initial metrics to include.
    """
    self.start(message=message, **(metrics or {}))
    start_time = time.perf_counter()
    try:
        yield
        duration = time.perf_counter() - start_time
        final_metrics = (metrics or {}).copy()
        final_metrics["duration_sec"] = duration
        self.complete(metrics=final_metrics)
    except Exception as e:
        duration = time.perf_counter() - start_time
        final_metrics = (metrics or {}).copy()
        final_metrics["duration_sec"] = duration
        self.fail(error=str(e), metrics=final_metrics)
        raise

Tunnel

Bases: SSHBase

Manages an SSH tunnel for port forwarding.

Source code in src/cloudmesh/ai/ssh/tunnel.py

class Tunnel(SSHBase):
    """Manages an SSH tunnel for port forwarding."""

    def __init__(
        self, 
        local_port: int, 
        remote_host: str, 
        remote_port: int, 
        ssh_host: str, 
        ssh_user: Optional[str] = None, 
        identity_file: Optional[Union[str, Path]] = None, 
        extra_args: Optional[List[str]] = None,
        debug: bool = False
    ):
        super().__init__(debug=debug)
        self.local_port = local_port
        self.remote_host = remote_host
        self.remote_port = remote_port
        self.ssh_host = ssh_host
        self.ssh_user = ssh_user
        self.identity_file = identity_file
        self.extra_args = extra_args or []
        self.process = None

    def _get_resolved_ssh_host(self) -> str:
        """Resolve ssh_host using SSHConfig if it's an alias."""
        cfg = SSHConfig()
        # If ssh_user is not provided, try to get it from config
        user = self.ssh_user or cfg.username(self.ssh_host)
        hostname = cfg.hostname(self.ssh_host)

        if user:
            return f"{user}@{hostname}"
        return hostname

    def start(self, timeout: int = 10) -> bool:
        """Starts the SSH tunnel in the background.

        Args:
            timeout: seconds to wait for the port to open.
        """
        if self.process and self.process.poll() is None:
            console.warn(f"Tunnel for port {self.local_port} is already running.")
            return True

        try:
            ssh_host = self._get_resolved_ssh_host()

            # -L local_port:remote_host:remote_port ssh_host -N
            # -N tells SSH not to execute a remote command.
            # ExitOnForwardFailure ensures the process exits if port forwarding fails.
            cmd = [
                "ssh",
                "-o", "ExitOnForwardFailure=yes",
                "-L", f"{self.local_port}:{self.remote_host}:{self.remote_port}",
            ]

            if self.identity_file:
                cmd.extend(["-i", self.identity_file])

            cmd.extend(self.extra_args)
            cmd.append(ssh_host)
            cmd.append("-N")

            if self.debug:
                console.debug(f"Executing: {' '.join(cmd)}")

            self.process = subprocess.Popen(
                cmd, 
                stdout=subprocess.DEVNULL, 
                stderr=subprocess.PIPE, 
                text=True,
                preexec_fn=os.setpgrp
            )

            # Wait for the port to become active
            if self._wait_for_port(timeout):
                console.ok(f"SSH tunnel established: localhost:{self.local_port} -> {self.remote_host}:{self.remote_port} via {self.ssh_host}")
                return True
            else:
                # If it didn't open, check for errors in stderr
                stderr = self.process.stderr.read() if self.process.stderr else "No stderr available"
                console.error(f"SSH tunnel failed to open port {self.local_port} within {timeout}s.")
                if stderr:
                    console.error(f"SSH Error: {stderr.strip()}")
                self.stop()
                return False

        except Exception as e:
            raise SSHTunnelError(f"Failed to start SSH tunnel: {e}")
            self.stop()
            return False

    def _wait_for_port(self, timeout: int) -> bool:
        """Polls the local port until it's open or timeout is reached."""
        start_time = time.time()
        while time.time() - start_time < timeout:
            if self.is_port_open("localhost", self.local_port):
                return True
            if self.process and self.process.poll() is not None:
                return False
            time.sleep(0.5)
        return False

    def stop(self):
        """Stops the SSH tunnel process."""
        if not self.process or self.process.poll() is not None:
            return False

        try:
            os.killpg(os.getpgid(self.process.pid), signal.SIGTERM)
            self.process = None
            console.ok(f"SSH tunnel on port {self.local_port} stopped.")
            return True
        except Exception as e:
            raise SSHTunnelError(f"Failed to stop SSH tunnel: {e}")
            return False

    def force_stop(self) -> bool:
        """Forcefully terminates any process bound to the local port using lsof.

        This is useful for cleaning up zombie tunnels that were not started by this 
        specific Tunnel instance.
        """
        if not self.is_port_open("localhost", self.local_port):
            console.debug(f"No active tunnel found on port {self.local_port}.")
            return False

        try:
            # -t: terse output (only PID), -i: port
            result = subprocess.run(
                ["lsof", "-t", f"-i:{self.local_port}"],
                capture_output=True,
                text=True,
                check=True
            )
            pids = result.stdout.strip().split()
            if pids:
                console.warn(f"Forcefully terminating process(es) {', '.join(pids)} on port {self.local_port}...")
                subprocess.run(["kill", "-9"] + pids, check=True)
                self.process = None
                console.ok(f"Successfully terminated tunnel process(es) on port {self.local_port}.")
                return True
        except subprocess.CalledProcessError:
            # lsof returns non-zero if no process is found
            console.debug(f"No process found on port {self.local_port} by lsof.")
        except Exception as e:
            raise SSHTunnelError(f"Error during force_stop on port {self.local_port}: {e}")

        return False


    def is_active(self) -> bool:
        """Checks if the tunnel process is still running and port is open."""
        return self.process is not None and self.process.poll() is None and self.is_port_open("localhost", self.local_port)

    def __enter__(self):
        """Context manager enter."""
        if self.start():
            return self
        raise RuntimeError(f"Could not start SSH tunnel on port {self.local_port}")

    def __exit__(self, exc_type, exc_val, exc_tb):
        """Context manager exit."""
        self.stop()

Methods:

__enter__

__enter__()

Context manager enter.

Source code in src/cloudmesh/ai/ssh/tunnel.py

def __enter__(self):
    """Context manager enter."""
    if self.start():
        return self
    raise RuntimeError(f"Could not start SSH tunnel on port {self.local_port}")

__exit__

__exit__(exc_type, exc_val, exc_tb)

Context manager exit.

Source code in src/cloudmesh/ai/ssh/tunnel.py

def __exit__(self, exc_type, exc_val, exc_tb):
    """Context manager exit."""
    self.stop()

force_stop

force_stop()

Forcefully terminates any process bound to the local port using lsof.

This is useful for cleaning up zombie tunnels that were not started by this specific Tunnel instance.

Source code in src/cloudmesh/ai/ssh/tunnel.py

def force_stop(self) -> bool:
    """Forcefully terminates any process bound to the local port using lsof.

    This is useful for cleaning up zombie tunnels that were not started by this 
    specific Tunnel instance.
    """
    if not self.is_port_open("localhost", self.local_port):
        console.debug(f"No active tunnel found on port {self.local_port}.")
        return False

    try:
        # -t: terse output (only PID), -i: port
        result = subprocess.run(
            ["lsof", "-t", f"-i:{self.local_port}"],
            capture_output=True,
            text=True,
            check=True
        )
        pids = result.stdout.strip().split()
        if pids:
            console.warn(f"Forcefully terminating process(es) {', '.join(pids)} on port {self.local_port}...")
            subprocess.run(["kill", "-9"] + pids, check=True)
            self.process = None
            console.ok(f"Successfully terminated tunnel process(es) on port {self.local_port}.")
            return True
    except subprocess.CalledProcessError:
        # lsof returns non-zero if no process is found
        console.debug(f"No process found on port {self.local_port} by lsof.")
    except Exception as e:
        raise SSHTunnelError(f"Error during force_stop on port {self.local_port}: {e}")

    return False

is_active

is_active()

Checks if the tunnel process is still running and port is open.

Source code in src/cloudmesh/ai/ssh/tunnel.py

def is_active(self) -> bool:
    """Checks if the tunnel process is still running and port is open."""
    return self.process is not None and self.process.poll() is None and self.is_port_open("localhost", self.local_port)

start

start(timeout=10)

Starts the SSH tunnel in the background.

Parameters:

Name	Type	Description	Default
timeout	int	seconds to wait for the port to open.	10

Source code in src/cloudmesh/ai/ssh/tunnel.py

def start(self, timeout: int = 10) -> bool:
    """Starts the SSH tunnel in the background.

    Args:
        timeout: seconds to wait for the port to open.
    """
    if self.process and self.process.poll() is None:
        console.warn(f"Tunnel for port {self.local_port} is already running.")
        return True

    try:
        ssh_host = self._get_resolved_ssh_host()

        # -L local_port:remote_host:remote_port ssh_host -N
        # -N tells SSH not to execute a remote command.
        # ExitOnForwardFailure ensures the process exits if port forwarding fails.
        cmd = [
            "ssh",
            "-o", "ExitOnForwardFailure=yes",
            "-L", f"{self.local_port}:{self.remote_host}:{self.remote_port}",
        ]

        if self.identity_file:
            cmd.extend(["-i", self.identity_file])

        cmd.extend(self.extra_args)
        cmd.append(ssh_host)
        cmd.append("-N")

        if self.debug:
            console.debug(f"Executing: {' '.join(cmd)}")

        self.process = subprocess.Popen(
            cmd, 
            stdout=subprocess.DEVNULL, 
            stderr=subprocess.PIPE, 
            text=True,
            preexec_fn=os.setpgrp
        )

        # Wait for the port to become active
        if self._wait_for_port(timeout):
            console.ok(f"SSH tunnel established: localhost:{self.local_port} -> {self.remote_host}:{self.remote_port} via {self.ssh_host}")
            return True
        else:
            # If it didn't open, check for errors in stderr
            stderr = self.process.stderr.read() if self.process.stderr else "No stderr available"
            console.error(f"SSH tunnel failed to open port {self.local_port} within {timeout}s.")
            if stderr:
                console.error(f"SSH Error: {stderr.strip()}")
            self.stop()
            return False

    except Exception as e:
        raise SSHTunnelError(f"Failed to start SSH tunnel: {e}")
        self.stop()
        return False

stop

stop()

Stops the SSH tunnel process.

Source code in src/cloudmesh/ai/ssh/tunnel.py

def stop(self):
    """Stops the SSH tunnel process."""
    if not self.process or self.process.poll() is not None:
        return False

    try:
        os.killpg(os.getpgid(self.process.pid), signal.SIGTERM)
        self.process = None
        console.ok(f"SSH tunnel on port {self.local_port} stopped.")
        return True
    except Exception as e:
        raise SSHTunnelError(f"Failed to stop SSH tunnel: {e}")
        return False

Functions:

_render_table

_render_table(rows)

Helper to render a list of dictionaries as a rich table.

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

def _render_table(rows):
    """Helper to render a list of dictionaries as a rich table."""
    if not rows:
        return None

    table = Table()

    # Use keys of the first dictionary as headers
    headers = list(rows[0].keys())
    for header in headers:
        table.add_column(header)

    # Add rows
    for row in rows:
        table.add_row(*[str(row.get(h, "")) for h in headers])

    return table

_save_fallback

_save_fallback(diag_data)

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

def _save_fallback(diag_data):
    diag_file = Path("ssh_diag.json")
    import json
    with open(diag_file, "w") as f:
        json.dump(diag_data, f, indent=4)
    console.print(f"[yellow]Diagnostic data saved to {diag_file} as fallback.[/yellow]")

check_ai

check_ai(api_key)

Gather diagnostic data and submit it to the vLLM server for repair.

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

@check_group.command(name="ai")
@click.option("--api-key", default=None, help="API key for the vLLM server.")
def check_ai(api_key):
    """Gather diagnostic data and submit it to the vLLM server for repair."""
    cfg = SSHConfig()

    console.info("Gathering SSH configuration diagnostics...")

    # 1. Get malformed entries
    errors = cfg.check()

    # 2. Get raw config content
    content = cfg.get_content()

    # 3. Package the data
    diag_data = {
        "config_file": str(cfg.filename),
        "raw_content": content,
        "errors": errors
    }

    # Load LLM config from YAML
    config = load_llm_config()

    # Determine base URL using url from config, default to http://localhost:17704/v1
    base_url = config.get("url", "http://localhost:17704/v1")

    # Determine API key priority: CLI -> YAML config -> legacy file
    if not api_key:
        api_key = config.get("key")
        if not api_key:
            try:
                key_path = Path("~/gemma/server_master_key.txt").expanduser()
                if key_path.exists():
                    api_key = key_path.read_text().strip()
            except Exception as e:
                logger.debug(f"Could not load default API key from ~/gemma/server_master_key.txt: {e}")

    headers = {}
    if api_key:
        headers["Authorization"] = f"Bearer {api_key}"

    try:
        # a. Get the model name from the vLLM server
        model_response = requests.get(f"{base_url}/models", headers=headers, timeout=10)
        model_response.raise_for_status()
        models = model_response.json().get("data", [])

        # Use model from config if specified, otherwise use the first available model from server
        model_name = config.get("model")
        if not model_name and models:
            model_name = models[0]["id"]

        if not model_name:
            raise Exception("No model specified in config and no models found on the vLLM server.")

        # b. Construct the prompt for the LLM
        system_prompt = (
            "You are an expert SSH configuration assistant. Your goal is to analyze the provided "
            "SSH config diagnostics and provide a corrected version of the config file. "
            "Explain the issues found and provide the final corrected content clearly."
        )
        user_prompt = f"Please analyze these SSH config diagnostics and provide a fix:\n\n{diag_data}"

        payload = {
            "model": model_name,
            "messages": [
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": user_prompt}
            ],
            "temperature": 0.2
        }

        console.info(f"Sending diagnostics to vLLM model {model_name} at {base_url}/chat/completions...")

        response = requests.post(f"{base_url}/chat/completions", json=payload, headers=headers, timeout=60)
        response.raise_for_status()

        result = response.json()
        answer = result["choices"][0]["message"]["content"]

        console.ok("AI Analysis complete!")
        console.print(f"\n[bold green]AI Repair Suggestion:[/bold green]\n\n{answer}")

    except requests.exceptions.ConnectionError:
        console.error(f"Could not connect to the vLLM server at {base_url}. Is it running?")
        _save_fallback(diag_data)
    except Exception as e:
        console.error(f"An error occurred during AI analysis: {e}")
        _save_fallback(diag_data)

check_basic

check_basic()

Run basic SSH config validation.

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

@check_group.command(name="basic")
def check_basic():
    """Run basic SSH config validation."""
    cfg = SSHConfig()
    errors = cfg.check()

    if not errors:
        console.ok("SSH config is valid!")
        return

    console.print(f"[yellow]Found {len(errors)} issue(s) in {cfg.filename}:[/yellow]")

    rows = []
    for err in errors:
        message = err["message"]
        if ": " in message:
            category, details = message.split(": ", 1)
        else:
            category, details = message, ""

        rows.append({
            "Line": err["line"],
            "Category": category,
            "Details": details
        })

    table = _render_table(rows)
    if table:
        console.print(table)

check_group

check_group(ctx)

Check the SSH config file for malformed entries.

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

@click.group(name="check", invoke_without_command=True)
@click.pass_context
def check_group(ctx):
    """Check the SSH config file for malformed entries."""
    if ctx.invoked_subcommand is None:
        ctx.invoke(check_basic)

get_contextual_logger

get_contextual_logger(name, initial_context=None)

Factory function to create a ContextualLogger.

Parameters:

Name	Type	Description	Default
name	str	The name of the logger.	required
initial_context	Optional[Dict[str, Any]]	Initial context to associate with the logger.	None

Returns:

Name	Type	Description
ContextualLogger	ContextualLogger	A logger adapter with the specified context.

Source code in cloudmesh/ai/common/logging_utils.py

def get_contextual_logger(name: str, initial_context: Optional[Dict[str, Any]] = None) -> ContextualLogger:
    """Factory function to create a ContextualLogger.

    Args:
        name: The name of the logger.
        initial_context: Initial context to associate with the logger.

    Returns:
        ContextualLogger: A logger adapter with the specified context.
    """
    logger = logging.getLogger(name)
    return ContextualLogger(logger, initial_context or {})

hosts_cmd

hosts_cmd()

List all hosts defined in the SSH config file.

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

@list_group.command(name="hosts")
def hosts_cmd():
    """List all hosts defined in the SSH config file."""
    cfg = SSHConfig()
    hosts = cfg.list()

    if not hosts:
        console.info("No hosts found in SSH config.")
        return

    rows = []
    for host in hosts:
        rows.append({
            "Host": host,
            "Hostname": cfg.hostname(host),
            "User": cfg.username(host),
            "Details": cfg.get_options(host)
        })

    table = _render_table(rows)
    if table:
        console.print(table)

list_group

list_group(ctx)

List SSH configurations and active tunnels.

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

@click.group(name="list", invoke_without_command=True)
@click.pass_context
def list_group(ctx):
    """List SSH configurations and active tunnels."""
    if ctx.invoked_subcommand is None:
        ctx.invoke(hosts_cmd)

load_llm_config

load_llm_config()

Load LLM configuration from ~/.config/cloudmesh/ai/llm.yaml, creating it if missing.

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

def load_llm_config():
    """Load LLM configuration from ~/.config/cloudmesh/ai/llm.yaml, creating it if missing."""
    config_path = Path("~/.config/cloudmesh/ai/llm.yaml").expanduser()

    if not config_path.exists():
        console.info(f"Creating default LLM config at {config_path}...")
        config_path.parent.mkdir(parents=True, exist_ok=True)
        default_config = {
            "url": "http://localhost:17704/v1",
            "model": "google/gemma-4-31B-it",
            "key": ""
        }
        with open(config_path, "w") as f:
            yaml.dump(default_config, f, default_flow_style=False)
        console.info("Default config created. Please edit the file to add your API key.")
        return default_config

    try:
        with open(config_path, "r") as f:
            return yaml.safe_load(f) or {}
    except Exception as e:
        logger.error(f"Error loading LLM config: {e}")
        return {}

path_expand

path_expand(text, slashreplace=True)

Standalone wrapper for console.expand_path.

Source code in cloudmesh/ai/common/io.py

def path_expand(text: str, slashreplace: bool = True) -> str:
    """Standalone wrapper for console.expand_path."""
    return console.expand_path(text, slashreplace)

register

register(cli)

Registers the ssh command group to the main CLI.

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

def register(cli):
    """Registers the ssh command group to the main CLI."""
    cli.add_command(ssh_group, name="ssh")

run_cmd

run_cmd()

Run the main functionality of ssh.

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

@ssh_group.command(name="run")
def run_cmd():
    """Run the main functionality of ssh."""
    logger.info("Executing ssh run command")
    console.ok(f"The ssh extension is running successfully!")

ssh_group

ssh_group()

ssh command group.

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

@click.group(name="ssh")
def ssh_group():
    """ssh command group."""
    pass

test_path_cmd

test_path_cmd(path)

Example command showing path expansion.

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

@ssh_group.command(name="test-path")
@click.argument("path")
def test_path_cmd(path):
    """Example command showing path expansion."""
    expanded = path_expand(path)
    console.info(f"Expanded path: {expanded}")

tunnel_cmd

tunnel_cmd(target, local_port, remote_host, ssh_user)

Create an SSH tunnel.

Target should be in the format HOST:PORT (e.g., my-server:8000).

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

@ssh_group.command(name="tunnel")
@click.argument("target")
@click.option("--local-port", type=int, help="Local port to bind to. Defaults to remote port.")
@click.option("--remote-host", default="localhost", help="Remote host relative to the SSH server. Defaults to localhost.")
@click.option("--ssh-user", help="SSH username to use.")
def tunnel_cmd(target, local_port, remote_host, ssh_user):
    """Create an SSH tunnel.

    Target should be in the format HOST:PORT (e.g., my-server:8000).
    """
    try:
        if ":" not in target:
            raise click.BadParameter("Target must be in the format HOST:PORT")

        ssh_host, remote_port_str = target.split(":", 1)
        remote_port = int(remote_port_str)

        # Default local port to remote port if not provided
        l_port = local_port if local_port else remote_port

        console.info(f"Setting up tunnel: localhost:{l_port} -> {remote_host}:{remote_port} via {ssh_host}")

        tunnel = Tunnel(
            local_port=l_port,
            remote_host=remote_host,
            remote_port=remote_port,
            ssh_host=ssh_host,
            ssh_user=ssh_user
        )

        if tunnel.start():
            console.ok(f"Tunnel is active. Press Ctrl+C to stop it.")
            try:
                while True:
                    if not tunnel.is_active():
                        console.warn("Tunnel process terminated unexpectedly.")
                        break
                    time.sleep(1)
            except KeyboardInterrupt:
                console.info("\nStopping tunnel...")
                tunnel.stop()
                console.ok("Tunnel closed.")
        else:
            console.error("Failed to start tunnel.")
            sys.exit(1)

    except ValueError:
        raise click.BadParameter("Port must be a valid integer.")
    except SSHTunnelError as e:
        console.error(f"SSH Tunnel Error: {e}")
        sys.exit(1)
    except Exception as e:
        console.error(f"Unexpected error: {e}")
        sys.exit(1)

tunnel_list_cmd

tunnel_list_cmd()

List all tunnels defined in config and their active status.

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

@list_group.command(name="tunnel")
def tunnel_list_cmd():
    """List all tunnels defined in config and their active status."""
    cfg = SSHConfig()
    tunnels = cfg.get_tunnels()

    if not tunnels:
        console.info("No tunnels defined in SSH config.")
        return

    from cloudmesh.ai.ssh.base import SSHBase
    checker = SSHBase()

    rows = []
    for t in tunnels:
        status = "Unknown"
        if t["type"] == "Local":
            try:
                is_open = checker.is_port_open("localhost", int(t["local_port"]))
                status = "Active" if is_open else "Inactive"
            except Exception:
                status = "Error"

        rows.append({
            "Host": t["host"],
            "Type": t["type"],
            "Local Port": t["local_port"],
            "Remote Target": t["remote_target"],
            "Status": status
        })

    table = _render_table(rows)
    if table:
        console.print(table)