diff --git a/txaws/newsfragments/78.feature b/txaws/newsfragments/78.feature
new file mode 100644
index 00000000..3db1bdc2
--- /dev/null
+++ b/txaws/newsfragments/78.feature
@@ -0,0 +1 @@
+txaws.s3.client.S3Client.get_bucket now accepts a ``prefix`` parameter for selecting a subset of S3 objects.
diff --git a/txaws/s3/client.py b/txaws/s3/client.py
index c2041717..c230cd1d 100644
--- a/txaws/s3/client.py
+++ b/txaws/s3/client.py
@@ -191,7 +191,7 @@ def delete_bucket(self, bucket):
         query = self._query_factory(details)
         return self._submit(query)
 
-    def get_bucket(self, bucket, marker=None, max_keys=None):
+    def get_bucket(self, bucket, marker=None, max_keys=None, prefix=None):
         """
         Get a list of all the objects in a bucket.
 
@@ -205,6 +205,10 @@ def get_bucket(self, bucket, marker=None, max_keys=None):
             return.
         @type max_keys: L{int} or L{NoneType}
 
+        @param prefix: If given, indicate that only objects with keys
+            beginning with this value should be returned.
+        @type prefix: L{bytes} or L{NoneType}
+
         @return: A L{Deferred} that fires with a L{BucketListing}
             describing the result.
 
@@ -215,6 +219,8 @@ def get_bucket(self, bucket, marker=None, max_keys=None):
             args.append(("marker", marker))
         if max_keys is not None:
             args.append(("max-keys", "%d" % (max_keys,)))
+        if prefix is not None:
+            args.append(("prefix", prefix))
         if args:
             object_name = "?" + urlencode(args)
         else:
diff --git a/txaws/s3/tests/test_client.py b/txaws/s3/tests/test_client.py
index 2fa8737a..cdd5fdd0 100644
--- a/txaws/s3/tests/test_client.py
+++ b/txaws/s3/tests/test_client.py
@@ -292,6 +292,25 @@ def check_query_args(passthrough):
         d.addCallback(check_query_args)
         return d
 
+    def test_get_bucket_prefix(self):
+        """
+        L{S3Client.get_bucket} accepts a C{prefix} argument to ask the server to
+        limit its response to objects beginning with a certain prefix.
+        """
+        query_factory = mock_query_factory(payload.sample_get_bucket_result)
+        def check_query_args(passthrough):
+            self.assertEqual(
+                b"http:///mybucket/?prefix=foobar",
+                query_factory.details.url_context.get_encoded_url(),
+            )
+            return passthrough
+
+        creds = AWSCredentials("foo", "bar")
+        s3 = client.S3Client(creds, query_factory=query_factory)
+        d = s3.get_bucket("mybucket", prefix=b"foobar")
+        d.addCallback(check_query_args)
+        return d
+
     def test_get_bucket_location(self):
         """
         L{S3Client.get_bucket_location} creates a L{Query} to get a bucket's
diff --git a/txaws/testing/s3.py b/txaws/testing/s3.py
index 3279745e..78f642dd 100644
--- a/txaws/testing/s3.py
+++ b/txaws/testing/s3.py
@@ -42,6 +42,9 @@ class S3ClientState(object):
     ``S3ClientState`` instances hold the ``_MemoryS3Client`` instance
     state that is specific to testing and does not exist on
     ``txaws.s3.S3Client`` instances.
+
+    @ivar buckets: A ``dict`` mapping bucket identifiers to ``dict`` of
+        ``Bucket`` and ``BucketListing`` details.
     """
     from time import time
 
@@ -91,12 +94,24 @@ def delete_bucket(self, bucket):
         return succeed(None)
 
     @_rate_limited
-    def get_bucket(self, bucket):
+    def get_bucket(self, bucket, prefix=None):
         try:
             pieces = self._state.buckets[bucket]
         except KeyError:
             return fail(S3Error("<nosuchbucket/>", 400))
-        return succeed(pieces["listing"])
+        listing = pieces["listing"]
+        if prefix is not None:
+            listing = attr.assoc(
+                listing,
+                contents=list(
+                    content
+                    for content
+                    in listing.contents
+                    if content.key.startswith(prefix)
+                ),
+                prefix=prefix,
+            )
+        return succeed(listing)
 
     @_rate_limited
     def get_bucket_location(self, bucket):
diff --git a/txaws/testing/s3_tests.py b/txaws/testing/s3_tests.py
index ee0e8bb0..27caa505 100644
--- a/txaws/testing/s3_tests.py
+++ b/txaws/testing/s3_tests.py
@@ -101,6 +101,21 @@ def test_objects(self):
                 "Expected to not find deleted objects in listing {}".format(objects),
             )
 
+        @inlineCallbacks
+        def test_get_bucket_prefix(self):
+            """
+            A subset of S3 objects in a bucket can be retrieved by specifying a value
+            for the ``prefix`` argument to ``get_bucket``.
+            """
+            bucket_name = str(uuid4())
+            client = get_client(self)
+            yield client.create_bucket(bucket_name)
+            yield client.put_object(bucket_name, b"a", b"foo")
+            yield client.put_object(bucket_name, b"b", b"bar")
+
+            objects = yield client.get_bucket(bucket_name, prefix=b"a")
+            self.assertEqual([b"a"], list(obj.key for obj in objects.contents))
+
         def test_get_bucket_location_empty(self):
             """
             When called for a bucket with no explicit location,