diff --git a/google/api_core/client_info.py b/google/api_core/client_info.py
index 4832679..90926be 100644
--- a/google/api_core/client_info.py
+++ b/google/api_core/client_info.py
@@ -57,7 +57,8 @@
         user_agent (Optional[str]): Prefix to the user agent header. This is
             used to supply information such as application name or partner tool.
             Recommended format: ``application-or-tool-ID/major.minor.version``.
-        rest_version (Optional[str]): The requests library version.
+        rest_version (Optional[str]): A string with labeled versions of the
+            dependencies used for REST transport.
     """
 
     def __init__(
diff --git a/google/api_core/gapic_v1/client_info.py b/google/api_core/gapic_v1/client_info.py
index 2de1be7..4516f33 100644
--- a/google/api_core/gapic_v1/client_info.py
+++ b/google/api_core/gapic_v1/client_info.py
@@ -45,6 +45,8 @@
         user_agent (Optional[str]): Prefix to the user agent header. This is
             used to supply information such as application name or partner tool.
             Recommended format: ``application-or-tool-ID/major.minor.version``.
+        rest_version (Optional[str]): A string with labeled versions of the
+            dependencies used for REST transport.
     """
 
     def to_grpc_metadata(self):
diff --git a/google/api_core/operations_v1/__init__.py b/google/api_core/operations_v1/__init__.py
index 8b75426..4db32a4 100644
--- a/google/api_core/operations_v1/__init__.py
+++ b/google/api_core/operations_v1/__init__.py
@@ -14,9 +14,7 @@
 
 """Package for interacting with the google.longrunning.operations meta-API."""
 
-from google.api_core.operations_v1.abstract_operations_client import (
-    AbstractOperationsClient,
-)
+from google.api_core.operations_v1.abstract_operations_client import AbstractOperationsClient
 from google.api_core.operations_v1.operations_async_client import OperationsAsyncClient
 from google.api_core.operations_v1.operations_client import OperationsClient
 from google.api_core.operations_v1.transports.rest import OperationsRestTransport
@@ -25,5 +23,18 @@
     "AbstractOperationsClient",
     "OperationsAsyncClient",
     "OperationsClient",
-    "OperationsRestTransport",
+    "OperationsRestTransport"
 ]
+
+try:
+    from google.api_core.operations_v1.transports.rest_asyncio import (
+        AsyncOperationsRestTransport,
+    )
+    from google.api_core.operations_v1.operations_rest_client_async import AsyncOperationsRestClient
+
+    __all__ += ["AsyncOperationsRestClient", "AsyncOperationsRestTransport"]
+except ImportError:
+    # This import requires the `async_rest` extra.
+    # Don't raise an exception if `AsyncOperationsRestTransport` cannot be imported
+    # as other transports are still available.
+    pass
diff --git a/google/api_core/operations_v1/abstract_operations_base_client.py b/google/api_core/operations_v1/abstract_operations_base_client.py
new file mode 100644
index 0000000..160c2a8
--- /dev/null
+++ b/google/api_core/operations_v1/abstract_operations_base_client.py
@@ -0,0 +1,370 @@
+# -*- coding: utf-8 -*-
+# Copyright 2024 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.
+
+from collections import OrderedDict
+import os
+import re
+from typing import Dict, Optional, Type, Union
+
+from google.api_core import client_options as client_options_lib  # type: ignore
+from google.api_core import gapic_v1  # type: ignore
+from google.api_core.operations_v1.transports.base import (
+    DEFAULT_CLIENT_INFO,
+    OperationsTransport,
+)
+from google.api_core.operations_v1.transports.rest import OperationsRestTransport
+
+try:
+    from google.api_core.operations_v1.transports.rest_asyncio import (
+        AsyncOperationsRestTransport,
+    )
+
+    HAS_ASYNC_REST_DEPENDENCIES = True
+except ImportError as e:
+    HAS_ASYNC_REST_DEPENDENCIES = False
+    ASYNC_REST_EXCEPTION = e
+
+from google.auth import credentials as ga_credentials  # type: ignore
+from google.auth.exceptions import MutualTLSChannelError  # type: ignore
+from google.auth.transport import mtls  # type: ignore
+
+
+class AbstractOperationsBaseClientMeta(type):
+    """Metaclass for the Operations Base client.
+
+    This provides base class-level methods for building and retrieving
+    support objects (e.g. transport) without polluting the client instance
+    objects.
+    """
+
+    _transport_registry = OrderedDict()  # type: Dict[str, Type[OperationsTransport]]
+    _transport_registry["rest"] = OperationsRestTransport
+    if HAS_ASYNC_REST_DEPENDENCIES:
+        _transport_registry["rest_asyncio"] = AsyncOperationsRestTransport
+
+    def get_transport_class(
+        cls,
+        label: Optional[str] = None,
+    ) -> Type[OperationsTransport]:
+        """Returns an appropriate transport class.
+
+        Args:
+            label: The name of the desired transport. If none is
+                provided, then the first transport in the registry is used.
+
+        Returns:
+            The transport class to use.
+        """
+        # If a specific transport is requested, return that one.
+        if (
+            label == "rest_asyncio" and not HAS_ASYNC_REST_DEPENDENCIES
+        ):  # pragma: NO COVER
+            raise ASYNC_REST_EXCEPTION
+
+        if label:
+            return cls._transport_registry[label]
+
+        # No transport is requested; return the default (that is, the first one
+        # in the dictionary).
+        return next(iter(cls._transport_registry.values()))
+
+
+class AbstractOperationsBaseClient(metaclass=AbstractOperationsBaseClientMeta):
+    """Manages long-running operations with an API service.
+
+    When an API method normally takes long time to complete, it can be
+    designed to return [Operation][google.api_core.operations_v1.Operation] to the
+    client, and the client can use this interface to receive the real
+    response asynchronously by polling the operation resource, or pass
+    the operation resource to another API (such as Google Cloud Pub/Sub
+    API) to receive the response. Any API service that returns
+    long-running operations should implement the ``Operations``
+    interface so developers can have a consistent client experience.
+    """
+
+    @staticmethod
+    def _get_default_mtls_endpoint(api_endpoint):
+        """Converts api endpoint to mTLS endpoint.
+
+        Convert "*.sandbox.googleapis.com" and "*.googleapis.com" to
+        "*.mtls.sandbox.googleapis.com" and "*.mtls.googleapis.com" respectively.
+        Args:
+            api_endpoint (Optional[str]): the api endpoint to convert.
+        Returns:
+            str: converted mTLS api endpoint.
+        """
+        if not api_endpoint:
+            return api_endpoint
+
+        mtls_endpoint_re = re.compile(
+            r"(?P<name>[^.]+)(?P<mtls>\.mtls)?(?P<sandbox>\.sandbox)?(?P<googledomain>\.googleapis\.com)?"
+        )
+
+        m = mtls_endpoint_re.match(api_endpoint)
+        name, mtls, sandbox, googledomain = m.groups()
+        if mtls or not googledomain:
+            return api_endpoint
+
+        if sandbox:
+            return api_endpoint.replace(
+                "sandbox.googleapis.com", "mtls.sandbox.googleapis.com"
+            )
+
+        return api_endpoint.replace(".googleapis.com", ".mtls.googleapis.com")
+
+    DEFAULT_ENDPOINT = "longrunning.googleapis.com"
+    DEFAULT_MTLS_ENDPOINT = _get_default_mtls_endpoint.__func__(  # type: ignore
+        DEFAULT_ENDPOINT
+    )
+
+    @classmethod
+    def from_service_account_info(cls, info: dict, *args, **kwargs):
+        """
+        This class method should be overridden by the subclasses.
+
+        Args:
+            info (dict): The service account private key info.
+            args: Additional arguments to pass to the constructor.
+            kwargs: Additional arguments to pass to the constructor.
+
+        Raises:
+            NotImplementedError: If the method is called on the base class.
+        """
+        raise NotImplementedError("`from_service_account_info` is not implemented.")
+
+    @classmethod
+    def from_service_account_file(cls, filename: str, *args, **kwargs):
+        """
+        This class method should be overridden by the subclasses.
+
+        Args:
+            filename (str): The path to the service account private key json
+                file.
+            args: Additional arguments to pass to the constructor.
+            kwargs: Additional arguments to pass to the constructor.
+
+        Raises:
+            NotImplementedError: If the method is called on the base class.
+        """
+        raise NotImplementedError("`from_service_account_file` is not implemented.")
+
+    from_service_account_json = from_service_account_file
+
+    @property
+    def transport(self) -> OperationsTransport:
+        """Returns the transport used by the client instance.
+
+        Returns:
+            OperationsTransport: The transport used by the client
+                instance.
+        """
+        return self._transport
+
+    @staticmethod
+    def common_billing_account_path(
+        billing_account: str,
+    ) -> str:
+        """Returns a fully-qualified billing_account string."""
+        return "billingAccounts/{billing_account}".format(
+            billing_account=billing_account,
+        )
+
+    @staticmethod
+    def parse_common_billing_account_path(path: str) -> Dict[str, str]:
+        """Parse a billing_account path into its component segments."""
+        m = re.match(r"^billingAccounts/(?P<billing_account>.+?)$", path)
+        return m.groupdict() if m else {}
+
+    @staticmethod
+    def common_folder_path(
+        folder: str,
+    ) -> str:
+        """Returns a fully-qualified folder string."""
+        return "folders/{folder}".format(
+            folder=folder,
+        )
+
+    @staticmethod
+    def parse_common_folder_path(path: str) -> Dict[str, str]:
+        """Parse a folder path into its component segments."""
+        m = re.match(r"^folders/(?P<folder>.+?)$", path)
+        return m.groupdict() if m else {}
+
+    @staticmethod
+    def common_organization_path(
+        organization: str,
+    ) -> str:
+        """Returns a fully-qualified organization string."""
+        return "organizations/{organization}".format(
+            organization=organization,
+        )
+
+    @staticmethod
+    def parse_common_organization_path(path: str) -> Dict[str, str]:
+        """Parse a organization path into its component segments."""
+        m = re.match(r"^organizations/(?P<organization>.+?)$", path)
+        return m.groupdict() if m else {}
+
+    @staticmethod
+    def common_project_path(
+        project: str,
+    ) -> str:
+        """Returns a fully-qualified project string."""
+        return "projects/{project}".format(
+            project=project,
+        )
+
+    @staticmethod
+    def parse_common_project_path(path: str) -> Dict[str, str]:
+        """Parse a project path into its component segments."""
+        m = re.match(r"^projects/(?P<project>.+?)$", path)
+        return m.groupdict() if m else {}
+
+    @staticmethod
+    def common_location_path(
+        project: str,
+        location: str,
+    ) -> str:
+        """Returns a fully-qualified location string."""
+        return "projects/{project}/locations/{location}".format(
+            project=project,
+            location=location,
+        )
+
+    @staticmethod
+    def parse_common_location_path(path: str) -> Dict[str, str]:
+        """Parse a location path into its component segments."""
+        m = re.match(r"^projects/(?P<project>.+?)/locations/(?P<location>.+?)$", path)
+        return m.groupdict() if m else {}
+
+    def __init__(
+        self,
+        *,
+        credentials: Optional[ga_credentials.Credentials] = None,
+        transport: Union[str, OperationsTransport, None] = None,
+        client_options: Optional[client_options_lib.ClientOptions] = None,
+        client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO,
+    ) -> None:
+        """Instantiates the operations client.
+
+        Args:
+            credentials (Optional[google.auth.credentials.Credentials]): The
+                authorization credentials to attach to requests. These
+                credentials identify the application to the service; if none
+                are specified, the client will attempt to ascertain the
+                credentials from the environment.
+            transport (Union[str, OperationsTransport]): The
+                transport to use. If set to None, a transport is chosen
+                automatically.
+            client_options (google.api_core.client_options.ClientOptions): Custom options for the
+                client. It won't take effect if a ``transport`` instance is provided.
+                (1) The ``api_endpoint`` property can be used to override the
+                default endpoint provided by the client. GOOGLE_API_USE_MTLS_ENDPOINT
+                environment variable can also be used to override the endpoint:
+                "always" (always use the default mTLS endpoint), "never" (always
+                use the default regular endpoint) and "auto" (auto switch to the
+                default mTLS endpoint if client certificate is present, this is
+                the default value). However, the ``api_endpoint`` property takes
+                precedence if provided.
+                (2) If GOOGLE_API_USE_CLIENT_CERTIFICATE environment variable
+                is "true", then the ``client_cert_source`` property can be used
+                to provide client certificate for mutual TLS transport. If
+                not provided, the default SSL client certificate will be used if
+                present. If GOOGLE_API_USE_CLIENT_CERTIFICATE is "false" or not
+                set, no client certificate will be used.
+            client_info (google.api_core.gapic_v1.client_info.ClientInfo):
+                The client info used to send a user-agent string along with
+                API requests. If ``None``, then default info will be used.
+                Generally, you only need to set this if you're developing
+                your own client library.
+
+        Raises:
+            google.auth.exceptions.MutualTLSChannelError: If mutual TLS transport
+                creation failed for any reason.
+        """
+        if isinstance(client_options, dict):
+            client_options = client_options_lib.from_dict(client_options)
+        if client_options is None:
+            client_options = client_options_lib.ClientOptions()
+
+        # Create SSL credentials for mutual TLS if needed.
+        use_client_cert = os.getenv(
+            "GOOGLE_API_USE_CLIENT_CERTIFICATE", "false"
+        ).lower()
+        if use_client_cert not in ("true", "false"):
+            raise ValueError(
+                "Environment variable `GOOGLE_API_USE_CLIENT_CERTIFICATE` must be either `true` or `false`"
+            )
+        client_cert_source_func = None
+        is_mtls = False
+        if use_client_cert == "true":
+            if client_options.client_cert_source:
+                is_mtls = True
+                client_cert_source_func = client_options.client_cert_source
+            else:
+                is_mtls = mtls.has_default_client_cert_source()
+                if is_mtls:
+                    client_cert_source_func = mtls.default_client_cert_source()
+                else:
+                    client_cert_source_func = None
+
+        # Figure out which api endpoint to use.
+        if client_options.api_endpoint is not None:
+            api_endpoint = client_options.api_endpoint
+        else:
+            use_mtls_env = os.getenv("GOOGLE_API_USE_MTLS_ENDPOINT", "auto")
+            if use_mtls_env == "never":
+                api_endpoint = self.DEFAULT_ENDPOINT
+            elif use_mtls_env == "always":
+                api_endpoint = self.DEFAULT_MTLS_ENDPOINT
+            elif use_mtls_env == "auto":
+                if is_mtls:
+                    api_endpoint = self.DEFAULT_MTLS_ENDPOINT
+                else:
+                    api_endpoint = self.DEFAULT_ENDPOINT
+            else:
+                raise MutualTLSChannelError(
+                    "Unsupported GOOGLE_API_USE_MTLS_ENDPOINT value. Accepted "
+                    "values: never, auto, always"
+                )
+
+        # Save or instantiate the transport.
+        # Ordinarily, we provide the transport, but allowing a custom transport
+        # instance provides an extensibility point for unusual situations.
+        if isinstance(transport, OperationsTransport):
+            # transport is a OperationsTransport instance.
+            if credentials or client_options.credentials_file:
+                raise ValueError(
+                    "When providing a transport instance, "
+                    "provide its credentials directly."
+                )
+            if client_options.scopes:
+                raise ValueError(
+                    "When providing a transport instance, provide its scopes "
+                    "directly."
+                )
+            self._transport = transport
+        else:
+            Transport = type(self).get_transport_class(transport)
+            self._transport = Transport(
+                credentials=credentials,
+                credentials_file=client_options.credentials_file,
+                host=api_endpoint,
+                scopes=client_options.scopes,
+                client_cert_source_for_mtls=client_cert_source_func,
+                quota_project_id=client_options.quota_project_id,
+                client_info=client_info,
+                always_use_jwt_access=True,
+            )
diff --git a/google/api_core/operations_v1/abstract_operations_client.py b/google/api_core/operations_v1/abstract_operations_client.py
index 38f532a..fc44536 100644
--- a/google/api_core/operations_v1/abstract_operations_client.py
+++ b/google/api_core/operations_v1/abstract_operations_client.py
@@ -13,10 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
-from collections import OrderedDict
-import os
-import re
-from typing import Dict, Optional, Sequence, Tuple, Type, Union
+from typing import Optional, Sequence, Tuple, Union
 
 from google.api_core import client_options as client_options_lib  # type: ignore
 from google.api_core import gapic_v1  # type: ignore
@@ -26,10 +23,10 @@
     DEFAULT_CLIENT_INFO,
     OperationsTransport,
 )
-from google.api_core.operations_v1.transports.rest import OperationsRestTransport
+from google.api_core.operations_v1.abstract_operations_base_client import (
+    AbstractOperationsBaseClient,
+)
 from google.auth import credentials as ga_credentials  # type: ignore
-from google.auth.exceptions import MutualTLSChannelError  # type: ignore
-from google.auth.transport import mtls  # type: ignore
 from google.longrunning import operations_pb2
 from google.oauth2 import service_account  # type: ignore
 import grpc
@@ -37,40 +34,7 @@
 OptionalRetry = Union[retries.Retry, object]
 
 
-class AbstractOperationsClientMeta(type):
-    """Metaclass for the Operations client.
-
-    This provides class-level methods for building and retrieving
-    support objects (e.g. transport) without polluting the client instance
-    objects.
-    """
-
-    _transport_registry = OrderedDict()  # type: Dict[str, Type[OperationsTransport]]
-    _transport_registry["rest"] = OperationsRestTransport
-
-    def get_transport_class(
-        cls,
-        label: Optional[str] = None,
-    ) -> Type[OperationsTransport]:
-        """Returns an appropriate transport class.
-
-        Args:
-            label: The name of the desired transport. If none is
-                provided, then the first transport in the registry is used.
-
-        Returns:
-            The transport class to use.
-        """
-        # If a specific transport is requested, return that one.
-        if label:
-            return cls._transport_registry[label]
-
-        # No transport is requested; return the default (that is, the first one
-        # in the dictionary).
-        return next(iter(cls._transport_registry.values()))
-
-
-class AbstractOperationsClient(metaclass=AbstractOperationsClientMeta):
+class AbstractOperationsClient(AbstractOperationsBaseClient):
     """Manages long-running operations with an API service.
 
     When an API method normally takes long time to complete, it can be
@@ -83,165 +47,6 @@
     interface so developers can have a consistent client experience.
     """
 
-    @staticmethod
-    def _get_default_mtls_endpoint(api_endpoint):
-        """Converts api endpoint to mTLS endpoint.
-
-        Convert "*.sandbox.googleapis.com" and "*.googleapis.com" to
-        "*.mtls.sandbox.googleapis.com" and "*.mtls.googleapis.com" respectively.
-        Args:
-            api_endpoint (Optional[str]): the api endpoint to convert.
-        Returns:
-            str: converted mTLS api endpoint.
-        """
-        if not api_endpoint:
-            return api_endpoint
-
-        mtls_endpoint_re = re.compile(
-            r"(?P<name>[^.]+)(?P<mtls>\.mtls)?(?P<sandbox>\.sandbox)?(?P<googledomain>\.googleapis\.com)?"
-        )
-
-        m = mtls_endpoint_re.match(api_endpoint)
-        name, mtls, sandbox, googledomain = m.groups()
-        if mtls or not googledomain:
-            return api_endpoint
-
-        if sandbox:
-            return api_endpoint.replace(
-                "sandbox.googleapis.com", "mtls.sandbox.googleapis.com"
-            )
-
-        return api_endpoint.replace(".googleapis.com", ".mtls.googleapis.com")
-
-    DEFAULT_ENDPOINT = "longrunning.googleapis.com"
-    DEFAULT_MTLS_ENDPOINT = _get_default_mtls_endpoint.__func__(  # type: ignore
-        DEFAULT_ENDPOINT
-    )
-
-    @classmethod
-    def from_service_account_info(cls, info: dict, *args, **kwargs):
-        """Creates an instance of this client using the provided credentials
-            info.
-
-        Args:
-            info (dict): The service account private key info.
-            args: Additional arguments to pass to the constructor.
-            kwargs: Additional arguments to pass to the constructor.
-
-        Returns:
-            AbstractOperationsClient: The constructed client.
-        """
-        credentials = service_account.Credentials.from_service_account_info(info)
-        kwargs["credentials"] = credentials
-        return cls(*args, **kwargs)
-
-    @classmethod
-    def from_service_account_file(cls, filename: str, *args, **kwargs):
-        """Creates an instance of this client using the provided credentials
-            file.
-
-        Args:
-            filename (str): The path to the service account private key json
-                file.
-            args: Additional arguments to pass to the constructor.
-            kwargs: Additional arguments to pass to the constructor.
-
-        Returns:
-            AbstractOperationsClient: The constructed client.
-        """
-        credentials = service_account.Credentials.from_service_account_file(filename)
-        kwargs["credentials"] = credentials
-        return cls(*args, **kwargs)
-
-    from_service_account_json = from_service_account_file
-
-    @property
-    def transport(self) -> OperationsTransport:
-        """Returns the transport used by the client instance.
-
-        Returns:
-            OperationsTransport: The transport used by the client
-                instance.
-        """
-        return self._transport
-
-    @staticmethod
-    def common_billing_account_path(
-        billing_account: str,
-    ) -> str:
-        """Returns a fully-qualified billing_account string."""
-        return "billingAccounts/{billing_account}".format(
-            billing_account=billing_account,
-        )
-
-    @staticmethod
-    def parse_common_billing_account_path(path: str) -> Dict[str, str]:
-        """Parse a billing_account path into its component segments."""
-        m = re.match(r"^billingAccounts/(?P<billing_account>.+?)$", path)
-        return m.groupdict() if m else {}
-
-    @staticmethod
-    def common_folder_path(
-        folder: str,
-    ) -> str:
-        """Returns a fully-qualified folder string."""
-        return "folders/{folder}".format(
-            folder=folder,
-        )
-
-    @staticmethod
-    def parse_common_folder_path(path: str) -> Dict[str, str]:
-        """Parse a folder path into its component segments."""
-        m = re.match(r"^folders/(?P<folder>.+?)$", path)
-        return m.groupdict() if m else {}
-
-    @staticmethod
-    def common_organization_path(
-        organization: str,
-    ) -> str:
-        """Returns a fully-qualified organization string."""
-        return "organizations/{organization}".format(
-            organization=organization,
-        )
-
-    @staticmethod
-    def parse_common_organization_path(path: str) -> Dict[str, str]:
-        """Parse a organization path into its component segments."""
-        m = re.match(r"^organizations/(?P<organization>.+?)$", path)
-        return m.groupdict() if m else {}
-
-    @staticmethod
-    def common_project_path(
-        project: str,
-    ) -> str:
-        """Returns a fully-qualified project string."""
-        return "projects/{project}".format(
-            project=project,
-        )
-
-    @staticmethod
-    def parse_common_project_path(path: str) -> Dict[str, str]:
-        """Parse a project path into its component segments."""
-        m = re.match(r"^projects/(?P<project>.+?)$", path)
-        return m.groupdict() if m else {}
-
-    @staticmethod
-    def common_location_path(
-        project: str,
-        location: str,
-    ) -> str:
-        """Returns a fully-qualified location string."""
-        return "projects/{project}/locations/{location}".format(
-            project=project,
-            location=location,
-        )
-
-    @staticmethod
-    def parse_common_location_path(path: str) -> Dict[str, str]:
-        """Parse a location path into its component segments."""
-        m = re.match(r"^projects/(?P<project>.+?)/locations/(?P<location>.+?)$", path)
-        return m.groupdict() if m else {}
-
     def __init__(
         self,
         *,
@@ -287,80 +92,49 @@
             google.auth.exceptions.MutualTLSChannelError: If mutual TLS transport
                 creation failed for any reason.
         """
-        if isinstance(client_options, dict):
-            client_options = client_options_lib.from_dict(client_options)
-        if client_options is None:
-            client_options = client_options_lib.ClientOptions()
+        super().__init__(
+            credentials=credentials,
+            transport=transport,
+            client_options=client_options,
+            client_info=client_info,
+        )
 
-        # Create SSL credentials for mutual TLS if needed.
-        use_client_cert = os.getenv(
-            "GOOGLE_API_USE_CLIENT_CERTIFICATE", "false"
-        ).lower()
-        if use_client_cert not in ("true", "false"):
-            raise ValueError(
-                "Environment variable `GOOGLE_API_USE_CLIENT_CERTIFICATE` must be either `true` or `false`"
-            )
-        client_cert_source_func = None
-        is_mtls = False
-        if use_client_cert == "true":
-            if client_options.client_cert_source:
-                is_mtls = True
-                client_cert_source_func = client_options.client_cert_source
-            else:
-                is_mtls = mtls.has_default_client_cert_source()
-                if is_mtls:
-                    client_cert_source_func = mtls.default_client_cert_source()
-                else:
-                    client_cert_source_func = None
+    @classmethod
+    def from_service_account_info(cls, info: dict, *args, **kwargs):
+        """Creates an instance of this client using the provided credentials
+            info.
 
-        # Figure out which api endpoint to use.
-        if client_options.api_endpoint is not None:
-            api_endpoint = client_options.api_endpoint
-        else:
-            use_mtls_env = os.getenv("GOOGLE_API_USE_MTLS_ENDPOINT", "auto")
-            if use_mtls_env == "never":
-                api_endpoint = self.DEFAULT_ENDPOINT
-            elif use_mtls_env == "always":
-                api_endpoint = self.DEFAULT_MTLS_ENDPOINT
-            elif use_mtls_env == "auto":
-                if is_mtls:
-                    api_endpoint = self.DEFAULT_MTLS_ENDPOINT
-                else:
-                    api_endpoint = self.DEFAULT_ENDPOINT
-            else:
-                raise MutualTLSChannelError(
-                    "Unsupported GOOGLE_API_USE_MTLS_ENDPOINT value. Accepted "
-                    "values: never, auto, always"
-                )
+        Args:
+            info (dict): The service account private key info.
+            args: Additional arguments to pass to the constructor.
+            kwargs: Additional arguments to pass to the constructor.
 
-        # Save or instantiate the transport.
-        # Ordinarily, we provide the transport, but allowing a custom transport
-        # instance provides an extensibility point for unusual situations.
-        if isinstance(transport, OperationsTransport):
-            # transport is a OperationsTransport instance.
-            if credentials or client_options.credentials_file:
-                raise ValueError(
-                    "When providing a transport instance, "
-                    "provide its credentials directly."
-                )
-            if client_options.scopes:
-                raise ValueError(
-                    "When providing a transport instance, provide its scopes "
-                    "directly."
-                )
-            self._transport = transport
-        else:
-            Transport = type(self).get_transport_class(transport)
-            self._transport = Transport(
-                credentials=credentials,
-                credentials_file=client_options.credentials_file,
-                host=api_endpoint,
-                scopes=client_options.scopes,
-                client_cert_source_for_mtls=client_cert_source_func,
-                quota_project_id=client_options.quota_project_id,
-                client_info=client_info,
-                always_use_jwt_access=True,
-            )
+        Returns:
+            AbstractOperationsClient: The constructed client.
+        """
+        credentials = service_account.Credentials.from_service_account_info(info)
+        kwargs["credentials"] = credentials
+        return cls(*args, **kwargs)
+
+    @classmethod
+    def from_service_account_file(cls, filename: str, *args, **kwargs):
+        """Creates an instance of this client using the provided credentials
+            file.
+
+        Args:
+            filename (str): The path to the service account private key json
+                file.
+            args: Additional arguments to pass to the constructor.
+            kwargs: Additional arguments to pass to the constructor.
+
+        Returns:
+            AbstractOperationsClient: The constructed client.
+        """
+        credentials = service_account.Credentials.from_service_account_file(filename)
+        kwargs["credentials"] = credentials
+        return cls(*args, **kwargs)
+
+    from_service_account_json = from_service_account_file
 
     def list_operations(
         self,
diff --git a/google/api_core/operations_v1/operations_rest_client_async.py b/google/api_core/operations_v1/operations_rest_client_async.py
new file mode 100644
index 0000000..7ab0cd3
--- /dev/null
+++ b/google/api_core/operations_v1/operations_rest_client_async.py
@@ -0,0 +1,345 @@
+# -*- coding: utf-8 -*-
+# Copyright 2024 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.
+#
+from typing import Optional, Sequence, Tuple, Union
+
+from google.api_core import client_options as client_options_lib  # type: ignore
+from google.api_core import gapic_v1  # type: ignore
+from google.api_core.operations_v1 import pagers_async as pagers
+from google.api_core.operations_v1.transports.base import (
+    DEFAULT_CLIENT_INFO,
+    OperationsTransport,
+)
+from google.api_core.operations_v1.abstract_operations_base_client import (
+    AbstractOperationsBaseClient,
+)
+from google.longrunning import operations_pb2
+
+try:
+    from google.auth.aio import credentials as ga_credentials  # type: ignore
+except ImportError as e:  # pragma: NO COVER
+    raise ImportError(
+        "The `async_rest` extra of `google-api-core` is required to use long-running operations.  Install it by running "
+        "`pip install google-api-core[async_rest]`."
+    ) from e
+
+
+class AsyncOperationsRestClient(AbstractOperationsBaseClient):
+    """Manages long-running operations with a REST API service for the asynchronous client.
+
+    When an API method normally takes long time to complete, it can be
+    designed to return [Operation][google.api_core.operations_v1.Operation] to the
+    client, and the client can use this interface to receive the real
+    response asynchronously by polling the operation resource, or pass
+    the operation resource to another API (such as Google Cloud Pub/Sub
+    API) to receive the response. Any API service that returns
+    long-running operations should implement the ``Operations``
+    interface so developers can have a consistent client experience.
+    """
+
+    def __init__(
+        self,
+        *,
+        credentials: Optional[ga_credentials.Credentials] = None,
+        transport: Union[str, OperationsTransport, None] = None,
+        client_options: Optional[client_options_lib.ClientOptions] = None,
+        client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO,
+    ) -> None:
+        """Instantiates the operations client.
+
+        Args:
+            credentials (Optional[google.auth.aio.credentials.Credentials]): The
+                authorization credentials to attach to requests. These
+                credentials identify the application to the service; if none
+                are specified, the client will attempt to ascertain the
+                credentials from the environment.
+            transport (Union[str, OperationsTransport]): The
+                transport to use. If set to None, this defaults to 'rest_asyncio'.
+            client_options (google.api_core.client_options.ClientOptions): Custom options for the
+                client. It won't take effect if a ``transport`` instance is provided.
+                (1) The ``api_endpoint`` property can be used to override the
+                default endpoint provided by the client. GOOGLE_API_USE_MTLS_ENDPOINT
+                environment variable can also be used to override the endpoint:
+                "always" (always use the default mTLS endpoint), "never" (always
+                use the default regular endpoint) and "auto" (auto switch to the
+                default mTLS endpoint if client certificate is present, this is
+                the default value). However, the ``api_endpoint`` property takes
+                precedence if provided.
+                (2) If GOOGLE_API_USE_CLIENT_CERTIFICATE environment variable
+                is "true", then the ``client_cert_source`` property can be used
+                to provide client certificate for mutual TLS transport. If
+                not provided, the default SSL client certificate will be used if
+                present. If GOOGLE_API_USE_CLIENT_CERTIFICATE is "false" or not
+                set, no client certificate will be used.
+            client_info (google.api_core.gapic_v1.client_info.ClientInfo):
+                The client info used to send a user-agent string along with
+                API requests. If ``None``, then default info will be used.
+                Generally, you only need to set this if you're developing
+                your own client library.
+
+        Raises:
+            google.auth.exceptions.MutualTLSChannelError: If mutual TLS transport
+                creation failed for any reason.
+        """
+        super().__init__(
+            credentials=credentials,  # type: ignore
+            # NOTE: If a transport is not provided, we force the client to use the async
+            # REST transport.
+            transport=transport or "rest_asyncio",
+            client_options=client_options,
+            client_info=client_info,
+        )
+
+    async def get_operation(
+        self,
+        name: str,
+        *,
+        # TODO(https://github.com/googleapis/python-api-core/issues/722): Leverage `retry`
+        # to allow configuring retryable error codes.
+        retry=gapic_v1.method_async.DEFAULT,
+        timeout: Optional[float] = None,
+        metadata: Sequence[Tuple[str, str]] = (),
+    ) -> operations_pb2.Operation:
+        r"""Gets the latest state of a long-running operation.
+        Clients can use this method to poll the operation result
+        at intervals as recommended by the API service.
+
+        Args:
+            name (str):
+                The name of the operation resource.
+            timeout (float): The timeout for this request.
+            metadata (Sequence[Tuple[str, str]]): Strings which should be
+                sent along with the request as metadata.
+
+        Returns:
+            google.longrunning.operations_pb2.Operation:
+                This resource represents a long-
+                running operation that is the result of a
+                network API call.
+
+        """
+
+        request = operations_pb2.GetOperationRequest(name=name)
+
+        # Wrap the RPC method; this adds retry and timeout information,
+        # and friendly error handling.
+        rpc = self._transport._wrapped_methods[self._transport.get_operation]
+
+        # Certain fields should be provided within the metadata header;
+        # add these here.
+        metadata = tuple(metadata or ()) + (
+            gapic_v1.routing_header.to_grpc_metadata((("name", request.name),)),
+        )
+
+        # Send the request.
+        response = await rpc(
+            request,
+            retry=retry,
+            timeout=timeout,
+            metadata=metadata,
+        )
+
+        # Done; return the response.
+        return response
+
+    async def list_operations(
+        self,
+        name: str,
+        filter_: Optional[str] = None,
+        *,
+        page_size: Optional[int] = None,
+        page_token: Optional[str] = None,
+        # TODO(https://github.com/googleapis/python-api-core/issues/722): Leverage `retry`
+        # to allow configuring retryable error codes.
+        retry=gapic_v1.method_async.DEFAULT,
+        timeout: Optional[float] = None,
+        metadata: Sequence[Tuple[str, str]] = (),
+    ) -> pagers.ListOperationsAsyncPager:
+        r"""Lists operations that match the specified filter in the request.
+        If the server doesn't support this method, it returns
+        ``UNIMPLEMENTED``.
+
+        NOTE: the ``name`` binding allows API services to override the
+        binding to use different resource name schemes, such as
+        ``users/*/operations``. To override the binding, API services
+        can add a binding such as ``"/v1/{name=users/*}/operations"`` to
+        their service configuration. For backwards compatibility, the
+        default name includes the operations collection id, however
+        overriding users must ensure the name binding is the parent
+        resource, without the operations collection id.
+
+        Args:
+            name (str):
+                The name of the operation's parent
+                resource.
+            filter_ (str):
+                The standard list filter.
+                This corresponds to the ``filter`` field
+                on the ``request`` instance; if ``request`` is provided, this
+                should not be set.
+            timeout (float): The timeout for this request.
+            metadata (Sequence[Tuple[str, str]]): Strings which should be
+                sent along with the request as metadata.
+
+        Returns:
+            google.api_core.operations_v1.pagers.ListOperationsPager:
+                The response message for
+                [Operations.ListOperations][google.api_core.operations_v1.Operations.ListOperations].
+
+                Iterating over this object will yield results and
+                resolve additional pages automatically.
+
+        """
+        # Create a protobuf request object.
+        request = operations_pb2.ListOperationsRequest(name=name, filter=filter_)
+        if page_size is not None:
+            request.page_size = page_size
+        if page_token is not None:
+            request.page_token = page_token
+
+        # Wrap the RPC method; this adds retry and timeout information,
+        # and friendly error handling.
+        rpc = self._transport._wrapped_methods[self._transport.list_operations]
+
+        # Certain fields should be provided within the metadata header;
+        # add these here.
+        metadata = tuple(metadata or ()) + (
+            gapic_v1.routing_header.to_grpc_metadata((("name", request.name),)),
+        )
+
+        # Send the request.
+        response = await rpc(
+            request,
+            retry=retry,
+            timeout=timeout,
+            metadata=metadata,
+        )
+
+        # This method is paged; wrap the response in a pager, which provides
+        # an `__iter__` convenience method.
+        response = pagers.ListOperationsAsyncPager(
+            method=rpc,
+            request=request,
+            response=response,
+            metadata=metadata,
+        )
+
+        # Done; return the response.
+        return response
+
+    async def delete_operation(
+        self,
+        name: str,
+        *,
+        # TODO(https://github.com/googleapis/python-api-core/issues/722): Leverage `retry`
+        # to allow configuring retryable error codes.
+        retry=gapic_v1.method_async.DEFAULT,
+        timeout: Optional[float] = None,
+        metadata: Sequence[Tuple[str, str]] = (),
+    ) -> None:
+        r"""Deletes a long-running operation. This method indicates that the
+        client is no longer interested in the operation result. It does
+        not cancel the operation. If the server doesn't support this
+        method, it returns ``google.rpc.Code.UNIMPLEMENTED``.
+
+        Args:
+            name (str):
+                The name of the operation resource to
+                be deleted.
+
+                This corresponds to the ``name`` field
+                on the ``request`` instance; if ``request`` is provided, this
+                should not be set.
+            timeout (float): The timeout for this request.
+            metadata (Sequence[Tuple[str, str]]): Strings which should be
+                sent along with the request as metadata.
+        """
+        # Create the request object.
+        request = operations_pb2.DeleteOperationRequest(name=name)
+
+        # Wrap the RPC method; this adds retry and timeout information,
+        # and friendly error handling.
+        rpc = self._transport._wrapped_methods[self._transport.delete_operation]
+
+        # Certain fields should be provided within the metadata header;
+        # add these here.
+        metadata = tuple(metadata or ()) + (
+            gapic_v1.routing_header.to_grpc_metadata((("name", request.name),)),
+        )
+
+        # Send the request.
+        await rpc(
+            request,
+            retry=retry,
+            timeout=timeout,
+            metadata=metadata,
+        )
+
+    async def cancel_operation(
+        self,
+        name: Optional[str] = None,
+        *,
+        # TODO(https://github.com/googleapis/python-api-core/issues/722): Leverage `retry`
+        # to allow configuring retryable error codes.
+        retry=gapic_v1.method_async.DEFAULT,
+        timeout: Optional[float] = None,
+        metadata: Sequence[Tuple[str, str]] = (),
+    ) -> None:
+        r"""Starts asynchronous cancellation on a long-running operation.
+        The server makes a best effort to cancel the operation, but
+        success is not guaranteed. If the server doesn't support this
+        method, it returns ``google.rpc.Code.UNIMPLEMENTED``. Clients
+        can use
+        [Operations.GetOperation][google.api_core.operations_v1.Operations.GetOperation]
+        or other methods to check whether the cancellation succeeded or
+        whether the operation completed despite cancellation. On
+        successful cancellation, the operation is not deleted; instead,
+        it becomes an operation with an
+        [Operation.error][google.api_core.operations_v1.Operation.error] value with
+        a [google.rpc.Status.code][google.rpc.Status.code] of 1,
+        corresponding to ``Code.CANCELLED``.
+
+        Args:
+            name (str):
+                The name of the operation resource to
+                be cancelled.
+
+                This corresponds to the ``name`` field
+                on the ``request`` instance; if ``request`` is provided, this
+                should not be set.
+            timeout (float): The timeout for this request.
+            metadata (Sequence[Tuple[str, str]]): Strings which should be
+                sent along with the request as metadata.
+        """
+        # Create the request object.
+        request = operations_pb2.CancelOperationRequest(name=name)
+
+        # Wrap the RPC method; this adds retry and timeout information,
+        # and friendly error handling.
+        rpc = self._transport._wrapped_methods[self._transport.cancel_operation]
+
+        # Certain fields should be provided within the metadata header;
+        # add these here.
+        metadata = tuple(metadata or ()) + (
+            gapic_v1.routing_header.to_grpc_metadata((("name", request.name),)),
+        )
+
+        # Send the request.
+        await rpc(
+            request,
+            retry=retry,
+            timeout=timeout,
+            metadata=metadata,
+        )
diff --git a/google/api_core/operations_v1/pagers.py b/google/api_core/operations_v1/pagers.py
index b8a4775..132f1c6 100644
--- a/google/api_core/operations_v1/pagers.py
+++ b/google/api_core/operations_v1/pagers.py
@@ -14,7 +14,6 @@
 # limitations under the License.
 #
 from typing import (
-    Any,
     Callable,
     Iterator,
     Sequence,
@@ -22,9 +21,10 @@
 )
 
 from google.longrunning import operations_pb2
+from google.api_core.operations_v1.pagers_base import ListOperationsPagerBase
 
 
-class ListOperationsPager:
+class ListOperationsPager(ListOperationsPagerBase):
     """A pager for iterating through ``list_operations`` requests.
 
     This class thinly wraps an initial
@@ -50,25 +50,9 @@
         *,
         metadata: Sequence[Tuple[str, str]] = ()
     ):
-        """Instantiate the pager.
-
-        Args:
-            method (Callable): The method that was originally called, and
-                which instantiated this pager.
-            request (google.longrunning.operations_pb2.ListOperationsRequest):
-                The initial request object.
-            response (google.longrunning.operations_pb2.ListOperationsResponse):
-                The initial response object.
-            metadata (Sequence[Tuple[str, str]]): Strings which should be
-                sent along with the request as metadata.
-        """
-        self._method = method
-        self._request = request
-        self._response = response
-        self._metadata = metadata
-
-    def __getattr__(self, name: str) -> Any:
-        return getattr(self._response, name)
+        super().__init__(
+            method=method, request=request, response=response, metadata=metadata
+        )
 
     @property
     def pages(self) -> Iterator[operations_pb2.ListOperationsResponse]:
@@ -81,6 +65,3 @@
     def __iter__(self) -> Iterator[operations_pb2.Operation]:
         for page in self.pages:
             yield from page.operations
-
-    def __repr__(self) -> str:
-        return "{0}<{1!r}>".format(self.__class__.__name__, self._response)
diff --git a/google/api_core/operations_v1/pagers_async.py b/google/api_core/operations_v1/pagers_async.py
new file mode 100644
index 0000000..e2909dd
--- /dev/null
+++ b/google/api_core/operations_v1/pagers_async.py
@@ -0,0 +1,71 @@
+# -*- coding: utf-8 -*-
+# Copyright 2024 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.
+#
+from typing import (
+    Callable,
+    AsyncIterator,
+    Sequence,
+    Tuple,
+)
+
+from google.longrunning import operations_pb2
+from google.api_core.operations_v1.pagers_base import ListOperationsPagerBase
+
+
+class ListOperationsAsyncPager(ListOperationsPagerBase):
+    """A pager for iterating through ``list_operations`` requests.
+
+    This class thinly wraps an initial
+    :class:`google.longrunning.operations_pb2.ListOperationsResponse` object, and
+    provides an ``__iter__`` method to iterate through its
+    ``operations`` field.
+
+    If there are more pages, the ``__iter__`` method will make additional
+    ``ListOperations`` requests and continue to iterate
+    through the ``operations`` field on the
+    corresponding responses.
+
+    All the usual :class:`google.longrunning.operations_pb2.ListOperationsResponse`
+    attributes are available on the pager. If multiple requests are made, only
+    the most recent response is retained, and thus used for attribute lookup.
+    """
+
+    def __init__(
+        self,
+        method: Callable[..., operations_pb2.ListOperationsResponse],
+        request: operations_pb2.ListOperationsRequest,
+        response: operations_pb2.ListOperationsResponse,
+        *,
+        metadata: Sequence[Tuple[str, str]] = ()
+    ):
+        super().__init__(
+            method=method, request=request, response=response, metadata=metadata
+        )
+
+    @property
+    async def pages(self) -> AsyncIterator[operations_pb2.ListOperationsResponse]:
+        yield self._response
+        while self._response.next_page_token:
+            self._request.page_token = self._response.next_page_token
+            self._response = await self._method(self._request, metadata=self._metadata)
+            yield self._response
+
+    def __aiter__(self) -> AsyncIterator[operations_pb2.Operation]:
+        async def async_generator():
+            async for page in self.pages:
+                for operation in page.operations:
+                    yield operation
+
+        return async_generator()
diff --git a/google/api_core/operations_v1/pagers_base.py b/google/api_core/operations_v1/pagers_base.py
new file mode 100644
index 0000000..24caf74
--- /dev/null
+++ b/google/api_core/operations_v1/pagers_base.py
@@ -0,0 +1,73 @@
+# -*- coding: utf-8 -*-
+# Copyright 2024 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.
+#
+from typing import (
+    Any,
+    Callable,
+    Sequence,
+    Tuple,
+)
+
+from google.longrunning import operations_pb2
+
+
+class ListOperationsPagerBase:
+    """A pager for iterating through ``list_operations`` requests.
+
+    This class thinly wraps an initial
+    :class:`google.longrunning.operations_pb2.ListOperationsResponse` object, and
+    provides an ``__iter__`` method to iterate through its
+    ``operations`` field.
+
+    If there are more pages, the ``__iter__`` method will make additional
+    ``ListOperations`` requests and continue to iterate
+    through the ``operations`` field on the
+    corresponding responses.
+
+    All the usual :class:`google.longrunning.operations_pb2.ListOperationsResponse`
+    attributes are available on the pager. If multiple requests are made, only
+    the most recent response is retained, and thus used for attribute lookup.
+    """
+
+    def __init__(
+        self,
+        method: Callable[..., operations_pb2.ListOperationsResponse],
+        request: operations_pb2.ListOperationsRequest,
+        response: operations_pb2.ListOperationsResponse,
+        *,
+        metadata: Sequence[Tuple[str, str]] = ()
+    ):
+        """Instantiate the pager.
+
+        Args:
+            method (Callable): The method that was originally called, and
+                which instantiated this pager.
+            request (google.longrunning.operations_pb2.ListOperationsRequest):
+                The initial request object.
+            response (google.longrunning.operations_pb2.ListOperationsResponse):
+                The initial response object.
+            metadata (Sequence[Tuple[str, str]]): Strings which should be
+                sent along with the request as metadata.
+        """
+        self._method = method
+        self._request = request
+        self._response = response
+        self._metadata = metadata
+
+    def __getattr__(self, name: str) -> Any:
+        return getattr(self._response, name)
+
+    def __repr__(self) -> str:
+        return "{0}<{1!r}>".format(self.__class__.__name__, self._response)
diff --git a/google/api_core/operations_v1/transports/__init__.py b/google/api_core/operations_v1/transports/__init__.py
index df53e15..8c24ce6 100644
--- a/google/api_core/operations_v1/transports/__init__.py
+++ b/google/api_core/operations_v1/transports/__init__.py
@@ -14,16 +14,26 @@
 # limitations under the License.
 #
 from collections import OrderedDict
+from typing import cast, Dict, Tuple
 
 from .base import OperationsTransport
 from .rest import OperationsRestTransport
 
-
 # Compile a registry of transports.
-_transport_registry = OrderedDict()
-_transport_registry["rest"] = OperationsRestTransport
+_transport_registry: Dict[str, OperationsTransport] = OrderedDict()
+_transport_registry["rest"] = cast(OperationsTransport, OperationsRestTransport)
 
-__all__ = (
-    "OperationsTransport",
-    "OperationsRestTransport",
-)
+__all__: Tuple[str, ...] = ("OperationsTransport", "OperationsRestTransport")
+
+try:
+    from .rest_asyncio import AsyncOperationsRestTransport
+
+    __all__ += ("AsyncOperationsRestTransport",)
+    _transport_registry["rest_asyncio"] = cast(
+        OperationsTransport, AsyncOperationsRestTransport
+    )
+except ImportError:
+    # This import requires the `async_rest` extra.
+    # Don't raise an exception if `AsyncOperationsRestTransport` cannot be imported
+    # as other transports are still available.
+    pass
diff --git a/google/api_core/operations_v1/transports/base.py b/google/api_core/operations_v1/transports/base.py
index fb1d4fc..50e1376 100644
--- a/google/api_core/operations_v1/transports/base.py
+++ b/google/api_core/operations_v1/transports/base.py
@@ -14,6 +14,7 @@
 # limitations under the License.
 #
 import abc
+import re
 from typing import Awaitable, Callable, Optional, Sequence, Union
 
 import google.api_core  # type: ignore
@@ -25,10 +26,13 @@
 from google.auth import credentials as ga_credentials  # type: ignore
 from google.longrunning import operations_pb2
 from google.oauth2 import service_account  # type: ignore
-from google.protobuf import empty_pb2  # type: ignore
+import google.protobuf
+from google.protobuf import empty_pb2, json_format  # type: ignore
 from grpc import Compression
 
 
+PROTOBUF_VERSION = google.protobuf.__version__
+
 DEFAULT_CLIENT_INFO = gapic_v1.client_info.ClientInfo(
     gapic_version=version.__version__,
 )
@@ -45,12 +49,14 @@
         self,
         *,
         host: str = DEFAULT_HOST,
+        # TODO(https://github.com/googleapis/python-api-core/issues/709): update type hint for credentials to include `google.auth.aio.Credentials`.
         credentials: Optional[ga_credentials.Credentials] = None,
         credentials_file: Optional[str] = None,
         scopes: Optional[Sequence[str]] = None,
         quota_project_id: Optional[str] = None,
         client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO,
         always_use_jwt_access: Optional[bool] = False,
+        url_scheme="https",
         **kwargs,
     ) -> None:
         """Instantiate the transport.
@@ -76,10 +82,23 @@
                 your own client library.
             always_use_jwt_access (Optional[bool]): Whether self signed JWT should
                 be used for service account credentials.
+            url_scheme: the protocol scheme for the API endpoint.  Normally
+                "https", but for testing or local servers,
+                "http" can be specified.
         """
+        maybe_url_match = re.match("^(?P<scheme>http(?:s)?://)?(?P<host>.*)$", host)
+        if maybe_url_match is None:
+            raise ValueError(
+                f"Unexpected hostname structure: {host}"
+            )  # pragma: NO COVER
+
+        url_match_items = maybe_url_match.groupdict()
+
+        host = f"{url_scheme}://{host}" if not url_match_items["scheme"] else host
+
         # Save the hostname. Default to port 443 (HTTPS) if none is specified.
         if ":" not in host:
-            host += ":443"
+            host += ":443"  # pragma: NO COVER
         self._host = host
 
         scopes_kwargs = {"scopes": scopes, "default_scopes": self.AUTH_SCOPES}
@@ -189,6 +208,37 @@
         """
         raise NotImplementedError()
 
+    def _convert_protobuf_message_to_dict(
+        self, message: google.protobuf.message.Message
+    ):
+        r"""Converts protobuf message to a dictionary.
+
+        When the dictionary is encoded to JSON, it conforms to proto3 JSON spec.
+
+        Args:
+            message(google.protobuf.message.Message): The protocol buffers message
+                instance to serialize.
+
+        Returns:
+            A dict representation of the protocol buffer message.
+        """
+        # TODO(https://github.com/googleapis/python-api-core/issues/643): For backwards compatibility
+        # with protobuf 3.x 4.x, Remove once support for protobuf 3.x and 4.x is dropped.
+        if PROTOBUF_VERSION[0:2] in ["3.", "4."]:
+            result = json_format.MessageToDict(
+                message,
+                preserving_proto_field_name=True,
+                including_default_value_fields=True,  # type: ignore # backward compatibility
+            )
+        else:
+            result = json_format.MessageToDict(
+                message,
+                preserving_proto_field_name=True,
+                always_print_fields_with_no_presence=True,
+            )
+
+        return result
+
     @property
     def list_operations(
         self,
diff --git a/google/api_core/operations_v1/transports/rest.py b/google/api_core/operations_v1/transports/rest.py
index f37bb34..766a668 100644
--- a/google/api_core/operations_v1/transports/rest.py
+++ b/google/api_core/operations_v1/transports/rest.py
@@ -14,7 +14,6 @@
 # limitations under the License.
 #
 
-import re
 from typing import Callable, Dict, Optional, Sequence, Tuple, Union
 
 from requests import __version__ as requests_version
@@ -41,7 +40,7 @@
 DEFAULT_CLIENT_INFO = gapic_v1.client_info.ClientInfo(
     gapic_version=BASE_DEFAULT_CLIENT_INFO.gapic_version,
     grpc_version=None,
-    rest_version=requests_version,
+    rest_version=f"requests@{requests_version}",
 )
 
 
@@ -123,16 +122,6 @@
         # TODO(yon-mg): resolve other ctor params i.e. scopes, quota, etc.
         # TODO: When custom host (api_endpoint) is set, `scopes` must *also* be set on the
         # credentials object
-        maybe_url_match = re.match("^(?P<scheme>http(?:s)?://)?(?P<host>.*)$", host)
-        if maybe_url_match is None:
-            raise ValueError(
-                f"Unexpected hostname structure: {host}"
-            )  # pragma: NO COVER
-
-        url_match_items = maybe_url_match.groupdict()
-
-        host = f"{url_scheme}://{host}" if not url_match_items["scheme"] else host
-
         super().__init__(
             host=host,
             credentials=credentials,
@@ -144,6 +133,7 @@
         )
         if client_cert_source_for_mtls:
             self._session.configure_mtls_channel(client_cert_source_for_mtls)
+        # TODO(https://github.com/googleapis/python-api-core/issues/720): Add wrap logic directly to the property methods for callables.
         self._prep_wrapped_messages(client_info)
         self._http_options = http_options or {}
         self._path_prefix = path_prefix
@@ -152,6 +142,8 @@
         self,
         request: operations_pb2.ListOperationsRequest,
         *,
+        # TODO(https://github.com/googleapis/python-api-core/issues/723): Leverage `retry`
+        # to allow configuring retryable error codes.
         retry: OptionalRetry = gapic_v1.method.DEFAULT,
         timeout: Optional[float] = None,
         compression: Optional[grpc.Compression] = gapic_v1.method.DEFAULT,
@@ -206,6 +198,7 @@
         # Send the request
         headers = dict(metadata)
         headers["Content-Type"] = "application/json"
+        # TODO(https://github.com/googleapis/python-api-core/issues/721): Update incorrect use of `uri`` variable name.
         response = getattr(self._session, method)(
             "{host}{uri}".format(host=self._host, uri=uri),
             timeout=timeout,
@@ -227,6 +220,8 @@
         self,
         request: operations_pb2.GetOperationRequest,
         *,
+        # TODO(https://github.com/googleapis/python-api-core/issues/723): Leverage `retry`
+        # to allow configuring retryable error codes.
         retry: OptionalRetry = gapic_v1.method.DEFAULT,
         timeout: Optional[float] = None,
         compression: Optional[grpc.Compression] = gapic_v1.method.DEFAULT,
@@ -282,6 +277,7 @@
         # Send the request
         headers = dict(metadata)
         headers["Content-Type"] = "application/json"
+        # TODO(https://github.com/googleapis/python-api-core/issues/721): Update incorrect use of `uri`` variable name.
         response = getattr(self._session, method)(
             "{host}{uri}".format(host=self._host, uri=uri),
             timeout=timeout,
@@ -303,6 +299,8 @@
         self,
         request: operations_pb2.DeleteOperationRequest,
         *,
+        # TODO(https://github.com/googleapis/python-api-core/issues/723): Leverage `retry`
+        # to allow configuring retryable error codes.
         retry: OptionalRetry = gapic_v1.method.DEFAULT,
         timeout: Optional[float] = None,
         compression: Optional[grpc.Compression] = gapic_v1.method.DEFAULT,
@@ -351,6 +349,7 @@
         # Send the request
         headers = dict(metadata)
         headers["Content-Type"] = "application/json"
+        # TODO(https://github.com/googleapis/python-api-core/issues/721): Update incorrect use of `uri`` variable name.
         response = getattr(self._session, method)(
             "{host}{uri}".format(host=self._host, uri=uri),
             timeout=timeout,
@@ -369,6 +368,8 @@
         self,
         request: operations_pb2.CancelOperationRequest,
         *,
+        # TODO(https://github.com/googleapis/python-api-core/issues/723): Leverage `retry`
+        # to allow configuring retryable error codes.
         retry: OptionalRetry = gapic_v1.method.DEFAULT,
         timeout: Optional[float] = None,
         compression: Optional[grpc.Compression] = gapic_v1.method.DEFAULT,
@@ -426,6 +427,7 @@
         # Send the request
         headers = dict(metadata)
         headers["Content-Type"] = "application/json"
+        # TODO(https://github.com/googleapis/python-api-core/issues/721): Update incorrect use of `uri`` variable name.
         response = getattr(self._session, method)(
             "{host}{uri}".format(host=self._host, uri=uri),
             timeout=timeout,
@@ -441,38 +443,6 @@
 
         return empty_pb2.Empty()
 
-    def _convert_protobuf_message_to_dict(
-        self, message: google.protobuf.message.Message
-    ):
-        r"""Converts protobuf message to a dictionary.
-
-        When the dictionary is encoded to JSON, it conforms to proto3 JSON spec.
-
-        Args:
-            message(google.protobuf.message.Message): The protocol buffers message
-                instance to serialize.
-
-        Returns:
-            A dict representation of the protocol buffer message.
-        """
-        # For backwards compatibility with protobuf 3.x 4.x
-        # Remove once support for protobuf 3.x and 4.x is dropped
-        # https://github.com/googleapis/python-api-core/issues/643
-        if PROTOBUF_VERSION[0:2] in ["3.", "4."]:
-            result = json_format.MessageToDict(
-                message,
-                preserving_proto_field_name=True,
-                including_default_value_fields=True,  # type: ignore # backward compatibility
-            )
-        else:
-            result = json_format.MessageToDict(
-                message,
-                preserving_proto_field_name=True,
-                always_print_fields_with_no_presence=True,
-            )
-
-        return result
-
     @property
     def list_operations(
         self,
diff --git a/google/api_core/operations_v1/transports/rest_asyncio.py b/google/api_core/operations_v1/transports/rest_asyncio.py
new file mode 100644
index 0000000..71c20eb
--- /dev/null
+++ b/google/api_core/operations_v1/transports/rest_asyncio.py
@@ -0,0 +1,560 @@
+# -*- coding: utf-8 -*-
+# Copyright 2024 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 json
+from typing import Any, Callable, Coroutine, Dict, Optional, Sequence, Tuple
+
+from google.auth import __version__ as auth_version
+
+try:
+    from google.auth.aio.transport.sessions import AsyncAuthorizedSession  # type: ignore
+except ImportError as e:  # pragma: NO COVER
+    raise ImportError(
+        "The `async_rest` extra of `google-api-core` is required to use long-running operations.  Install it by running "
+        "`pip install google-api-core[async_rest]`."
+    ) from e
+
+from google.api_core import exceptions as core_exceptions  # type: ignore
+from google.api_core import gapic_v1  # type: ignore
+from google.api_core import path_template  # type: ignore
+from google.api_core import rest_helpers  # type: ignore
+from google.api_core import retry_async as retries_async  # type: ignore
+from google.auth.aio import credentials as ga_credentials_async  # type: ignore
+from google.longrunning import operations_pb2  # type: ignore
+from google.protobuf import empty_pb2  # type: ignore
+from google.protobuf import json_format  # type: ignore
+
+from .base import DEFAULT_CLIENT_INFO as BASE_DEFAULT_CLIENT_INFO, OperationsTransport
+
+DEFAULT_CLIENT_INFO = gapic_v1.client_info.ClientInfo(
+    gapic_version=BASE_DEFAULT_CLIENT_INFO.gapic_version,
+    grpc_version=None,
+    rest_version=f"google-auth@{auth_version}",
+)
+
+
+class AsyncOperationsRestTransport(OperationsTransport):
+    """Asynchronous REST backend transport for Operations.
+
+    Manages async long-running operations with an API service.
+
+    When an API method normally takes long time to complete, it can be
+    designed to return [Operation][google.api_core.operations_v1.Operation] to the
+    client, and the client can use this interface to receive the real
+    response asynchronously by polling the operation resource, or pass
+    the operation resource to another API (such as Google Cloud Pub/Sub
+    API) to receive the response. Any API service that returns
+    long-running operations should implement the ``Operations``
+    interface so developers can have a consistent client experience.
+
+    This class defines the same methods as the primary client, so the
+    primary client can load the underlying transport implementation
+    and call it.
+
+    It sends JSON representations of protocol buffers over HTTP/1.1
+    """
+
+    def __init__(
+        self,
+        *,
+        host: str = "longrunning.googleapis.com",
+        credentials: Optional[ga_credentials_async.Credentials] = None,
+        credentials_file: Optional[str] = None,
+        scopes: Optional[Sequence[str]] = None,
+        client_cert_source_for_mtls: Optional[Callable[[], Tuple[bytes, bytes]]] = None,
+        quota_project_id: Optional[str] = None,
+        client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO,
+        always_use_jwt_access: Optional[bool] = False,
+        url_scheme: str = "https",
+        http_options: Optional[Dict] = None,
+        path_prefix: str = "v1",
+        # TODO(https://github.com/googleapis/python-api-core/issues/715): Add docstring for `credentials_file` to async REST transport.
+        # TODO(https://github.com/googleapis/python-api-core/issues/716): Add docstring for `scopes` to async REST transport.
+        # TODO(https://github.com/googleapis/python-api-core/issues/717): Add docstring for `quota_project_id` to async REST transport.
+        # TODO(https://github.com/googleapis/python-api-core/issues/718): Add docstring for `client_cert_source` to async REST transport.
+    ) -> None:
+        """Instantiate the transport.
+
+        Args:
+            host (Optional[str]):
+                 The hostname to connect to.
+            credentials (Optional[google.auth.aio.credentials.Credentials]): The
+                authorization credentials to attach to requests. These
+                credentials identify the application to the service; if none
+                are specified, the client will attempt to ascertain the
+                credentials from the environment.
+            client_info (google.api_core.gapic_v1.client_info.ClientInfo):
+                The client info used to send a user-agent string along with
+                API requests. If ``None``, then default info will be used.
+                Generally, you only need to set this if you're developing
+                your own client library.
+            always_use_jwt_access (Optional[bool]): Whether self signed JWT should
+                be used for service account credentials.
+            url_scheme: the protocol scheme for the API endpoint.  Normally
+                "https", but for testing or local servers,
+                "http" can be specified.
+            http_options: a dictionary of http_options for transcoding, to override
+                the defaults from operations.proto.  Each method has an entry
+                with the corresponding http rules as value.
+            path_prefix: path prefix (usually represents API version). Set to
+                "v1" by default.
+
+        """
+        unsupported_params = {
+            # TODO(https://github.com/googleapis/python-api-core/issues/715): Add support for `credentials_file` to async REST transport.
+            "google.api_core.client_options.ClientOptions.credentials_file": credentials_file,
+            # TODO(https://github.com/googleapis/python-api-core/issues/716): Add support for `scopes` to async REST transport.
+            "google.api_core.client_options.ClientOptions.scopes": scopes,
+            # TODO(https://github.com/googleapis/python-api-core/issues/717): Add support for `quota_project_id` to async REST transport.
+            "google.api_core.client_options.ClientOptions.quota_project_id": quota_project_id,
+            # TODO(https://github.com/googleapis/python-api-core/issues/718): Add support for `client_cert_source` to async REST transport.
+            "google.api_core.client_options.ClientOptions.client_cert_source": client_cert_source_for_mtls,
+            # TODO(https://github.com/googleapis/python-api-core/issues/718): Add support for `client_cert_source` to async REST transport.
+            "google.api_core.client_options.ClientOptions.client_cert_source": client_cert_source_for_mtls,
+        }
+        provided_unsupported_params = [
+            name for name, value in unsupported_params.items() if value is not None
+        ]
+        if provided_unsupported_params:
+            raise core_exceptions.AsyncRestUnsupportedParameterError(
+                f"The following provided parameters are not supported for `transport=rest_asyncio`: {', '.join(provided_unsupported_params)}"
+            )
+
+        super().__init__(
+            host=host,
+            # TODO(https://github.com/googleapis/python-api-core/issues/709): Remove `type: ignore` when the linked issue is resolved.
+            credentials=credentials,  # type: ignore
+            client_info=client_info,
+            # TODO(https://github.com/googleapis/python-api-core/issues/725): Set always_use_jwt_access token when supported.
+            always_use_jwt_access=False,
+        )
+        # TODO(https://github.com/googleapis/python-api-core/issues/708): add support for
+        # `default_host` in AsyncAuthorizedSession for feature parity with the synchronous
+        # code.
+        # TODO(https://github.com/googleapis/python-api-core/issues/709): Remove `type: ignore` when the linked issue is resolved.
+        self._session = AsyncAuthorizedSession(self._credentials)  # type: ignore
+        # TODO(https://github.com/googleapis/python-api-core/issues/720): Add wrap logic directly to the property methods for callables.
+        self._prep_wrapped_messages(client_info)
+        self._http_options = http_options or {}
+        self._path_prefix = path_prefix
+
+    def _prep_wrapped_messages(self, client_info):
+        # Precompute the wrapped methods.
+        self._wrapped_methods = {
+            self.list_operations: gapic_v1.method_async.wrap_method(
+                self.list_operations,
+                default_retry=retries_async.AsyncRetry(
+                    initial=0.5,
+                    maximum=10.0,
+                    multiplier=2.0,
+                    predicate=retries_async.if_exception_type(
+                        core_exceptions.ServiceUnavailable,
+                    ),
+                    deadline=10.0,
+                ),
+                default_timeout=10.0,
+                client_info=client_info,
+                kind="rest_asyncio",
+            ),
+            self.get_operation: gapic_v1.method_async.wrap_method(
+                self.get_operation,
+                default_retry=retries_async.AsyncRetry(
+                    initial=0.5,
+                    maximum=10.0,
+                    multiplier=2.0,
+                    predicate=retries_async.if_exception_type(
+                        core_exceptions.ServiceUnavailable,
+                    ),
+                    deadline=10.0,
+                ),
+                default_timeout=10.0,
+                client_info=client_info,
+                kind="rest_asyncio",
+            ),
+            self.delete_operation: gapic_v1.method_async.wrap_method(
+                self.delete_operation,
+                default_retry=retries_async.AsyncRetry(
+                    initial=0.5,
+                    maximum=10.0,
+                    multiplier=2.0,
+                    predicate=retries_async.if_exception_type(
+                        core_exceptions.ServiceUnavailable,
+                    ),
+                    deadline=10.0,
+                ),
+                default_timeout=10.0,
+                client_info=client_info,
+                kind="rest_asyncio",
+            ),
+            self.cancel_operation: gapic_v1.method_async.wrap_method(
+                self.cancel_operation,
+                default_retry=retries_async.AsyncRetry(
+                    initial=0.5,
+                    maximum=10.0,
+                    multiplier=2.0,
+                    predicate=retries_async.if_exception_type(
+                        core_exceptions.ServiceUnavailable,
+                    ),
+                    deadline=10.0,
+                ),
+                default_timeout=10.0,
+                client_info=client_info,
+                kind="rest_asyncio",
+            ),
+        }
+
+    async def _list_operations(
+        self,
+        request: operations_pb2.ListOperationsRequest,
+        *,
+        # TODO(https://github.com/googleapis/python-api-core/issues/722): Leverage `retry`
+        # to allow configuring retryable error codes.
+        retry=gapic_v1.method_async.DEFAULT,
+        timeout: Optional[float] = None,
+        metadata: Sequence[Tuple[str, str]] = (),
+    ) -> operations_pb2.ListOperationsResponse:
+        r"""Asynchronously call the list operations method over HTTP.
+
+        Args:
+            request (~.operations_pb2.ListOperationsRequest):
+                The request object. The request message for
+                [Operations.ListOperations][google.api_core.operations_v1.Operations.ListOperations].
+            timeout (float): The timeout for this request.
+            metadata (Sequence[Tuple[str, str]]): Strings which should be
+                sent along with the request as metadata.
+
+        Returns:
+            ~.operations_pb2.ListOperationsResponse:
+                The response message for
+                [Operations.ListOperations][google.api_core.operations_v1.Operations.ListOperations].
+
+        """
+
+        http_options = [
+            {
+                "method": "get",
+                "uri": "/{}/{{name=**}}/operations".format(self._path_prefix),
+            },
+        ]
+        if "google.longrunning.Operations.ListOperations" in self._http_options:
+            http_options = self._http_options[
+                "google.longrunning.Operations.ListOperations"
+            ]
+
+        request_kwargs = self._convert_protobuf_message_to_dict(request)
+        transcoded_request = path_template.transcode(http_options, **request_kwargs)
+
+        uri = transcoded_request["uri"]
+        method = transcoded_request["method"]
+
+        # Jsonify the query params
+        query_params_request = operations_pb2.ListOperationsRequest()
+        json_format.ParseDict(transcoded_request["query_params"], query_params_request)
+        query_params = json_format.MessageToDict(
+            query_params_request,
+            preserving_proto_field_name=False,
+            use_integers_for_enums=False,
+        )
+
+        # Send the request
+        headers = dict(metadata)
+        headers["Content-Type"] = "application/json"
+        # TODO(https://github.com/googleapis/python-api-core/issues/721): Update incorrect use of `uri`` variable name.
+        response = await getattr(self._session, method)(
+            "{host}{uri}".format(host=self._host, uri=uri),
+            timeout=timeout,
+            headers=headers,
+            params=rest_helpers.flatten_query_params(query_params),
+        )
+        content = await response.read()
+
+        # In case of error, raise the appropriate core_exceptions.GoogleAPICallError exception
+        # subclass.
+        if response.status_code >= 400:
+            payload = json.loads(content.decode("utf-8"))
+            request_url = "{host}{uri}".format(host=self._host, uri=uri)
+            raise core_exceptions.format_http_response_error(response, method, request_url, payload)  # type: ignore
+
+        # Return the response
+        api_response = operations_pb2.ListOperationsResponse()
+        json_format.Parse(content, api_response, ignore_unknown_fields=False)
+        return api_response
+
+    async def _get_operation(
+        self,
+        request: operations_pb2.GetOperationRequest,
+        *,
+        # TODO(https://github.com/googleapis/python-api-core/issues/722): Leverage `retry`
+        # to allow configuring retryable error codes.
+        retry=gapic_v1.method_async.DEFAULT,
+        timeout: Optional[float] = None,
+        metadata: Sequence[Tuple[str, str]] = (),
+    ) -> operations_pb2.Operation:
+        r"""Asynchronously call the get operation method over HTTP.
+
+        Args:
+            request (~.operations_pb2.GetOperationRequest):
+                The request object. The request message for
+                [Operations.GetOperation][google.api_core.operations_v1.Operations.GetOperation].
+            timeout (float): The timeout for this request.
+            metadata (Sequence[Tuple[str, str]]): Strings which should be
+                sent along with the request as metadata.
+
+        Returns:
+            ~.operations_pb2.Operation:
+                This resource represents a long-
+                running operation that is the result of a
+                network API call.
+
+        """
+
+        http_options = [
+            {
+                "method": "get",
+                "uri": "/{}/{{name=**/operations/*}}".format(self._path_prefix),
+            },
+        ]
+        if "google.longrunning.Operations.GetOperation" in self._http_options:
+            http_options = self._http_options[
+                "google.longrunning.Operations.GetOperation"
+            ]
+
+        request_kwargs = self._convert_protobuf_message_to_dict(request)
+        transcoded_request = path_template.transcode(http_options, **request_kwargs)
+
+        uri = transcoded_request["uri"]
+        method = transcoded_request["method"]
+
+        # Jsonify the query params
+        query_params_request = operations_pb2.GetOperationRequest()
+        json_format.ParseDict(transcoded_request["query_params"], query_params_request)
+        query_params = json_format.MessageToDict(
+            query_params_request,
+            preserving_proto_field_name=False,
+            use_integers_for_enums=False,
+        )
+
+        # Send the request
+        headers = dict(metadata)
+        headers["Content-Type"] = "application/json"
+        # TODO(https://github.com/googleapis/python-api-core/issues/721): Update incorrect use of `uri`` variable name.
+        response = await getattr(self._session, method)(
+            "{host}{uri}".format(host=self._host, uri=uri),
+            timeout=timeout,
+            headers=headers,
+            params=rest_helpers.flatten_query_params(query_params),
+        )
+        content = await response.read()
+
+        # In case of error, raise the appropriate core_exceptions.GoogleAPICallError exception
+        # subclass.
+        if response.status_code >= 400:
+            payload = json.loads(content.decode("utf-8"))
+            request_url = "{host}{uri}".format(host=self._host, uri=uri)
+            raise core_exceptions.format_http_response_error(response, method, request_url, payload)  # type: ignore
+
+        # Return the response
+        api_response = operations_pb2.Operation()
+        json_format.Parse(content, api_response, ignore_unknown_fields=False)
+        return api_response
+
+    async def _delete_operation(
+        self,
+        request: operations_pb2.DeleteOperationRequest,
+        *,
+        # TODO(https://github.com/googleapis/python-api-core/issues/722): Leverage `retry`
+        # to allow configuring retryable error codes.
+        retry=gapic_v1.method_async.DEFAULT,
+        timeout: Optional[float] = None,
+        metadata: Sequence[Tuple[str, str]] = (),
+    ) -> empty_pb2.Empty:
+        r"""Asynchronously call the delete operation method over HTTP.
+
+        Args:
+            request (~.operations_pb2.DeleteOperationRequest):
+                The request object. The request message for
+                [Operations.DeleteOperation][google.api_core.operations_v1.Operations.DeleteOperation].
+
+            retry (google.api_core.retry.Retry): Designation of what errors, if any,
+                should be retried.
+            timeout (float): The timeout for this request.
+            metadata (Sequence[Tuple[str, str]]): Strings which should be
+                sent along with the request as metadata.
+        """
+
+        http_options = [
+            {
+                "method": "delete",
+                "uri": "/{}/{{name=**/operations/*}}".format(self._path_prefix),
+            },
+        ]
+        if "google.longrunning.Operations.DeleteOperation" in self._http_options:
+            http_options = self._http_options[
+                "google.longrunning.Operations.DeleteOperation"
+            ]
+
+        request_kwargs = self._convert_protobuf_message_to_dict(request)
+        transcoded_request = path_template.transcode(http_options, **request_kwargs)
+
+        uri = transcoded_request["uri"]
+        method = transcoded_request["method"]
+
+        # Jsonify the query params
+        query_params_request = operations_pb2.DeleteOperationRequest()
+        json_format.ParseDict(transcoded_request["query_params"], query_params_request)
+        query_params = json_format.MessageToDict(
+            query_params_request,
+            preserving_proto_field_name=False,
+            use_integers_for_enums=False,
+        )
+
+        # Send the request
+        headers = dict(metadata)
+        headers["Content-Type"] = "application/json"
+        # TODO(https://github.com/googleapis/python-api-core/issues/721): Update incorrect use of `uri`` variable name.
+        response = await getattr(self._session, method)(
+            "{host}{uri}".format(host=self._host, uri=uri),
+            timeout=timeout,
+            headers=headers,
+            params=rest_helpers.flatten_query_params(query_params),
+        )
+
+        # In case of error, raise the appropriate core_exceptions.GoogleAPICallError exception
+        # subclass.
+        if response.status_code >= 400:
+            content = await response.read()
+            payload = json.loads(content.decode("utf-8"))
+            request_url = "{host}{uri}".format(host=self._host, uri=uri)
+            raise core_exceptions.format_http_response_error(response, method, request_url, payload)  # type: ignore
+
+        return empty_pb2.Empty()
+
+    async def _cancel_operation(
+        self,
+        request: operations_pb2.CancelOperationRequest,
+        *,
+        # TODO(https://github.com/googleapis/python-api-core/issues/722): Leverage `retry`
+        # to allow configuring retryable error codes.
+        retry=gapic_v1.method_async.DEFAULT,
+        timeout: Optional[float] = None,
+        metadata: Sequence[Tuple[str, str]] = (),
+        # TODO(https://github.com/googleapis/python-api-core/issues/722): Add `retry` parameter
+        # to allow configuring retryable error codes.
+    ) -> empty_pb2.Empty:
+        r"""Asynchronously call the cancel operation method over HTTP.
+
+        Args:
+            request (~.operations_pb2.CancelOperationRequest):
+                The request object. The request message for
+                [Operations.CancelOperation][google.api_core.operations_v1.Operations.CancelOperation].
+            timeout (float): The timeout for this request.
+            metadata (Sequence[Tuple[str, str]]): Strings which should be
+                sent along with the request as metadata.
+        """
+
+        http_options = [
+            {
+                "method": "post",
+                "uri": "/{}/{{name=**/operations/*}}:cancel".format(self._path_prefix),
+                "body": "*",
+            },
+        ]
+        if "google.longrunning.Operations.CancelOperation" in self._http_options:
+            http_options = self._http_options[
+                "google.longrunning.Operations.CancelOperation"
+            ]
+
+        request_kwargs = self._convert_protobuf_message_to_dict(request)
+        transcoded_request = path_template.transcode(http_options, **request_kwargs)
+
+        # Jsonify the request body
+        body_request = operations_pb2.CancelOperationRequest()
+        json_format.ParseDict(transcoded_request["body"], body_request)
+        body = json_format.MessageToDict(
+            body_request,
+            preserving_proto_field_name=False,
+            use_integers_for_enums=False,
+        )
+        uri = transcoded_request["uri"]
+        method = transcoded_request["method"]
+
+        # Jsonify the query params
+        query_params_request = operations_pb2.CancelOperationRequest()
+        json_format.ParseDict(transcoded_request["query_params"], query_params_request)
+        query_params = json_format.MessageToDict(
+            query_params_request,
+            preserving_proto_field_name=False,
+            use_integers_for_enums=False,
+        )
+
+        # Send the request
+        headers = dict(metadata)
+        headers["Content-Type"] = "application/json"
+        # TODO(https://github.com/googleapis/python-api-core/issues/721): Update incorrect use of `uri`` variable name.
+        response = await getattr(self._session, method)(
+            "{host}{uri}".format(host=self._host, uri=uri),
+            timeout=timeout,
+            headers=headers,
+            params=rest_helpers.flatten_query_params(query_params),
+            data=body,
+        )
+
+        # In case of error, raise the appropriate core_exceptions.GoogleAPICallError exception
+        # subclass.
+        if response.status_code >= 400:
+            content = await response.read()
+            payload = json.loads(content.decode("utf-8"))
+            request_url = "{host}{uri}".format(host=self._host, uri=uri)
+            raise core_exceptions.format_http_response_error(response, method, request_url, payload)  # type: ignore
+
+        return empty_pb2.Empty()
+
+    @property
+    def list_operations(
+        self,
+    ) -> Callable[
+        [operations_pb2.ListOperationsRequest],
+        Coroutine[Any, Any, operations_pb2.ListOperationsResponse],
+    ]:
+        return self._list_operations
+
+    @property
+    def get_operation(
+        self,
+    ) -> Callable[
+        [operations_pb2.GetOperationRequest],
+        Coroutine[Any, Any, operations_pb2.Operation],
+    ]:
+        return self._get_operation
+
+    @property
+    def delete_operation(
+        self,
+    ) -> Callable[
+        [operations_pb2.DeleteOperationRequest], Coroutine[Any, Any, empty_pb2.Empty]
+    ]:
+        return self._delete_operation
+
+    @property
+    def cancel_operation(
+        self,
+    ) -> Callable[
+        [operations_pb2.CancelOperationRequest], Coroutine[Any, Any, empty_pb2.Empty]
+    ]:
+        return self._cancel_operation
+
+
+__all__ = ("AsyncOperationsRestTransport",)
diff --git a/tests/unit/operations_v1/test_operations_rest_client.py b/tests/unit/operations_v1/test_operations_rest_client.py
index 4ab4f1f..644cf26 100644
--- a/tests/unit/operations_v1/test_operations_rest_client.py
+++ b/tests/unit/operations_v1/test_operations_rest_client.py
@@ -17,21 +17,24 @@
 
 import mock
 import pytest
+from typing import Any, List
 
 try:
     import grpc  # noqa: F401
 except ImportError:  # pragma: NO COVER
     pytest.skip("No GRPC", allow_module_level=True)
 from requests import Response  # noqa I201
-from requests.sessions import Session
+from google.auth.transport.requests import AuthorizedSession
 
 from google.api_core import client_options
 from google.api_core import exceptions as core_exceptions
 from google.api_core import gapic_v1
 from google.api_core.operations_v1 import AbstractOperationsClient
-from google.api_core.operations_v1 import pagers
-from google.api_core.operations_v1 import transports
+
 import google.auth
+from google.api_core.operations_v1 import pagers
+from google.api_core.operations_v1 import pagers_async
+from google.api_core.operations_v1 import transports
 from google.auth import credentials as ga_credentials
 from google.auth.exceptions import MutualTLSChannelError
 from google.longrunning import operations_pb2
@@ -39,6 +42,16 @@
 from google.protobuf import json_format  # type: ignore
 from google.rpc import status_pb2  # type: ignore
 
+try:
+    import aiohttp  # noqa: F401
+    import google.auth.aio.transport
+    from google.auth.aio.transport.sessions import AsyncAuthorizedSession
+    from google.api_core.operations_v1 import AsyncOperationsRestClient
+    from google.auth.aio import credentials as ga_credentials_async
+
+    GOOGLE_AUTH_AIO_INSTALLED = True
+except ImportError:
+    GOOGLE_AUTH_AIO_INSTALLED = False
 
 HTTP_OPTIONS = {
     "google.longrunning.Operations.CancelOperation": [
@@ -55,17 +68,62 @@
     ],
 }
 
+PYPARAM_CLIENT: List[Any] = [
+    AbstractOperationsClient,
+]
+PYPARAM_CLIENT_TRANSPORT_NAME = [
+    [AbstractOperationsClient, transports.OperationsRestTransport, "rest"],
+]
+PYPARAM_CLIENT_TRANSPORT_CREDENTIALS = [
+    [
+        AbstractOperationsClient,
+        transports.OperationsRestTransport,
+        ga_credentials.AnonymousCredentials(),
+    ],
+]
+
+if GOOGLE_AUTH_AIO_INSTALLED:
+    PYPARAM_CLIENT.append(AsyncOperationsRestClient)
+    PYPARAM_CLIENT_TRANSPORT_NAME.append(
+        [
+            AsyncOperationsRestClient,
+            transports.AsyncOperationsRestTransport,
+            "rest_asyncio",
+        ]
+    )
+    PYPARAM_CLIENT_TRANSPORT_CREDENTIALS.append(
+        [
+            AsyncOperationsRestClient,
+            transports.AsyncOperationsRestTransport,
+            ga_credentials_async.AnonymousCredentials(),
+        ]
+    )
+
 
 def client_cert_source_callback():
     return b"cert bytes", b"key bytes"
 
 
-def _get_operations_client(http_options=HTTP_OPTIONS):
-    transport = transports.rest.OperationsRestTransport(
-        credentials=ga_credentials.AnonymousCredentials(), http_options=http_options
+def _get_session_type(is_async: bool):
+    return (
+        AsyncAuthorizedSession
+        if is_async and GOOGLE_AUTH_AIO_INSTALLED
+        else AuthorizedSession
     )
 
-    return AbstractOperationsClient(transport=transport)
+
+def _get_operations_client(is_async: bool, http_options=HTTP_OPTIONS):
+    if is_async and GOOGLE_AUTH_AIO_INSTALLED:
+        async_transport = transports.rest_asyncio.AsyncOperationsRestTransport(
+            credentials=ga_credentials_async.AnonymousCredentials(),
+            http_options=http_options,
+        )
+        return AsyncOperationsRestClient(transport=async_transport)
+    else:
+        sync_transport = transports.rest.OperationsRestTransport(
+            credentials=ga_credentials.AnonymousCredentials(), http_options=http_options
+        )
+        return AbstractOperationsClient(transport=sync_transport)
 
 
 # If default endpoint is localhost, then default mtls endpoint will be the same.
@@ -79,57 +137,69 @@
     )
 
 
-def test__get_default_mtls_endpoint():
+# TODO: Add support for mtls in async rest
+@pytest.mark.parametrize(
+    "client_class",
+    [
+        AbstractOperationsClient,
+    ],
+)
+def test__get_default_mtls_endpoint(client_class):
     api_endpoint = "example.googleapis.com"
     api_mtls_endpoint = "example.mtls.googleapis.com"
     sandbox_endpoint = "example.sandbox.googleapis.com"
     sandbox_mtls_endpoint = "example.mtls.sandbox.googleapis.com"
     non_googleapi = "api.example.com"
 
-    assert AbstractOperationsClient._get_default_mtls_endpoint(None) is None
+    assert client_class._get_default_mtls_endpoint(None) is None
+    assert client_class._get_default_mtls_endpoint(api_endpoint) == api_mtls_endpoint
     assert (
-        AbstractOperationsClient._get_default_mtls_endpoint(api_endpoint)
-        == api_mtls_endpoint
+        client_class._get_default_mtls_endpoint(api_mtls_endpoint) == api_mtls_endpoint
     )
     assert (
-        AbstractOperationsClient._get_default_mtls_endpoint(api_mtls_endpoint)
-        == api_mtls_endpoint
-    )
-    assert (
-        AbstractOperationsClient._get_default_mtls_endpoint(sandbox_endpoint)
+        client_class._get_default_mtls_endpoint(sandbox_endpoint)
         == sandbox_mtls_endpoint
     )
     assert (
-        AbstractOperationsClient._get_default_mtls_endpoint(sandbox_mtls_endpoint)
+        client_class._get_default_mtls_endpoint(sandbox_mtls_endpoint)
         == sandbox_mtls_endpoint
     )
-    assert (
-        AbstractOperationsClient._get_default_mtls_endpoint(non_googleapi)
-        == non_googleapi
-    )
-
-
-@pytest.mark.parametrize("client_class", [AbstractOperationsClient])
-def test_operations_client_from_service_account_info(client_class):
-    creds = ga_credentials.AnonymousCredentials()
-    with mock.patch.object(
-        service_account.Credentials, "from_service_account_info"
-    ) as factory:
-        factory.return_value = creds
-        info = {"valid": True}
-        client = client_class.from_service_account_info(info)
-        assert client.transport._credentials == creds
-        assert isinstance(client, client_class)
-
-        assert client.transport._host == "https://longrunning.googleapis.com"
+    assert client_class._get_default_mtls_endpoint(non_googleapi) == non_googleapi
 
 
 @pytest.mark.parametrize(
-    "transport_class,transport_name", [(transports.OperationsRestTransport, "rest")]
+    "client_class",
+    PYPARAM_CLIENT,
 )
-def test_operations_client_service_account_always_use_jwt(
-    transport_class, transport_name
-):
+def test_operations_client_from_service_account_info(client_class):
+    creds = ga_credentials.AnonymousCredentials()
+    if "async" in str(client_class):
+        # TODO(): Add support for service account info to async REST transport.
+        with pytest.raises(NotImplementedError):
+            info = {"valid": True}
+            client_class.from_service_account_info(info)
+    else:
+        with mock.patch.object(
+            service_account.Credentials, "from_service_account_info"
+        ) as factory:
+            factory.return_value = creds
+            info = {"valid": True}
+            client = client_class.from_service_account_info(info)
+            assert client.transport._credentials == creds
+            assert isinstance(client, client_class)
+
+            assert client.transport._host == "https://longrunning.googleapis.com"
+
+
+@pytest.mark.parametrize(
+    "transport_class",
+    [
+        transports.OperationsRestTransport,
+        # TODO(https://github.com/googleapis/python-api-core/issues/706): Add support for
+        # service account credentials in transports.AsyncOperationsRestTransport
+    ],
+)
+def test_operations_client_service_account_always_use_jwt(transport_class):
     with mock.patch.object(
         service_account.Credentials, "with_always_use_jwt_access", create=True
     ) as use_jwt:
@@ -145,35 +215,53 @@
         use_jwt.assert_not_called()
 
 
-@pytest.mark.parametrize("client_class", [AbstractOperationsClient])
+@pytest.mark.parametrize(
+    "client_class",
+    PYPARAM_CLIENT,
+)
 def test_operations_client_from_service_account_file(client_class):
-    creds = ga_credentials.AnonymousCredentials()
-    with mock.patch.object(
-        service_account.Credentials, "from_service_account_file"
-    ) as factory:
-        factory.return_value = creds
-        client = client_class.from_service_account_file("dummy/file/path.json")
-        assert client.transport._credentials == creds
-        assert isinstance(client, client_class)
 
-        client = client_class.from_service_account_json("dummy/file/path.json")
-        assert client.transport._credentials == creds
-        assert isinstance(client, client_class)
+    if "async" in str(client_class):
+        # TODO(): Add support for service account creds to async REST transport.
+        with pytest.raises(NotImplementedError):
+            client_class.from_service_account_file("dummy/file/path.json")
+    else:
+        creds = ga_credentials.AnonymousCredentials()
+        with mock.patch.object(
+            service_account.Credentials, "from_service_account_file"
+        ) as factory:
+            factory.return_value = creds
+            client = client_class.from_service_account_file("dummy/file/path.json")
+            assert client.transport._credentials == creds
+            assert isinstance(client, client_class)
 
-        assert client.transport._host == "https://longrunning.googleapis.com"
+            client = client_class.from_service_account_json("dummy/file/path.json")
+            assert client.transport._credentials == creds
+            assert isinstance(client, client_class)
+
+            assert client.transport._host == "https://longrunning.googleapis.com"
 
 
-def test_operations_client_get_transport_class():
-    transport = AbstractOperationsClient.get_transport_class()
+@pytest.mark.parametrize(
+    "client_class,transport_class,transport_name",
+    PYPARAM_CLIENT_TRANSPORT_NAME,
+)
+def test_operations_client_get_transport_class(
+    client_class, transport_class, transport_name
+):
+    transport = client_class.get_transport_class()
     available_transports = [
         transports.OperationsRestTransport,
     ]
+    if GOOGLE_AUTH_AIO_INSTALLED:
+        available_transports.append(transports.AsyncOperationsRestTransport)
     assert transport in available_transports
 
-    transport = AbstractOperationsClient.get_transport_class("rest")
-    assert transport == transports.OperationsRestTransport
+    transport = client_class.get_transport_class(transport_name)
+    assert transport == transport_class
 
 
+# TODO(): Update this test case to include async REST once we have support for MTLS.
 @pytest.mark.parametrize(
     "client_class,transport_class,transport_name",
     [(AbstractOperationsClient, transports.OperationsRestTransport, "rest")],
@@ -186,22 +274,21 @@
 def test_operations_client_client_options(
     client_class, transport_class, transport_name
 ):
-    # Check that if channel is provided we won't create a new one.
-    with mock.patch.object(AbstractOperationsClient, "get_transport_class") as gtc:
-        transport = transport_class(credentials=ga_credentials.AnonymousCredentials())
-        client = client_class(transport=transport)
-        gtc.assert_not_called()
+    # # Check that if channel is provided we won't create a new one.
+    # with mock.patch.object(AbstractOperationsBaseClient, "get_transport_class") as gtc:
+    #     client = client_class(transport=transport_class())
+    #     gtc.assert_not_called()
 
-    # Check that if channel is provided via str we will create a new one.
-    with mock.patch.object(AbstractOperationsClient, "get_transport_class") as gtc:
-        client = client_class(transport=transport_name)
-        gtc.assert_called()
+    # # Check that if channel is provided via str we will create a new one.
+    # with mock.patch.object(AbstractOperationsBaseClient, "get_transport_class") as gtc:
+    #     client = client_class(transport=transport_name)
+    #     gtc.assert_called()
 
     # Check the case api_endpoint is provided.
     options = client_options.ClientOptions(api_endpoint="squid.clam.whelk")
     with mock.patch.object(transport_class, "__init__") as patched:
         patched.return_value = None
-        client = client_class(client_options=options)
+        client = client_class(client_options=options, transport=transport_name)
         patched.assert_called_once_with(
             credentials=None,
             credentials_file=None,
@@ -218,7 +305,7 @@
     with mock.patch.dict(os.environ, {"GOOGLE_API_USE_MTLS_ENDPOINT": "never"}):
         with mock.patch.object(transport_class, "__init__") as patched:
             patched.return_value = None
-            client = client_class()
+            client = client_class(transport=transport_name)
             patched.assert_called_once_with(
                 credentials=None,
                 credentials_file=None,
@@ -235,7 +322,7 @@
     with mock.patch.dict(os.environ, {"GOOGLE_API_USE_MTLS_ENDPOINT": "always"}):
         with mock.patch.object(transport_class, "__init__") as patched:
             patched.return_value = None
-            client = client_class()
+            client = client_class(transport=transport_name)
             patched.assert_called_once_with(
                 credentials=None,
                 credentials_file=None,
@@ -264,7 +351,7 @@
     options = client_options.ClientOptions(quota_project_id="octopus")
     with mock.patch.object(transport_class, "__init__") as patched:
         patched.return_value = None
-        client = client_class(client_options=options)
+        client = client_class(client_options=options, transport=transport_name)
         patched.assert_called_once_with(
             credentials=None,
             credentials_file=None,
@@ -277,6 +364,7 @@
         )
 
 
+# TODO: Add support for mtls in async REST
 @pytest.mark.parametrize(
     "client_class,transport_class,transport_name,use_client_cert_env",
     [
@@ -393,7 +481,7 @@
 
 @pytest.mark.parametrize(
     "client_class,transport_class,transport_name",
-    [(AbstractOperationsClient, transports.OperationsRestTransport, "rest")],
+    PYPARAM_CLIENT_TRANSPORT_NAME,
 )
 def test_operations_client_client_options_scopes(
     client_class, transport_class, transport_name
@@ -402,52 +490,59 @@
     options = client_options.ClientOptions(
         scopes=["1", "2"],
     )
-    with mock.patch.object(transport_class, "__init__") as patched:
-        patched.return_value = None
-        client = client_class(client_options=options)
-        patched.assert_called_once_with(
-            credentials=None,
-            credentials_file=None,
-            host=client.DEFAULT_ENDPOINT,
-            scopes=["1", "2"],
-            client_cert_source_for_mtls=None,
-            quota_project_id=None,
-            client_info=transports.base.DEFAULT_CLIENT_INFO,
-            always_use_jwt_access=True,
-        )
+    if "async" in str(client_class):
+        # TODO(): Add support for scopes to async REST transport.
+        with pytest.raises(core_exceptions.AsyncRestUnsupportedParameterError):
+            client_class(client_options=options, transport=transport_name)
+    else:
+        with mock.patch.object(transport_class, "__init__") as patched:
+            patched.return_value = None
+            client = client_class(client_options=options, transport=transport_name)
+            patched.assert_called_once_with(
+                credentials=None,
+                credentials_file=None,
+                host=client.DEFAULT_ENDPOINT,
+                scopes=["1", "2"],
+                client_cert_source_for_mtls=None,
+                quota_project_id=None,
+                client_info=transports.base.DEFAULT_CLIENT_INFO,
+                always_use_jwt_access=True,
+            )
 
 
 @pytest.mark.parametrize(
     "client_class,transport_class,transport_name",
-    [(AbstractOperationsClient, transports.OperationsRestTransport, "rest")],
+    PYPARAM_CLIENT_TRANSPORT_NAME,
 )
 def test_operations_client_client_options_credentials_file(
     client_class, transport_class, transport_name
 ):
     # Check the case credentials file is provided.
     options = client_options.ClientOptions(credentials_file="credentials.json")
-    with mock.patch.object(transport_class, "__init__") as patched:
-        patched.return_value = None
-        client = client_class(client_options=options)
-        patched.assert_called_once_with(
-            credentials=None,
-            credentials_file="credentials.json",
-            host=client.DEFAULT_ENDPOINT,
-            scopes=None,
-            client_cert_source_for_mtls=None,
-            quota_project_id=None,
-            client_info=transports.base.DEFAULT_CLIENT_INFO,
-            always_use_jwt_access=True,
-        )
+    if "async" in str(client_class):
+        # TODO(): Add support for credentials file to async REST transport.
+        with pytest.raises(core_exceptions.AsyncRestUnsupportedParameterError):
+            client_class(client_options=options, transport=transport_name)
+    else:
+        with mock.patch.object(transport_class, "__init__") as patched:
+            patched.return_value = None
+            client = client_class(client_options=options, transport=transport_name)
+            patched.assert_called_once_with(
+                credentials=None,
+                credentials_file="credentials.json",
+                host=client.DEFAULT_ENDPOINT,
+                scopes=None,
+                client_cert_source_for_mtls=None,
+                quota_project_id=None,
+                client_info=transports.base.DEFAULT_CLIENT_INFO,
+                always_use_jwt_access=True,
+            )
 
 
-def test_list_operations_rest(
-    transport: str = "rest", request_type=operations_pb2.ListOperationsRequest
-):
-    client = _get_operations_client()
-
+def test_list_operations_rest():
+    client = _get_operations_client(is_async=False)
     # Mock the http request call within the method and fake a response.
-    with mock.patch.object(Session, "request") as req:
+    with mock.patch.object(_get_session_type(is_async=False), "request") as req:
         # Designate an appropriate value for the returned response.
         return_value = operations_pb2.ListOperationsResponse(
             next_page_token="next_page_token_value",
@@ -477,10 +572,49 @@
     assert response.next_page_token == "next_page_token_value"
 
 
-def test_list_operations_rest_failure():
-    client = _get_operations_client(http_options=None)
+@pytest.mark.asyncio
+async def test_list_operations_rest_async():
+    if not GOOGLE_AUTH_AIO_INSTALLED:
+        pytest.skip("Skipped because google-api-core[async_rest] is not installed")
 
-    with mock.patch.object(Session, "request") as req:
+    client = _get_operations_client(is_async=True)
+    # Mock the http request call within the method and fake a response.
+    with mock.patch.object(_get_session_type(is_async=True), "request") as req:
+        # Designate an appropriate value for the returned response.
+        return_value = operations_pb2.ListOperationsResponse(
+            next_page_token="next_page_token_value",
+        )
+
+        # Wrap the value into a proper Response obj
+        response_value = mock.Mock()
+        response_value.status_code = 200
+        json_return_value = json_format.MessageToJson(return_value)
+        response_value.read = mock.AsyncMock(
+            return_value=json_return_value.encode("UTF-8")
+        )
+        req.return_value = response_value
+        response = await client.list_operations(
+            name="operations", filter_="my_filter", page_size=10, page_token="abc"
+        )
+
+        actual_args = req.call_args
+        assert actual_args.args[0] == "GET"
+        assert actual_args.args[1] == "https://longrunning.googleapis.com/v3/operations"
+        assert actual_args.kwargs["params"] == [
+            ("filter", "my_filter"),
+            ("pageSize", 10),
+            ("pageToken", "abc"),
+        ]
+
+    # Establish that the response is the type that we expect.
+    assert isinstance(response, pagers_async.ListOperationsAsyncPager)
+    assert response.next_page_token == "next_page_token_value"
+
+
+def test_list_operations_rest_failure():
+    client = _get_operations_client(is_async=False, http_options=None)
+
+    with mock.patch.object(_get_session_type(is_async=False), "request") as req:
         response_value = Response()
         response_value.status_code = 400
         mock_request = mock.MagicMock()
@@ -492,13 +626,31 @@
             client.list_operations(name="operations")
 
 
+@pytest.mark.asyncio
+async def test_list_operations_rest_failure_async():
+    if not GOOGLE_AUTH_AIO_INSTALLED:
+        pytest.skip("Skipped because google-api-core[async_rest] is not installed")
+
+    client = _get_operations_client(is_async=True, http_options=None)
+
+    with mock.patch.object(_get_session_type(is_async=True), "request") as req:
+        response_value = mock.Mock()
+        response_value.status_code = 400
+        response_value.read = mock.AsyncMock(return_value=b"{}")
+        mock_request = mock.MagicMock()
+        mock_request.method = "GET"
+        mock_request.url = "https://longrunning.googleapis.com:443/v1/operations"
+        response_value.request = mock_request
+        req.return_value = response_value
+        with pytest.raises(core_exceptions.GoogleAPIError):
+            await client.list_operations(name="operations")
+
+
 def test_list_operations_rest_pager():
-    client = AbstractOperationsClient(
-        credentials=ga_credentials.AnonymousCredentials(),
-    )
+    client = _get_operations_client(is_async=False, http_options=None)
 
     # Mock the http request call within the method and fake a response.
-    with mock.patch.object(Session, "request") as req:
+    with mock.patch.object(_get_session_type(is_async=False), "request") as req:
         # TODO(kbandes): remove this mock unless there's a good reason for it.
         # with mock.patch.object(path_template, 'transcode') as transcode:
         # Set the response as a series of pages
@@ -545,13 +697,80 @@
             assert page_.next_page_token == token
 
 
-def test_get_operation_rest(
-    transport: str = "rest", request_type=operations_pb2.GetOperationRequest
-):
-    client = _get_operations_client()
+@pytest.mark.asyncio
+async def test_list_operations_rest_pager_async():
+    if not GOOGLE_AUTH_AIO_INSTALLED:
+        pytest.skip("Skipped because google-api-core[async_rest] is not installed")
+    client = _get_operations_client(is_async=True, http_options=None)
 
     # Mock the http request call within the method and fake a response.
-    with mock.patch.object(Session, "request") as req:
+    with mock.patch.object(_get_session_type(is_async=True), "request") as req:
+        # TODO(kbandes): remove this mock unless there's a good reason for it.
+        # with mock.patch.object(path_template, 'transcode') as transcode:
+        # Set the response as a series of pages
+        response = (
+            operations_pb2.ListOperationsResponse(
+                operations=[
+                    operations_pb2.Operation(),
+                    operations_pb2.Operation(),
+                    operations_pb2.Operation(),
+                ],
+                next_page_token="abc",
+            ),
+            operations_pb2.ListOperationsResponse(
+                operations=[],
+                next_page_token="def",
+            ),
+            operations_pb2.ListOperationsResponse(
+                operations=[operations_pb2.Operation()],
+                next_page_token="ghi",
+            ),
+            operations_pb2.ListOperationsResponse(
+                operations=[operations_pb2.Operation(), operations_pb2.Operation()],
+            ),
+        )
+        # Two responses for two calls
+        response = response + response
+
+        # Wrap the values into proper Response objs
+        response = tuple(json_format.MessageToJson(x) for x in response)
+        return_values = tuple(mock.Mock() for i in response)
+        for return_val, response_val in zip(return_values, response):
+            return_val.read = mock.AsyncMock(return_value=response_val.encode("UTF-8"))
+            return_val.status_code = 200
+        req.side_effect = return_values
+
+        pager = await client.list_operations(name="operations")
+
+        responses = []
+        async for response in pager:
+            responses.append(response)
+
+        results = list(responses)
+        assert len(results) == 6
+        assert all(isinstance(i, operations_pb2.Operation) for i in results)
+        pager = await client.list_operations(name="operations")
+
+        responses = []
+        async for response in pager:
+            responses.append(response)
+
+        assert len(responses) == 6
+        assert all(isinstance(i, operations_pb2.Operation) for i in results)
+
+        pages = []
+
+        async for page in pager.pages:
+            pages.append(page)
+        for page_, token in zip(pages, ["", "", "", "abc", "def", "ghi", ""]):
+            assert page_.next_page_token == token
+
+
+def test_get_operation_rest():
+    client = _get_operations_client(is_async=False)
+
+    # Mock the http request call within the method and fake a response.
+    with mock.patch.object(_get_session_type(is_async=False), "request") as req:
         # Designate an appropriate value for the returned response.
         return_value = operations_pb2.Operation(
             name="operations/sample1",
@@ -580,10 +799,46 @@
     assert response.done is True
 
 
-def test_get_operation_rest_failure():
-    client = _get_operations_client(http_options=None)
+@pytest.mark.asyncio
+async def test_get_operation_rest_async():
+    if not GOOGLE_AUTH_AIO_INSTALLED:
+        pytest.skip("Skipped because google-api-core[async_rest] is not installed")
+    client = _get_operations_client(is_async=True)
 
-    with mock.patch.object(Session, "request") as req:
+    # Mock the http request call within the method and fake a response.
+    with mock.patch.object(_get_session_type(is_async=True), "request") as req:
+        # Designate an appropriate value for the returned response.
+        return_value = operations_pb2.Operation(
+            name="operations/sample1",
+            done=True,
+            error=status_pb2.Status(code=411),
+        )
+
+        # Wrap the value into a proper Response obj
+        response_value = mock.Mock()
+        response_value.status_code = 200
+        json_return_value = json_format.MessageToJson(return_value)
+        response_value.read = mock.AsyncMock(return_value=json_return_value)
+        req.return_value = response_value
+        response = await client.get_operation("operations/sample1")
+
+    actual_args = req.call_args
+    assert actual_args.args[0] == "GET"
+    assert (
+        actual_args.args[1]
+        == "https://longrunning.googleapis.com/v3/operations/sample1"
+    )
+
+    # Establish that the response is the type that we expect.
+    assert isinstance(response, operations_pb2.Operation)
+    assert response.name == "operations/sample1"
+    assert response.done is True
+
+
+def test_get_operation_rest_failure():
+    client = _get_operations_client(is_async=False, http_options=None)
+
+    with mock.patch.object(_get_session_type(is_async=False), "request") as req:
         response_value = Response()
         response_value.status_code = 400
         mock_request = mock.MagicMock()
@@ -595,13 +850,30 @@
             client.get_operation("sample0/operations/sample1")
 
 
-def test_delete_operation_rest(
-    transport: str = "rest", request_type=operations_pb2.DeleteOperationRequest
-):
-    client = _get_operations_client()
+@pytest.mark.asyncio
+async def test_get_operation_rest_failure_async():
+    if not GOOGLE_AUTH_AIO_INSTALLED:
+        pytest.skip("Skipped because google-api-core[async_rest] is not installed")
+    client = _get_operations_client(is_async=True, http_options=None)
+
+    with mock.patch.object(_get_session_type(is_async=True), "request") as req:
+        response_value = mock.Mock()
+        response_value.status_code = 400
+        response_value.read = mock.AsyncMock(return_value=b"{}")
+        mock_request = mock.MagicMock()
+        mock_request.method = "GET"
+        mock_request.url = "https://longrunning.googleapis.com/v1/operations/sample1"
+        response_value.request = mock_request
+        req.return_value = response_value
+        with pytest.raises(core_exceptions.GoogleAPIError):
+            await client.get_operation("sample0/operations/sample1")
+
+
+def test_delete_operation_rest():
+    client = _get_operations_client(is_async=False)
 
     # Mock the http request call within the method and fake a response.
-    with mock.patch.object(Session, "request") as req:
+    with mock.patch.object(_get_session_type(is_async=False), "request") as req:
         # Wrap the value into a proper Response obj
         response_value = Response()
         response_value.status_code = 200
@@ -618,10 +890,36 @@
         )
 
 
-def test_delete_operation_rest_failure():
-    client = _get_operations_client(http_options=None)
+@pytest.mark.asyncio
+async def test_delete_operation_rest_async():
+    if not GOOGLE_AUTH_AIO_INSTALLED:
+        pytest.skip("Skipped because google-api-core[async_rest] is not installed")
+    client = _get_operations_client(is_async=True)
 
-    with mock.patch.object(Session, "request") as req:
+    # Mock the http request call within the method and fake a response.
+    with mock.patch.object(_get_session_type(is_async=True), "request") as req:
+        # Wrap the value into a proper Response obj
+        response_value = mock.Mock()
+        response_value.status_code = 200
+        json_return_value = ""
+        response_value.read = mock.AsyncMock(
+            return_value=json_return_value.encode("UTF-8")
+        )
+        req.return_value = response_value
+        await client.delete_operation(name="operations/sample1")
+        assert req.call_count == 1
+        actual_args = req.call_args
+        assert actual_args.args[0] == "DELETE"
+        assert (
+            actual_args.args[1]
+            == "https://longrunning.googleapis.com/v3/operations/sample1"
+        )
+
+
+def test_delete_operation_rest_failure():
+    client = _get_operations_client(is_async=False, http_options=None)
+
+    with mock.patch.object(_get_session_type(is_async=False), "request") as req:
         response_value = Response()
         response_value.status_code = 400
         mock_request = mock.MagicMock()
@@ -633,11 +931,30 @@
             client.delete_operation(name="sample0/operations/sample1")
 
 
-def test_cancel_operation_rest(transport: str = "rest"):
-    client = _get_operations_client()
+@pytest.mark.asyncio
+async def test_delete_operation_rest_failure_async():
+    if not GOOGLE_AUTH_AIO_INSTALLED:
+        pytest.skip("Skipped because google-api-core[async_rest] is not installed")
+    client = _get_operations_client(is_async=True, http_options=None)
+
+    with mock.patch.object(_get_session_type(is_async=True), "request") as req:
+        response_value = mock.Mock()
+        response_value.status_code = 400
+        response_value.read = mock.AsyncMock(return_value=b"{}")
+        mock_request = mock.MagicMock()
+        mock_request.method = "DELETE"
+        mock_request.url = "https://longrunning.googleapis.com/v1/operations/sample1"
+        response_value.request = mock_request
+        req.return_value = response_value
+        with pytest.raises(core_exceptions.GoogleAPIError):
+            await client.delete_operation(name="sample0/operations/sample1")
+
+
+def test_cancel_operation_rest():
+    client = _get_operations_client(is_async=False)
 
     # Mock the http request call within the method and fake a response.
-    with mock.patch.object(Session, "request") as req:
+    with mock.patch.object(_get_session_type(is_async=False), "request") as req:
         # Wrap the value into a proper Response obj
         response_value = Response()
         response_value.status_code = 200
@@ -654,10 +971,36 @@
         )
 
 
-def test_cancel_operation_rest_failure():
-    client = _get_operations_client(http_options=None)
+@pytest.mark.asyncio
+async def test_cancel_operation_rest_async():
+    if not GOOGLE_AUTH_AIO_INSTALLED:
+        pytest.skip("Skipped because google-api-core[async_rest] is not installed")
+    client = _get_operations_client(is_async=True)
 
-    with mock.patch.object(Session, "request") as req:
+    # Mock the http request call within the method and fake a response.
+    with mock.patch.object(_get_session_type(is_async=True), "request") as req:
+        # Wrap the value into a proper Response obj
+        response_value = mock.Mock()
+        response_value.status_code = 200
+        json_return_value = ""
+        response_value.read = mock.AsyncMock(
+            return_value=json_return_value.encode("UTF-8")
+        )
+        req.return_value = response_value
+        await client.cancel_operation(name="operations/sample1")
+        assert req.call_count == 1
+        actual_args = req.call_args
+        assert actual_args.args[0] == "POST"
+        assert (
+            actual_args.args[1]
+            == "https://longrunning.googleapis.com/v3/operations/sample1:cancel"
+        )
+
+
+def test_cancel_operation_rest_failure():
+    client = _get_operations_client(is_async=False, http_options=None)
+
+    with mock.patch.object(_get_session_type(is_async=False), "request") as req:
         response_value = Response()
         response_value.status_code = 400
         mock_request = mock.MagicMock()
@@ -671,52 +1014,79 @@
             client.cancel_operation(name="sample0/operations/sample1")
 
 
-def test_credentials_transport_error():
+@pytest.mark.asyncio
+async def test_cancel_operation_rest_failure_async():
+    if not GOOGLE_AUTH_AIO_INSTALLED:
+        pytest.skip("Skipped because google-api-core[async_rest] is not installed")
+    client = _get_operations_client(is_async=True, http_options=None)
+
+    with mock.patch.object(_get_session_type(is_async=True), "request") as req:
+        response_value = mock.Mock()
+        response_value.status_code = 400
+        response_value.read = mock.AsyncMock(return_value=b"{}")
+        mock_request = mock.MagicMock()
+        mock_request.method = "POST"
+        mock_request.url = (
+            "https://longrunning.googleapis.com/v1/operations/sample1:cancel"
+        )
+        response_value.request = mock_request
+        req.return_value = response_value
+        with pytest.raises(core_exceptions.GoogleAPIError):
+            await client.cancel_operation(name="sample0/operations/sample1")
+
+
+@pytest.mark.parametrize(
+    "client_class,transport_class,credentials",
+    PYPARAM_CLIENT_TRANSPORT_CREDENTIALS,
+)
+def test_credentials_transport_error(client_class, transport_class, credentials):
+
     # It is an error to provide credentials and a transport instance.
-    transport = transports.OperationsRestTransport(
-        credentials=ga_credentials.AnonymousCredentials(),
-    )
+    transport = transport_class(credentials=credentials)
     with pytest.raises(ValueError):
-        AbstractOperationsClient(
+        client_class(
             credentials=ga_credentials.AnonymousCredentials(),
             transport=transport,
         )
 
     # It is an error to provide a credentials file and a transport instance.
-    transport = transports.OperationsRestTransport(
-        credentials=ga_credentials.AnonymousCredentials(),
-    )
+    transport = transport_class(credentials=credentials)
     with pytest.raises(ValueError):
-        AbstractOperationsClient(
+        client_class(
             client_options={"credentials_file": "credentials.json"},
             transport=transport,
         )
 
     # It is an error to provide scopes and a transport instance.
-    transport = transports.OperationsRestTransport(
-        credentials=ga_credentials.AnonymousCredentials(),
-    )
+    transport = transport_class(credentials=credentials)
     with pytest.raises(ValueError):
-        AbstractOperationsClient(
+        client_class(
             client_options={"scopes": ["1", "2"]},
             transport=transport,
         )
 
 
-def test_transport_instance():
+@pytest.mark.parametrize(
+    "client_class,transport_class,credentials",
+    PYPARAM_CLIENT_TRANSPORT_CREDENTIALS,
+)
+def test_transport_instance(client_class, transport_class, credentials):
     # A client may be instantiated with a custom transport instance.
-    transport = transports.OperationsRestTransport(
-        credentials=ga_credentials.AnonymousCredentials(),
+    transport = transport_class(
+        credentials=credentials,
     )
-    client = AbstractOperationsClient(transport=transport)
+    client = client_class(transport=transport)
     assert client.transport is transport
 
 
-@pytest.mark.parametrize("transport_class", [transports.OperationsRestTransport])
-def test_transport_adc(transport_class):
+@pytest.mark.parametrize(
+    "client_class,transport_class,credentials",
+    PYPARAM_CLIENT_TRANSPORT_CREDENTIALS,
+)
+def test_transport_adc(client_class, transport_class, credentials):
     # Test default credentials are used if not provided.
     with mock.patch.object(google.auth, "default") as adc:
-        adc.return_value = (ga_credentials.AnonymousCredentials(), None)
+        adc.return_value = (credentials, None)
         transport_class()
         adc.assert_called_once()
 
@@ -788,32 +1158,59 @@
         adc.assert_called_once()
 
 
-def test_operations_auth_adc():
+@pytest.mark.parametrize(
+    "client_class",
+    PYPARAM_CLIENT,
+)
+def test_operations_auth_adc(client_class):
     # If no credentials are provided, we should use ADC credentials.
     with mock.patch.object(google.auth, "default", autospec=True) as adc:
         adc.return_value = (ga_credentials.AnonymousCredentials(), None)
-        AbstractOperationsClient()
-        adc.assert_called_once_with(
-            scopes=None,
-            default_scopes=(),
-            quota_project_id=None,
-        )
+
+        if "async" in str(client_class).lower():
+            # TODO(): Add support for adc to async REST transport.
+            # NOTE: Ideally, the logic for adc shouldn't be called if transport
+            # is set to async REST. If the user does not configure credentials
+            # of type `google.auth.aio.credentials.Credentials`,
+            # we should raise an exception to avoid the adc workflow.
+            with pytest.raises(google.auth.exceptions.InvalidType):
+                client_class()
+        else:
+            client_class()
+            adc.assert_called_once_with(
+                scopes=None,
+                default_scopes=(),
+                quota_project_id=None,
+            )
 
 
-def test_operations_http_transport_client_cert_source_for_mtls():
+# TODO(https://github.com/googleapis/python-api-core/issues/705): Add
+# testing for `transports.AsyncOperationsRestTransport` once MTLS is supported
+# in `google.auth.aio.transport`.
+@pytest.mark.parametrize(
+    "transport_class",
+    [
+        transports.OperationsRestTransport,
+    ],
+)
+def test_operations_http_transport_client_cert_source_for_mtls(transport_class):
     cred = ga_credentials.AnonymousCredentials()
     with mock.patch(
         "google.auth.transport.requests.AuthorizedSession.configure_mtls_channel"
     ) as mock_configure_mtls_channel:
-        transports.OperationsRestTransport(
+        transport_class(
             credentials=cred, client_cert_source_for_mtls=client_cert_source_callback
         )
         mock_configure_mtls_channel.assert_called_once_with(client_cert_source_callback)
 
 
-def test_operations_host_no_port():
-    client = AbstractOperationsClient(
-        credentials=ga_credentials.AnonymousCredentials(),
+@pytest.mark.parametrize(
+    "client_class,transport_class,credentials",
+    PYPARAM_CLIENT_TRANSPORT_CREDENTIALS,
+)
+def test_operations_host_no_port(client_class, transport_class, credentials):
+    client = client_class(
+        credentials=credentials,
         client_options=client_options.ClientOptions(
             api_endpoint="longrunning.googleapis.com"
         ),
@@ -821,9 +1218,13 @@
     assert client.transport._host == "https://longrunning.googleapis.com"
 
 
-def test_operations_host_with_port():
-    client = AbstractOperationsClient(
-        credentials=ga_credentials.AnonymousCredentials(),
+@pytest.mark.parametrize(
+    "client_class,transport_class,credentials",
+    PYPARAM_CLIENT_TRANSPORT_CREDENTIALS,
+)
+def test_operations_host_with_port(client_class, transport_class, credentials):
+    client = client_class(
+        credentials=credentials,
         client_options=client_options.ClientOptions(
             api_endpoint="longrunning.googleapis.com:8000"
         ),
@@ -831,127 +1232,165 @@
     assert client.transport._host == "https://longrunning.googleapis.com:8000"
 
 
-def test_common_billing_account_path():
+@pytest.mark.parametrize(
+    "client_class",
+    PYPARAM_CLIENT,
+)
+def test_common_billing_account_path(client_class):
     billing_account = "squid"
     expected = "billingAccounts/{billing_account}".format(
         billing_account=billing_account,
     )
-    actual = AbstractOperationsClient.common_billing_account_path(billing_account)
+    actual = client_class.common_billing_account_path(billing_account)
     assert expected == actual
 
 
-def test_parse_common_billing_account_path():
+@pytest.mark.parametrize(
+    "client_class",
+    PYPARAM_CLIENT,
+)
+def test_parse_common_billing_account_path(client_class):
     expected = {
         "billing_account": "clam",
     }
-    path = AbstractOperationsClient.common_billing_account_path(**expected)
+    path = client_class.common_billing_account_path(**expected)
 
     # Check that the path construction is reversible.
-    actual = AbstractOperationsClient.parse_common_billing_account_path(path)
+    actual = client_class.parse_common_billing_account_path(path)
     assert expected == actual
 
 
-def test_common_folder_path():
+@pytest.mark.parametrize(
+    "client_class",
+    PYPARAM_CLIENT,
+)
+def test_common_folder_path(client_class):
     folder = "whelk"
     expected = "folders/{folder}".format(
         folder=folder,
     )
-    actual = AbstractOperationsClient.common_folder_path(folder)
+    actual = client_class.common_folder_path(folder)
     assert expected == actual
 
 
-def test_parse_common_folder_path():
+@pytest.mark.parametrize(
+    "client_class",
+    PYPARAM_CLIENT,
+)
+def test_parse_common_folder_path(client_class):
     expected = {
         "folder": "octopus",
     }
-    path = AbstractOperationsClient.common_folder_path(**expected)
+    path = client_class.common_folder_path(**expected)
 
     # Check that the path construction is reversible.
-    actual = AbstractOperationsClient.parse_common_folder_path(path)
+    actual = client_class.parse_common_folder_path(path)
     assert expected == actual
 
 
-def test_common_organization_path():
+@pytest.mark.parametrize(
+    "client_class",
+    PYPARAM_CLIENT,
+)
+def test_common_organization_path(client_class):
     organization = "oyster"
     expected = "organizations/{organization}".format(
         organization=organization,
     )
-    actual = AbstractOperationsClient.common_organization_path(organization)
+    actual = client_class.common_organization_path(organization)
     assert expected == actual
 
 
-def test_parse_common_organization_path():
+@pytest.mark.parametrize(
+    "client_class",
+    PYPARAM_CLIENT,
+)
+def test_parse_common_organization_path(client_class):
     expected = {
         "organization": "nudibranch",
     }
-    path = AbstractOperationsClient.common_organization_path(**expected)
+    path = client_class.common_organization_path(**expected)
 
     # Check that the path construction is reversible.
-    actual = AbstractOperationsClient.parse_common_organization_path(path)
+    actual = client_class.parse_common_organization_path(path)
     assert expected == actual
 
 
-def test_common_project_path():
+@pytest.mark.parametrize(
+    "client_class",
+    PYPARAM_CLIENT,
+)
+def test_common_project_path(client_class):
     project = "cuttlefish"
     expected = "projects/{project}".format(
         project=project,
     )
-    actual = AbstractOperationsClient.common_project_path(project)
+    actual = client_class.common_project_path(project)
     assert expected == actual
 
 
-def test_parse_common_project_path():
+@pytest.mark.parametrize(
+    "client_class",
+    PYPARAM_CLIENT,
+)
+def test_parse_common_project_path(client_class):
     expected = {
         "project": "mussel",
     }
-    path = AbstractOperationsClient.common_project_path(**expected)
+    path = client_class.common_project_path(**expected)
 
     # Check that the path construction is reversible.
-    actual = AbstractOperationsClient.parse_common_project_path(path)
+    actual = client_class.parse_common_project_path(path)
     assert expected == actual
 
 
-def test_common_location_path():
+@pytest.mark.parametrize(
+    "client_class",
+    PYPARAM_CLIENT,
+)
+def test_common_location_path(client_class):
     project = "winkle"
     location = "nautilus"
     expected = "projects/{project}/locations/{location}".format(
         project=project,
         location=location,
     )
-    actual = AbstractOperationsClient.common_location_path(project, location)
+    actual = client_class.common_location_path(project, location)
     assert expected == actual
 
 
-def test_parse_common_location_path():
+@pytest.mark.parametrize(
+    "client_class",
+    PYPARAM_CLIENT,
+)
+def test_parse_common_location_path(client_class):
     expected = {
         "project": "scallop",
         "location": "abalone",
     }
-    path = AbstractOperationsClient.common_location_path(**expected)
+    path = client_class.common_location_path(**expected)
 
     # Check that the path construction is reversible.
-    actual = AbstractOperationsClient.parse_common_location_path(path)
+    actual = client_class.parse_common_location_path(path)
     assert expected == actual
 
 
-def test_client_withDEFAULT_CLIENT_INFO():
+@pytest.mark.parametrize(
+    "client_class,transport_class,credentials",
+    PYPARAM_CLIENT_TRANSPORT_CREDENTIALS,
+)
+def test_client_withDEFAULT_CLIENT_INFO(client_class, transport_class, credentials):
     client_info = gapic_v1.client_info.ClientInfo()
-
-    with mock.patch.object(
-        transports.OperationsTransport, "_prep_wrapped_messages"
-    ) as prep:
-        AbstractOperationsClient(
-            credentials=ga_credentials.AnonymousCredentials(),
+    with mock.patch.object(transport_class, "_prep_wrapped_messages") as prep:
+        client_class(
+            credentials=credentials,
             client_info=client_info,
         )
         prep.assert_called_once_with(client_info)
 
-    with mock.patch.object(
-        transports.OperationsTransport, "_prep_wrapped_messages"
-    ) as prep:
-        transport_class = AbstractOperationsClient.get_transport_class()
+    with mock.patch.object(transport_class, "_prep_wrapped_messages") as prep:
         transport_class(
-            credentials=ga_credentials.AnonymousCredentials(),
+            credentials=credentials,
             client_info=client_info,
         )
         prep.assert_called_once_with(client_info)