MCP SSH Session

Instructions

Implementation Reference

• mcp_ssh_session/server.py:12-112 (handler)

  The primary MCP tool handler for 'execute_command'. Decorated with @mcp.tool() which serves as both registration and schema definition via type hints and docstring. Calls SSHSessionManager.execute_command() and handles responses, including async transitions (exit 124) and input awaiting.
  @mcp.tool() def execute_command( host: str, command: str, username: Optional[str] = None, password: Optional[str] = None, key_filename: Optional[str] = None, port: Optional[int] = None, enable_password: Optional[str] = None, enable_command: str = "enable", sudo_password: Optional[str] = None, timeout: int = 30 ) -> str: """Execute a command on an SSH host using a persistent session. Starts synchronously and waits for completion. If the command doesn't complete within the timeout, it automatically transitions to async mode and returns a command ID for tracking. The host parameter can be either a hostname/IP or an SSH config alias. If an SSH config alias is provided, configuration will be read from ~/.ssh/config. For network devices (routers, switches), use enable_password to automatically enter privileged/enable mode before executing commands. For Unix/Linux hosts requiring sudo, use sudo_password to automatically handle the sudo password prompt. The command will be automatically prefixed with 'sudo' if not already present. Args: host: Hostname, IP address, or SSH config alias (e.g., "myserver") command: Command to execute username: SSH username (optional, will use SSH config or current user) password: Password (optional) key_filename: Path to SSH key file (optional, will use SSH config) port: SSH port (optional, will use SSH config or default 22) enable_password: Enable mode password for network devices (optional) enable_command: Command to enter enable mode (default: "enable") sudo_password: Password for sudo commands on Unix/Linux hosts (optional) timeout: Timeout in seconds for command execution (default: 30) """ logger = session_manager.logger.getChild('tool_execute_command') logger.info(f"Executing command on {host}: {command[:100]}...") try: stdout, stderr, exit_status = session_manager.execute_command( host=host, username=username, command=command, password=password, key_filename=key_filename, port=port, enable_password=enable_password, enable_command=enable_command, sudo_password=sudo_password, timeout=timeout, ) except ConnectionError as e: logger.error(f"Connection error: {e}") return f"Connection Error: {e}" except Exception as e: logger.error(f"Exception during execute_command: {e}", exc_info=True) return f"Error: {e}" # Check if command transitioned to async mode if exit_status == 124: if stderr.startswith("ASYNC:"): command_id = stderr.split(":", 1)[1] response = ( f"Command exceeded timeout of {timeout}s and is now running in background.\n\n" f"Command ID: {command_id}\n\n" f"Use get_command_status('{command_id}') to check progress.\n" f"Use interrupt_command_by_id('{command_id}') to stop it." ) logger.warning(f"Command timed out, returning async response for command_id {command_id}") return response if stderr.startswith("AWAITING_INPUT:"): # Format: AWAITING_INPUT:command_id:reason parts = stderr.split(":", 2) command_id = parts[1] reason = parts[2] if len(parts) > 2 else "unknown" response = ( f"Command paused waiting for user input ({reason}).\n\n" f"Command ID: {command_id}\n\n" f"Use send_input('{command_id}', 'your_input\\n') to provide input.\n" f"For example, if it's a password, provide the password followed by \\n." ) logger.info(f"Command awaiting input, returning instructions for command_id {command_id}") return response result = f"Exit Status: {exit_status}\n\n" if stdout: result += f"STDOUT:\n{stdout}\n" if stderr: result += f"STDERR:\n{stderr}\n" logger.info(f"Command finished with exit status {exit_status}.") logger.debug(f"Returning result:\n{result}") return result
• mcp_ssh_session/command_executor.py:26-94 (helper)

  Core execution logic in CommandExecutor.execute_command(). Starts execution asynchronously via execute_command_async(), then polls for completion, handles timeouts by transitioning to background async, and detects awaiting input. Called by SSHSessionManager.execute_command().
  def execute_command(self, host: str, username: Optional[str] = None, command: str = "", password: Optional[str] = None, key_filename: Optional[str] = None, port: Optional[int] = None, enable_password: Optional[str] = None, enable_command: str = "enable", sudo_password: Optional[str] = None, timeout: int = 30) -> tuple[str, str, int]: """Execute a command on a host using persistent session.""" logger = self.logger.getChild('execute_command') logger.info(f"[EXEC_REQ] host={host}, cmd={command[:100]}..., timeout={timeout}") # Validate command is_valid, error_msg = self._session_manager._command_validator.validate_command(command) if not is_valid: logger.warning(f"[EXEC_INVALID] {error_msg}") return "", error_msg, 1 # Start async logger.debug(f"[EXEC_ASYNC_START] Starting async execution") try: command_id = self.execute_command_async( host, username, command, password, key_filename, port, sudo_password, enable_password, enable_command, timeout ) except Exception as e: return "", str(e), 1 logger.debug(f"[EXEC_ASYNC_ID] command_id={command_id}") # Poll until done or timeout start = time.time() poll_count = 0 idle_threshold = getattr(self._session_manager, 'SYNC_IDLE_TO_ASYNC', 0) last_activity = start last_stdout = "" last_stderr = "" while time.time() - start < timeout: status = self.get_command_status(command_id) poll_count += 1 if 'error' in status: logger.error(f"[EXEC_ERROR] {status['error']}") return "", status['error'], 1 if status['status'] == 'awaiting_input': reason = status.get('awaiting_input_reason', 'unknown') logger.info(f"[EXEC_AWAIT] Command {command_id} waiting for input: {reason}") # Return immediately for awaiting input, let the tool handle it. return "", f"AWAITING_INPUT:{command_id}:{reason}", 124 if status['status'] != 'running': logger.info(f"[EXEC_DONE] status={status['status']}, polls={poll_count}, duration={time.time() - start:.2f}s") return status['stdout'], status['stderr'], status['exit_code'] or 0 # If the command is running but has been idle for SYNC_IDLE_TO_ASYNC, it means it *should* transition to async # and we should continue polling until the full timeout for this sync execute_command call. # The `_execute_standard_command_internal` function will return `ASYNC:{command_id}` if it hits its idle timeout. # The outer `execute_command` (sync tool) should then continue to poll this async ID. # This block is for when the *internal* async transition happens, but the outer sync call should keep waiting. # If we reached here, it means the internal worker is still 'running' but might be idle or waiting internally. time.sleep(0.1) # If we reach here, the command genuinely timed out based on the outer `timeout` parameter. logger.warning(f"[EXEC_TIMEOUT] Command {command_id} timed out after {timeout}s") status_on_timeout = self.get_command_status(command_id) return status_on_timeout.get('stdout', ''), f"ASYNC:{command_id}", 124
• mcp_ssh_session/session_manager.py:1534-1547 (helper)

  Wrapper method in SSHSessionManager.execute_command() that delegates to self.command_executor.execute_command(). Provides the entry point from the MCP handler.
  def execute_command(self, host: str, username: Optional[str] = None, command: str = "", password: Optional[str] = None, key_filename: Optional[str] = None, port: Optional[int] = None, enable_password: Optional[str] = None, enable_command: str = "enable", sudo_password: Optional[str] = None, timeout: int = 30) -> tuple[str, str, int]: """Execute a command on a host using persistent session.""" return self.command_executor.execute_command( host, username, command, password, key_filename, port, enable_password, enable_command, sudo_password, timeout )
• mcp_ssh_session/server.py:12-12 (registration)

  The @mcp.tool() decorator registers the execute_command function as an MCP tool with the name 'execute_command'.
  @mcp.tool()