Sympathy-MCP

Instructions

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)