diff --git a/ipykernel/kernelbase.py b/ipykernel/kernelbase.py
index e391411e4..bfd107127 100644
--- a/ipykernel/kernelbase.py
+++ b/ipykernel/kernelbase.py
@@ -10,6 +10,7 @@
 import itertools
 import logging
 import inspect
+import os
 from signal import signal, default_int_handler, SIGINT
 import sys
 import time
@@ -198,7 +199,7 @@ def _parent_header(self):
         'inspect_request', 'history_request',
         'comm_info_request', 'kernel_info_request',
         'connect_request', 'shutdown_request',
-        'is_complete_request',
+        'is_complete_request', 'interrupt_request',
         # deprecated:
         'apply_request',
     ]
@@ -780,6 +781,30 @@ async def comm_info_request(self, stream, ident, parent):
                                 reply_content, parent, ident)
         self.log.debug("%s", msg)
 
+    async def interrupt_request(self, stream, ident, parent):
+        pid = os.getpid()
+        pgid = os.getpgid(pid)
+
+        if os.name == "nt":
+            self.log.error("Interrupt message not supported on Windows")
+
+        else:
+            # Prefer process-group over process
+            if pgid and hasattr(os, "killpg"):
+                try:
+                    os.killpg(pgid, SIGINT)
+                    return
+                except OSError:
+                    pass
+            try:
+                os.kill(pid, SIGINT)
+            except OSError:
+                pass
+
+        content = parent['content']
+        self.session.send(stream, 'interrupt_reply', content, parent, ident=ident)
+        return
+
     async def shutdown_request(self, stream, ident, parent):
         content = self.do_shutdown(parent['content']['restart'])
         if inspect.isawaitable(content):
diff --git a/ipykernel/tests/test_kernel.py b/ipykernel/tests/test_kernel.py
index a2df85fc6..6724d0f6d 100644
--- a/ipykernel/tests/test_kernel.py
+++ b/ipykernel/tests/test_kernel.py
@@ -391,6 +391,26 @@ def test_interrupt_during_input():
         validate_message(reply, 'execute_reply', msg_id)
 
 
+@pytest.mark.skipif(
+    os.name == "nt",
+    reason="Message based interrupt not supported on Windows"
+)
+def test_interrupt_with_message():
+    """
+
+    """
+    with new_kernel() as kc:
+        km = kc.parent
+        km.kernel_spec.interrupt_mode = "message"
+        msg_id = kc.execute("input()")
+        time.sleep(1)  # Make sure it's actually waiting for input.
+        km.interrupt_kernel()
+        from .test_message_spec import validate_message
+        # If we failed to interrupt interrupt, this will timeout:
+        reply = get_reply(kc, msg_id, TIMEOUT)
+        validate_message(reply, 'execute_reply', msg_id)
+
+
 @pytest.mark.skipif(
     version.parse(IPython.__version__) < version.parse("7.14.0"),
     reason="Need new IPython"