code: def __init__(
      self,
      interact: Callable[
          [PromptToolkitSSHSession], Coroutine[Any, Any, None]
      ],
      *,
      enable_cpr: bool,
  ) -> None:
      self.interact = interact
      self.enable_cpr = enable_cpr
      self.interact_task: asyncio.Task[None] | None = None
      self._chan: Any | None = None
      self.app_session: AppSession | None = None
  
      # PipInput object, for sending input in the CLI.
      # (This is something that we can use in the prompt_toolkit event loop,
      # but still write date in manually.)
      self._input: PipeInput | None = None
      self._output: Vt100_Output | None = None
  
      # Output object. Don't render to the real stdout, but write everything
      # in the SSH channel.
      class Stdout:
          def write(s, data: str) -> None:
              try:
                  if self._chan is not None:
                      self._chan.write(data.replace("\n", "\r\n"))
              except BrokenPipeError:
                  pass  # Channel not open for sending.
  
          def isatty(s) -> bool:
              return True
  
          def flush(s) -> None:
              pass
  
          @property
          def encoding(s) -> str:
              assert self._chan is not None
              return str(self._chan._orig_chan.get_encoding()[0])
  
      self.stdout = cast(TextIO, Stdout())
  
imports: from __future__ import annotations
  import asyncio
  import traceback
  from asyncio import get_running_loop
  from typing import Any, Callable, Coroutine, TextIO, cast
  import asyncssh
  from prompt_toolkit.application.current import AppSession, create_app_session
  from prompt_toolkit.data_structures import Size
  from prompt_toolkit.input import PipeInput, create_pipe_input
  from prompt_toolkit.output.vt100 import Vt100_Output
label_desc: Use image node_id 1 to create a new PromptToolkitSSHSession object from inherited base classes: asyncssh with example: obj = PromptToolkitSSHSession(interact)