diff --git a/Lib/asyncio/base_events.py b/Lib/asyncio/base_events.py
index 6e6e5aaac15caf..6fe6328be6610e 100644
--- a/Lib/asyncio/base_events.py
+++ b/Lib/asyncio/base_events.py
@@ -50,7 +50,7 @@
 from .log import logger
 
 
-__all__ = 'BaseEventLoop','Server',
+__all__ = 'BaseEventLoop','Server','crash_exception_handler'
 
 
 # Minimum number of _scheduled timer handles before cleanup of
@@ -414,6 +414,21 @@ async def wait_closed(self):
         await waiter
 
 
+def crash_exception_handler(self, context):
+    exception = context.get("exception")
+    if exception is None:
+        message = context.get("message")
+        if message is None:
+            message = "Unhandled exception in event loop"
+        exception = RuntimeError(message)
+
+    if events._get_running_loop() is self:
+        self.crash(exception)
+        return
+
+    self.call_soon_threadsafe(self.crash, exception)
+
+
 class BaseEventLoop(events.AbstractEventLoop):
 
     def __init__(self):
@@ -447,6 +462,7 @@ def __init__(self):
         self._asyncgens_shutdown_called = False
         # Set to True when `loop.shutdown_default_executor` is called.
         self._executor_shutdown_called = False
+        self._exceptions = []
 
     def __repr__(self):
         return (
@@ -682,6 +698,14 @@ def run_forever(self):
             while True:
                 self._run_once()
                 if self._stopping:
+                    if self._exceptions:
+                        try:
+                            raise BaseExceptionGroup(
+                                "errors occured in asyncio callbacks",
+                                self._exceptions,
+                            )
+                        finally:
+                            self._exceptions = []
                     break
         finally:
             self._run_forever_cleanup()
@@ -724,6 +748,10 @@ def run_until_complete(self, future):
 
         return future.result()
 
+    def crash(self, exception):
+        self._exceptions.append(exception)
+        self.stop()
+
     def stop(self):
         """Stop running the event loop.
 
diff --git a/Lib/asyncio/events.py b/Lib/asyncio/events.py
index 2ee9870e80f20b..7435dbae8793a6 100644
--- a/Lib/asyncio/events.py
+++ b/Lib/asyncio/events.py
@@ -665,6 +665,9 @@ def get_debug(self):
     def set_debug(self, enabled):
         raise NotImplementedError
 
+    def crash(self, exception):
+        raise NotImplementedError
+
 
 class _AbstractEventLoopPolicy:
     """Abstract policy for accessing the event loop."""
diff --git a/Lib/test/test_asyncio/test_base_events.py b/Lib/test/test_asyncio/test_base_events.py
index 1e063c1352ecb9..6e87141d350e0d 100644
--- a/Lib/test/test_asyncio/test_base_events.py
+++ b/Lib/test/test_asyncio/test_base_events.py
@@ -674,6 +674,43 @@ def handler(loop, context):
                     'Unhandled error in exception handler'),
                 exc_info=(AttributeError, MOCK_ANY, MOCK_ANY))
 
+    def test_set_exception_handler_crash_handler(self):
+        self.loop._process_events = mock.Mock()
+        self.loop.set_exception_handler(asyncio.crash_exception_handler)
+
+        def crash(e):
+            raise e
+
+        self.loop.call_soon(crash, RuntimeError("hello"))
+        self.loop.call_soon(crash, ValueError("world"))
+
+        with self.assertRaises(ExceptionGroup) as exc_info:
+            self.loop.run_forever()
+
+        self.assertIsInstance(exc_info.exception.exceptions[0], RuntimeError)
+        self.assertIsInstance(exc_info.exception.exceptions[1], ValueError)
+
+    def test_set_exception_handler_crash_handler_be(self):
+        self.loop._process_events = mock.Mock()
+        self.loop.set_exception_handler(asyncio.crash_exception_handler)
+
+        class MyBaseException(BaseException):
+            pass
+
+        def crash(e):
+            raise e
+
+        self.loop.call_soon(crash, RuntimeError("hello"))
+        self.loop.call_soon(crash, ValueError("world"))
+        self.loop.call_soon(crash, MyBaseException("mbe"))
+
+        with self.assertRaises(BaseExceptionGroup) as exc_info:
+            self.loop.run_forever()
+
+        self.assertIsInstance(exc_info.exception.exceptions[0], RuntimeError)
+        self.assertIsInstance(exc_info.exception.exceptions[1], ValueError)
+        self.assertIsInstance(exc_info.exception.exceptions[2], MyBaseException)
+
     def test_default_exc_handler_broken(self):
         _context = None