diff --git a/docs/proposed/http-storage-node-protocol.rst b/docs/proposed/http-storage-node-protocol.rst
index 3926d9f4a..693ce9290 100644
--- a/docs/proposed/http-storage-node-protocol.rst
+++ b/docs/proposed/http-storage-node-protocol.rst
@@ -738,7 +738,7 @@ Reading
 ``GET /v1/mutable/:storage_index/shares``
 !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
 
-Retrieve a list indicating all shares available for the indicated storage index.
+Retrieve a set indicating all shares available for the indicated storage index.
 For example::
 
   [1, 5]
diff --git a/src/allmydata/storage/http_client.py b/src/allmydata/storage/http_client.py
index da350e0c6..5920d5a5b 100644
--- a/src/allmydata/storage/http_client.py
+++ b/src/allmydata/storage/http_client.py
@@ -106,6 +106,11 @@ _SCHEMAS = {
         share_number = uint
         """
     ),
+    "mutable_list_shares": Schema(
+        """
+        response = #6.258([* uint])
+        """
+    ),
 }
 
 
@@ -720,3 +725,18 @@ class StorageClientMutables:
         return read_share_chunk(
             self._client, "mutable", storage_index, share_number, offset, length
         )
+
+    @async_to_deferred
+    async def list_shares(self, storage_index: bytes) -> set[int]:
+        """
+        List the share numbers for a given storage index.
+        """
+        # TODO unit test all the things
+        url = self._client.relative_url(
+            "/v1/mutable/{}/shares".format(_encode_si(storage_index))
+        )
+        response = await self._client.request("GET", url)
+        if response.code == http.OK:
+            return await _decode_cbor(response, _SCHEMAS["mutable_list_shares"])
+        else:
+            raise ClientException(response.code)
diff --git a/src/allmydata/storage/http_server.py b/src/allmydata/storage/http_server.py
index 0169d1463..0b407a1c4 100644
--- a/src/allmydata/storage/http_server.py
+++ b/src/allmydata/storage/http_server.py
@@ -645,6 +645,20 @@ class HTTPServer(object):
             )
         return data
 
+    @_authorized_route(
+        _app,
+        set(),
+        "/v1/mutable/<storage_index:storage_index>/shares",
+        methods=["GET"],
+    )
+    def list_mutable_shares(self, request, authorization, storage_index):
+        """List mutable shares for a storage index."""
+        try:
+            shares = self._storage_server.list_mutable_shares(storage_index)
+        except KeyError:
+            raise _HTTPError(http.NOT_FOUND)
+        return self._send_encoded(request, shares)
+
 
 @implementer(IStreamServerEndpoint)
 @attr.s
diff --git a/src/allmydata/storage/server.py b/src/allmydata/storage/server.py
index 9d1a3d6a4..1a0255601 100644
--- a/src/allmydata/storage/server.py
+++ b/src/allmydata/storage/server.py
@@ -1,18 +1,9 @@
 """
 Ported to Python 3.
 """
-from __future__ import division
-from __future__ import absolute_import
-from __future__ import print_function
-from __future__ import unicode_literals
-
-from future.utils import bytes_to_native_str, PY2
-if PY2:
-    # Omit open() to get native behavior where open("w") always accepts native
-    # strings. Omit bytes so we don't leak future's custom bytes.
-    from future.builtins import filter, map, zip, ascii, chr, hex, input, next, oct, pow, round, super, dict, list, object, range, str, max, min  # noqa: F401
-else:
-    from typing import Dict, Tuple
+from __future__ import annotations
+from future.utils import bytes_to_native_str
+from typing import Dict, Tuple
 
 import os, re
 
@@ -699,6 +690,25 @@ class StorageServer(service.MultiService):
                                          self)
         return share
 
+    def list_mutable_shares(self, storage_index) -> set[int]:
+        """List all share numbers for the given mutable.
+
+        Raises ``KeyError`` if the storage index is not known.
+        """
+        # TODO unit test
+        si_dir = storage_index_to_dir(storage_index)
+        # shares exist if there is a file for them
+        bucketdir = os.path.join(self.sharedir, si_dir)
+        if not os.path.isdir(bucketdir):
+            raise KeyError("Not found")
+        result = set()
+        for sharenum_s in os.listdir(bucketdir):
+            try:
+                result.add(int(sharenum_s))
+            except ValueError:
+                continue
+        return result
+
     def slot_readv(self, storage_index, shares, readv):
         start = self._clock.seconds()
         self.count("readv")
diff --git a/src/allmydata/storage_client.py b/src/allmydata/storage_client.py
index 68164e697..8b2f68a9e 100644
--- a/src/allmydata/storage_client.py
+++ b/src/allmydata/storage_client.py
@@ -1196,9 +1196,10 @@ class _HTTPStorageServer(object):
         mutable_client = StorageClientMutables(self._http_client)
         pending_reads = {}
         reads = {}
-        # TODO if shares list is empty, that means list all shares, so we need
+        # If shares list is empty, that means list all shares, so we need
         # to do a query to get that.
-        assert shares  # TODO replace with call to list shares if and only if it's empty
+        if not shares:
+            shares = yield mutable_client.list_shares(storage_index)
 
         # Start all the queries in parallel:
         for share_number in shares:
diff --git a/src/allmydata/test/test_istorageserver.py b/src/allmydata/test/test_istorageserver.py
index e7b869713..d9fd13acb 100644
--- a/src/allmydata/test/test_istorageserver.py
+++ b/src/allmydata/test/test_istorageserver.py
@@ -1154,5 +1154,4 @@ class HTTPMutableAPIsTests(
         "test_add_lease_renewal",
         "test_add_new_lease",
         "test_advise_corrupt_share",
-        "test_slot_readv_no_shares",
     }