googleapis
@@ -0,0 +1,89 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""An async client for interacting with Google Cloud Storage using the gRPC API."""
+
+from google.cloud import _storage_v2 as storage_v2
+
+
+class AsyncGrpcClient:
+    """An asynchronous client for interacting with Google Cloud Storage using the gRPC API.
+
+    :type credentials: :class:`~google.auth.credentials.Credentials`
+    :param credentials: (Optional) The OAuth2 Credentials to use for this
+                        client. If not passed, falls back to the default
+                        inferred from the environment.
+
+    :type client_info: :class:`~google.api_core.client_info.ClientInfo`
+    :param client_info:
+        The client info used to send a user-agent string along with API
+        requests. If ``None``, then default info will be used.
+
+    :type client_options: :class:`~google.api_core.client_options.ClientOptions` or :class:`dict`
+    :param client_options: (Optional) Client options used to set user options
+        on the client.
+
+    :type attempt_direct_path: bool
+    :param attempt_direct_path:
+        (Optional) Whether to attempt to use DirectPath for gRPC connections.
+        Defaults to ``True``.
+    """
+
+    def __init__(
+        self,
+        credentials=None,
+        client_info=None,
+        client_options=None,
+        *,
+        attempt_direct_path=True,
+    ):
+        self._grpc_client = self._create_async_grpc_client(
+            credentials=credentials,
+            client_info=client_info,
+            client_options=client_options,
+            attempt_direct_path=attempt_direct_path,
+        )
+
+    def _create_async_grpc_client(
+        self,
+        credentials=None,
+        client_info=None,
+        client_options=None,
+        attempt_direct_path=True,
+    ):
+        transport_cls = storage_v2.StorageAsyncClient.get_transport_class(
+            "grpc_asyncio"
+        )
+        channel = transport_cls.create_channel(attempt_direct_path=attempt_direct_path)
+        transport = transport_cls(credentials=credentials, channel=channel)
+
+        return storage_v2.StorageAsyncClient(
+            credentials=credentials,
+            transport=transport,
+            client_info=client_info,
+            client_options=client_options,
+        )
+
+    @property
+    def grpc_client(self):
+        """The underlying gRPC client.
+
+        This property gives users direct access to the `_storage_v2.StorageAsyncClient`
+         instance. This can be useful for accessing
+        newly added or experimental RPCs that are not yet exposed through
+        the high-level GrpcClient.
+        Returns:
+            google.cloud._storage_v2.StorageAsyncClient: The configured GAPIC client.
+        """
+        return self._grpc_client
@@ -0,0 +1,85 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import unittest
+from unittest import mock
+from google.auth import credentials as auth_credentials
+
+
+def _make_credentials(spec=None):
+    if spec is None:
+        return mock.Mock(spec=auth_credentials.Credentials)
+    return mock.Mock(spec=spec)
+
+
+class TestAsyncGrpcClient(unittest.TestCase):
+    @mock.patch("google.cloud._storage_v2.StorageAsyncClient")
+    def test_constructor_default_options(self, mock_async_storage_client):
+        from google.cloud.storage._experimental import async_grpc_client
+
+        mock_transport_cls = mock.MagicMock()
+        mock_async_storage_client.get_transport_class.return_value = mock_transport_cls
+        mock_creds = _make_credentials()
+
+        async_grpc_client.AsyncGrpcClient(credentials=mock_creds)
+
+        mock_async_storage_client.get_transport_class.assert_called_once_with(
+            "grpc_asyncio"
+        )
+        mock_transport_cls.create_channel.assert_called_once_with(
+            attempt_direct_path=True
+        )
+        mock_channel = mock_transport_cls.create_channel.return_value
+        mock_transport_cls.assert_called_once_with(
+            credentials=mock_creds, channel=mock_channel
+        )
+        mock_transport = mock_transport_cls.return_value
+        mock_async_storage_client.assert_called_once_with(
+            credentials=mock_creds,
+            transport=mock_transport,
+            client_options=None,
+            client_info=None,
+        )
+
+    @mock.patch("google.cloud._storage_v2.StorageAsyncClient")
+    def test_constructor_disables_directpath(self, mock_async_storage_client):
+        from google.cloud.storage._experimental import async_grpc_client
+
+        mock_transport_cls = mock.MagicMock()
+        mock_async_storage_client.get_transport_class.return_value = mock_transport_cls
+        mock_creds = _make_credentials()
+
+        async_grpc_client.AsyncGrpcClient(
+            credentials=mock_creds, attempt_direct_path=False
+        )
+
+        mock_transport_cls.create_channel.assert_called_once_with(
+            attempt_direct_path=False
+        )
+        mock_channel = mock_transport_cls.create_channel.return_value
+        mock_transport_cls.assert_called_once_with(
+            credentials=mock_creds, channel=mock_channel
+        )
+
+    @mock.patch("google.cloud._storage_v2.StorageAsyncClient")
+    def test_grpc_client_property(self, mock_async_storage_client):
+        from google.cloud.storage._experimental import async_grpc_client
+
+        mock_creds = _make_credentials()
+
+        client = async_grpc_client.AsyncGrpcClient(credentials=mock_creds)
+
+        retrieved_client = client.grpc_client
+
+        self.assertIs(retrieved_client, mock_async_storage_client.return_value)