Sympathy-MCP

Instructions

Runs the command via `sh -c` so shell features (pipes, redirects, etc.)
work as expected. Captures both stdout and stderr.

Args:
    vm: Name of the host (as configured in hosts.toml).
    command: Shell command to execute.
    workdir: Working directory on the remote host (default: /).
    timeout: Maximum seconds to wait (default: 120).

Returns:
    Command output with stdout, stderr, and exit code.

Implementation Reference

• src/sympathy_mcp/server.py:101-126 (registration)

  MCP tool registration and handler for vm_exec. This is decorated with @mcp.tool() and serves as the entry point for the tool. It validates parameters, calls the underlying implementation, formats the result, and handles exceptions.

  @mcp.tool()
  async def vm_exec(
      vm: str,
      command: str,
      workdir: str = "/",
      timeout: int = 120,
  ) -> str:
      """Execute a command on a remote host via SSH.
  
      Runs the command via `sh -c` so shell features (pipes, redirects, etc.)
      work as expected. Captures both stdout and stderr.
  
      Args:
          vm: Name of the host (as configured in hosts.toml).
          command: Shell command to execute.
          workdir: Working directory on the remote host (default: /).
          timeout: Maximum seconds to wait (default: 120).
  
      Returns:
          Command output with stdout, stderr, and exit code.
      """
      try:
          result = await _vm_exec(vm, command, workdir=workdir, timeout=timeout)
          return _format_exec_result(result)
      except (ValueError, KeyError, TimeoutError, RuntimeError, OSError) as e:
          return f"ERROR: {e}"

• src/sympathy_mcp/transport.py:204-236 (handler)

  Core implementation of vm_exec that performs SSH command execution. Resolves the host config, validates the command, wraps it with cd to workdir, and executes via SSH using sh -c for shell features.

  async def vm_exec(
      vm: str,
      command: str,
      workdir: str = "/",
      timeout: int = 120,
  ) -> ExecResult:
      """Execute a command on a remote host via SSH.
  
      Uses `ssh user@host sh -c <command>` to run the command in a shell
      so pipes, redirects, etc. work as expected.
  
      Args:
          vm: Name of the host (as configured in hosts.toml).
          command: Shell command to execute.
          workdir: Working directory on the remote host (default: /).
          timeout: Maximum seconds to wait (default: 120).
  
      Returns:
          ExecResult with stdout, stderr, and exit code.
      """
      host = _resolve_host(vm)
      if not command or not command.strip():
          raise ValueError("Command cannot be empty")
  
      # Wrap the command with cd to workdir.
      # SSH concatenates remote args into a single string, so we pass
      # the entire wrapped command as one argument to avoid splitting.
      wrapped = f"cd {workdir} && {command}"
      return await _run_ssh(
          host,
          [wrapped],
          timeout=timeout,
      )

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

  Low-level SSH execution helper that runs commands on remote hosts using asyncio.create_subprocess_exec. Handles timeout, stdin piping, and returns an 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:27-34 (schema)

  ExecResult dataclass schema that defines the return type for command execution, containing stdout, stderr, and exit_code fields.

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

• src/sympathy_mcp/server.py:58-66 (helper)

  Helper function that formats an ExecResult into a readable string for MCP responses, combining stdout, stderr, and exit code into a single formatted output.

  def _format_exec_result(result: ExecResult) -> str:
      """Format an ExecResult as a readable string for MCP response."""
      parts = []
      if result.stdout:
          parts.append(f"STDOUT:\n{result.stdout}")
      if result.stderr:
          parts.append(f"STDERR:\n{result.stderr}")
      parts.append(f"EXIT CODE: {result.exit_code}")
      return "\n".join(parts)