diff --git a/doc/develop/test/twister.rst b/doc/develop/test/twister.rst
index d70d062719633..58cdbf778bdef 100644
--- a/doc/develop/test/twister.rst
+++ b/doc/develop/test/twister.rst
@@ -1285,6 +1285,25 @@ In this case you can run twister with the following options:
 The script is user-defined and handles delivering the messages which can be
 used by twister to determine the test execution status.
 
+To support devices that communicate via RTT, use the ``--device-rtt`` option. Twister
+will connect to the device with ``west rtt`` command and capture the log messages.
+In this case you can run twister with the following options:
+
+.. tabs::
+
+   .. group-tab:: Linux
+
+      .. code-block:: bash
+
+         scripts/twister --device-testing --device-rtt \
+         -p nrf7002dk/nrf5340/cpuapp -T tests/kernel
+
+   .. group-tab:: Windows
+
+      .. note::
+
+         Not supported on Windows OS
+
 The ``--device-flash-timeout`` option allows to set explicit timeout on the
 device flash operation, for example when device flashing takes significantly
 large time.
@@ -1480,6 +1499,23 @@ work. It is equivalent to following west and twister commands.
   manually according to above example. This is because the serial port
   of the PTY is not fixed and being allocated in the system at runtime.
 
+RTT support using ``--device-rtt``  can also be used in the
+hardware map:
+
+.. code-block:: yaml
+
+   - connected: true
+     id: 001050795550
+     platform: nrf7002dk/nrf5340/cpuapp
+     product: J-Link
+     runner: nrfutil
+     rtt: true
+     rtt_runner: jlink
+
+If a different runner should be used for RTT connection than for flashing, specify
+``rtt_runner`` field, like shown above. If a different runner is not needed, then
+``rtt_runner`` can be omitted.
+
 If west is not available or does not know how to flash your system, a custom
 flash command can be specified using the ``flash-command`` flag. The script is
 called with a ``--build-dir`` with the path of the current build, as well as a
diff --git a/samples/subsys/testsuite/pytest/shell/testcase.yaml b/samples/subsys/testsuite/pytest/shell/testcase.yaml
index 60b80817620d8..2e7690076af26 100644
--- a/samples/subsys/testsuite/pytest/shell/testcase.yaml
+++ b/samples/subsys/testsuite/pytest/shell/testcase.yaml
@@ -18,6 +18,12 @@ tests:
     harness: pytest
     extra_configs:
       - CONFIG_SHELL_VT100_COLORS=n
+  sample.pytest.rtt:
+    harness: pytest
+    extra_configs:
+      - CONFIG_USE_SEGGER_RTT=y
+      - CONFIG_LOG_MODE_DEFERRED=y
+      - CONFIG_SHELL_BACKEND_RTT=y
   sample.harness.shell:
     harness: shell
     harness_config:
diff --git a/scripts/pylib/pytest-twister-harness/README.rst b/scripts/pylib/pytest-twister-harness/README.rst
index 81e8300d14095..099501125d710 100644
--- a/scripts/pylib/pytest-twister-harness/README.rst
+++ b/scripts/pylib/pytest-twister-harness/README.rst
@@ -26,6 +26,9 @@ Run exemplary test shell application by Twister:
   # hardware
   ./scripts/twister -p nrf52840dk/nrf52840 --device-testing --device-serial /dev/ttyACM0 -T samples/subsys/testsuite/pytest/shell
 
+  # hardware over RTT
+  ./scripts/twister -p nrf52840dk/nrf52840 --device-testing --device-rtt -T samples/subsys/testsuite/pytest/shell --test sample.pytest.rtt
+
 or build shell application by west and call pytest directly:
 
 .. code-block:: sh
@@ -45,3 +48,7 @@ or build shell application by west and call pytest directly:
   # hardware
   west build -p -b nrf52840dk/nrf52840
   pytest --twister-harness --device-type=hardware --device-serial=/dev/ttyACM0 --build-dir=build -p twister_harness.plugin
+
+  # hardware over RTT
+  west build -p -b nrf52840dk/nrf52840 . -T sample.pytest.rtt
+  pytest --twister-harness --device-type=hardware --device-rtt=True --build-dir=build -p twister_harness.plugin
diff --git a/scripts/pylib/pytest-twister-harness/src/twister_harness/device/device_adapter.py b/scripts/pylib/pytest-twister-harness/src/twister_harness/device/device_adapter.py
index 87ecac0f75a9d..c38ee6526a566 100644
--- a/scripts/pylib/pytest-twister-harness/src/twister_harness/device/device_adapter.py
+++ b/scripts/pylib/pytest-twister-harness/src/twister_harness/device/device_adapter.py
@@ -81,8 +81,9 @@ def launch(self) -> None:
         self._start_reader_thread()
 
         if self.device_config.flash_before:
-            # For hardware devices with shared USB or software USB, connect after flashing.
-            # Retry for up to 10 seconds for USB-CDC based devices to enumerate.
+            # For hardware devices with shared USB or software USB or RTT, connect after
+            # flashing. Retry for up to 10 seconds for USB-CDC based devices to
+            # enumerate.
             self._flash_and_run()
             self.connect(retry_s = 10)
         else:
diff --git a/scripts/pylib/pytest-twister-harness/src/twister_harness/device/hardware_adapter.py b/scripts/pylib/pytest-twister-harness/src/twister_harness/device/hardware_adapter.py
index d0d200d6225ff..3c26f9ad6dd5e 100644
--- a/scripts/pylib/pytest-twister-harness/src/twister_harness/device/hardware_adapter.py
+++ b/scripts/pylib/pytest-twister-harness/src/twister_harness/device/hardware_adapter.py
@@ -32,7 +32,7 @@ def __init__(self, device_config: DeviceConfig) -> None:
         super().__init__(device_config)
         self._flashing_timeout: float = device_config.flash_timeout
         self._serial_connection: serial.Serial | None = None
-        self._serial_pty_proc: subprocess.Popen | None = None
+        self._proc: subprocess.Popen | None = None
         self._serial_buffer: bytearray = bytearray()
 
         self.device_log_path: Path = device_config.build_dir / 'device.log'
@@ -76,6 +76,25 @@ def generate_command(self) -> None:
             command.extend(command_extra_args)
         self.command = command
 
+    def generate_rtt_command(self):
+        """Return command to connect to the device via RTT."""
+
+        command = ["west", "rtt", "--skip-rebuild", "-d",
+                               str(self.device_config.build_dir)]
+
+        rtt_runner = self.device_config.rtt_runner
+        if rtt_runner:
+            command.append("--runner")
+            command.append(rtt_runner)
+
+            if rtt_runner in ("jlink", "pyocd"):
+                command.append("--dev-id")
+                command.append(self.device_config.id)
+
+        # Since _start_pty expects a string, we need to convert the command list into
+        # one.
+        return " ".join(command)
+
     def _prepare_runner_args(self) -> tuple[list[str], list[str]]:
         base_args: list[str] = []
         extra_args: list[str] = []
@@ -154,7 +173,13 @@ def _flash_and_run(self) -> None:
                 raise TwisterHarnessException(msg)
 
     def _connect_device(self) -> None:
-        serial_name = self._open_serial_pty() or self.device_config.serial
+        if self.device_config.serial_pty:
+            serial_name = self._open_pty(self.device_config.serial_pty)
+        elif self.device_config.use_rtt:
+            serial_name = self._open_pty(self.generate_rtt_command())
+        else:
+            serial_name = self.device_config.serial
+
         logger.debug('Opening serial connection for %s', serial_name)
         try:
             self._serial_connection = serial.Serial(
@@ -167,17 +192,15 @@ def _connect_device(self) -> None:
             )
         except serial.SerialException as exc:
             logger.exception('Cannot open connection: %s', exc)
-            self._close_serial_pty()
+            self._close_pty()
             raise
 
         self._serial_connection.flush()
         self._serial_connection.reset_input_buffer()
         self._serial_connection.reset_output_buffer()
 
-    def _open_serial_pty(self) -> str | None:
-        """Open a pty pair, run process and return tty name"""
-        if not self.device_config.serial_pty:
-            return None
+    def _open_pty(self, command: str) -> str | None:
+        """Open a pty pair, run process with command and return tty name."""
 
         try:
             master, slave = pty.openpty()
@@ -186,14 +209,14 @@ def _open_serial_pty(self) -> str | None:
             raise exc
 
         try:
-            self._serial_pty_proc = subprocess.Popen(
-                re.split(',| ', self.device_config.serial_pty),
+            self._proc = subprocess.Popen(
+                re.split(',| ', command),
                 stdout=master,
                 stdin=master,
                 stderr=master
             )
         except subprocess.CalledProcessError as exc:
-            logger.exception('Failed to run subprocess %s, error %s', self.device_config.serial_pty, str(exc))
+            logger.exception('Failed to run subprocess %s, error %s', command, str(exc))
             raise
         return os.ttyname(slave)
 
@@ -201,17 +224,22 @@ def _disconnect_device(self) -> None:
         if self._serial_connection:
             serial_name = self._serial_connection.port
             self._serial_connection.close()
-            # self._serial_connection = None
             logger.debug('Closed serial connection for %s', serial_name)
-        self._close_serial_pty()
+        self._close_pty()
+
+    def _close_pty(self) -> None:
+        """Terminate the process opened for serial pty script or RTT."""
+        if self._proc:
+            terminate_process(self._proc)
+            self._proc.communicate(timeout=self.base_timeout)
+
+            if self.device_config.serial_pty:
+                proc = self.device_config.serial_pty
+            else:
+                proc = "west rtt"
 
-    def _close_serial_pty(self) -> None:
-        """Terminate the process opened for serial pty script"""
-        if self._serial_pty_proc:
-            self._serial_pty_proc.terminate()
-            self._serial_pty_proc.communicate(timeout=self.base_timeout)
-            logger.debug('Process %s terminated', self.device_config.serial_pty)
-            self._serial_pty_proc = None
+            logger.debug('Process %s terminated', proc)
+            self._proc = None
 
     def _close_device(self) -> None:
         if self.device_config.post_script:
@@ -282,7 +310,7 @@ def _flush_device_output(self) -> None:
     def _clear_internal_resources(self) -> None:
         super()._clear_internal_resources()
         self._serial_connection = None
-        self._serial_pty_proc = None
+        self._proc = None
         self._serial_buffer.clear()
 
     @staticmethod
diff --git a/scripts/pylib/pytest-twister-harness/src/twister_harness/fixtures.py b/scripts/pylib/pytest-twister-harness/src/twister_harness/fixtures.py
index 765da163312ad..413f22a03a2e0 100644
--- a/scripts/pylib/pytest-twister-harness/src/twister_harness/fixtures.py
+++ b/scripts/pylib/pytest-twister-harness/src/twister_harness/fixtures.py
@@ -73,8 +73,11 @@ def dut(request: pytest.FixtureRequest, device_object: DeviceAdapter) -> Generat
 def shell(dut: DeviceAdapter) -> Shell:
     """Return ready to use shell interface"""
     shell = Shell(dut, timeout=20.0)
+
+    symbol = 'CONFIG_SHELL_PROMPT_RTT' if dut.device_config.use_rtt else 'CONFIG_SHELL_PROMPT_UART'
+
     if prompt := find_in_config(Path(dut.device_config.app_build_dir) / 'zephyr' / '.config',
-                                'CONFIG_SHELL_PROMPT_UART'):
+                                symbol):
         shell.prompt = prompt
     logger.info('Wait for prompt')
     if not shell.wait_for_prompt():
diff --git a/scripts/pylib/pytest-twister-harness/src/twister_harness/helpers/shell.py b/scripts/pylib/pytest-twister-harness/src/twister_harness/helpers/shell.py
index 0960db7743ae8..ceaa9f3d049be 100644
--- a/scripts/pylib/pytest-twister-harness/src/twister_harness/helpers/shell.py
+++ b/scripts/pylib/pytest-twister-harness/src/twister_harness/helpers/shell.py
@@ -54,24 +54,30 @@ def exec_command(
         self, command: str, timeout: float | None = None, print_output: bool = True
     ) -> list[str]:
         """
-        Send shell command to a device and return response. Passed command
-        is extended by double enter sings - first one to execute this command
-        on a device, second one to receive next prompt what is a signal that
-        execution was finished. Method returns printout of the executed command.
+        Send shell command to a device and return response. Method returns printout of
+        the executed command.
         """
         timeout = timeout or self.base_timeout
-        command_ext = f'{command}\n\n'
+        command_ext = f'{command}\n'
         regex_prompt = re.escape(self.prompt)
         regex_command = f'.*{re.escape(command)}'
+
+        # Execute command
         self._device.clear_buffer()
         self._device.write(command_ext.encode())
         lines: list[str] = []
-        # wait for device command print - it should be done immediately after sending command to device
+        # wait for device command print - it should be done immediately after sending
+        # command to device.
         lines.extend(
             self._device.readlines_until(
                 regex=regex_command, timeout=1.0, print_output=print_output
             )
         )
+
+        # Send single enter to get next prompt after command execution, as it signals
+        # that execution finished.
+        self._device.write(b"\n")
+
         # wait for device command execution
         lines.extend(
             self._device.readlines_until(
diff --git a/scripts/pylib/pytest-twister-harness/src/twister_harness/plugin.py b/scripts/pylib/pytest-twister-harness/src/twister_harness/plugin.py
index 0c471428b6db1..de539e587b781 100644
--- a/scripts/pylib/pytest-twister-harness/src/twister_harness/plugin.py
+++ b/scripts/pylib/pytest-twister-harness/src/twister_harness/plugin.py
@@ -70,13 +70,17 @@ def pytest_addoption(parser: pytest.Parser):
     )
     twister_harness_group.addoption(
         '--runner',
-        help='Use the specified west runner (pyocd, nrfjprog, etc.).'
+        help='Use the specified west runner (pyocd, nrfjprog, etc.) when running west flash.'
     )
     twister_harness_group.addoption(
         '--runner-params',
         action='append',
         help='Use the specified west runner params.'
     )
+    twister_harness_group.addoption(
+        '--rtt-runner',
+        help='Use the specified west runner (pyocd, nrfjprog, etc.) when running west rtt.'
+    )
     twister_harness_group.addoption(
         '--device-id',
         help='ID of connected hardware device (for example 000682459367).'
@@ -89,6 +93,11 @@ def pytest_addoption(parser: pytest.Parser):
         '--device-serial-pty',
         help='Script for controlling pseudoterminal.'
     )
+    twister_harness_group.addoption(
+        '--device-rtt',
+        type=bool,
+        help='Use RTT as communication transport.',
+    )
     twister_harness_group.addoption(
         '--flash-before',
         type=bool,
diff --git a/scripts/pylib/pytest-twister-harness/src/twister_harness/twister_harness_config.py b/scripts/pylib/pytest-twister-harness/src/twister_harness/twister_harness_config.py
index 807b72f10ff6c..31bf6fa0406ca 100644
--- a/scripts/pylib/pytest-twister-harness/src/twister_harness/twister_harness_config.py
+++ b/scripts/pylib/pytest-twister-harness/src/twister_harness/twister_harness_config.py
@@ -26,6 +26,8 @@ class DeviceConfig:
     baud: int = 115200
     runner: str = ''
     runner_params: list[str] = field(default_factory=list, repr=False)
+    rtt_runner: str = ''
+    use_rtt: bool = False
     id: str = ''
     product: str = ''
     serial_pty: str = ''
@@ -68,6 +70,10 @@ def create(cls, config: pytest.Config) -> TwisterHarnessConfig:
         runner_params: list[str] = []
         if config.option.runner_params:
             runner_params = [w.strip() for w in config.option.runner_params]
+        if config.option.device_rtt:
+            flash_before = True
+        else:
+            flash_before = bool(config.option.flash_before)
         device_from_cli = DeviceConfig(
             type=config.option.device_type,
             build_dir=_cast_to_path(config.option.build_dir),
@@ -78,10 +84,12 @@ def create(cls, config: pytest.Config) -> TwisterHarnessConfig:
             baud=config.option.device_serial_baud,
             runner=config.option.runner,
             runner_params=runner_params,
+            rtt_runner=config.option.rtt_runner,
+            use_rtt=config.option.device_rtt,
             id=config.option.device_id,
             product=config.option.device_product,
             serial_pty=config.option.device_serial_pty,
-            flash_before=bool(config.option.flash_before),
+            flash_before=flash_before,
             west_flash_extra_args=west_flash_extra_args,
             flash_command=flash_command,
             pre_script=_cast_to_path(config.option.pre_script),
diff --git a/scripts/pylib/pytest-twister-harness/tests/device/hardware_adapter_test.py b/scripts/pylib/pytest-twister-harness/tests/device/hardware_adapter_test.py
index eac79478074b9..958fe09c85abd 100644
--- a/scripts/pylib/pytest-twister-harness/tests/device/hardware_adapter_test.py
+++ b/scripts/pylib/pytest-twister-harness/tests/device/hardware_adapter_test.py
@@ -227,6 +227,7 @@ def test_if_hardware_adapter_uses_serial_pty(
 
     monkeypatch.setattr('twister_harness.device.hardware_adapter.pty.openpty', lambda: (123, 456))
     monkeypatch.setattr('twister_harness.device.hardware_adapter.os.ttyname', lambda x: f'/pty/ttytest/{x}')
+    monkeypatch.setattr('twister_harness.device.hardware_adapter.terminate_process', lambda x: None)
 
     serial_mock = mock.Mock()
     serial_mock.port = '/pty/ttytest/456'
@@ -235,7 +236,7 @@ def test_if_hardware_adapter_uses_serial_pty(
     device._device_run.set()
     device.connect()
     assert device._serial_connection.port == '/pty/ttytest/456'  # type: ignore[union-attr]
-    assert device._serial_pty_proc
+    assert device._proc
     patched_popen.assert_called_with(
         ['script.py'],
         stdout=123,
@@ -244,7 +245,7 @@ def test_if_hardware_adapter_uses_serial_pty(
     )
 
     device.disconnect()
-    assert not device._serial_pty_proc
+    assert not device._proc
 
 
 def test_if_hardware_adapter_properly_send_data_to_subprocess(
diff --git a/scripts/pylib/twister/twisterlib/environment.py b/scripts/pylib/twister/twisterlib/environment.py
index 47942ba1cd139..cfb099d6b2036 100644
--- a/scripts/pylib/twister/twisterlib/environment.py
+++ b/scripts/pylib/twister/twisterlib/environment.py
@@ -174,8 +174,9 @@ def add_parse_arguments(parser = None) -> argparse.ArgumentParser:
 
     run_group_option.add_argument(
         "--device-testing", action="store_true",
-        help="Test on device directly. Specify the serial device to "
-             "use with the --device-serial option.")
+        help="Test on device directly. Specify the serial device to use with "
+            "the --device-serial option or specify script to use with the "
+            "--device-serial-pty or specify to use with --device-rtt option.")
 
     run_group_option.add_argument("--generate-hardware-map",
                         help="""Probe serial devices connected to this platform
@@ -204,6 +205,15 @@ def add_parse_arguments(parser = None) -> argparse.ArgumentParser:
                         --device-serial-pty <script>
                         """)
 
+    device.add_argument("--device-rtt", action="store_true", default=False,
+                        help="""Use RTT as communication transport.
+                        Instead of using serial console, Twister will connect to the
+                        device with "west rtt" command to capture logs and interact with
+                        the test application. Setting this option automatically sets
+                        --flash-before option, as we can connect to the device only
+                        after it was flashed.
+                        """)
+
     device.add_argument("--hardware-map",
                         help="""Load hardware map from a file. This will be used
                         for testing on hardware that is listed in the file.
@@ -959,20 +969,23 @@ def parse_arguments(
 
     if (
         (not options.device_testing)
-        and (options.device_serial or options.device_serial_pty or options.hardware_map)
+        and (options.device_serial or options.device_serial_pty or options.device_rtt or
+             options.hardware_map)
     ):
         logger.error(
-            "Use --device-testing with --device-serial, or --device-serial-pty, or --hardware-map."
+            "Use --device-testing with --device-serial, or --device-serial-pty, or "
+            "--device-rtt, or --hardware-map."
         )
         sys.exit(1)
 
     if (
         options.device_testing
-        and (options.device_serial or options.device_serial_pty) and len(options.platform) != 1
+        and (options.device_serial or options.device_serial_pty or
+             options.device_rtt) and len(options.platform) != 1
     ):
         logger.error("When --device-testing is used with --device-serial "
-                     "or --device-serial-pty, exactly one platform must "
-                     "be specified")
+                     "or --device-serial-pty, or --device-rtt, "
+                     "exactly one platform must be specified")
         sys.exit(1)
 
     if options.device_flash_with_test and not options.device_testing:
diff --git a/scripts/pylib/twister/twisterlib/handlers.py b/scripts/pylib/twister/twisterlib/handlers.py
index 695888cc695cd..c084a072b48bf 100755
--- a/scripts/pylib/twister/twisterlib/handlers.py
+++ b/scripts/pylib/twister/twisterlib/handlers.py
@@ -515,7 +515,8 @@ def device_is_available(self, instance):
         for d in self.duts:
             if fixture and fixture not in map(lambda f: f.split(sep=':')[0], d.fixtures):
                 continue
-            if d.platform != device or (d.serial is None and d.serial_pty is None):
+            if d.platform != device or (d.serial is None and d.serial_pty is None and
+                                        d.use_rtt is False):
                 continue
             duts_found.append(d)
 
@@ -565,7 +566,7 @@ def run_custom_script(script, timeout):
                 proc.communicate()
                 logger.error(f"{script} timed out")
 
-    def _create_flash_command(self, hardware):
+    def _create_flash_command_from_option(self, hardware):
         flash_command = next(csv.reader([self.options.flash_command]))
 
         command = [flash_command[0]]
@@ -579,9 +580,28 @@ def _create_flash_command(self, hardware):
 
         return command
 
-    def _create_command(self, runner, hardware):
+    def _create_rtt_command(self, hardware):
+
+        command = ["west", "rtt", "--skip-rebuild", "-d", self.build_dir]
+        rtt_runner = hardware.rtt_runner
+
+        if rtt_runner:
+            command.append("--runner")
+            command.append(rtt_runner)
+
+            if rtt_runner in ("jlink", "pyocd", "probe-rs"):
+                board_id = hardware.probe_id or hardware.id
+
+                command.append("--dev-id")
+                command.append(board_id)
+
+        # Since _start_pty expects a string, we need to convert the command list into
+        # one.
+        return " ".join(command)
+
+    def _create_flash_command(self, runner, hardware):
         if self.options.flash_command:
-            return self._create_flash_command(hardware)
+            return self._create_flash_command_from_option(hardware)
 
         command = ["west"]
         if self.options.verbose > 2:
@@ -648,18 +668,18 @@ def _create_command(self, runner, hardware):
 
         return command
 
-    def _terminate_pty(self, ser_pty, ser_pty_process):
-        logger.debug(f"Terminating serial-pty:'{ser_pty}'")
-        terminate_process(ser_pty_process)
+    def _terminate_pty(self, pty_name, pty_process):
+        logger.debug(f"Terminating pty:'{pty_name}'")
+        terminate_process(pty_process)
         try:
-            (stdout, stderr) = ser_pty_process.communicate(timeout=self.get_test_timeout())
-            logger.debug(f"Terminated serial-pty:'{ser_pty}', stdout:'{stdout}', stderr:'{stderr}'")
+            (stdout, stderr) = pty_process.communicate(timeout=self.get_test_timeout())
+            logger.debug(f"Terminated pty:'{pty_name}', stdout:'{stdout}', stderr:'{stderr}'")
         except subprocess.TimeoutExpired:
-            logger.debug(f"Terminated serial-pty:'{ser_pty}'")
-    #
+            logger.debug(f"Terminated pty:'{pty_name}'")
+
 
     def _create_serial_connection(self, dut, serial_device, hardware_baud,
-                                  flash_timeout, serial_pty, ser_pty_process):
+                                  flash_timeout, pty_name, pty_process):
         try:
             ser = serial.Serial(
                 serial_device,
@@ -671,20 +691,20 @@ def _create_serial_connection(self, dut, serial_device, hardware_baud,
                 timeout=max(flash_timeout, self.get_test_timeout())
             )
         except serial.SerialException as e:
-            self._handle_serial_exception(e, dut, serial_pty, ser_pty_process)
+            self._handle_serial_exception(e, dut, pty_name, pty_process)
             raise
 
         return ser
 
 
-    def _handle_serial_exception(self, exception, dut, serial_pty, ser_pty_process):
+    def _handle_serial_exception(self, exception, dut, pty_name, pty_process):
         self.instance.status = TwisterStatus.FAIL
         self.instance.reason = "Serial Device Error"
         logger.error(f"Serial device error: {exception!s}")
 
         self.instance.add_missing_case_status(TwisterStatus.BLOCK, "Serial Device Error")
-        if serial_pty and ser_pty_process:
-            self._terminate_pty(serial_pty, ser_pty_process)
+        if pty_name and pty_process:
+            self._terminate_pty(pty_name, pty_process)
 
         self.make_dut_available(dut)
 
@@ -706,21 +726,21 @@ def get_hardware(self):
             logger.error(self.instance.reason)
         return hardware
 
-    def _start_serial_pty(self, serial_pty, serial_pty_master):
-        ser_pty_process = None
+    def _start_pty(self, command, pty_master):
+        pty_process = None
         try:
-            ser_pty_process = subprocess.Popen(
-                re.split('[, ]', serial_pty),
-                stdout=serial_pty_master,
-                stdin=serial_pty_master,
-                stderr=serial_pty_master
+            pty_process = subprocess.Popen(
+                re.split('[, ]', command),
+                stdout=pty_master,
+                stdin=pty_master,
+                stderr=pty_master
             )
         except subprocess.CalledProcessError as error:
             logger.error(
-                f"Failed to run subprocess {serial_pty}, error {error.output}"
+                f"Failed to run subprocess {command}, error {error.output}"
             )
 
-        return ser_pty_process
+        return pty_process
 
     def handle(self, harness):
         runner = None
@@ -732,16 +752,19 @@ def handle(self, harness):
 
         runner = hardware.runner or self.options.west_runner
         serial_pty = hardware.serial_pty
-
-        if not serial_pty:
-            serial_device = hardware.serial
+        use_rtt  = hardware.use_rtt
+        pty_name = None
+
+        if serial_pty or use_rtt:
+            pty_master, slave = pty.openpty()
+            pty_name = os.ttyname(slave)
+            reason = "RTT" if use_rtt else "serial PTY"
+            logger.debug(f"Using pty {pty_name} for {reason}")
         else:
-            ser_pty_master, slave = pty.openpty()
-            serial_device = os.ttyname(slave)
-
-        logger.debug(f"Using serial device {serial_device} @ {hardware.baud} baud")
+            serial_device = hardware.serial
+            logger.debug(f"Using serial device {serial_device} @ {hardware.baud} baud")
 
-        command = self._create_command(runner, hardware)
+        flash_command = self._create_flash_command(runner, hardware)
 
         pre_script = hardware.pre_script
         post_flash_script = hardware.post_flash_script
@@ -759,12 +782,16 @@ def handle(self, harness):
             flash_timeout += self.get_test_timeout()
 
         serial_port = None
-        ser_pty_process = None
+        pty_process = None
         if hardware.flash_before is False:
             if serial_pty:
-                ser_pty_process = self._start_serial_pty(serial_pty, ser_pty_master)
-            serial_port = serial_device
+                pty_process = self._start_pty(serial_pty, pty_master)
+                serial_port = pty_name
+            else:
+                serial_port = serial_device
 
+        # Open connection to serial device or if using serial pty just create ser object
+        # and open it later
         try:
             ser = self._create_serial_connection(
                 hardware,
@@ -772,7 +799,7 @@ def handle(self, harness):
                 hardware.baud,
                 flash_timeout,
                 serial_pty,
-                ser_pty_process
+                pty_process
             )
         except serial.SerialException:
             return
@@ -785,11 +812,13 @@ def handle(self, harness):
         t.start()
 
         d_log = f"{self.instance.build_dir}/device.log"
-        logger.debug(f'Flash command: {command}', )
+        logger.debug(f'Flash command: {flash_command}')
         failure_type = Handler.FailureType.NONE
         try:
             stdout = stderr = None
-            with subprocess.Popen(command, stderr=subprocess.PIPE, stdout=subprocess.PIPE) as proc:
+            with subprocess.Popen(flash_command,
+                                  stderr=subprocess.PIPE,
+                                  stdout=subprocess.PIPE) as proc:
                 try:
                     (stdout, stderr) = proc.communicate(timeout=flash_timeout)
                     # ignore unencodable unicode chars
@@ -828,10 +857,18 @@ def handle(self, harness):
         # Connect to device after flashing it
         if hardware.flash_before:
             try:
-                if serial_pty:
-                    ser_pty_process = self._start_serial_pty(serial_pty, ser_pty_master)
-                logger.debug(f"Attach serial device {serial_device} @ {hardware.baud} baud")
-                ser.port = serial_device
+                if use_rtt:
+                    rtt_command = self._create_rtt_command(hardware)
+                    pty_process = self._start_pty(rtt_command, pty_master)
+                    logger.debug(f"Start RTT connection on pty {pty_name}")
+                    ser.port = pty_name
+                elif serial_pty:
+                    pty_process = self._start_pty(serial_pty, pty_master)
+                    logger.debug(f"Start serial connection on pty {pty_name}")
+                    ser.port = pty_name
+                else:
+                    logger.debug(f"Attach serial device {serial_device} @ {hardware.baud} baud")
+                    ser.port = serial_device
 
                 # Apply ESP32-specific RTS/DTR reset logic
                 if runner == "esp32":
@@ -854,7 +891,7 @@ def handle(self, harness):
                     ser.open()
 
             except serial.SerialException as e:
-                self._handle_serial_exception(e, hardware, serial_pty, ser_pty_process)
+                self._handle_serial_exception(e, hardware, pty_name, pty_process)
                 return
 
         if failure_type != Handler.FailureType.FLASH:
@@ -876,8 +913,8 @@ def handle(self, harness):
         if ser.isOpen():
             ser.close()
 
-        if serial_pty:
-            self._terminate_pty(serial_pty, ser_pty_process)
+        if pty_name or use_rtt:
+            self._terminate_pty(pty_name, pty_process)
 
         self.execution_time = time.time() - start_time
 
diff --git a/scripts/pylib/twister/twisterlib/hardwaremap.py b/scripts/pylib/twister/twisterlib/hardwaremap.py
index 74c384ab5d6eb..6e31ef1d332cb 100644
--- a/scripts/pylib/twister/twisterlib/hardwaremap.py
+++ b/scripts/pylib/twister/twisterlib/hardwaremap.py
@@ -40,6 +40,7 @@ def __init__(self,
                  platform=None,
                  product=None,
                  serial_pty=None,
+                 use_rtt=False,
                  connected=False,
                  runner_params=None,
                  pre_script=None,
@@ -47,6 +48,7 @@ def __init__(self,
                  post_flash_script=None,
                  script_param=None,
                  runner=None,
+                 rtt_runner=None,
                  flash_timeout=60,
                  flash_with_test=False,
                  flash_before=False):
@@ -55,6 +57,7 @@ def __init__(self,
         self.baud = serial_baud or 115200
         self.platform = platform
         self.serial_pty = serial_pty
+        self.use_rtt = use_rtt
         self._counter = Value("i", 0)
         self._available = Value("i", 1)
         self._failures = Value("i", 0)
@@ -64,6 +67,7 @@ def __init__(self,
         self.product = product
         self.runner = runner
         self.runner_params = runner_params
+        self.rtt_runner = rtt_runner
         self.flash_before = flash_before
         self.fixtures = []
         self.post_flash_script = post_flash_script
@@ -126,7 +130,8 @@ def to_dict(self):
 
 
     def __repr__(self):
-        return f"<{self.platform} ({self.product}) on {self.serial}>"
+        comm_type = "RTT" if self.use_rtt else self.serial
+        return f"<{self.platform} ({self.product}) on {comm_type}>"
 
 class HardwareMap:
     schema_path = os.path.join(ZEPHYR_BASE, "scripts", "schemas", "twister", "hwmap-schema.yaml")
@@ -199,6 +204,7 @@ def discover(self):
                                 self.options.platform[0],
                                 self.options.pre_script,
                                 False,
+                                False,
                                 baud=self.options.device_serial_baud,
                                 flash_timeout=self.options.device_flash_timeout,
                                 flash_with_test=self.options.device_flash_with_test,
@@ -210,11 +216,24 @@ def discover(self):
                                 self.options.platform[0],
                                 self.options.pre_script,
                                 True,
+                                False,
                                 flash_timeout=self.options.device_flash_timeout,
                                 flash_with_test=self.options.device_flash_with_test,
                                 flash_before=self.options.flash_before,
                                 )
 
+            elif self.options.device_rtt:
+                # RTT always requires flashing before connecting to work properly.
+                self.add_device(self.options.device_serial_pty,
+                                self.options.platform[0],
+                                self.options.pre_script,
+                                False,
+                                True,
+                                flash_timeout=self.options.device_flash_timeout,
+                                flash_with_test=self.options.device_flash_with_test,
+                                flash_before=True,
+                                )
+
             # the fixtures given by twister command explicitly should be assigned to each DUT
             if self.options.fixture:
                 for d in self.duts:
@@ -239,6 +258,7 @@ def add_device(
         platform,
         pre_script,
         is_pty,
+        uses_rtt,
         baud=None,
         flash_timeout=60,
         flash_with_test=False,
@@ -253,6 +273,9 @@ def add_device(
             flash_with_test=flash_with_test,
             flash_before=flash_before
         )
+
+        if uses_rtt:
+            device.use_rtt = True
         if is_pty:
             device.serial_pty = serial
         else:
@@ -273,7 +296,11 @@ def load(self, map_file):
             if flash_with_test is None:
                 flash_with_test = self.options.device_flash_with_test
             serial_pty = dut.get('serial_pty')
+            use_rtt = dut.get('rtt')
             flash_before = dut.get('flash_before')
+            if use_rtt:
+                # RTT always requires flashing before connecting to work properly.
+                flash_before = True
             if flash_before is None:
                 flash_before = self.options.flash_before and (not flash_with_test)
             platform = dut.get('platform')
@@ -286,11 +313,12 @@ def load(self, map_file):
             id = dut.get('id')
             runner = dut.get('runner')
             runner_params = dut.get('runner_params')
+            rtt_runner = dut.get('rtt_runner')
             serial = dut.get('serial')
             baud = dut.get('baud', None)
             product = dut.get('product')
             fixtures = dut.get('fixtures', [])
-            connected = dut.get('connected') and ((serial or serial_pty) is not None)
+            connected = dut.get('connected') and ((serial or serial_pty or use_rtt) is not None)
             if not connected:
                 continue
             for plat in platforms:
@@ -298,7 +326,9 @@ def load(self, map_file):
                               product=product,
                               runner=runner,
                               runner_params=runner_params,
+                              rtt_runner=rtt_runner,
                               id=id,
+                              use_rtt=use_rtt,
                               serial_pty=serial_pty,
                               serial=serial,
                               serial_baud=baud,
@@ -472,7 +502,7 @@ def dump(self, filtered=None, header=None, connected_only=False, detected=False)
             to_show = self.duts
 
         if not header:
-            header = ["Platform", "ID", "Serial device"]
+            header = ["Platform", "ID", "Communication type"]
         for p in to_show:
             platform = p.platform
             connected = p.connected
@@ -480,6 +510,13 @@ def dump(self, filtered=None, header=None, connected_only=False, detected=False)
                 continue
 
             if not connected_only or connected:
-                table.append([platform, p.id, p.serial])
+                if p.use_rtt:
+                    comm_type = "RTT"
+                elif p.serial_pty:
+                    comm_type = "PTY"
+                else:
+                    comm_type = f"Serial: {p.serial}"
+
+                table.append([platform, p.id, comm_type])
 
         print(tabulate(table, headers=header, tablefmt="github"))
diff --git a/scripts/pylib/twister/twisterlib/harness.py b/scripts/pylib/twister/twisterlib/harness.py
index 3e8511e9d618d..4dbf201144485 100644
--- a/scripts/pylib/twister/twisterlib/harness.py
+++ b/scripts/pylib/twister/twisterlib/harness.py
@@ -480,6 +480,12 @@ def _generate_parameters_for_hardware(self, handler: Handler):
             for param in hardware.runner_params:
                 command.append(f'--runner-params={param}')
 
+        if hardware.use_rtt:
+            command.append(f'--device-rtt={hardware.use_rtt}')
+
+        if hardware.rtt_runner:
+            command.append(f'--rtt-runner={hardware.rtt_runner}')
+
         if options.west_flash and options.west_flash != []:
             command.append(f'--west-flash-extra-args={options.west_flash}')
 
diff --git a/scripts/schemas/twister/hwmap-schema.yaml b/scripts/schemas/twister/hwmap-schema.yaml
index 142d4a1969bfa..3b12514d00274 100644
--- a/scripts/schemas/twister/hwmap-schema.yaml
+++ b/scripts/schemas/twister/hwmap-schema.yaml
@@ -32,6 +32,12 @@ sequence:
         required: false
         sequence:
           - type: str
+      "rtt_runner":
+        type: str
+        required: false
+      "rtt":
+        type: bool
+        required: false
       "serial_pty":
         type: str
         required: false
diff --git a/scripts/tests/twister/test_environment.py b/scripts/tests/twister/test_environment.py
index 33f0c3ab9cd1e..faa76894964ec 100644
--- a/scripts/tests/twister/test_environment.py
+++ b/scripts/tests/twister/test_environment.py
@@ -52,8 +52,8 @@
             'dummy',
         ],
         'When --device-testing is used with --device-serial' \
-        ' or --device-serial-pty, exactly one platform must' \
-        ' be specified'
+        ' or --device-serial-pty, or --device-rtt, exactly ' \
+        'one platform must be specified'
     ),
     (
         None,
@@ -69,8 +69,8 @@
             'dummy_platform2'
         ],
         'When --device-testing is used with --device-serial' \
-        ' or --device-serial-pty, exactly one platform must' \
-        ' be specified'
+        ' or --device-serial-pty, or --device-rtt, exactly ' \
+        'one platform must be specified'
     ),
 # Note the underscore.
     (
diff --git a/scripts/tests/twister/test_handlers.py b/scripts/tests/twister/test_handlers.py
index 5bec199c642b6..9fb8db1f55808 100644
--- a/scripts/tests/twister/test_handlers.py
+++ b/scripts/tests/twister/test_handlers.py
@@ -761,6 +761,7 @@ def test_devicehandler_monitor_serial_splitlines(mocked_instance):
                 platform='dummy_platform',
                 serial_pty=None,
                 serial=None,
+                use_rtt=False,
                 available=1,
                 failures=0,
                 counter_increment=mock.Mock(),
@@ -812,6 +813,7 @@ def test_devicehandler_monitor_serial_splitlines(mocked_instance):
                 platform='dummy_platform',
                 serial_pty=None,
                 serial=None,
+                use_rtt=False,
                 available=1,
                 failures=0,
                 counter_increment=mock.Mock(),
@@ -1200,7 +1202,7 @@ def test_devicehandler_create_command(
         runner_params=['param1', 'param2']
     )
 
-    command = handler._create_command(runner, hardware)
+    command = handler._create_flash_command(runner, hardware)
 
     assert command == expected
 
@@ -1348,7 +1350,7 @@ def mock_popen(command, *args, **kwargs):
     popen_mock = mock.Mock(side_effect=mock_popen)
 
     with mock.patch('subprocess.Popen', popen_mock):
-        result = handler._start_serial_pty(serial_pty, 'master')
+        result = handler._start_pty(serial_pty, 'master')
 
     if popen_exception:
         assert result is None
@@ -1369,8 +1371,8 @@ def mock_popen(command, *args, **kwargs):
     (True, False, False, False, 0, True, False,
      TwisterStatus.NONE, None, ['Timed out while monitoring serial output on IPName']),
     (True, False, False, False, 0, False, True,
-     TwisterStatus.NONE, None, ["Terminating serial-pty:'Serial PTY'",
-                                "Terminated serial-pty:'Serial PTY', stdout:'', stderr:''"]),
+     TwisterStatus.NONE, None, ["Terminating pty:'slave name'",
+                                "Terminated pty:'slave name', stdout:'', stderr:''"]),
 ]
 
 @pytest.mark.parametrize(
@@ -1441,6 +1443,7 @@ def mock_popen(command, *args, **kwargs):
         runner='dummy runner',
         serial_pty='Serial PTY' if use_pty else None,
         serial='dummy serial',
+        use_rtt=False,
         pre_script='dummy pre script',
         post_script='dummy post script',
         post_flash_script='dummy post flash script',
@@ -1456,8 +1459,8 @@ def mock_popen(command, *args, **kwargs):
         west_runner=None,
         verbose=0
     )
-    handler._start_serial_pty = mock.Mock(side_effect=mock_start_serial_pty)
-    handler._create_command = mock.Mock(return_value=['dummy', 'command'])
+    handler._start_pty = mock.Mock(side_effect=mock_start_serial_pty)
+    handler._create_flash_command = mock.Mock(return_value=['dummy', 'command'])
     handler.run_custom_script = mock.Mock()
     handler._create_serial_connection = mock.Mock(
         side_effect=mock_create_serial
diff --git a/scripts/tests/twister/test_hardwaremap.py b/scripts/tests/twister/test_hardwaremap.py
index c8f8267921157..468f1ec93372e 100644
--- a/scripts/tests/twister/test_hardwaremap.py
+++ b/scripts/tests/twister/test_hardwaremap.py
@@ -253,7 +253,8 @@ def test_hardwaremap_add_device(is_pty):
     serial = 'dummy'
     platform = 'p0'
     pre_script = 'dummy pre script'
-    hm.add_device(serial, platform, pre_script, is_pty)
+    uses_rtt = False
+    hm.add_device(serial, platform, pre_script, is_pty, uses_rtt)
 
     assert len(hm.duts) == 1
     if is_pty:
@@ -686,11 +687,11 @@ def mock_open(filename, mode='r'):
         True,
         True,
 """
-| Platform   |   ID | Serial device   |
-|------------|------|-----------------|
-| p1         |    1 | s1              |
-| p3         |    3 | s3              |
-| p5         |    5 | s5              |
+| Platform   |   ID | Communication type   |
+|------------|------|----------------------|
+| p1         |    1 | Serial: s1           |
+| p3         |    3 | Serial: s3           |
+| p5         |    5 | Serial: s5           |
 """
     ),
     (
@@ -699,16 +700,16 @@ def mock_open(filename, mode='r'):
         False,
         False,
 """
-| ?   |   ?? | ???   |
-|-----|------|-------|
-| p1  |    1 | s1    |
-| p2  |    2 | s2    |
-| p3  |    3 | s3    |
-| p4  |    4 | s4    |
-| p5  |    5 | s5    |
-| p6  |    6 | s6    |
-| p7  |    7 | s7    |
-| p8  |    8 | s8    |
+| ?   |   ?? | ???        |
+|-----|------|------------|
+| p1  |    1 | Serial: s1 |
+| p2  |    2 | Serial: s2 |
+| p3  |    3 | Serial: s3 |
+| p4  |    4 | Serial: s4 |
+| p5  |    5 | Serial: s5 |
+| p6  |    6 | Serial: s6 |
+| p7  |    7 | Serial: s7 |
+| p8  |    8 | Serial: s8 |
 """
     ),
 ]
diff --git a/scripts/west_commands/runners/core.py b/scripts/west_commands/runners/core.py
index 611848f671cd9..e4b0fe8e4e669 100644
--- a/scripts/west_commands/runners/core.py
+++ b/scripts/west_commands/runners/core.py
@@ -13,6 +13,7 @@
 
 import abc
 import argparse
+import contextlib
 import errno
 import logging
 import os
@@ -965,20 +966,138 @@ def run_telnet_client(self, host: str, port: int, active_sock=None) -> None:
             sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
             sock.connect((host, port))
 
-        # Otherwise, use a pure python implementation. This will work well for logging,
-        # but input is line based only.
+        if platform.system() == 'Windows':
+            self._windows_telnet_client(sock)
+        else:
+            self._unix_telnet_client(sock)
+
+
+    def _unix_telnet_client(self, sock: socket.socket) -> None:
+        """
+        Start a telnet client on Unix.
+
+        Selectors are used to register both the given socket and the stdin and wait on
+        incoming data, to later process it.
+        """
+        import termios
+        import tty
+
         sel = selectors.DefaultSelector()
         sel.register(sys.stdin, selectors.EVENT_READ)
         sel.register(sock, selectors.EVENT_READ)
-        while True:
-            events = sel.select()
-            for key, _ in events:
-                if key.fileobj == sys.stdin:
-                    text = sys.stdin.readline()
-                    if text:
-                        sock.send(text.encode())
-
-                elif key.fileobj == sock:
+
+
+        # We can't type check functions from tty and termios on Windows operating
+        # systems: mypy thinks the modules have no such attributes.
+        fd = sys.stdin.fileno()
+        orig_attr = termios.tcgetattr(fd) # type: ignore
+        try:
+            # Enable cbreak mode on the keyboard input, that way all shell editing
+            # commands (arrow keys, backspace, etc.) work as expected.
+            tty.setcbreak(fd) # type: ignore
+            while True:
+                events = sel.select()
+                for key, _ in events:
+                    if key.fileobj == sys.stdin:
+                        data = os.read(fd, 2048)
+                        if data:
+                            sock.send(data)
+
+                    elif key.fileobj == sock:
+                        resp = sock.recv(2048)
+                        if resp:
+                            print(resp.decode(), end='', flush=True)
+        except OSError as e:
+            print("\n\nCommunication was closed, reason:", e)
+        finally:
+            termios.tcsetattr(fd, termios.TCSADRAIN, orig_attr) # type: ignore
+
+
+    def _windows_telnet_client(self, sock):
+        """
+        Start a telnet client on Windows.
+
+        Since we don't have tty package on the Windows, a different approach to
+        establish two-way communction:
+        - Two threads are started.
+        - One uses the kbhit() and getch() from msvcrt package to read every single
+          character from input.
+        - The other one reads data from the socket and prints it to the screen.
+
+        Additional logic is need to convert arrow keys and navigation keys into
+        corresponding ANSI escape sequences
+        Additional logic with the stop event is needed to handle CTRL+C and any errors.
+        """
+        import msvcrt
+        import threading
+
+        def send_keyboard_input(sock: socket.socket,
+                                stop: threading.Event,
+                                err: threading.Event):
+            KEY_MAP = {
+                # Arrow keys
+                b'\xe0H': b'\x1b[A',  # Up
+                b'\xe0P': b'\x1b[B',  # Down
+                b'\xe0K': b'\x1b[D',  # Left
+                b'\xe0M': b'\x1b[C',  # Right
+
+                # Navigation keys
+                b'\xe0G': b'\x1b[H',  # Home
+                b'\xe0O': b'\x1b[F',  # End
+                b'\xe0R': b'\x1b[2~', # Insert
+                b'\xe0S': b'\x1b[3~', # Delete
+            }
+
+            try:
+                # We can't type check functions from msvcrt on Unix operating systems:
+                # mypy thinks the module has no such attributes.
+                while not stop.is_set():
+                    if msvcrt.kbhit(): # type: ignore
+                        ch = msvcrt.getch() # type: ignore
+                        if ch == b'\xe0':
+                            ch2 = msvcrt.getch() # type: ignore
+                            if ansi := KEY_MAP.get(ch + ch2):
+                                sock.sendall(ansi)
+                        elif ch == b'\r':
+                            # Normalize CR to CRLF
+                            sock.sendall(b"\r\n")
+                        else:
+                            sock.sendall(ch)
+            except OSError as e:
+                if not err.is_set() and not stop.is_set():
+                    err.set()
+                    print("\n\nCommunication was closed, reason:", e)
+                stop.set()
+
+        def receive_rtt_data(sock: socket.socket,
+                            stop: threading.Event,
+                            err: threading.Event):
+            try:
+                while not stop.is_set():
                     resp = sock.recv(2048)
-                    if resp:
-                        print(resp.decode(), end='')
+                    if not resp:
+                        print("\n\nConnection closed by the remote host, exiting!")
+                        stop.set()
+                        break
+                    print(resp.decode(errors="replace"), end='', flush=True)
+            except OSError as e:
+                if not err.is_set() and not stop.is_set():
+                    err.set()
+                    print("\n\nCommunication was closed, reason:", e)
+                stop.set()
+
+        stop = threading.Event()
+        err = threading.Event()
+        threading.Thread(target=send_keyboard_input, args=(sock, stop, err)).start()
+        threading.Thread(target=receive_rtt_data, args=(sock, stop, err)).start()
+
+        try:
+            while not stop.is_set():
+                # Wait for an event in the loop. Without timeout the
+                # KeyboardInterrupt would not be detected.
+                stop.wait(timeout=0.1)
+        except KeyboardInterrupt:
+            stop.set()
+        finally:
+            with contextlib.suppress(OSError):
+                sock.close()