diff --git a/google/api_core/exceptions.py b/google/api_core/exceptions.py
index 74f46ef3..8dc8c08f 100644
--- a/google/api_core/exceptions.py
+++ b/google/api_core/exceptions.py
@@ -22,7 +22,7 @@
 from __future__ import unicode_literals
 
 import http.client
-from typing import Dict
+from typing import Optional, Dict
 from typing import Union
 import warnings
 
@@ -476,22 +476,37 @@ def from_http_status(status_code, message, **kwargs):
     return error
 
 
-def from_http_response(response):
-    """Create a :class:`GoogleAPICallError` from a :class:`requests.Response`.
+def _format_rest_error_message(error, method, url):
+    method = method.upper() if method else None
+    message = "{method} {url}: {error}".format(
+        method=method,
+        url=url,
+        error=error,
+    )
+    return message
+
+
+# NOTE: We're moving away from `from_http_status` because it expects an aiohttp response compared
+# to `format_http_response_error` which expects a more abstract response from google.auth and is
+# compatible with both sync and async response types.
+# TODO(https://github.com/googleapis/python-api-core/issues/691): Add type hint for response.
+def format_http_response_error(
+    response, method: str, url: str, payload: Optional[Dict] = None
+):
+    """Create a :class:`GoogleAPICallError` from a google auth rest response.
 
     Args:
-        response (requests.Response): The HTTP response.
+        response Union[google.auth.transport.Response, google.auth.aio.transport.Response]: The HTTP response.
+        method Optional(str): The HTTP request method.
+        url Optional(str): The HTTP request url.
+        payload Optional(dict): The HTTP response payload. If not passed in, it is read from response for a response type of google.auth.transport.Response.
 
     Returns:
         GoogleAPICallError: An instance of the appropriate subclass of
             :class:`GoogleAPICallError`, with the message and errors populated
             from the response.
     """
-    try:
-        payload = response.json()
-    except ValueError:
-        payload = {"error": {"message": response.text or "unknown error"}}
-
+    payload = {} if not payload else payload
     error_message = payload.get("error", {}).get("message", "unknown error")
     errors = payload.get("error", {}).get("errors", ())
     # In JSON, details are already formatted in developer-friendly way.
@@ -504,12 +519,7 @@ def from_http_response(response):
         )
     )
     error_info = error_info[0] if error_info else None
-
-    message = "{method} {url}: {error}".format(
-        method=response.request.method,
-        url=response.request.url,
-        error=error_message,
-    )
+    message = _format_rest_error_message(error_message, method, url)
 
     exception = from_http_status(
         response.status_code,
@@ -522,6 +532,26 @@ def from_http_response(response):
     return exception
 
 
+def from_http_response(response):
+    """Create a :class:`GoogleAPICallError` from a :class:`requests.Response`.
+
+    Args:
+        response (requests.Response): The HTTP response.
+
+    Returns:
+        GoogleAPICallError: An instance of the appropriate subclass of
+            :class:`GoogleAPICallError`, with the message and errors populated
+            from the response.
+    """
+    try:
+        payload = response.json()
+    except ValueError:
+        payload = {"error": {"message": response.text or "unknown error"}}
+    return format_http_response_error(
+        response, response.request.method, response.request.url, payload
+    )
+
+
 def exception_class_for_grpc_status(status_code):
     """Return the exception class for a specific :class:`grpc.StatusCode`.
 
diff --git a/google/api_core/gapic_v1/method_async.py b/google/api_core/gapic_v1/method_async.py
index 24880756..c0f38c0e 100644
--- a/google/api_core/gapic_v1/method_async.py
+++ b/google/api_core/gapic_v1/method_async.py
@@ -25,6 +25,8 @@
 from google.api_core.gapic_v1.method import DEFAULT  # noqa: F401
 from google.api_core.gapic_v1.method import USE_DEFAULT_METADATA  # noqa: F401
 
+_DEFAULT_ASYNC_TRANSPORT_KIND = "grpc_asyncio"
+
 
 def wrap_method(
     func,
@@ -32,6 +34,7 @@ def wrap_method(
     default_timeout=None,
     default_compression=None,
     client_info=client_info.DEFAULT_CLIENT_INFO,
+    kind=_DEFAULT_ASYNC_TRANSPORT_KIND,
 ):
     """Wrap an async RPC method with common behavior.
 
@@ -40,7 +43,8 @@ def wrap_method(
             and ``compression`` arguments and applies the common error mapping,
             retry, timeout, metadata, and compression behavior to the low-level RPC method.
     """
-    func = grpc_helpers_async.wrap_errors(func)
+    if kind == _DEFAULT_ASYNC_TRANSPORT_KIND:
+        func = grpc_helpers_async.wrap_errors(func)
 
     metadata = [client_info.to_grpc_metadata()] if client_info is not None else None
 
diff --git a/tests/asyncio/gapic/test_method_async.py b/tests/asyncio/gapic/test_method_async.py
index ee206979..f64157b4 100644
--- a/tests/asyncio/gapic/test_method_async.py
+++ b/tests/asyncio/gapic/test_method_async.py
@@ -252,3 +252,14 @@ async def test_wrap_method_with_overriding_timeout_as_a_number():
 
     assert result == 42
     method.assert_called_once_with(timeout=22, metadata=mock.ANY)
+
+
+@pytest.mark.asyncio
+async def test_wrap_method_without_wrap_errors():
+    fake_call = mock.AsyncMock()
+
+    wrapped_method = gapic_v1.method_async.wrap_method(fake_call, kind="rest")
+    with mock.patch("google.api_core.grpc_helpers_async.wrap_errors") as method:
+        await wrapped_method()
+
+        method.assert_not_called()