Skip to main content
Glama

vm_file_read

Read files from remote hosts via SSH to access content on virtual machines managed through the Incus CLI.

Instructions

Read a file from a remote host via SSH.

Args:
    vm: Name of the host (as configured in hosts.toml).
    path: Absolute path to the file on the remote host.

Returns:
    File contents as a string, or an error message.

Input Schema

TableJSON Schema
NameRequiredDescriptionDefault
vmYes
pathYes

Implementation Reference

  • MCP tool handler registration and execution logic for vm_file_read. Decorated with @mcp.tool() and calls the transport implementation (_vm_file_read) with error handling.
    @mcp.tool()
    async def vm_file_read(vm: str, path: str) -> str:
        """Read a file from a remote host via SSH.
    
        Args:
            vm: Name of the host (as configured in hosts.toml).
            path: Absolute path to the file on the remote host.
    
        Returns:
            File contents as a string, or an error message.
        """
        try:
            content = await _vm_file_read(vm, path)
            return content
        except (ValueError, KeyError, RuntimeError, OSError) as e:
            return f"ERROR: {e}"
  • Core implementation of vm_file_read that reads a file from remote host via SSH using 'cat' command. Validates inputs, resolves host config, executes SSH command, and returns file contents or raises RuntimeError on failure.
    async def vm_file_read(vm: str, path: str) -> str:
        """Read a file from a remote host via SSH.
    
        Uses `ssh user@host cat <path>`.
    
        Args:
            vm: Name of the host (as configured in hosts.toml).
            path: Absolute path to the file on the remote host.
    
        Returns:
            File contents as a string.
    
        Raises:
            RuntimeError: If the file cannot be read.
        """
        host = _resolve_host(vm)
        _validate_path(path)
    
        result = await _run_ssh(host, ["cat", path])
        if result.exit_code != 0:
            raise RuntimeError(
                f"Failed to read {path} from {vm}: {result.stderr.strip()}"
            )
        return result.stdout
  • Low-level SSH transport function (_run_ssh) used by vm_file_read to execute remote commands. Handles subprocess creation, timeout management, and returns ExecResult with stdout, stderr, and exit code.
    async def _run_ssh(
        host_config: HostConfig,
        remote_command: list[str],
        timeout: int = 120,
        stdin_data: bytes | None = None,
    ) -> ExecResult:
        """Run a command on a remote host via SSH.
    
        Args:
            host_config: SSH connection details for the target host.
            remote_command: Command and arguments to run remotely.
            timeout: Maximum seconds to wait.
            stdin_data: Optional bytes to pipe to stdin.
    
        Returns:
            ExecResult with stdout, stderr, and exit code.
    
        Raises:
            TimeoutError: If the command exceeds the timeout.
            OSError: If the ssh binary is not found.
        """
        ssh_args = host_config.ssh_args()
        full_cmd = ["ssh"] + ssh_args + remote_command
    
        proc = await asyncio.create_subprocess_exec(
            *full_cmd,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
            stdin=asyncio.subprocess.PIPE if stdin_data else asyncio.subprocess.DEVNULL,
        )
    
        try:
            stdout_bytes, stderr_bytes = await asyncio.wait_for(
                proc.communicate(input=stdin_data),
                timeout=timeout,
            )
        except asyncio.TimeoutError:
            proc.kill()
            await proc.wait()
            raise TimeoutError(
                f"SSH command timed out after {timeout}s on {host_config.name}"
            )
    
        return ExecResult(
            stdout=stdout_bytes.decode("utf-8", errors="replace"),
            stderr=stderr_bytes.decode("utf-8", errors="replace"),
            exit_code=proc.returncode or 0,
        )
  • Path validation helper (_validate_path) used by vm_file_read to ensure paths are absolute, non-empty, and don't contain null bytes for security.
    def _validate_path(path: str) -> None:
        """Validate a file path on the remote host (basic safety checks)."""
        if not path or not path.strip():
            raise ValueError("Path cannot be empty")
        if "\x00" in path:
            raise ValueError("Path cannot contain null bytes")
        if not path.startswith("/"):
            raise ValueError(f"Path must be absolute (start with /): '{path}'")
  • Host resolution function (_resolve_host) that validates VM name and looks up HostConfig from the loaded configuration, used by vm_file_read to get SSH connection details.
    def _resolve_host(vm: str) -> HostConfig:
        """Resolve a VM/host name to its SSH config."""
        _validate_host_name(vm)
        config = _get_config()
        return config.get_host(vm)

Latest Blog Posts

MCP directory API

We provide all the information about MCP servers via our MCP API.

curl -X GET 'https://glama.ai/api/mcp/v1/servers/bobbyhiddn/Sympathy-MCP'

If you have feedback or need assistance with the MCP directory API, please join our Discord server