diff --git a/google/cloud/pubsub_v1/publisher/client.py b/google/cloud/pubsub_v1/publisher/client.py
index efdb1faad..43305afcc 100644
--- a/google/cloud/pubsub_v1/publisher/client.py
+++ b/google/cloud/pubsub_v1/publisher/client.py
@@ -22,6 +22,7 @@
 import time
 import typing
 from typing import Any, Dict, Optional, Sequence, Tuple, Type, Union
+import warnings
 
 from google.api_core import gapic_v1
 from google.auth.credentials import AnonymousCredentials  # type: ignore
@@ -193,6 +194,26 @@ def target(self) -> str:
         """
         return self._target
 
+    @property
+    def api(self):
+        """The underlying gapic API client.
+
+        .. versionchanged:: 2.10.0
+            Instead of a GAPIC ``PublisherClient`` client instance, this property is a
+            proxy object to it with the same interface.
+
+        .. deprecated:: 2.10.0
+            Use the GAPIC methods and properties on the client instance directly
+            instead of through the :attr:`api` attribute.
+        """
+        msg = (
+            'The "api" property only exists for backward compatibility, access its '
+            'attributes directly thorugh the client instance (e.g. "client.foo" '
+            'instead of "client.api.foo").'
+        )
+        warnings.warn(msg, category=DeprecationWarning)
+        return super()
+
     def _get_or_create_sequencer(self, topic: str, ordering_key: str) -> SequencerType:
         """ Get an existing sequencer or create a new one given the (topic,
             ordering_key) pair.
diff --git a/google/cloud/pubsub_v1/subscriber/client.py b/google/cloud/pubsub_v1/subscriber/client.py
index 637e56a33..9c12a0bfb 100644
--- a/google/cloud/pubsub_v1/subscriber/client.py
+++ b/google/cloud/pubsub_v1/subscriber/client.py
@@ -18,6 +18,7 @@
 import pkg_resources
 import typing
 from typing import cast, Any, Callable, Optional, Sequence, Union
+import warnings
 
 from google.auth.credentials import AnonymousCredentials  # type: ignore
 from google.oauth2 import service_account  # type: ignore
@@ -125,6 +126,26 @@ def closed(self) -> bool:
         """
         return self._closed
 
+    @property
+    def api(self):
+        """The underlying gapic API client.
+
+        .. versionchanged:: 2.10.0
+            Instead of a GAPIC ``SubscriberClient`` client instance, this property is a
+            proxy object to it with the same interface.
+
+        .. deprecated:: 2.10.0
+            Use the GAPIC methods and properties on the client instance directly
+            instead of through the :attr:`api` attribute.
+        """
+        msg = (
+            'The "api" property only exists for backward compatibility, access its '
+            'attributes directly thorugh the client instance (e.g. "client.foo" '
+            'instead of "client.api.foo").'
+        )
+        warnings.warn(msg, category=DeprecationWarning)
+        return super()
+
     def subscribe(
         self,
         subscription: str,
diff --git a/tests/unit/pubsub_v1/publisher/test_publisher_client.py b/tests/unit/pubsub_v1/publisher/test_publisher_client.py
index 4c27665cd..20d5b328c 100644
--- a/tests/unit/pubsub_v1/publisher/test_publisher_client.py
+++ b/tests/unit/pubsub_v1/publisher/test_publisher_client.py
@@ -22,6 +22,7 @@
 import mock
 import pytest
 import time
+import warnings
 
 from google.api_core import gapic_v1
 from google.api_core import retry as retries
@@ -50,6 +51,38 @@ def _assert_retries_equal(retry, retry2):
     assert inspect.getclosurevars(pred) == inspect.getclosurevars(pred2)
 
 
+def test_api_property_deprecated(creds):
+    client = publisher.Client(credentials=creds)
+
+    with warnings.catch_warnings(record=True) as warned:
+        client.api
+
+    assert len(warned) == 1
+    assert issubclass(warned[0].category, DeprecationWarning)
+    warning_msg = str(warned[0].message)
+    assert "client.api" in warning_msg
+
+
+def test_api_property_proxy_to_generated_client(creds):
+    client = publisher.Client(credentials=creds)
+
+    with warnings.catch_warnings(record=True):
+        api_object = client.api
+
+    # Not a perfect check, but we are satisficed if the returned API object indeed
+    # contains all methods of the generated class.
+    superclass_attrs = (attr for attr in dir(type(client).__mro__[1]))
+    assert all(
+        hasattr(api_object, attr)
+        for attr in superclass_attrs
+        if callable(getattr(client, attr))
+    )
+
+    # The resume_publish() method only exists on the hand-written wrapper class.
+    assert hasattr(client, "resume_publish")
+    assert not hasattr(api_object, "resume_publish")
+
+
 def test_init(creds):
     client = publisher.Client(credentials=creds)
 
diff --git a/tests/unit/pubsub_v1/subscriber/test_subscriber_client.py b/tests/unit/pubsub_v1/subscriber/test_subscriber_client.py
index dbf2f0033..1f60b536d 100644
--- a/tests/unit/pubsub_v1/subscriber/test_subscriber_client.py
+++ b/tests/unit/pubsub_v1/subscriber/test_subscriber_client.py
@@ -208,6 +208,38 @@ def test_context_manager_raises_if_closed(creds):
             pass
 
 
+def test_api_property_deprecated(creds):
+    client = subscriber.Client(credentials=creds)
+
+    with warnings.catch_warnings(record=True) as warned:
+        client.api
+
+    assert len(warned) == 1
+    assert issubclass(warned[0].category, DeprecationWarning)
+    warning_msg = str(warned[0].message)
+    assert "client.api" in warning_msg
+
+
+def test_api_property_proxy_to_generated_client(creds):
+    client = subscriber.Client(credentials=creds)
+
+    with warnings.catch_warnings(record=True):
+        api_object = client.api
+
+    # Not a perfect check, but we are satisficed if the returned API object indeed
+    # contains all methods of the generated class.
+    superclass_attrs = (attr for attr in dir(type(client).__mro__[1]))
+    assert all(
+        hasattr(api_object, attr)
+        for attr in superclass_attrs
+        if callable(getattr(client, attr))
+    )
+
+    # The close() method only exists on the hand-written wrapper class.
+    assert hasattr(client, "close")
+    assert not hasattr(api_object, "close")
+
+
 def test_streaming_pull_gapic_monkeypatch(creds):
     client = subscriber.Client(credentials=creds)