diff --git a/Doc/library/asyncio-eventloop.rst b/Doc/library/asyncio-eventloop.rst
index 3911d2caf5c..ea1d146f06c 100644
--- a/Doc/library/asyncio-eventloop.rst
+++ b/Doc/library/asyncio-eventloop.rst
@@ -778,7 +778,7 @@ Creating network servers
                           *, sock=None, backlog=100, ssl=None, \
                           ssl_handshake_timeout=None, \
                           ssl_shutdown_timeout=None, \
-                          start_serving=True)
+                          start_serving=True, cleanup_socket=True)
 
    Similar to :meth:`loop.create_server` but works with the
    :py:const:`~socket.AF_UNIX` socket family.
@@ -788,6 +788,10 @@ Creating network servers
    :class:`str`, :class:`bytes`, and :class:`~pathlib.Path` paths
    are supported.
 
+   If *cleanup_socket* is True then the Unix socket will automatically
+   be removed from the filesystem when the server is closed, unless the
+   socket has been replaced after the server has been created.
+
    See the documentation of the :meth:`loop.create_server` method
    for information about arguments to this method.
 
@@ -802,6 +806,10 @@ Creating network servers
 
       Added the *ssl_shutdown_timeout* parameter.
 
+   .. versionchanged:: 3.13
+
+      Added the *cleanup_socket* parameter.
+
 
 .. coroutinemethod:: loop.connect_accepted_socket(protocol_factory, \
                         sock, *, ssl=None, ssl_handshake_timeout=None, \
diff --git a/Doc/whatsnew/3.13.rst b/Doc/whatsnew/3.13.rst
index 291e276dc67..428b648d9e1 100644
--- a/Doc/whatsnew/3.13.rst
+++ b/Doc/whatsnew/3.13.rst
@@ -149,6 +149,13 @@ array
   It can be used instead of ``'u'`` type code, which is deprecated.
   (Contributed by Inada Naoki in :gh:`80480`.)
 
+asyncio
+-------
+
+* :meth:`asyncio.loop.create_unix_server` will now automatically remove
+  the Unix socket when the server is closed.
+  (Contributed by Pierre Ossman in :gh:`111246`.)
+
 copy
 ----
 
diff --git a/Lib/asyncio/unix_events.py b/Lib/asyncio/unix_events.py
index c9441918ecf..41ccf1b78fb 100644
--- a/Lib/asyncio/unix_events.py
+++ b/Lib/asyncio/unix_events.py
@@ -64,6 +64,7 @@ class _UnixSelectorEventLoop(selector_events.BaseSelectorEventLoop):
     def __init__(self, selector=None):
         super().__init__(selector)
         self._signal_handlers = {}
+        self._unix_server_sockets = {}
 
     def close(self):
         super().close()
@@ -284,7 +285,7 @@ class _UnixSelectorEventLoop(selector_events.BaseSelectorEventLoop):
             sock=None, backlog=100, ssl=None,
             ssl_handshake_timeout=None,
             ssl_shutdown_timeout=None,
-            start_serving=True):
+            start_serving=True, cleanup_socket=True):
         if isinstance(ssl, bool):
             raise TypeError('ssl argument must be an SSLContext or None')
 
@@ -340,6 +341,15 @@ class _UnixSelectorEventLoop(selector_events.BaseSelectorEventLoop):
                 raise ValueError(
                     f'A UNIX Domain Stream Socket was expected, got {sock!r}')
 
+        if cleanup_socket:
+            path = sock.getsockname()
+            # Check for abstract socket. `str` and `bytes` paths are supported.
+            if path[0] not in (0, '\x00'):
+                try:
+                    self._unix_server_sockets[sock] = os.stat(path).st_ino
+                except FileNotFoundError:
+                    pass
+
         sock.setblocking(False)
         server = base_events.Server(self, [sock], protocol_factory,
                                     ssl, backlog, ssl_handshake_timeout,
@@ -460,6 +470,27 @@ class _UnixSelectorEventLoop(selector_events.BaseSelectorEventLoop):
                     self.remove_writer(fd)
         fut.add_done_callback(cb)
 
+    def _stop_serving(self, sock):
+        # Is this a unix socket that needs cleanup?
+        if sock in self._unix_server_sockets:
+            path = sock.getsockname()
+        else:
+            path = None
+
+        super()._stop_serving(sock)
+
+        if path is not None:
+            prev_ino = self._unix_server_sockets[sock]
+            del self._unix_server_sockets[sock]
+            try:
+                if os.stat(path).st_ino == prev_ino:
+                    os.unlink(path)
+            except FileNotFoundError:
+                pass
+            except OSError as err:
+                logger.error('Unable to clean up listening UNIX socket '
+                             '%r: %r', path, err)
+
 
 class _UnixReadPipeTransport(transports.ReadTransport):
 
diff --git a/Lib/test/test_asyncio/test_server.py b/Lib/test/test_asyncio/test_server.py
index 7ff3f55f4f0..f22cf3026e2 100644
--- a/Lib/test/test_asyncio/test_server.py
+++ b/Lib/test/test_asyncio/test_server.py
@@ -1,4 +1,6 @@
 import asyncio
+import os
+import socket
 import time
 import threading
 import unittest
@@ -177,6 +179,80 @@ class TestServer2(unittest.IsolatedAsyncioTestCase):
 
 
 
+# Test the various corner cases of Unix server socket removal
+class UnixServerCleanupTests(unittest.IsolatedAsyncioTestCase):
+    @socket_helper.skip_unless_bind_unix_socket
+    async def test_unix_server_addr_cleanup(self):
+        # Default scenario
+        with test_utils.unix_socket_path() as addr:
+            async def serve(*args):
+                pass
+
+            srv = await asyncio.start_unix_server(serve, addr)
+
+            srv.close()
+            self.assertFalse(os.path.exists(addr))
+
+    @socket_helper.skip_unless_bind_unix_socket
+    async def test_unix_server_sock_cleanup(self):
+        # Using already bound socket
+        with test_utils.unix_socket_path() as addr:
+            async def serve(*args):
+                pass
+
+            sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+            sock.bind(addr)
+
+            srv = await asyncio.start_unix_server(serve, sock=sock)
+
+            srv.close()
+            self.assertFalse(os.path.exists(addr))
+
+    @socket_helper.skip_unless_bind_unix_socket
+    async def test_unix_server_cleanup_gone(self):
+        # Someone else has already cleaned up the socket
+        with test_utils.unix_socket_path() as addr:
+            async def serve(*args):
+                pass
+
+            sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+            sock.bind(addr)
+
+            srv = await asyncio.start_unix_server(serve, sock=sock)
+
+            os.unlink(addr)
+
+            srv.close()
+
+    @socket_helper.skip_unless_bind_unix_socket
+    async def test_unix_server_cleanup_replaced(self):
+        # Someone else has replaced the socket with their own
+        with test_utils.unix_socket_path() as addr:
+            async def serve(*args):
+                pass
+
+            srv = await asyncio.start_unix_server(serve, addr)
+
+            os.unlink(addr)
+            sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+            sock.bind(addr)
+
+            srv.close()
+            self.assertTrue(os.path.exists(addr))
+
+    @socket_helper.skip_unless_bind_unix_socket
+    async def test_unix_server_cleanup_prevented(self):
+        # Automatic cleanup explicitly disabled
+        with test_utils.unix_socket_path() as addr:
+            async def serve(*args):
+                pass
+
+            srv = await asyncio.start_unix_server(serve, addr, cleanup_socket=False)
+
+            srv.close()
+            self.assertTrue(os.path.exists(addr))
+
+
 @unittest.skipUnless(hasattr(asyncio, 'ProactorEventLoop'), 'Windows only')
 class ProactorStartServerTests(BaseStartServer, unittest.TestCase):
 
diff --git a/Misc/NEWS.d/next/Library/2023-10-30-14-47-23.gh-issue-111246.QJ_ehs.rst b/Misc/NEWS.d/next/Library/2023-10-30-14-47-23.gh-issue-111246.QJ_ehs.rst
new file mode 100644
index 00000000000..a9630de5cc6
--- /dev/null
+++ b/Misc/NEWS.d/next/Library/2023-10-30-14-47-23.gh-issue-111246.QJ_ehs.rst
@@ -0,0 +1,2 @@
+:meth:`asyncio.loop.create_unix_server` will now automatically remove the
+Unix socket when the server is closed.