vm_file_write

Instructions

Handles arbitrary content safely by piping through SSH stdin.

Args:
    vm: Name of the host (as configured in hosts.toml).
    path: Absolute path where the file should be written.
    content: The content to write to the file.

Returns:
    Success confirmation or error message.

Implementation Reference

• src/sympathy_mcp/transport.py:265-294 (handler)

  Core implementation of vm_file_write that writes content to a remote file via SSH using tee for safe content transfer without shell escaping issues.

  async def vm_file_write(vm: str, path: str, content: str) -> ExecResult:
      """Write content to a file on a remote host.
  
      Pipes content via stdin to `ssh user@host tee <path>` for safe transfer
      of arbitrary content (no shell escaping issues).
  
      Args:
          vm: Name of the host (as configured in hosts.toml).
          path: Absolute path where the file should be written.
          content: The content to write.
  
      Returns:
          ExecResult with operation details.
      """
      host = _resolve_host(vm)
      _validate_path(path)
  
      # Use ssh with stdin piped to tee — handles arbitrary content safely
      # Redirect tee's stdout to /dev/null so we don't echo the content back
      result = await _run_ssh(
          host,
          ["tee", path],
          stdin_data=content.encode("utf-8"),
      )
      # tee echoes stdin to stdout, so clear that from the result
      return ExecResult(
          stdout="",
          stderr=result.stderr,
          exit_code=result.exit_code,
      )

• src/sympathy_mcp/server.py:147-167 (registration)

  MCP tool registration using @mcp.tool() decorator that wraps the transport implementation and returns a user-friendly success/error message.

  @mcp.tool()
  async def vm_file_write(vm: str, path: str, content: str) -> str:
      """Write content to a file on a remote host via SSH.
  
      Handles arbitrary content safely by piping through SSH stdin.
  
      Args:
          vm: Name of the host (as configured in hosts.toml).
          path: Absolute path where the file should be written.
          content: The content to write to the file.
  
      Returns:
          Success confirmation or error message.
      """
      try:
          result = await _vm_file_write(vm, path, content)
          if result.exit_code == 0:
              return f"Successfully wrote {len(content)} bytes to {vm}:{path}"
          return f"ERROR writing file: {result.stderr.strip()}"
      except (ValueError, KeyError, RuntimeError, OSError) as e:
          return f"ERROR: {e}"

• src/sympathy_mcp/transport.py:28-33 (schema)

  ExecResult class definition - the schema for command execution results containing stdout, stderr, and exit_code fields.

  class ExecResult:
      """Result of running a command on a remote host."""
  
      stdout: str
      stderr: str
      exit_code: int

• src/sympathy_mcp/transport.py:77-124 (helper)

  _run_ssh helper function that executes SSH commands with optional stdin data, handling timeouts and process management.

  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 (helper)

  _validate_path helper function that performs safety checks on file paths before writing.

  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}'")