Sympathy-MCP

Instructions

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