ssh-mcp

Instructions

Implementation Reference

• src/ssh_mcp/server.py:180-209 (handler)

  The execute tool handler decorated with @mcp.tool(). This is the MCP tool entry point that initializes the server, calls the SSHManager.execute method, and formats the result.

  @mcp.tool()
  async def execute(
      server: str,
      command: str,
      timeout: int = 30,
      working_dir: str | None = None,
      force: bool = False,
  ) -> str:
      """Execute a shell command on a single SSH server.
  
      Args:
          server: Server name (e.g. 'pro-dicentra', 'inf-ai'). Must match a configured
                  server. Use list_servers to see available servers.
          command: Shell command to execute on the remote server.
          timeout: Command timeout in seconds. Default 30.
          working_dir: Remote directory to execute from. Uses server default if omitted.
          force: Bypass dangerous command detection. Use with extreme caution. Default false.
  
      Returns:
          Formatted command execution result with stdout, stderr, and exit code.
      """
      try:
          _init()
  
          result = await _ssh.execute(server, command, timeout, working_dir, force)
          return format_exec_result(result)
  
      except Exception as e:
          logger.error(f"Error executing command on {server}: {e}")
          return f"Error executing command on {server}: {e}"

• src/ssh_mcp/ssh.py:122-273 (handler)

  The core implementation of the execute method in SSHManager class. Handles dangerous command detection, SSH connection management, command execution with timeout, output truncation, and comprehensive error handling.

  async def execute(
      self,
      server_name: str,
      command: str,
      timeout: int = 30,
      working_dir: str | None = None,
      force: bool = False,
  ) -> ExecResult:
      """Execute command on a remote server.
  
      Args:
          server_name: Server name from registry
          command: Command to execute
          timeout: Command timeout in seconds
          working_dir: Working directory for command execution
          force: Bypass dangerous command detection (use with caution)
  
      Returns:
          ExecResult with command output and metadata
      """
      try:
          server = self.registry.get_server(server_name)
  
          # Check for dangerous commands unless force is enabled
          if not force and _is_dangerous_command(command):
              logger.warning(
                  f"Blocked potentially destructive command on {server_name}: {command}"
              )
              return ExecResult(
                  server=server_name,
                  command=command,
                  stdout="",
                  stderr="",
                  exit_code=None,
                  error="Blocked: potentially destructive command detected. Review and use with caution.",
              )
  
          # Use server-specific timeout if configured
          effective_timeout = server.timeout or timeout
  
          # Prepend working directory if specified
          effective_command = command
          if working_dir:
              effective_command = f"cd {shlex.quote(working_dir)} && {command}"
          elif server.default_dir:
              effective_command = f"cd {shlex.quote(server.default_dir)} && {command}"
  
          # Get or create connection
          conn = await self._get_connection(server_name)
  
          # Execute command and track duration
          start_time = time.monotonic()
          try:
              result = await conn.run(effective_command, timeout=effective_timeout)
              duration_ms = int((time.monotonic() - start_time) * 1000)
  
              # Truncate output if needed
              stdout = result.stdout or ""
              stderr = result.stderr or ""
  
              if len(stdout) > self.settings.max_output_bytes:
                  stdout = (
                      stdout[: self.settings.max_output_bytes]
                      + f"\n[... output truncated at {self.settings.max_output_bytes} bytes]"
                  )
  
              if len(stderr) > self.settings.max_output_bytes:
                  stderr = (
                      stderr[: self.settings.max_output_bytes]
                      + f"\n[... output truncated at {self.settings.max_output_bytes} bytes]"
                  )
  
              exec_result = ExecResult(
                  server=server_name,
                  command=command,
                  stdout=stdout,
                  stderr=stderr,
                  exit_code=result.exit_status,
                  duration_ms=duration_ms,
              )
  
              # Audit log successful execution
              self._audit.info(
                  "server=%s command=%s exit_code=%s duration_ms=%s",
                  server_name,
                  command,
                  exec_result.exit_code,
                  exec_result.duration_ms,
              )
  
              return exec_result
  
          except asyncio.TimeoutError as e:
              duration_ms = int((time.monotonic() - start_time) * 1000)
              logger.error(f"Command timeout on {server_name}: {command}")
              exec_result = ExecResult(
                  server=server_name,
                  command=command,
                  stdout="",
                  stderr="",
                  exit_code=None,
                  error=f"Command timeout after {effective_timeout}s: {e}",
                  duration_ms=duration_ms,
              )
  
              # Audit log timeout
              self._audit.info(
                  "server=%s command=%s exit_code=%s duration_ms=%s error=timeout",
                  server_name,
                  command,
                  exec_result.exit_code,
                  exec_result.duration_ms,
              )
  
              return exec_result
  
      except KeyError as e:
          logger.error(f"Server not found: {server_name}")
          return ExecResult(
              server=server_name,
              command=command,
              stdout="",
              stderr="",
              exit_code=None,
              error=f"Server not found: {e}",
          )
  
      except (
          asyncssh.DisconnectError,
          asyncssh.PermissionDenied,
          OSError,
      ) as e:
          logger.error(f"SSH error on {server_name}: {e}")
          return ExecResult(
              server=server_name,
              command=command,
              stdout="",
              stderr="",
              exit_code=None,
              error=f"SSH error: {e}",
          )
  
      except Exception as e:
          logger.error(f"Unexpected error on {server_name}: {e}")
          return ExecResult(
              server=server_name,
              command=command,
              stdout="",
              stderr="",
              exit_code=None,
              error=f"Unexpected error: {e}",
          )

• src/ssh_mcp/models.py:79-100 (schema)

  ExecResult dataclass defining the result structure returned by the execute tool, containing server, command, stdout, stderr, exit_code, error, and duration_ms fields.

  class ExecResult:
      """Result from executing a command on a remote server.
  
      Mutable to allow construction during execution.
  
      Attributes:
          server: Server name where command was executed
          command: The command that was executed
          stdout: Standard output captured from command
          stderr: Standard error captured from command
          exit_code: Process exit code (None if execution failed)
          error: Error message if execution failed
          duration_ms: Command execution duration in milliseconds
      """
  
      server: str
      command: str
      stdout: str
      stderr: str
      exit_code: int | None
      error: str | None = None
      duration_ms: int = 0

• src/ssh_mcp/server.py:180-180 (registration)

  The @mcp.tool() decorator registers the execute function as an MCP tool with the FastMCP server instance.

  @mcp.tool()