Implement an interactive shell over ssh in Python using Paramiko?

I needed this for a Cisco router, which is rather different from a Linux machine. Credit to @misha for identifying the big challenge when running multiple commands during a single session, namely detecting the end of one command output so you know when to stop reading from the router's stdout object. If you don't detect that, the read loop just hangs. Sending two commands every time and using the second command as a sentinel is kind of a clever hack, so I copied it! This uses a well-known error response from the IOS command prompt as a sentinel.

import logging
import paramiko
import socket

logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)

def inter_handler(
        title: str,
        instructions: str,
        prompts: List[Tuple[str, bool]]) -> List[str]:
    """
    Proprietary details omitted. Here is the relevant doc from
    paramiko/transport.py, function `auth_interactive`:

        The handler is expected to be a callable that will handle calls of the
        form: ``handler(title, instructions, prompt_list)``.  The ``title`` is
        meant to be a dialog-window title, and the ``instructions`` are user
        instructions (both are strings).  ``prompt_list`` will be a list of
        prompts, each prompt being a tuple of ``(str, bool)``.  The string is
        the prompt and the boolean indicates whether the user text should be
        echoed.

        A sample call would thus be:
        ``handler('title', 'instructions', [('Password:', False)])``.

        The handler should return a list or tuple of answers to the server's
        questions.
    """
    pass


def ssh_run_cmds(
        host: str,
        port: int,
        user: str,
        commands: list) -> None:
    """
    Connect to the router, authenticate by computing a challenge
    response, and run commands. A major challenge is detecting
    the end of the command output, to know when to stop reading
    from the router session. This code uses an ugly hack of
    sending an invalid command and checking for a well-known
    error message.
    """
    # Create a socket and connect it to port 22 on the remote host
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    # the argument must be a tuple
    sock.connect((host, port))
    # Wrap the socket in a paramiko Transport object
    ts = paramiko.Transport(sock)
    # Tell Paramiko that the Transport is going to be used as a client
    ts.start_client(timeout=10)
    # Authenticate the specified user via the handler
    ts.auth_interactive(user, inter_handler)
    # Open a channel
    chan = ts.open_channel(kind='session', timeout=10)
    # Associate a pseudo tty
    chan.get_pty()
    # Request an interactive shell session
    chan.invoke_shell()
    # Create writer/reader file-like objects
    stdin = chan.makefile('wb')
    stdout = chan.makefile('r')
    # Use the output from this invalid command as a sentinel
    bogus_cmd = 'show bogus'
    for cmd in commands:
        # Send the command AND a bogus command to detect end of output
        cmds = f'{cmd}\n{bogus_cmd}\n'
        logger.debug('Send commands: %s', cmds)
        stdin.write(cmds)
        stdin.flush()

        # Read the response
        for line in stdout:
            line = line.strip()
            logger.debug('Output line: %s', line)
            # the response from the bogus command is the last line
            if line.startswith("% Invalid input detected at '^' marker."):
                break
        # for line
    # for cmd

    stdin.close()
    stdout.close()
    chan.close()
    ts.close()
    sock.close()

I tried the answer above, and it didn't work because ECHO command returned error in Python CLI, which I was using with SSH.

So I wrote another code that is applicable for Python CLI, assuming that the output is in one line.

And I also think something like f"print('{finish}')" can do same thing (terminator??) as ECHO in the answer above. But I didn't make use of it because my output always has to be in one line.

class MusicPlayer:
def __init__(self, host='', username='pi', password=''):
    self.ssh = paramiko.SSHClient()
    self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    self.ssh.connect(host, username=username, password=password)
    channel = self.ssh.invoke_shell()
    self.stdin = channel.makefile('wb')
    self.stdout = channel.makefile('r')
    self.in_history = []
    self.out_history = []
    self.init_vlc()
    self.print()
    # atexit.register(self.__del__)

def __del__(self):
    self.ssh.close()

def execute(self, cmd):
    self.in_history.append(cmd)
    self.stdin.write(cmd + '\n')

def print(self, lines=1):
    for line in self.stdout:
        lined = line.strip()
        print(lined)
        self.out_history.append(lined)
        if self.in_history[-1] in lined:
            next_one = self.stdout.__next__().strip()
            print(next_one)
            self.out_history.append(next_one)
            return next_one

def init_vlc(self):
    for command in ['python', 'import vlc', 'import time', 'media_player = vlc.MediaPlayer()']:
        self.execute(command)