MCP SSH Session

Instructions

Implementation Reference

• mcp_ssh_session/server.py:129-152 (handler)

  MCP tool handler for 'close_session'. Registers the tool, defines input schema via type hints and docstring, logs the action, delegates to session_manager.close_session(), resolves connection details for user-friendly response, and returns confirmation.

  @mcp.tool()
  def close_session(host: str, username: Optional[str] = None, port: Optional[int] = None) -> str:
      """Close a specific SSH session.
  
      The host parameter can be either a hostname/IP or an SSH config alias.
  
      Args:
          host: Hostname, IP address, or SSH config alias
          username: SSH username (optional, will use SSH config or current user)
          port: SSH port (optional, will use SSH config or default 22)
      """
      logger = session_manager.logger.getChild('tool_close_session')
      logger.info(f"Closing session for host={host}, user={username}, port={port}")
      session_manager.close_session(host, username, port)
  
      # Get the resolved values for the response message
      host_config = session_manager._ssh_config.lookup(host)
      resolved_host = host_config.get('hostname', host)
      resolved_username = username or host_config.get('user', 'current user')
      resolved_port = port or int(host_config.get('port', 22))
  
      response = f"Closed session: {resolved_username}@{resolved_host}:{resolved_port}"
      logger.info(f"Session closed successfully. Response: {response}")
      return response

• mcp_ssh_session/session_manager.py:281-332 (helper)

  Core implementation of session closing in SSHSessionManager. 'close_session' resolves the session_key and calls internal '_close_session'. '_close_session' clears associated commands, closes shell and SSH client, removes tracking data from dictionaries, and cleans up enable mode state.

  def close_session(self, host: str, username: Optional[str] = None, port: Optional[int] = None):
      """Close a specific session.
  
      Args:
          host: Hostname or SSH config alias
          username: SSH username (optional, will use config if available)
          port: SSH port (optional, will use config if available)
      """
      logger = self.logger.getChild('close_session')
      _, _, _, _, session_key = self._resolve_connection(host, username, port)
      logger.info(f"Request to close session: {session_key}")
      with self._lock:
          self._close_session(session_key)
  
  def _close_session(self, session_key: str):
      """Internal method to close a session (not thread-safe)."""
      logger = self.logger.getChild('internal_close')
      logger.debug(f"Closing session resources for {session_key}")
  
      # Clear any commands for this session first
      logger.debug(f"Clearing commands for {session_key}")
      self.command_executor.clear_session_commands(session_key)
  
      # Close persistent shell if exists
      if session_key in self._session_shells:
          logger.debug(f"Closing persistent shell for {session_key}")
          try:
              self._session_shells[session_key].close()
          except Exception as e:
              logger.warning(f"Error closing shell for {session_key}: {e}")
          del self._session_shells[session_key]
  
      if session_key in self._sessions:
          logger.debug(f"Closing SSH client for {session_key}")
          try:
              self._sessions[session_key].close()
          except Exception as e:
              logger.warning(f"Error closing client for {session_key}: {e}")
          del self._sessions[session_key]
      self._session_shell_types.pop(session_key, None)
      self._session_prompt_patterns.pop(session_key, None)
      self._session_prompts.pop(session_key, None)
      if session_key in self._session_shell_types:
          del self._session_shell_types[session_key]
  
      # Clean up enable mode tracking
      if session_key in self._enable_mode:
          logger.debug(f"Cleaning up enable mode tracking for {session_key}")
          del self._enable_mode[session_key]
  
      logger.info(f"Session closed: {session_key}")