class_docstr:     Telnet server implementation.
  
      Example::
  
          async def interact(connection):
              connection.send("Welcome")
              session = PromptSession()
              result = await session.prompt_async(message="Say something: ")
              connection.send(f"You said: {result}
  ")
  
          async def main():
              server = TelnetServer(interact=interact, port=2323)
              await server.run()
      
docstr: Accept new incoming connection.
code: def _accept(self, listen_socket: socket.socket) -> None:
      """
      Accept new incoming connection.
      """
      conn, addr = listen_socket.accept()
      logger.info("New connection %r %r", *addr)
  
      # Run application for this connection.
      async def run() -> None:
          try:
              with create_pipe_input() as vt100_input:
                  connection = TelnetConnection(
                      conn,
                      addr,
                      self.interact,
                      self,
                      encoding=self.encoding,
                      style=self.style,
                      vt100_input=vt100_input,
                      enable_cpr=self.enable_cpr,
                  )
                  self.connections.add(connection)
  
                  logger.info("Starting interaction %r %r", *addr)
                  try:
                      await connection.run_application()
                  finally:
                      self.connections.remove(connection)
                      logger.info("Stopping interaction %r %r", *addr)
          except EOFError:
              # Happens either when the connection is closed by the client
              # (e.g., when the user types 'control-]', then 'quit' in the
              # telnet client) or when the user types control-d in a prompt
              # and this is not handled by the interact function.
              logger.info("Unhandled EOFError in telnet application.")
          except KeyboardInterrupt:
              # Unhandled control-c propagated by a prompt.
              logger.info(
                  "Unhandled KeyboardInterrupt in telnet application."
              )
          except BaseException as e:
              print("Got %s" % type(e).__name__, e)
              import traceback
  
              traceback.print_exc()
          finally:
              self._application_tasks.remove(task)
  
      task = get_running_loop().create_task(run())
      self._application_tasks.append(task)
  
imports: from __future__ import annotations
  import asyncio
  import contextvars
  import socket
  from asyncio import get_running_loop
  from typing import Any, Callable, Coroutine, TextIO, cast
  from prompt_toolkit.application.current import create_app_session, get_app
  from prompt_toolkit.application.run_in_terminal import run_in_terminal
  from prompt_toolkit.data_structures import Size
  from prompt_toolkit.formatted_text import AnyFormattedText, to_formatted_text
  from prompt_toolkit.input import PipeInput, create_pipe_input
  from prompt_toolkit.output.vt100 import Vt100_Output
  from prompt_toolkit.renderer import print_formatted_text
  from prompt_toolkit.styles import BaseStyle, DummyStyle
  from .log import logger
  from .protocol import DO, ECHO, IAC, LINEMODE, MODE, NAWS, SB, SE, SEND, SUPPRESS_GO_AHEAD, TTYPE, WILL, TelnetProtocolParser
label_desc: Use image node_id 4 for calling the TelnetServer obj's underlying member method code with example usage: obj._accept(listen_socket) without return types