Sympathy-MCP

Instructions

Implementation Reference

• src/sympathy_mcp/server.py:129-144 (handler)

  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}"
• src/sympathy_mcp/transport.py:239-262 (handler)

  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
• src/sympathy_mcp/transport.py:77-124 (helper)

  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, )
• src/sympathy_mcp/transport.py:55-62 (schema)

  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}'")
• src/sympathy_mcp/transport.py:193-197 (helper)

  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)