{
"$type": "site.standard.document",
"bskyPostRef": {
"cid": "bafyreidvatnpgz4kdy4h7vf2lpirzdyvrtqdl6voxs5gkb4ikejwr5iu7m",
"uri": "at://did:plc:lk3jfj3zq4k4wxnk474axylu/app.bsky.feed.post/3mpu5yid74jd2"
},
"path": "/t/my-python-input-utility-for-console-chatbots-multiline-input-method-with-editing-and-mid-prompt-commands/1385734#post_2",
"publishedAt": "2026-07-04T21:43:09.000Z",
"site": "https://community.openai.com",
"tags": [
"@staticmethod",
"@classmethod",
"@kb.add"
],
"textContent": "# `chatbot_input_util` library file\n\n\n \"\"\"Multiline input utilities for chatbot-style console programs.\n\n The module exposes :class:`MultilineInput`, a standalone stateful input\n collector that returns complete multiline messages while still allowing\n single-word slash commands to interrupt the draft. Interrupted drafts are\n retained by the session so callers can run an application command and then\n return the user to the message they were composing.\n\n The module also exposes :func:`input_multiline`, a stateless, input-like helper.\n It creates an internal session for one completed input operation, disables\n mid-input application command interruption, and discards the session when the\n operation finishes.\n\n Supported runtime:\n\n - Python 3.10 or newer. The module uses built-in generic annotations and PEP\n 604 unions without postponed annotation evaluation.\n - prompt_toolkit is optional. When installed, the supported floor is\n prompt_toolkit 3.0.0. The integration uses PromptSession, KeyBindings,\n multiline prompts, default text, application exit results, current buffer\n access, and document cursor properties that exist in the 3.0 API.\n - keyboard is optional. It is used as a best-effort modified-Enter\n detector where the active frontend cannot expose that key combination\n directly; when unavailable or unsupported, the slash send marker remains the\n portable send path.\n\n The input session handles its own editor command, `/back`. Applications using\n this class do not need to implement that command; they can simply document it\n if it is useful for users. On a terminal, /back reprints draft input, and may\n return caret to end-of-line.\n \"\"\"\n\n __all__ = [\"MultilineInput\", \"input_multiline\"]\n\n\n class MultilineInput:\n \"\"\"docstring\"\"\"\n class _PromptToolkitFrontendUnavailable(Exception):\n \"\"\"Internal signal to retry the prompt without prompt_toolkit.\"\"\"\n\n SEND_MARKER = \"/\"\n BACK_COMMAND = \"/back\"\n DEFAULT_PROMPT = \"-- your message? --\"\n PROMPT_SEND_HINT = \"{send_marker!r} alone on a line to send.\"\n PROMPT_MODIFIED_ENTER_HINT = (\n \"{modified_enter} to send, or {send_marker!r} alone on a line.\"\n )\n PROMPT_ADDITIONAL_HELP: tuple[str, ...] = (\"/help for commands.\",)\n MODIFIED_ENTER_BRIDGE_SECONDS = 0.35\n\n def __init__(self, prompt_message: str | None = None) -> None:\n self.prompt_message = (\n self.DEFAULT_PROMPT if prompt_message is None else prompt_message\n )\n self.help_enabled = True\n self.sending_help = True\n self.command_interrupts_enabled = True\n self.transcript_echo_enabled = True\n self._draft = \"\"\n self._keyboard_loaded = False\n self._keyboard: object | None = None\n self._hotkey_support_cache: dict[str, bool] = {}\n self._prompt_toolkit_available_cache: bool | None = None\n self._prompt_toolkit_modified_enter_cache: tuple[str, str] | None = None\n self._prompt_toolkit_modified_enter_checked = False\n self._prompt_toolkit_runtime_disabled = False\n # Internal debugging handle:\n # force prompt_toolkit even in IDLE / non-TTY / unsupported TERM contexts.\n self._FORCE_PTK = False\n\n @staticmethod\n def is_slash_command(line: str) -> bool:\n \"\"\"\n Return True when a console line is exactly one slash command token.\n \"\"\"\n value = line.rstrip()\n return (\n value.startswith(\"/\")\n and not value.startswith(\"//\")\n and len(value) > 1\n and not any(ch.isspace() for ch in value)\n )\n\n @classmethod\n def _is_slash_command(cls, line: str) -> bool:\n return cls.is_slash_command(line)\n\n @classmethod\n def _is_back_command(cls, line: str) -> bool:\n return (\n cls._is_slash_command(line)\n and line.rstrip().lower() == cls.BACK_COMMAND\n )\n\n @classmethod\n def _is_send_marker(cls, line: str) -> bool:\n return line.rstrip() == cls.SEND_MARKER\n\n @staticmethod\n def _postprocess(raw: str) -> str:\n lines = raw.splitlines()\n for i, line in enumerate(lines):\n if line.startswith(\"//\"):\n lines[i] = \"/\" + line[2:]\n return \"\\n\".join(lines)\n\n def _finalize_message(self, raw: str, *, strip: bool) -> str:\n self._draft = \"\"\n out = self._postprocess(raw)\n return out.strip() if strip else out\n\n @staticmethod\n def _modified_enter_candidates() -> tuple[tuple[str, str, str], ...]:\n \"\"\"\n Return hotkey name, modifier name, and display label candidates.\n \"\"\"\n import sys\n\n if sys.platform == \"darwin\":\n return (\n (\"command+enter\", \"command\", \"Command+Enter\"),\n (\"cmd+enter\", \"cmd\", \"Command+Enter\"),\n (\"ctrl+enter\", \"ctrl\", \"Ctrl+Enter\"),\n )\n\n return ((\"ctrl+enter\", \"ctrl\", \"Ctrl+Enter\"),)\n\n @staticmethod\n def _is_keyboard_backend_error(exc: Exception) -> bool:\n error_types: tuple[type[BaseException], ...] = (\n AttributeError,\n ImportError,\n KeyError,\n RuntimeError,\n OSError,\n ValueError,\n )\n\n try:\n from subprocess import CalledProcessError\n except ImportError:\n return isinstance(exc, error_types)\n\n return isinstance(exc, error_types + (CalledProcessError,))\n\n def _load_keyboard(self) -> object | None:\n if self._keyboard_loaded:\n return self._keyboard\n\n self._keyboard_loaded = True\n try:\n import keyboard # type: ignore # pylint: disable=C0415\n except ImportError as exc:\n if exc.name and exc.name.startswith(\"keyboard\"):\n return None\n raise\n\n self._keyboard = keyboard\n return self._keyboard\n\n def _keyboard_hotkey_supported(self, hotkey_name: str) -> bool:\n cached = self._hotkey_support_cache.get(hotkey_name)\n if cached is not None:\n return cached\n\n keyboard = self._load_keyboard()\n if keyboard is None:\n self._hotkey_support_cache[hotkey_name] = False\n return False\n\n try:\n hotkey = keyboard.add_hotkey( # type: ignore[attr-defined]\n hotkey_name,\n lambda: None,\n suppress=False,\n )\n except Exception as exc:\n if not self._is_keyboard_backend_error(exc):\n raise\n self._hotkey_support_cache[hotkey_name] = False\n return False\n\n try:\n keyboard.remove_hotkey(hotkey) # type: ignore[attr-defined]\n except Exception as exc:\n if not self._is_keyboard_backend_error(exc):\n raise\n\n self._hotkey_support_cache[hotkey_name] = True\n return True\n\n def _supported_modified_enter_candidate(\n self,\n ) -> tuple[str, str, str] | None:\n for candidate in self._modified_enter_candidates():\n hotkey_name, _, _ = candidate\n if self._keyboard_hotkey_supported(hotkey_name):\n return candidate\n return None\n\n def _add_modified_enter_hotkey(\n self,\n callback: object,\n ) -> tuple[object | None, list[object], tuple[str, str, str] | None]:\n keyboard = self._load_keyboard()\n if keyboard is None:\n return None, [], None\n\n for candidate in self._modified_enter_candidates():\n hotkey_name, _, _ = candidate\n if self._hotkey_support_cache.get(hotkey_name) is False:\n continue\n\n try:\n hotkey = keyboard.add_hotkey( # type: ignore[attr-defined]\n hotkey_name,\n callback,\n suppress=False,\n )\n except Exception as exc:\n if not self._is_keyboard_backend_error(exc):\n raise\n self._hotkey_support_cache[hotkey_name] = False\n continue\n\n self._hotkey_support_cache[hotkey_name] = True\n return keyboard, [hotkey], candidate\n\n return keyboard, [], None\n\n @staticmethod\n def _remove_keyboard_hotkeys(\n keyboard: object | None,\n hotkeys: list[object],\n ) -> None:\n if keyboard is None:\n return\n\n for hotkey in hotkeys:\n try:\n keyboard.remove_hotkey(hotkey) # type: ignore[attr-defined]\n except Exception as exc:\n if not MultilineInput._is_keyboard_backend_error(exc):\n raise\n\n @staticmethod\n def _modified_enter_pressed(\n keyboard: object | None,\n candidate: tuple[str, str, str] | None,\n ) -> bool:\n if keyboard is None or candidate is None:\n return False\n\n _, modifier_name, _ = candidate\n try:\n return bool(keyboard.is_pressed(modifier_name)) # type: ignore[attr-defined]\n except Exception as exc:\n if not MultilineInput._is_keyboard_backend_error(exc):\n raise\n return False\n\n def _modified_enter_prompt_label(self, *, prompt_toolkit_mode: bool) -> str | None:\n if prompt_toolkit_mode:\n ptk_candidate = self._prompt_toolkit_modified_enter_candidate()\n if ptk_candidate is not None:\n _, label = ptk_candidate\n return label\n\n candidate = self._supported_modified_enter_candidate()\n if candidate is None:\n return None\n\n _, _, label = candidate\n return label\n\n @staticmethod\n def _prompt_toolkit_modified_enter_candidates() -> tuple[tuple[str, str], ...]:\n \"\"\"\n Return prompt_toolkit key names that may represent modified Enter.\n\n Plain Enter is Control-M in many terminals, so this deliberately avoids\n aliases such as \"c-m\" and \"c-j\"; binding those would make ordinary\n Enter send the message. A two-key Escape-then-Enter sequence is\n intentionally not a candidate because the send chord is Ctrl+Enter.\n \"\"\"\n return ((\"c-enter\", \"Ctrl+Enter\"),)\n\n def _prompt_toolkit_modified_enter_candidate(self) -> tuple[str, str] | None:\n if self._prompt_toolkit_modified_enter_checked:\n return self._prompt_toolkit_modified_enter_cache\n\n try:\n from prompt_toolkit.key_binding import KeyBindings # pylint: disable=C0415\n except ImportError as exc:\n if exc.name and exc.name.startswith(\"prompt_toolkit\"):\n self._prompt_toolkit_modified_enter_checked = True\n return None\n raise\n\n for key_name, label in self._prompt_toolkit_modified_enter_candidates():\n bindings = KeyBindings()\n try:\n bindings.add(key_name)(lambda event: None)\n except ValueError:\n continue\n\n self._prompt_toolkit_modified_enter_cache = (key_name, label)\n self._prompt_toolkit_modified_enter_checked = True\n return key_name, label\n\n self._prompt_toolkit_modified_enter_checked = True\n return None\n\n def _format_prompt_fragment(\n self,\n template: str,\n *,\n modified_enter: str | None,\n ) -> str:\n return template.format(\n send_marker=self.SEND_MARKER,\n back_command=self.BACK_COMMAND,\n modified_enter=modified_enter or \"\",\n ).strip()\n\n def _prompt_additional_help(self) -> tuple[str, ...]:\n additional_help = self.PROMPT_ADDITIONAL_HELP\n if isinstance(additional_help, str):\n return (additional_help,)\n return tuple(additional_help)\n\n def _store_draft_and_return_command(\n self,\n draft_lines: list[str],\n command_line: str,\n *,\n strip: bool,\n ) -> str:\n # Keep the draft in its raw input form. Escaped slash lines such as\n # \"//help\" must remain escaped until the user finally sends the message;\n # otherwise a restored literal slash line could be mistaken for a command.\n self._draft = \"\\n\".join(draft_lines)\n return command_line.strip() if strip else command_line\n\n def _should_return_command(\n self,\n command_line: str,\n draft_lines: list[str],\n ) -> bool:\n if not self._is_slash_command(command_line):\n return False\n\n if self._is_back_command(command_line):\n return True\n\n return self.command_interrupts_enabled or not draft_lines\n\n def _split_latest_command(\n self,\n raw: str,\n *,\n strip: bool,\n ) -> str | None:\n \"\"\"\n Return a latest-line command and preserve prior text as draft.\n \"\"\"\n lines = raw.splitlines()\n if not lines:\n return None\n\n command_line = lines[-1]\n draft_lines = lines[:-1]\n if not self._should_return_command(command_line, draft_lines):\n return None\n\n return self._store_draft_and_return_command(\n draft_lines,\n command_line,\n strip=strip,\n )\n\n def _format_prompt(self, *, prompt_toolkit_mode: bool) -> str:\n modified_enter_label = (\n self._modified_enter_prompt_label(\n prompt_toolkit_mode=prompt_toolkit_mode,\n )\n if self.sending_help\n else None\n )\n prompt_parts: list[str] = []\n\n base_prompt = self.prompt_message.strip()\n if base_prompt:\n prompt_parts.append(base_prompt)\n\n if self.sending_help:\n if modified_enter_label is None:\n send_help = self._format_prompt_fragment(\n self.PROMPT_SEND_HINT,\n modified_enter=None,\n )\n else:\n send_help = self._format_prompt_fragment(\n self.PROMPT_MODIFIED_ENTER_HINT,\n modified_enter=modified_enter_label,\n )\n if send_help:\n prompt_parts.append(send_help)\n\n if self.help_enabled:\n for additional_help in self._prompt_additional_help():\n help_text = self._format_prompt_fragment(\n additional_help,\n modified_enter=modified_enter_label,\n )\n if help_text:\n prompt_parts.append(help_text)\n\n return \" \".join(prompt_parts)\n\n def _print_prompt(self, *, prompt_toolkit_mode: bool) -> None:\n prompt = self._format_prompt(prompt_toolkit_mode=prompt_toolkit_mode)\n if prompt:\n print(prompt)\n\n if not self._draft:\n return\n\n # prompt_toolkit renders the restored draft in the editable input\n # buffer. Plain input has no editable buffer, so echo the retained\n # draft directly after the normal prompt header.\n if not prompt_toolkit_mode:\n print(self._draft)\n\n def _print_prompt_toolkit_transcript(self, text: str) -> None:\n if not self.transcript_echo_enabled or not text:\n return\n\n import sys\n\n sys.stdout.write(text)\n if not text.endswith(\"\\n\"):\n sys.stdout.write(\"\\n\")\n sys.stdout.flush()\n\n @staticmethod\n def _terminal_supports_prompt_toolkit() -> bool:\n import os\n import sys\n\n try:\n stdin_tty = sys.stdin.isatty()\n stdout_tty = sys.stdout.isatty()\n except (AttributeError, OSError, ValueError):\n return False\n\n in_idle = \"idlelib\" in sys.modules or (\n hasattr(sys.stdin, \"__class__\")\n and getattr(sys.stdin.__class__, \"__module__\", \"\").startswith(\"idlelib\")\n )\n\n if os.name == \"nt\":\n term_ok = True\n else:\n term_ok = os.environ.get(\"TERM\") not in (None, \"\", \"dumb\")\n\n return stdin_tty and stdout_tty and term_ok and not in_idle\n\n def _prompt_toolkit_available(self) -> bool:\n cached = self._prompt_toolkit_available_cache\n if cached is not None:\n return cached\n\n try:\n from prompt_toolkit import PromptSession # pylint: disable=C0415\n from prompt_toolkit.key_binding import KeyBindings # pylint: disable=C0415\n except ImportError as exc:\n if exc.name and exc.name.startswith(\"prompt_toolkit\"):\n self._prompt_toolkit_available_cache = False\n return False\n raise\n\n self._prompt_toolkit_available_cache = bool(PromptSession and KeyBindings)\n return self._prompt_toolkit_available_cache\n\n @staticmethod\n def _prompt_toolkit_frontend_error_types() -> tuple[type[BaseException], ...]:\n \"\"\"Return prompt_toolkit frontend failures that merit plain fallback.\"\"\"\n import sys\n\n error_types: tuple[type[BaseException], ...] = (OSError, RuntimeError)\n if sys.platform != \"win32\":\n return error_types\n\n try:\n from prompt_toolkit.output.win32 import ( # pylint: disable=C0415\n NoConsoleScreenBufferError,\n )\n except ImportError as exc:\n if exc.name and exc.name.startswith(\"prompt_toolkit\"):\n return error_types\n raise\n\n return error_types + (NoConsoleScreenBufferError,)\n\n def _collect_with_plain_input(self, *, strip: bool) -> str | None:\n # Line-based collector that works in IDLE and any non-TTY stdin environment.\n # Best-effort modified-Enter detection via optional `keyboard`.\n send_requested = False\n\n def _mark_send() -> None:\n nonlocal send_requested\n send_requested = True\n\n keyboard, hotkeys, active_candidate = self._add_modified_enter_hotkey(\n _mark_send,\n )\n\n def _mod_enter_send_requested() -> bool:\n nonlocal send_requested\n if send_requested:\n send_requested = False\n return True\n return self._modified_enter_pressed(keyboard, active_candidate)\n\n try:\n lines: list[str] = self._draft.splitlines() if self._draft else []\n\n try:\n first = input()\n except (EOFError, KeyboardInterrupt):\n return None\n\n if self._is_send_marker(first):\n return self._finalize_message(\"\\n\".join(lines), strip=strip)\n\n if self._should_return_command(first, lines):\n return self._store_draft_and_return_command(\n lines,\n first,\n strip=strip,\n )\n\n lines.append(first)\n if _mod_enter_send_requested():\n return self._finalize_message(\"\\n\".join(lines), strip=strip)\n\n while True:\n try:\n line = input()\n except EOFError:\n break\n except KeyboardInterrupt:\n return None\n\n if self._is_send_marker(line):\n break\n\n if self._should_return_command(line, lines):\n return self._store_draft_and_return_command(\n lines,\n line,\n strip=strip,\n )\n\n lines.append(line)\n if _mod_enter_send_requested():\n break\n\n return self._finalize_message(\"\\n\".join(lines), strip=strip)\n finally:\n self._remove_keyboard_hotkeys(keyboard, hotkeys)\n\n def _collect_with_prompt_toolkit(self, *, strip: bool) -> str | None:\n # Real terminal: prompt_toolkit provides editable restored drafts.\n try:\n from prompt_toolkit import PromptSession # pylint: disable=C0415\n from prompt_toolkit.key_binding import KeyBindings # pylint: disable=C0415\n except ImportError as exc:\n if exc.name and exc.name.startswith(\"prompt_toolkit\"):\n return self._collect_with_plain_input(strip=strip)\n raise\n\n keyboard: object | None = None\n hotkeys: list[object] = []\n ptk_frontend_error_types = self._prompt_toolkit_frontend_error_types()\n\n try:\n session = PromptSession(\n erase_when_done=self.transcript_echo_enabled,\n )\n kb = KeyBindings()\n ptk_modified_enter = self._prompt_toolkit_modified_enter_candidate()\n send_requested_at: float | None = None\n\n def _send_current_buffer(event):\n event.app.exit(result=event.app.current_buffer.text)\n\n def _mark_send() -> None:\n nonlocal send_requested_at\n from time import monotonic\n\n send_requested_at = monotonic()\n\n if ptk_modified_enter is None:\n keyboard, hotkeys, _ = self._add_modified_enter_hotkey(_mark_send)\n\n def _keyboard_send_requested() -> bool:\n nonlocal send_requested_at\n if send_requested_at is None:\n return False\n\n requested_at = send_requested_at\n send_requested_at = None\n\n from time import monotonic\n\n return (\n monotonic() - requested_at\n <= self.MODIFIED_ENTER_BRIDGE_SECONDS\n )\n\n @kb.add(\"c-d\")\n def _ctrl_d(event): # type: ignore[no-redef]\n _send_current_buffer(event)\n\n if ptk_modified_enter is not None:\n ptk_modified_enter_key, _ = ptk_modified_enter\n\n @kb.add(ptk_modified_enter_key)\n def _modified_enter(event): # type: ignore[no-redef]\n _send_current_buffer(event)\n\n @kb.add(\"enter\")\n def _enter(event): # type: ignore[no-redef]\n buf = event.app.current_buffer\n doc = buf.document\n text = buf.text\n\n if _keyboard_send_requested():\n event.app.exit(result=text)\n return\n\n lines = text.splitlines()\n row = doc.cursor_position_row\n col = doc.cursor_position_col\n if lines and row == (len(lines) - 1):\n current_line = doc.current_line\n if (\n self._is_send_marker(current_line)\n and col == len(current_line)\n ):\n event.app.exit(result=\"\\n\".join(lines[:-1]))\n return\n if (\n self._should_return_command(\n current_line,\n lines[:-1],\n )\n and col == len(current_line)\n ):\n event.app.exit(result=text)\n return\n\n buf.insert_text(\"\\n\")\n\n try:\n raw = session.prompt(\n \"\",\n multiline=True,\n key_bindings=kb,\n default=self._draft,\n )\n except (EOFError, KeyboardInterrupt):\n return None\n\n command = self._split_latest_command(raw, strip=strip)\n if command is not None:\n if not self._is_back_command(command):\n self._print_prompt_toolkit_transcript(command)\n return command\n\n message = self._finalize_message(raw, strip=strip)\n self._print_prompt_toolkit_transcript(message)\n return message\n except ptk_frontend_error_types as exc:\n if self._FORCE_PTK:\n raise\n self._prompt_toolkit_runtime_disabled = True\n raise self._PromptToolkitFrontendUnavailable() from exc\n finally:\n self._remove_keyboard_hotkeys(keyboard, hotkeys)\n\n def get_input(self, strip: bool = True) -> str | None:\n while True:\n use_ptk = (\n (\n self._FORCE_PTK\n or (\n not self._prompt_toolkit_runtime_disabled\n and self._terminal_supports_prompt_toolkit()\n )\n )\n and self._prompt_toolkit_available()\n )\n self._print_prompt(prompt_toolkit_mode=use_ptk)\n try:\n if use_ptk:\n result = self._collect_with_prompt_toolkit(strip=strip)\n else:\n result = self._collect_with_plain_input(strip=strip)\n except self._PromptToolkitFrontendUnavailable:\n continue\n\n if result is None:\n return None\n\n if self._is_back_command(result):\n continue\n\n return result\n\n\n def input_multiline(\n prompt: object = None,\n /,\n *,\n help: bool = True, # pylint: disable=redefined-builtin\n sending_help: bool = True,\n ) -> str | None:\n \"\"\"\n Return one complete multiline input value using an input-like API.\n\n The helper is stateless across calls. It creates a fresh\n MultilineInput, applies the prompt and guidance settings, disables\n mid-input application slash-command interruption, and discards the session\n after the input operation completes. A single-word slash command entered as\n the first line is still returned immediately. The built-in /back editor\n command remains available inside the one input operation.\n \"\"\"\n session = MultilineInput(\n prompt_message=None if prompt is None else str(prompt),\n )\n session.help_enabled = help\n session.sending_help = sending_help\n session.command_interrupts_enabled = False\n return session.get_input(strip=False)\n\n\n def multi_input_demo() -> None:\n \"\"\"\n Run a small receiver loop demonstrating MultilineInput.\n \"\"\"\n message_input = MultilineInput(\n prompt_message=MultilineInput.DEFAULT_PROMPT,\n )\n message_input.help_enabled = True\n message_input.sending_help = True\n message_input.command_interrupts_enabled = True\n\n def print_help() -> None:\n print(\"Demo commands:\")\n print(\" /help Show this help text.\")\n print(\" /exit Stop the demo.\")\n print(\" /quit Stop the demo.\")\n print(\" /back Restore the current draft in the input editor.\")\n print(\"Any other single-word slash command is reported as received.\")\n\n def handle_command(command: str) -> bool:\n \"\"\"\n Return True when the receiver loop should continue.\n \"\"\"\n print(f\"Received command {command}\")\n if command.lower() == \"/help\":\n print_help()\n return True\n if command.lower() in [\"/exit\", \"/quit\"]:\n print(\"Exiting...\")\n return False\n return True\n\n print(\"MultilineInput() demo\")\n\n while True:\n value = message_input.get_input()\n if value is None:\n print(\"Input closed.\")\n return\n\n command = value.strip()\n if MultilineInput.is_slash_command(command):\n if not handle_command(command):\n return\n continue\n\n line_count = len(value.splitlines()) if value else 0\n print(f\"Received message, {line_count} lines.\")\n\n\n if __name__ == \"__main__\":\n multi_input_demo()\n",
"title": "My Python input utility for console chatbots - multiline input method, with editing and mid-prompt commands"
}