diff --git a/gcloud/bigtable/client.py b/gcloud/bigtable/client.py
index f5a27c277e81..59eb0596d294 100644
--- a/gcloud/bigtable/client.py
+++ b/gcloud/bigtable/client.py
@@ -400,3 +400,23 @@ def list_zones(self):
                     zone.display_name,))
             result.append(zone.display_name)
         return result
+
+    def list_clusters(self):
+        """Lists clusters owned by the project.
+
+        :rtype: tuple
+        :returns: A pair of results, the first is a list of :class:`.Cluster` s
+                  returned and the second is a list of strings (the failed
+                  zones in the request).
+        """
+        request_pb = messages_pb2.ListClustersRequest(name=self.project_name)
+        response = self._cluster_stub.ListClusters.async(request_pb,
+                                                         self.timeout_seconds)
+        # We expect a `.messages_pb2.ListClustersResponse`
+        list_clusters_response = response.result()
+
+        failed_zones = [zone.display_name
+                        for zone in list_clusters_response.failed_zones]
+        clusters = [Cluster.from_pb(cluster_pb, self)
+                    for cluster_pb in list_clusters_response.clusters]
+        return clusters, failed_zones
diff --git a/gcloud/bigtable/test_client.py b/gcloud/bigtable/test_client.py
index 5684975f4baa..c2e12bd601ad 100644
--- a/gcloud/bigtable/test_client.py
+++ b/gcloud/bigtable/test_client.py
@@ -547,6 +547,70 @@ def test_list_zones_failure(self):
         with self.assertRaises(ValueError):
             self._list_zones_helper(data_pb2.Zone.EMERGENCY_MAINENANCE)
 
+    def test_list_clusters(self):
+        from gcloud.bigtable._generated import (
+            bigtable_cluster_data_pb2 as data_pb2)
+        from gcloud.bigtable._generated import (
+            bigtable_cluster_service_messages_pb2 as messages_pb2)
+
+        credentials = _Credentials()
+        project = 'PROJECT'
+        timeout_seconds = 8004
+        client = self._makeOne(project=project, credentials=credentials,
+                               admin=True, timeout_seconds=timeout_seconds)
+
+        # Create request_pb
+        request_pb = messages_pb2.ListClustersRequest(
+            name='projects/' + project,
+        )
+
+        # Create response_pb
+        zone = 'foo'
+        failed_zone = 'bar'
+        cluster_id1 = 'cluster-id1'
+        cluster_id2 = 'cluster-id2'
+        cluster_name1 = ('projects/' + project + '/zones/' + zone +
+                         '/clusters/' + cluster_id1)
+        cluster_name2 = ('projects/' + project + '/zones/' + zone +
+                         '/clusters/' + cluster_id2)
+        response_pb = messages_pb2.ListClustersResponse(
+            failed_zones=[
+                data_pb2.Zone(display_name=failed_zone),
+            ],
+            clusters=[
+                data_pb2.Cluster(
+                    name=cluster_name1,
+                    display_name=cluster_name1,
+                    serve_nodes=3,
+                ),
+                data_pb2.Cluster(
+                    name=cluster_name2,
+                    display_name=cluster_name2,
+                    serve_nodes=3,
+                ),
+            ],
+        )
+
+        # Patch the stub used by the API method.
+        client._cluster_stub_internal = stub = _FakeStub(response_pb)
+
+        # Create expected_result.
+        failed_zones = [failed_zone]
+        clusters = [
+            client.cluster(zone, cluster_id1),
+            client.cluster(zone, cluster_id2),
+        ]
+        expected_result = (clusters, failed_zones)
+
+        # Perform the method and check the result.
+        result = client.list_clusters()
+        self.assertEqual(result, expected_result)
+        self.assertEqual(stub.method_calls, [(
+            'ListClusters',
+            (request_pb, timeout_seconds),
+            {},
+        )])
+
 
 class _Credentials(object):