From 2e6861f9990b1b50ab9ff7c20621474e3aae5fa1 Mon Sep 17 00:00:00 2001
From: chalmer lowe <chalmerlowe@google.com>
Date: Thu, 27 Aug 2026 05:59:55 -0400
Subject: [PATCH 01/16] feat(api-core): add ClientInterceptor and
 apply_interceptors helper

---
 .../google/api_core/grpc_helpers.py           | 35 ++++++-
 .../tests/unit/test_grpc_helpers.py           | 94 ++++++++++++++++++-
 2 files changed, 125 insertions(+), 4 deletions(-)

diff --git a/packages/google-api-core/google/api_core/grpc_helpers.py b/packages/google-api-core/google/api_core/grpc_helpers.py
index 263079e7d1f7..ab944b240198 100644
--- a/packages/google-api-core/google/api_core/grpc_helpers.py
+++ b/packages/google-api-core/google/api_core/grpc_helpers.py
@@ -17,7 +17,7 @@
 import collections
 import functools
 import warnings
-from typing import Generic, Iterator, Optional, TypeVar
+from typing import Generic, Iterator, Optional, Sequence, TypeVar, Union
 
 import google.auth
 import google.auth.credentials
@@ -25,7 +25,6 @@
 import google.auth.transport.requests
 import google.protobuf
 import grpc
-
 from google.api_core import exceptions, general_helpers
 
 # The list of gRPC Callable interfaces that return iterators.
@@ -34,6 +33,14 @@
 # denotes the proto response type for grpc calls
 P = TypeVar("P")
 
+# Type alias representing any client-side gRPC interceptor
+ClientInterceptor = Union[
+    grpc.UnaryUnaryClientInterceptor,
+    grpc.UnaryStreamClientInterceptor,
+    grpc.StreamUnaryClientInterceptor,
+    grpc.StreamStreamClientInterceptor,
+]
+
 
 def _patch_callable_name(callable_):
     """Fix-up gRPC callable attributes.
@@ -419,6 +426,30 @@ def _modify_target_for_direct_path(target: str) -> str:
     return target
 
 
+def apply_interceptors(
+    channel: grpc.Channel,
+    interceptors: Optional[Sequence[ClientInterceptor]] = None,
+) -> grpc.Channel:
+    """Applies a sequence of interceptors to a gRPC channel.
+
+    The interceptors are applied in the order provided, wrapping the channel
+    sequentially.
+
+    Args:
+        channel (grpc.Channel): The channel to intercept.
+        interceptors (Optional[Sequence[ClientInterceptor]]): An optional sequence
+            of client interceptors to apply.
+
+    Returns:
+        grpc.Channel: The intercepted channel, or the original channel if no
+            interceptors were provided.
+    """
+    if interceptors:
+        for interceptor in interceptors:
+            channel = grpc.intercept_channel(channel, interceptor)
+    return channel
+
+
 _MethodCall = collections.namedtuple(
     "_MethodCall", ("request", "timeout", "metadata", "credentials", "compression")
 )
diff --git a/packages/google-api-core/tests/unit/test_grpc_helpers.py b/packages/google-api-core/tests/unit/test_grpc_helpers.py
index 69281d58109b..677fc15ce2d3 100644
--- a/packages/google-api-core/tests/unit/test_grpc_helpers.py
+++ b/packages/google-api-core/tests/unit/test_grpc_helpers.py
@@ -24,9 +24,8 @@
     pytest.skip("No GRPC", allow_module_level=True)
 
 import google.auth.credentials
-from google.longrunning import operations_pb2
-
 from google.api_core import exceptions, grpc_helpers
+from google.longrunning import operations_pb2
 
 
 def test__patch_callable_name():
@@ -932,3 +931,94 @@ def test_subscribe_unsubscribe(self):
     def test_close(self):
         channel = grpc_helpers.ChannelStub()
         assert channel.close() is None
+
+
+@pytest.mark.parametrize("falsy_interceptors", [None, [], ()])
+def test_apply_interceptors_passthrough(falsy_interceptors):
+    """Verify that falsy or empty interceptor sequences return the channel unmodified."""
+    mock_channel = mock.Mock()
+    result = grpc_helpers.apply_interceptors(mock_channel, falsy_interceptors)
+    assert result is mock_channel
+
+
+@pytest.mark.parametrize("count", [1, 2, 3])
+def test_apply_interceptors_wrapping(count):
+    """Verify that interceptors are wrapped sequentially in the order provided.
+
+    When given a sequence of N interceptors [i_0, i_1, ..., i_{N-1}], apply_interceptors
+    must pass the base channel and i_0 to grpc.intercept_channel, then pass the
+    resulting wrapped channel and i_1 to grpc.intercept_channel, and so on.
+    This ensures each subsequent interceptor wraps the preceding channel state.
+    """
+    mock_channel = mock.Mock(name="base_channel")
+    # Generate distinct mock interceptors and the expected wrapped channel returns for each step
+    interceptors = [mock.Mock(name=f"interceptor_{i}") for i in range(count)]
+    wrapped_channels = [mock.Mock(name=f"wrapped_channel_{i}") for i in range(count)]
+
+    with mock.patch(
+        "grpc.intercept_channel", side_effect=wrapped_channels
+    ) as mock_intercept:
+        result = grpc_helpers.apply_interceptors(mock_channel, interceptors)
+
+    # The final return value must be the outermost wrapped channel from the final loop iteration
+    assert result is wrapped_channels[-1]
+    assert mock_intercept.call_count == count
+
+    # Construct the expected sequential chaining: (base, i_0) -> (wrapped_0, i_1) -> ...
+    expected_calls = []
+    current_channel = mock_channel
+    for i, interceptor in enumerate(interceptors):
+        expected_calls.append(mock.call(current_channel, interceptor))
+        current_channel = wrapped_channels[i]
+
+    mock_intercept.assert_has_calls(expected_calls)
+
+
+def test_apply_interceptors_execution_order():
+    """Verify runtime execution order (onion model) when invoking an RPC on an intercepted channel.
+
+    In gRPC Python, sequential wrapping via grpc.intercept_channel(channel, i) creates an
+    'onion' layer where the LAST applied interceptor becomes the OUTSIDE layer.
+    Therefore, given [i1, i2]:
+      - i1 wraps the raw channel (innermost layer)
+      - i2 wraps the result of (channel + i1) (outermost layer)
+
+    During an RPC invocation:
+      1. i2 intercepts the call first (request inbound / pre-call)
+      2. i2 calls continuation(), which triggers i1
+      3. i1 calls continuation(), which reaches the channel stub / network
+      4. i1 post-call logic finishes
+      5. i2 post-call logic finishes
+    """
+    execution_order = []
+
+    class OrderInterceptor(grpc.UnaryUnaryClientInterceptor):
+        def __init__(self, name):
+            self.name = name
+
+        def intercept_unary_unary(self, continuation, client_call_details, request):
+            execution_order.append(f"{self.name}_start")
+            response = continuation(client_call_details, request)
+            execution_order.append(f"{self.name}_end")
+            return response
+
+    i1 = OrderInterceptor("i1")
+    i2 = OrderInterceptor("i2")
+
+    mock_channel = mock.Mock(spec=grpc.Channel)
+    mock_callable = mock.Mock(spec=grpc.UnaryUnaryMultiCallable)
+    mock_call = mock.Mock(spec=grpc.Call)
+    expected_response = operations_pb2.Operation(name="test_op")
+    mock_callable.with_call.return_value = (expected_response, mock_call)
+    mock_channel.unary_unary.return_value = mock_callable
+
+    # Apply interceptors in sequence [i1, i2]
+    intercepted_channel = grpc_helpers.apply_interceptors(mock_channel, [i1, i2])
+
+    # Trigger a unary RPC through the intercepted channel
+    stub = operations_pb2.OperationsStub(intercepted_channel)
+    response = stub.GetOperation(operations_pb2.GetOperationRequest(name="test_op"))
+
+    assert response.name == "test_op"
+    # Verify i2 executed as the outer layer surrounding i1
+    assert execution_order == ["i2_start", "i1_start", "i1_end", "i2_end"]

From 84e4cd55babd5de780aaf61b978963ab9ac1d4e3 Mon Sep 17 00:00:00 2001
From: chalmer lowe <chalmerlowe@google.com>
Date: Thu, 27 Aug 2026 06:30:02 -0400
Subject: [PATCH 02/16] fix(api-core): unpack interceptors directly into
 grpc.intercept_channel

---
 .../google/api_core/grpc_helpers.py           |  7 ++-
 .../tests/unit/test_grpc_helpers.py           | 49 +++++++------------
 2 files changed, 22 insertions(+), 34 deletions(-)

diff --git a/packages/google-api-core/google/api_core/grpc_helpers.py b/packages/google-api-core/google/api_core/grpc_helpers.py
index ab944b240198..bb2a3523d735 100644
--- a/packages/google-api-core/google/api_core/grpc_helpers.py
+++ b/packages/google-api-core/google/api_core/grpc_helpers.py
@@ -432,8 +432,8 @@ def apply_interceptors(
 ) -> grpc.Channel:
     """Applies a sequence of interceptors to a gRPC channel.
 
-    The interceptors are applied in the order provided, wrapping the channel
-    sequentially.
+    The interceptors are applied in the order provided, such that the first
+    interceptor in the sequence is the outermost layer (executes first).
 
     Args:
         channel (grpc.Channel): The channel to intercept.
@@ -445,8 +445,7 @@ def apply_interceptors(
             interceptors were provided.
     """
     if interceptors:
-        for interceptor in interceptors:
-            channel = grpc.intercept_channel(channel, interceptor)
+        return grpc.intercept_channel(channel, *interceptors)
     return channel
 
 
diff --git a/packages/google-api-core/tests/unit/test_grpc_helpers.py b/packages/google-api-core/tests/unit/test_grpc_helpers.py
index 677fc15ce2d3..b9fc553e6555 100644
--- a/packages/google-api-core/tests/unit/test_grpc_helpers.py
+++ b/packages/google-api-core/tests/unit/test_grpc_helpers.py
@@ -943,52 +943,41 @@ def test_apply_interceptors_passthrough(falsy_interceptors):
 
 @pytest.mark.parametrize("count", [1, 2, 3])
 def test_apply_interceptors_wrapping(count):
-    """Verify that interceptors are wrapped sequentially in the order provided.
+    """Verify that interceptors are passed to grpc.intercept_channel in a single call.
 
     When given a sequence of N interceptors [i_0, i_1, ..., i_{N-1}], apply_interceptors
-    must pass the base channel and i_0 to grpc.intercept_channel, then pass the
-    resulting wrapped channel and i_1 to grpc.intercept_channel, and so on.
-    This ensures each subsequent interceptor wraps the preceding channel state.
+    must pass the base channel and all interceptors unpacked (*interceptors) to
+    grpc.intercept_channel. This creates a single intercepted channel wrapper rather than
+    multiple nested wrappers.
     """
     mock_channel = mock.Mock(name="base_channel")
-    # Generate distinct mock interceptors and the expected wrapped channel returns for each step
+    mock_intercepted = mock.Mock(name="intercepted_channel")
     interceptors = [mock.Mock(name=f"interceptor_{i}") for i in range(count)]
-    wrapped_channels = [mock.Mock(name=f"wrapped_channel_{i}") for i in range(count)]
 
     with mock.patch(
-        "grpc.intercept_channel", side_effect=wrapped_channels
+        "grpc.intercept_channel", return_value=mock_intercepted
     ) as mock_intercept:
         result = grpc_helpers.apply_interceptors(mock_channel, interceptors)
 
-    # The final return value must be the outermost wrapped channel from the final loop iteration
-    assert result is wrapped_channels[-1]
-    assert mock_intercept.call_count == count
-
-    # Construct the expected sequential chaining: (base, i_0) -> (wrapped_0, i_1) -> ...
-    expected_calls = []
-    current_channel = mock_channel
-    for i, interceptor in enumerate(interceptors):
-        expected_calls.append(mock.call(current_channel, interceptor))
-        current_channel = wrapped_channels[i]
-
-    mock_intercept.assert_has_calls(expected_calls)
+    assert result is mock_intercepted
+    mock_intercept.assert_called_once_with(mock_channel, *interceptors)
 
 
 def test_apply_interceptors_execution_order():
     """Verify runtime execution order (onion model) when invoking an RPC on an intercepted channel.
 
-    In gRPC Python, sequential wrapping via grpc.intercept_channel(channel, i) creates an
-    'onion' layer where the LAST applied interceptor becomes the OUTSIDE layer.
+    In standard gRPC Python, grpc.intercept_channel(channel, *interceptors) processes
+    the interceptor list such that the first interceptor in the sequence is the outermost layer.
     Therefore, given [i1, i2]:
-      - i1 wraps the raw channel (innermost layer)
-      - i2 wraps the result of (channel + i1) (outermost layer)
+      - i1 is the outermost layer (executes first on outbound request)
+      - i2 is the inner layer (executes second on outbound request)
 
     During an RPC invocation:
-      1. i2 intercepts the call first (request inbound / pre-call)
-      2. i2 calls continuation(), which triggers i1
-      3. i1 calls continuation(), which reaches the channel stub / network
-      4. i1 post-call logic finishes
-      5. i2 post-call logic finishes
+      1. i1 intercepts the call first (request inbound / pre-call)
+      2. i1 calls continuation(), which triggers i2
+      3. i2 calls continuation(), which reaches the channel stub / network
+      4. i2 post-call logic finishes
+      5. i1 post-call logic finishes
     """
     execution_order = []
 
@@ -1020,5 +1009,5 @@ def intercept_unary_unary(self, continuation, client_call_details, request):
     response = stub.GetOperation(operations_pb2.GetOperationRequest(name="test_op"))
 
     assert response.name == "test_op"
-    # Verify i2 executed as the outer layer surrounding i1
-    assert execution_order == ["i2_start", "i1_start", "i1_end", "i2_end"]
+    # Verify i1 executed as the outer layer surrounding i2
+    assert execution_order == ["i1_start", "i2_start", "i2_end", "i1_end"]

From 9ea6cc6339debf68c268ceaaddec711512a677f9 Mon Sep 17 00:00:00 2001
From: chalmer lowe <chalmerlowe@google.com>
Date: Thu, 27 Aug 2026 09:26:26 -0400
Subject: [PATCH 03/16] docs(api-core): clarify apply_interceptors execution
 order in docstring

---
 packages/google-api-core/google/api_core/grpc_helpers.py | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/packages/google-api-core/google/api_core/grpc_helpers.py b/packages/google-api-core/google/api_core/grpc_helpers.py
index bb2a3523d735..2f0ef9631dd8 100644
--- a/packages/google-api-core/google/api_core/grpc_helpers.py
+++ b/packages/google-api-core/google/api_core/grpc_helpers.py
@@ -430,10 +430,10 @@ def apply_interceptors(
     channel: grpc.Channel,
     interceptors: Optional[Sequence[ClientInterceptor]] = None,
 ) -> grpc.Channel:
-    """Applies a sequence of interceptors to a gRPC channel.
+    """Applies client interceptors to a gRPC channel.
 
-    The interceptors are applied in the order provided, such that the first
-    interceptor in the sequence is the outermost layer (executes first).
+    The first interceptor in the sequence is the outermost layer: it
+    executes first on outbound requests and last on inbound responses.
 
     Args:
         channel (grpc.Channel): The channel to intercept.

From ea38a9866b087bbd9ba1de3f20fdebe2707f321c Mon Sep 17 00:00:00 2001
From: chalmer lowe <chalmerlowe@google.com>
Date: Thu, 27 Aug 2026 09:45:16 -0400
Subject: [PATCH 04/16] test(api-core): remove redundant execution order test
 and simplify interceptor unit tests

---
 .../tests/unit/test_grpc_helpers.py           | 55 +------------------
 1 file changed, 2 insertions(+), 53 deletions(-)

diff --git a/packages/google-api-core/tests/unit/test_grpc_helpers.py b/packages/google-api-core/tests/unit/test_grpc_helpers.py
index b9fc553e6555..0dad58217545 100644
--- a/packages/google-api-core/tests/unit/test_grpc_helpers.py
+++ b/packages/google-api-core/tests/unit/test_grpc_helpers.py
@@ -943,12 +943,11 @@ def test_apply_interceptors_passthrough(falsy_interceptors):
 
 @pytest.mark.parametrize("count", [1, 2, 3])
 def test_apply_interceptors_wrapping(count):
-    """Verify that interceptors are passed to grpc.intercept_channel in a single call.
+    """Verify that interceptors are passed to grpc.intercept_channel unpacked in a single call.
 
     When given a sequence of N interceptors [i_0, i_1, ..., i_{N-1}], apply_interceptors
     must pass the base channel and all interceptors unpacked (*interceptors) to
-    grpc.intercept_channel. This creates a single intercepted channel wrapper rather than
-    multiple nested wrappers.
+    grpc.intercept_channel.
     """
     mock_channel = mock.Mock(name="base_channel")
     mock_intercepted = mock.Mock(name="intercepted_channel")
@@ -961,53 +960,3 @@ def test_apply_interceptors_wrapping(count):
 
     assert result is mock_intercepted
     mock_intercept.assert_called_once_with(mock_channel, *interceptors)
-
-
-def test_apply_interceptors_execution_order():
-    """Verify runtime execution order (onion model) when invoking an RPC on an intercepted channel.
-
-    In standard gRPC Python, grpc.intercept_channel(channel, *interceptors) processes
-    the interceptor list such that the first interceptor in the sequence is the outermost layer.
-    Therefore, given [i1, i2]:
-      - i1 is the outermost layer (executes first on outbound request)
-      - i2 is the inner layer (executes second on outbound request)
-
-    During an RPC invocation:
-      1. i1 intercepts the call first (request inbound / pre-call)
-      2. i1 calls continuation(), which triggers i2
-      3. i2 calls continuation(), which reaches the channel stub / network
-      4. i2 post-call logic finishes
-      5. i1 post-call logic finishes
-    """
-    execution_order = []
-
-    class OrderInterceptor(grpc.UnaryUnaryClientInterceptor):
-        def __init__(self, name):
-            self.name = name
-
-        def intercept_unary_unary(self, continuation, client_call_details, request):
-            execution_order.append(f"{self.name}_start")
-            response = continuation(client_call_details, request)
-            execution_order.append(f"{self.name}_end")
-            return response
-
-    i1 = OrderInterceptor("i1")
-    i2 = OrderInterceptor("i2")
-
-    mock_channel = mock.Mock(spec=grpc.Channel)
-    mock_callable = mock.Mock(spec=grpc.UnaryUnaryMultiCallable)
-    mock_call = mock.Mock(spec=grpc.Call)
-    expected_response = operations_pb2.Operation(name="test_op")
-    mock_callable.with_call.return_value = (expected_response, mock_call)
-    mock_channel.unary_unary.return_value = mock_callable
-
-    # Apply interceptors in sequence [i1, i2]
-    intercepted_channel = grpc_helpers.apply_interceptors(mock_channel, [i1, i2])
-
-    # Trigger a unary RPC through the intercepted channel
-    stub = operations_pb2.OperationsStub(intercepted_channel)
-    response = stub.GetOperation(operations_pb2.GetOperationRequest(name="test_op"))
-
-    assert response.name == "test_op"
-    # Verify i1 executed as the outer layer surrounding i2
-    assert execution_order == ["i1_start", "i2_start", "i2_end", "i1_end"]

From 8e65db065a23465d500f59d00aa7efa3246f18e1 Mon Sep 17 00:00:00 2001
From: chalmer lowe <chalmerlowe@google.com>
Date: Thu, 27 Aug 2026 10:08:49 -0400
Subject: [PATCH 05/16] test(api-core): align mock variable naming to Option 1
 convention

---
 .../tests/unit/test_grpc_helpers.py           | 24 ++++++++++---------
 1 file changed, 13 insertions(+), 11 deletions(-)

diff --git a/packages/google-api-core/tests/unit/test_grpc_helpers.py b/packages/google-api-core/tests/unit/test_grpc_helpers.py
index 0dad58217545..9e41bab853df 100644
--- a/packages/google-api-core/tests/unit/test_grpc_helpers.py
+++ b/packages/google-api-core/tests/unit/test_grpc_helpers.py
@@ -936,9 +936,9 @@ def test_close(self):
 @pytest.mark.parametrize("falsy_interceptors", [None, [], ()])
 def test_apply_interceptors_passthrough(falsy_interceptors):
     """Verify that falsy or empty interceptor sequences return the channel unmodified."""
-    mock_channel = mock.Mock()
-    result = grpc_helpers.apply_interceptors(mock_channel, falsy_interceptors)
-    assert result is mock_channel
+    mock_base_channel = mock.Mock(name="base_channel")
+    result = grpc_helpers.apply_interceptors(mock_base_channel, falsy_interceptors)
+    assert result is mock_base_channel
 
 
 @pytest.mark.parametrize("count", [1, 2, 3])
@@ -949,14 +949,16 @@ def test_apply_interceptors_wrapping(count):
     must pass the base channel and all interceptors unpacked (*interceptors) to
     grpc.intercept_channel.
     """
-    mock_channel = mock.Mock(name="base_channel")
-    mock_intercepted = mock.Mock(name="intercepted_channel")
-    interceptors = [mock.Mock(name=f"interceptor_{i}") for i in range(count)]
+    mock_base_channel = mock.Mock(name="base_channel")
+    mock_wrapped_channel = mock.Mock(name="wrapped_channel")
+    mock_interceptors = [mock.Mock(name=f"interceptor_{i}") for i in range(count)]
 
     with mock.patch(
-        "grpc.intercept_channel", return_value=mock_intercepted
-    ) as mock_intercept:
-        result = grpc_helpers.apply_interceptors(mock_channel, interceptors)
+        "grpc.intercept_channel", return_value=mock_wrapped_channel
+    ) as mock_intercept_channel:
+        result = grpc_helpers.apply_interceptors(mock_base_channel, mock_interceptors)
 
-    assert result is mock_intercepted
-    mock_intercept.assert_called_once_with(mock_channel, *interceptors)
+    assert result is mock_wrapped_channel
+    mock_intercept_channel.assert_called_once_with(
+        mock_base_channel, *mock_interceptors
+    )

From b41b0c1cad30d94609cae0ad095f4b9ffb93b993 Mon Sep 17 00:00:00 2001
From: chalmer lowe <chalmerlowe@google.com>
Date: Fri, 28 Aug 2026 09:51:22 -0400
Subject: [PATCH 06/16] feat(api-core): update default env var to
 GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED

- Set is_otel_capabilities_enabled default env_var to GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED

- Activate fail-fast FeatureGatingError experimental path when tracer_provider is set without env var

- Update unit tests to verify experimental gating behavior
---
 .../google/api_core/_observability.py         |  2 +-
 .../tests/unit/test_observability.py          | 43 +++++++++++++++++--
 2 files changed, 41 insertions(+), 4 deletions(-)

diff --git a/packages/google-api-core/google/api_core/_observability.py b/packages/google-api-core/google/api_core/_observability.py
index a36df8b39599..b1b71b056658 100644
--- a/packages/google-api-core/google/api_core/_observability.py
+++ b/packages/google-api-core/google/api_core/_observability.py
@@ -26,7 +26,7 @@
 
 def is_otel_capabilities_enabled(
     client_options: Optional[ClientOptions | dict[str, Any]] = None,
-    env_var: str = "GOOGLE_CLOUD_PYTHON_TRACING_ENABLED",
+    env_var: str = "GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED",
 ) -> bool:
     """Checks if OTel capabilities are enabled and installed.
 
diff --git a/packages/google-api-core/tests/unit/test_observability.py b/packages/google-api-core/tests/unit/test_observability.py
index fc63023aadcd..d39edb806040 100644
--- a/packages/google-api-core/tests/unit/test_observability.py
+++ b/packages/google-api-core/tests/unit/test_observability.py
@@ -15,17 +15,19 @@
 import sys
 from unittest import mock
 
+import pytest
 from google.api_core import _observability
+from google.api_core._feature_gating_helpers import FeatureGatingError
 from google.api_core.client_options import ClientOptions
 
 
 def test_is_otel_capabilities_enabled_disabled(monkeypatch):
-    monkeypatch.setenv("GOOGLE_CLOUD_PYTHON_TRACING_ENABLED", "false")
+    monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", "false")
     assert not _observability.is_otel_capabilities_enabled()
 
 
 def test_is_otel_capabilities_enabled_otel_missing(monkeypatch):
-    monkeypatch.setenv("GOOGLE_CLOUD_PYTHON_TRACING_ENABLED", "true")
+    monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", "true")
     # Simulate OTel not being installed by blocking imports
     monkeypatch.setitem(sys.modules, "opentelemetry.instrumentation.grpc", None)
 
@@ -33,7 +35,7 @@ def test_is_otel_capabilities_enabled_otel_missing(monkeypatch):
 
 
 def test_is_otel_capabilities_enabled_otel_installed(monkeypatch):
-    monkeypatch.setenv("GOOGLE_CLOUD_PYTHON_TRACING_ENABLED", "true")
+    monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", "true")
 
     mock_otel = mock.Mock()
     mock_otel_grpc = mock_otel.instrumentation.grpc
@@ -49,6 +51,41 @@ def test_is_otel_capabilities_enabled_otel_installed(monkeypatch):
     assert _observability.is_otel_capabilities_enabled()
 
 
+def test_is_otel_capabilities_enabled_experimental_requires_env_var(monkeypatch):
+    """Proves that passing client_options with tracer_provider without the experimental
+    env var set to 'true' raises FeatureGatingError (Fail Fast).
+    """
+    monkeypatch.delenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", raising=False)
+    options = ClientOptions(tracer_provider=object())
+
+    with pytest.raises(
+        FeatureGatingError,
+        match="requires GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED",
+    ):
+        _observability.is_otel_capabilities_enabled(options)
+
+
+def test_is_otel_capabilities_enabled_experimental_enabled_with_config(monkeypatch):
+    """Proves that when GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED=true and tracer_provider
+    is supplied via client_options, is_otel_capabilities_enabled returns True.
+    """
+    monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", "true")
+
+    mock_otel = mock.Mock()
+    mock_otel_grpc = mock_otel.instrumentation.grpc
+
+    monkeypatch.setitem(sys.modules, "opentelemetry", mock_otel)
+    monkeypatch.setitem(
+        sys.modules, "opentelemetry.instrumentation", mock_otel.instrumentation
+    )
+    monkeypatch.setitem(
+        sys.modules, "opentelemetry.instrumentation.grpc", mock_otel_grpc
+    )
+
+    options = ClientOptions(tracer_provider=object())
+    assert _observability.is_otel_capabilities_enabled(options)
+
+
 def test_apply_otel_capabilities_to_channel_enabled_otel_installed(monkeypatch):
     mock_channel = mock.Mock()
     mock_intercepted_channel = mock.Mock()

From 8b2dc1fc458d505bf6bfc5ac1032d24a14f51018 Mon Sep 17 00:00:00 2001
From: chalmer lowe <chalmerlowe@google.com>
Date: Mon, 31 Aug 2026 08:15:09 -0400
Subject: [PATCH 07/16] feat(api-core): add ChannelWrapper and
 apply_channel_wrappers helper

---
 .../google/api_core/grpc_helpers.py           |  63 ++++++++---
 .../tests/unit/test_grpc_helpers.py           | 106 ++++++++++++++----
 2 files changed, 136 insertions(+), 33 deletions(-)

diff --git a/packages/google-api-core/google/api_core/grpc_helpers.py b/packages/google-api-core/google/api_core/grpc_helpers.py
index 2f0ef9631dd8..5f9baf4cf12f 100644
--- a/packages/google-api-core/google/api_core/grpc_helpers.py
+++ b/packages/google-api-core/google/api_core/grpc_helpers.py
@@ -17,7 +17,16 @@
 import collections
 import functools
 import warnings
-from typing import Generic, Iterator, Optional, Sequence, TypeVar, Union
+from typing import (
+    Callable,
+    Generic,
+    Iterator,
+    Optional,
+    Sequence,
+    TypeVar,
+    Union,
+    get_args,
+)
 
 import google.auth
 import google.auth.credentials
@@ -41,6 +50,15 @@
     grpc.StreamStreamClientInterceptor,
 ]
 
+# Runtime tuple of gRPC client interceptor base classes for isinstance checks
+_CLIENT_INTERCEPTOR_CLASSES = get_args(ClientInterceptor)
+
+# Type alias representing a channel-wrapping callable
+ChannelWrapperCallable = Callable[[grpc.Channel], grpc.Channel]
+
+# Generic type alias representing any channel wrapper (interceptor or callable)
+ChannelWrapper = Union[ClientInterceptor, ChannelWrapperCallable]
+
 
 def _patch_callable_name(callable_):
     """Fix-up gRPC callable attributes.
@@ -426,27 +444,44 @@ def _modify_target_for_direct_path(target: str) -> str:
     return target
 
 
-def apply_interceptors(
+def apply_channel_wrappers(
     channel: grpc.Channel,
-    interceptors: Optional[Sequence[ClientInterceptor]] = None,
+    wrappers: Optional[Sequence[ChannelWrapper]] = None,
 ) -> grpc.Channel:
-    """Applies client interceptors to a gRPC channel.
+    """Applies channel wrappers (client interceptors or channel-wrapping callables) to a gRPC channel.
 
-    The first interceptor in the sequence is the outermost layer: it
-    executes first on outbound requests and last on inbound responses.
+    Executes in reverse order so the first wrapper in the sequence becomes the
+    outermost layer on outbound requests and the innermost layer on inbound responses.
 
     Args:
-        channel (grpc.Channel): The channel to intercept.
-        interceptors (Optional[Sequence[ClientInterceptor]]): An optional sequence
-            of client interceptors to apply.
+        channel (grpc.Channel): The channel to wrap.
+        wrappers (Optional[Sequence[ChannelWrapper]]):
+            An optional sequence of client interceptors or channel-wrapping
+            callables to apply.
 
     Returns:
-        grpc.Channel: The intercepted channel, or the original channel if no
-            interceptors were provided.
+        grpc.Channel: The wrapped channel, or the original channel if no
+            wrappers were provided.
+
+    Raises:
+        TypeError: If an item in ``wrappers`` is neither a gRPC ClientInterceptor
+            nor a Callable[[Channel], Channel].
     """
-    if interceptors:
-        return grpc.intercept_channel(channel, *interceptors)
-    return channel
+    if not wrappers:
+        return channel
+
+    modified_channel = channel
+    for wrapper in reversed(list(wrappers)):
+        if isinstance(wrapper, _CLIENT_INTERCEPTOR_CLASSES):
+            modified_channel = grpc.intercept_channel(modified_channel, wrapper)
+        elif callable(wrapper):
+            modified_channel = wrapper(modified_channel)
+        else:
+            raise TypeError(
+                f"Expected ChannelWrapper (ClientInterceptor or Callable[[Channel], Channel]), got {type(wrapper).__name__}"
+            )
+
+    return modified_channel
 
 
 _MethodCall = collections.namedtuple(
diff --git a/packages/google-api-core/tests/unit/test_grpc_helpers.py b/packages/google-api-core/tests/unit/test_grpc_helpers.py
index 9e41bab853df..8d91e63f15f7 100644
--- a/packages/google-api-core/tests/unit/test_grpc_helpers.py
+++ b/packages/google-api-core/tests/unit/test_grpc_helpers.py
@@ -933,32 +933,100 @@ def test_close(self):
         assert channel.close() is None
 
 
-@pytest.mark.parametrize("falsy_interceptors", [None, [], ()])
-def test_apply_interceptors_passthrough(falsy_interceptors):
-    """Verify that falsy or empty interceptor sequences return the channel unmodified."""
+@pytest.mark.parametrize("falsy_wrappers", [None, [], ()])
+def test_apply_channel_wrappers_passthrough(falsy_wrappers):
+    """Verify that falsy or empty wrapper sequences return the channel unmodified."""
     mock_base_channel = mock.Mock(name="base_channel")
-    result = grpc_helpers.apply_interceptors(mock_base_channel, falsy_interceptors)
+    result = grpc_helpers.apply_channel_wrappers(mock_base_channel, falsy_wrappers)
     assert result is mock_base_channel
 
 
-@pytest.mark.parametrize("count", [1, 2, 3])
-def test_apply_interceptors_wrapping(count):
-    """Verify that interceptors are passed to grpc.intercept_channel unpacked in a single call.
+def test_apply_channel_wrappers_grpc_client_interceptors():
+    """Verify that standard gRPC ClientInterceptor instances are applied via grpc.intercept_channel."""
+
+    class DummyUnaryInterceptor(grpc.UnaryUnaryClientInterceptor):
+        def intercept_unary_unary(self, continuation, client_call_details, request):
+            return continuation(client_call_details, request)
+
+    class DummyStreamInterceptor(grpc.StreamStreamClientInterceptor):
+        def intercept_stream_stream(
+            self, continuation, client_call_details, request_iterator
+        ):
+            return continuation(client_call_details, request_iterator)
+
+    interceptor1 = DummyUnaryInterceptor()
+    interceptor2 = DummyStreamInterceptor()
 
-    When given a sequence of N interceptors [i_0, i_1, ..., i_{N-1}], apply_interceptors
-    must pass the base channel and all interceptors unpacked (*interceptors) to
-    grpc.intercept_channel.
-    """
     mock_base_channel = mock.Mock(name="base_channel")
-    mock_wrapped_channel = mock.Mock(name="wrapped_channel")
-    mock_interceptors = [mock.Mock(name=f"interceptor_{i}") for i in range(count)]
+    mock_chan_after_i2 = mock.Mock(name="chan_after_i2")
+    mock_chan_after_i1 = mock.Mock(name="chan_after_i1")
 
     with mock.patch(
-        "grpc.intercept_channel", return_value=mock_wrapped_channel
-    ) as mock_intercept_channel:
-        result = grpc_helpers.apply_interceptors(mock_base_channel, mock_interceptors)
+        "grpc.intercept_channel",
+        side_effect=[mock_chan_after_i2, mock_chan_after_i1],
+    ) as mock_intercept:
+        result = grpc_helpers.apply_channel_wrappers(
+            mock_base_channel, [interceptor1, interceptor2]
+        )
 
-    assert result is mock_wrapped_channel
-    mock_intercept_channel.assert_called_once_with(
-        mock_base_channel, *mock_interceptors
+    assert result is mock_chan_after_i1
+    assert mock_intercept.call_count == 2
+    # Executed in reverse order so interceptor1 is outermost
+    mock_intercept.assert_has_calls(
+        [
+            mock.call(mock_base_channel, interceptor2),
+            mock.call(mock_chan_after_i2, interceptor1),
+        ]
     )
+
+
+def test_apply_channel_wrappers_callables():
+    """Verify that channel-wrapping callables Callable[[Channel], Channel] are invoked in sequence."""
+    mock_base_channel = mock.Mock(name="base_channel")
+    mock_chan_1 = mock.Mock(name="chan_1")
+    mock_chan_2 = mock.Mock(name="chan_2")
+
+    wrapper1 = mock.Mock(side_effect=lambda ch: mock_chan_2)
+    wrapper2 = mock.Mock(side_effect=lambda ch: mock_chan_1)
+
+    result = grpc_helpers.apply_channel_wrappers(
+        mock_base_channel, [wrapper1, wrapper2]
+    )
+
+    assert result is mock_chan_2
+    # Executed in reverse order: wrapper2 runs first on base channel, then wrapper1
+    wrapper2.assert_called_once_with(mock_base_channel)
+    wrapper1.assert_called_once_with(mock_chan_1)
+
+
+def test_apply_channel_wrappers_interspersed():
+    """Verify that a mixed sequence of gRPC interceptors and channel wrapper callables are applied."""
+
+    class DummyUnaryInterceptor(grpc.UnaryUnaryClientInterceptor):
+        def intercept_unary_unary(self, continuation, client_call_details, request):
+            return continuation(client_call_details, request)
+
+    interceptor = DummyUnaryInterceptor()
+    mock_base_channel = mock.Mock(name="base_channel")
+    mock_chan_after_wrapper = mock.Mock(name="chan_after_wrapper")
+    mock_chan_after_interceptor = mock.Mock(name="chan_after_interceptor")
+
+    wrapper = mock.Mock(return_value=mock_chan_after_wrapper)
+
+    with mock.patch(
+        "grpc.intercept_channel", return_value=mock_chan_after_interceptor
+    ) as mock_intercept:
+        result = grpc_helpers.apply_channel_wrappers(
+            mock_base_channel, [interceptor, wrapper]
+        )
+
+    assert result is mock_chan_after_interceptor
+    wrapper.assert_called_once_with(mock_base_channel)
+    mock_intercept.assert_called_once_with(mock_chan_after_wrapper, interceptor)
+
+
+def test_apply_channel_wrappers_invalid_type_raises():
+    """Verify that passing an invalid object that is neither an interceptor nor callable raises TypeError."""
+    mock_base_channel = mock.Mock(name="base_channel")
+    with pytest.raises(TypeError, match="Expected ChannelWrapper"):
+        grpc_helpers.apply_channel_wrappers(mock_base_channel, [12345])

From 7aeb5a2a43237ca93073e0facf4552adb01e39c1 Mon Sep 17 00:00:00 2001
From: chalmer lowe <chalmerlowe@google.com>
Date: Mon, 31 Aug 2026 08:39:10 -0400
Subject: [PATCH 08/16] style(api-core): format imports with isort via ruff

---
 packages/google-api-core/google/api_core/grpc_helpers.py   | 1 +
 packages/google-api-core/tests/unit/test_client_options.py | 1 +
 packages/google-api-core/tests/unit/test_grpc_helpers.py   | 3 ++-
 packages/google-api-core/tests/unit/test_observability.py  | 1 +
 packages/google-api-core/tests/unit/test_path_template.py  | 1 +
 5 files changed, 6 insertions(+), 1 deletion(-)

diff --git a/packages/google-api-core/google/api_core/grpc_helpers.py b/packages/google-api-core/google/api_core/grpc_helpers.py
index 5f9baf4cf12f..7e7777766f7b 100644
--- a/packages/google-api-core/google/api_core/grpc_helpers.py
+++ b/packages/google-api-core/google/api_core/grpc_helpers.py
@@ -34,6 +34,7 @@
 import google.auth.transport.requests
 import google.protobuf
 import grpc
+
 from google.api_core import exceptions, general_helpers
 
 # The list of gRPC Callable interfaces that return iterators.
diff --git a/packages/google-api-core/tests/unit/test_client_options.py b/packages/google-api-core/tests/unit/test_client_options.py
index c15e83174ed4..49fb35e8ba2f 100644
--- a/packages/google-api-core/tests/unit/test_client_options.py
+++ b/packages/google-api-core/tests/unit/test_client_options.py
@@ -15,6 +15,7 @@
 from re import match
 
 import pytest
+
 from google.api_core import client_options
 
 from ..helpers import warn_deprecated_credentials_file
diff --git a/packages/google-api-core/tests/unit/test_grpc_helpers.py b/packages/google-api-core/tests/unit/test_grpc_helpers.py
index 8d91e63f15f7..4367552651d7 100644
--- a/packages/google-api-core/tests/unit/test_grpc_helpers.py
+++ b/packages/google-api-core/tests/unit/test_grpc_helpers.py
@@ -24,9 +24,10 @@
     pytest.skip("No GRPC", allow_module_level=True)
 
 import google.auth.credentials
-from google.api_core import exceptions, grpc_helpers
 from google.longrunning import operations_pb2
 
+from google.api_core import exceptions, grpc_helpers
+
 
 def test__patch_callable_name():
     callable = mock.Mock(spec=["__class__"])
diff --git a/packages/google-api-core/tests/unit/test_observability.py b/packages/google-api-core/tests/unit/test_observability.py
index d39edb806040..f0ebe0afc14d 100644
--- a/packages/google-api-core/tests/unit/test_observability.py
+++ b/packages/google-api-core/tests/unit/test_observability.py
@@ -16,6 +16,7 @@
 from unittest import mock
 
 import pytest
+
 from google.api_core import _observability
 from google.api_core._feature_gating_helpers import FeatureGatingError
 from google.api_core.client_options import ClientOptions
diff --git a/packages/google-api-core/tests/unit/test_path_template.py b/packages/google-api-core/tests/unit/test_path_template.py
index fb67973549f7..f053fc952176 100644
--- a/packages/google-api-core/tests/unit/test_path_template.py
+++ b/packages/google-api-core/tests/unit/test_path_template.py
@@ -18,6 +18,7 @@
 
 import pytest
 from google.api import auth_pb2
+
 from google.api_core import path_template
 
 

From 9149df8dda89fcd6c63088100b8bbc7b6dd6c9cb Mon Sep 17 00:00:00 2001
From: chalmer lowe <chalmerlowe@google.com>
Date: Mon, 31 Aug 2026 11:57:28 -0400
Subject: [PATCH 09/16] refactor(api-core): use PEP 604 union syntax in
 grpc_helpers.py and clarify reverse execution

---
 .../google/api_core/grpc_helpers.py            | 18 +++++++++---------
 1 file changed, 9 insertions(+), 9 deletions(-)

diff --git a/packages/google-api-core/google/api_core/grpc_helpers.py b/packages/google-api-core/google/api_core/grpc_helpers.py
index 7e7777766f7b..447d521bfd8f 100644
--- a/packages/google-api-core/google/api_core/grpc_helpers.py
+++ b/packages/google-api-core/google/api_core/grpc_helpers.py
@@ -24,7 +24,6 @@
     Optional,
     Sequence,
     TypeVar,
-    Union,
     get_args,
 )
 
@@ -44,12 +43,12 @@
 P = TypeVar("P")
 
 # Type alias representing any client-side gRPC interceptor
-ClientInterceptor = Union[
-    grpc.UnaryUnaryClientInterceptor,
-    grpc.UnaryStreamClientInterceptor,
-    grpc.StreamUnaryClientInterceptor,
-    grpc.StreamStreamClientInterceptor,
-]
+ClientInterceptor = (
+    grpc.UnaryUnaryClientInterceptor
+    | grpc.UnaryStreamClientInterceptor
+    | grpc.StreamUnaryClientInterceptor
+    | grpc.StreamStreamClientInterceptor
+)
 
 # Runtime tuple of gRPC client interceptor base classes for isinstance checks
 _CLIENT_INTERCEPTOR_CLASSES = get_args(ClientInterceptor)
@@ -58,7 +57,7 @@
 ChannelWrapperCallable = Callable[[grpc.Channel], grpc.Channel]
 
 # Generic type alias representing any channel wrapper (interceptor or callable)
-ChannelWrapper = Union[ClientInterceptor, ChannelWrapperCallable]
+ChannelWrapper = ClientInterceptor | ChannelWrapperCallable
 
 
 def _patch_callable_name(callable_):
@@ -447,7 +446,7 @@ def _modify_target_for_direct_path(target: str) -> str:
 
 def apply_channel_wrappers(
     channel: grpc.Channel,
-    wrappers: Optional[Sequence[ChannelWrapper]] = None,
+    wrappers: Sequence[ChannelWrapper] | None = None,
 ) -> grpc.Channel:
     """Applies channel wrappers (client interceptors or channel-wrapping callables) to a gRPC channel.
 
@@ -472,6 +471,7 @@ def apply_channel_wrappers(
         return channel
 
     modified_channel = channel
+    # Reverse the inputs to align with the behavior of grpc.create_channel(*interceptors)
     for wrapper in reversed(list(wrappers)):
         if isinstance(wrapper, _CLIENT_INTERCEPTOR_CLASSES):
             modified_channel = grpc.intercept_channel(modified_channel, wrapper)

From f89cfbd7eda0b77fd838d0ad53e53b65f33e54e9 Mon Sep 17 00:00:00 2001
From: chalmer lowe <chalmerlowe@google.com>
Date: Mon, 31 Aug 2026 12:05:11 -0400
Subject: [PATCH 10/16] fix(api-core): add TypeAlias and typing.cast to satisfy
 mypy

---
 .../google-api-core/google/api_core/grpc_helpers.py   | 11 +++++++----
 1 file changed, 7 insertions(+), 4 deletions(-)

diff --git a/packages/google-api-core/google/api_core/grpc_helpers.py b/packages/google-api-core/google/api_core/grpc_helpers.py
index 447d521bfd8f..13f89c64a13f 100644
--- a/packages/google-api-core/google/api_core/grpc_helpers.py
+++ b/packages/google-api-core/google/api_core/grpc_helpers.py
@@ -23,7 +23,9 @@
     Iterator,
     Optional,
     Sequence,
+    TypeAlias,
     TypeVar,
+    cast,
     get_args,
 )
 
@@ -43,7 +45,7 @@
 P = TypeVar("P")
 
 # Type alias representing any client-side gRPC interceptor
-ClientInterceptor = (
+ClientInterceptor: TypeAlias = (
     grpc.UnaryUnaryClientInterceptor
     | grpc.UnaryStreamClientInterceptor
     | grpc.StreamUnaryClientInterceptor
@@ -54,10 +56,10 @@
 _CLIENT_INTERCEPTOR_CLASSES = get_args(ClientInterceptor)
 
 # Type alias representing a channel-wrapping callable
-ChannelWrapperCallable = Callable[[grpc.Channel], grpc.Channel]
+ChannelWrapperCallable: TypeAlias = Callable[[grpc.Channel], grpc.Channel]
 
 # Generic type alias representing any channel wrapper (interceptor or callable)
-ChannelWrapper = ClientInterceptor | ChannelWrapperCallable
+ChannelWrapper: TypeAlias = ClientInterceptor | ChannelWrapperCallable
 
 
 def _patch_callable_name(callable_):
@@ -476,7 +478,8 @@ def apply_channel_wrappers(
         if isinstance(wrapper, _CLIENT_INTERCEPTOR_CLASSES):
             modified_channel = grpc.intercept_channel(modified_channel, wrapper)
         elif callable(wrapper):
-            modified_channel = wrapper(modified_channel)
+            wrapper_callable = cast(ChannelWrapperCallable, wrapper)
+            modified_channel = wrapper_callable(modified_channel)
         else:
             raise TypeError(
                 f"Expected ChannelWrapper (ClientInterceptor or Callable[[Channel], Channel]), got {type(wrapper).__name__}"

From 36875330780703f1a2007d820b3f34caa19e3744 Mon Sep 17 00:00:00 2001
From: Chalmer Lowe <chalmerlowe@google.com>
Date: Tue, 1 Sep 2026 08:59:30 -0400
Subject: [PATCH 11/16] feat(api-core): add channel orchestration for
 OpenTelemetry (B) (#18237)

This pull request introduces OpenTelemetry helper functions in
`google.api_core._observability` to produce channel wrappers for
synchronous gRPC channels and interceptors for asynchronous gRPC
channels.

### Problem
Generated client libraries need a consistent and maintainable way to
instrument gRPC channels with OpenTelemetry tracing when enabled via
environment variables or client options.
Because synchronous gRPC channels can be wrapped post-creation while
asynchronous gRPC channels require interceptors at channel creation
time, client transports need helpers that return the appropriate channel
wrapper or async interceptors without duplicating OpenTelemetry
resolution logic across client libraries.

### Solution
This pull request introduces the following helper functions in
`google.api_core._observability`:
1. `get_otel_channel_wrapper(client_options)`:
* Returns a channel-wrapping function (`Callable[[Channel], Channel]`)
for synchronous gRPC channels when OpenTelemetry tracing is enabled and
installed.
* Integrates with `grpc_helpers.apply_channel_wrappers` to wrap raw
channels using OpenTelemetry's `intercept_channel`.
2. `get_otel_async_interceptor(client_options)`:
* Returns a list of OpenTelemetry asynchronous client interceptors
(`aio_client_interceptors`) for use when constructing `grpc.aio`
channels.
3. `_get_otel_interceptor(client_options, is_async)`:
* Internal helper that extracts `tracer_provider` from `ClientOptions`
and creates the appropriate OpenTelemetry sync or async interceptors.

### Testing
* Added unit tests in `tests/unit/test_observability.py` covering:
* Sync and async interceptor extraction and `tracer_provider`
configuration.
* `get_otel_channel_wrapper` behavior when tracing is disabled, when
OpenTelemetry is not installed, and when tracing is enabled.
* Integration between `get_otel_channel_wrapper` and
`grpc_helpers.apply_channel_wrappers`.
* `get_otel_async_interceptor` behavior across disabled, missing, and
enabled states.

### Notes for Reviewers
* This PR builds upon PR #18236 (`ChannelWrapper` and
`apply_channel_wrappers`).
* `get_otel_channel_wrapper` returns a callable rather than modifying
the channel immediately, allowing transport layers to combine
OpenTelemetry wrapping with user-supplied custom channel wrappers.
---
 .../google/api_core/_observability.py         |  80 ++++++--
 .../tests/unit/test_observability.py          | 183 ++++++++++++++----
 2 files changed, 210 insertions(+), 53 deletions(-)

diff --git a/packages/google-api-core/google/api_core/_observability.py b/packages/google-api-core/google/api_core/_observability.py
index b1b71b056658..16f9917e736d 100644
--- a/packages/google-api-core/google/api_core/_observability.py
+++ b/packages/google-api-core/google/api_core/_observability.py
@@ -16,16 +16,21 @@
 
 """OpenTelemetry helpers for resolving and instantiating interceptors."""
 
-from typing import Any, Optional
+from typing import TYPE_CHECKING, Any, Callable
 
 from google.api_core import _feature_gating_helpers
 from google.api_core.client_options import ClientOptions
 
+if TYPE_CHECKING:
+    from google.api_core.grpc_helpers import ChannelWrapperCallable
+else:
+    ChannelWrapperCallable = Callable[[Any], Any]
+
 _TRACER_PROVIDER = "tracer_provider"
 
 
 def is_otel_capabilities_enabled(
-    client_options: Optional[ClientOptions | dict[str, Any]] = None,
+    client_options: ClientOptions | dict[str, Any] | None = None,
     env_var: str = "GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED",
 ) -> bool:
     """Checks if OTel capabilities are enabled and installed.
@@ -54,26 +59,19 @@ def is_otel_capabilities_enabled(
     return False
 
 
-def apply_otel_capabilities_to_channel(
-    channel: Any,
-    client_options: Optional[ClientOptions | dict[str, Any]] = None,
+def _get_otel_interceptor(
+    client_options: ClientOptions | dict[str, Any] | None = None,
+    is_async: bool = False,
 ) -> Any:
-    """Applies OTel capabilities (like tracing) to the channel.
-
-    Precondition: This function assumes `is_otel_capabilities_enabled` has already
-    been called and returned `True`, i.e. in the Client. At this time
-    this function is not intended to be standalone.
+    """Instantiates a sync or async OpenTelemetry gRPC client interceptor.
 
     Args:
-        channel: The raw gRPC channel to wrap.
         client_options: The client options object or dictionary.
+        is_async: If True, returns an async interceptor (`aio_client_interceptor`),
+            otherwise returns a sync interceptor (`client_interceptor`).
 
     Returns:
-        Any: The intercepted channel.
-
-    Raises:
-        ImportError: If OpenTelemetry packages are not installed and this function
-            is called directly (bypassing the precondition).
+        Any: The instantiated OpenTelemetry client interceptor.
     """
     import opentelemetry.instrumentation.grpc as otel_grpc  # type: ignore[import-not-found]
 
@@ -83,7 +81,51 @@ def apply_otel_capabilities_to_channel(
     elif client_options is not None:
         tracer_provider = getattr(client_options, _TRACER_PROVIDER, None)
 
-    interceptor = otel_grpc.client_interceptor(tracer_provider=tracer_provider)
+    if is_async:
+        return otel_grpc.aio_client_interceptors(tracer_provider=tracer_provider)
+    return otel_grpc.client_interceptor(tracer_provider=tracer_provider)
+
+
+def get_otel_channel_wrapper(
+    client_options: ClientOptions | dict[str, Any] | None = None,
+) -> ChannelWrapperCallable | None:
+    """Returns a channel wrapper callable that wraps a sync gRPC channel with OpenTelemetry tracing.
+
+    Args:
+        client_options: The client options object or dictionary used for feature gating
+            and extracting the tracer provider.
+
+    Returns:
+        Optional[ChannelWrapperCallable]: A channel-wrapping callable if OpenTelemetry
+            tracing is enabled and installed, None otherwise.
+    """
+    if not is_otel_capabilities_enabled(client_options):
+        return None
+
+    import opentelemetry.instrumentation.grpc as otel_grpc  # type: ignore[import-not-found]
+
+    interceptor = _get_otel_interceptor(client_options, is_async=False)
+
+    def channel_wrapper(channel: Any) -> Any:
+        return otel_grpc.intercept_channel(channel, interceptor)
+
+    return channel_wrapper
+
+
+def get_otel_async_interceptor(
+    client_options: ClientOptions | dict[str, Any] | None = None,
+) -> Any | None:
+    """Returns an async gRPC client interceptor for OpenTelemetry tracing.
+
+    Args:
+        client_options: The client options object or dictionary used for feature gating
+            and extracting the tracer provider.
+
+    Returns:
+        Optional[Any]: An instantiated OpenTelemetry async client interceptor
+            if tracing is enabled and installed, None otherwise.
+    """
+    if not is_otel_capabilities_enabled(client_options):
+        return None
 
-    # We use OTel's own compatible applier to avoid standard gRPC TypeError.
-    return otel_grpc.intercept_channel(channel, interceptor)
+    return _get_otel_interceptor(client_options, is_async=True)
diff --git a/packages/google-api-core/tests/unit/test_observability.py b/packages/google-api-core/tests/unit/test_observability.py
index f0ebe0afc14d..6b2d2fc7005f 100644
--- a/packages/google-api-core/tests/unit/test_observability.py
+++ b/packages/google-api-core/tests/unit/test_observability.py
@@ -87,16 +87,57 @@ def test_is_otel_capabilities_enabled_experimental_enabled_with_config(monkeypat
     assert _observability.is_otel_capabilities_enabled(options)
 
 
-def test_apply_otel_capabilities_to_channel_enabled_otel_installed(monkeypatch):
-    mock_channel = mock.Mock()
-    mock_intercepted_channel = mock.Mock()
+def test_get_otel_interceptor_sync_default(monkeypatch):
+    mock_otel = mock.Mock()
+    mock_otel_grpc = mock_otel.instrumentation.grpc
+    mock_interceptor = mock.Mock()
+    mock_otel_grpc.client_interceptor.return_value = mock_interceptor
+
+    monkeypatch.setitem(sys.modules, "opentelemetry", mock_otel)
+    monkeypatch.setitem(
+        sys.modules, "opentelemetry.instrumentation", mock_otel.instrumentation
+    )
+    monkeypatch.setitem(
+        sys.modules, "opentelemetry.instrumentation.grpc", mock_otel_grpc
+    )
+
+    result = _observability._get_otel_interceptor()
+    assert result is mock_interceptor
+    mock_otel_grpc.client_interceptor.assert_called_once_with(tracer_provider=None)
+
+
+def test_get_otel_interceptor_sync_config(monkeypatch):
+    mock_tracer_provider = object()
+    options = ClientOptions(tracer_provider=mock_tracer_provider)
 
     mock_otel = mock.Mock()
     mock_otel_grpc = mock_otel.instrumentation.grpc
     mock_interceptor = mock.Mock()
+    mock_otel_grpc.client_interceptor.return_value = mock_interceptor
+
+    monkeypatch.setitem(sys.modules, "opentelemetry", mock_otel)
+    monkeypatch.setitem(
+        sys.modules, "opentelemetry.instrumentation", mock_otel.instrumentation
+    )
+    monkeypatch.setitem(
+        sys.modules, "opentelemetry.instrumentation.grpc", mock_otel_grpc
+    )
 
+    result = _observability._get_otel_interceptor(client_options=options)
+    assert result is mock_interceptor
+    mock_otel_grpc.client_interceptor.assert_called_once_with(
+        tracer_provider=mock_tracer_provider
+    )
+
+
+def test_get_otel_interceptor_sync_dict_config(monkeypatch):
+    mock_tracer_provider = object()
+    options = {"tracer_provider": mock_tracer_provider}
+
+    mock_otel = mock.Mock()
+    mock_otel_grpc = mock_otel.instrumentation.grpc
+    mock_interceptor = mock.Mock()
     mock_otel_grpc.client_interceptor.return_value = mock_interceptor
-    mock_otel_grpc.intercept_channel.return_value = mock_intercepted_channel
 
     monkeypatch.setitem(sys.modules, "opentelemetry", mock_otel)
     monkeypatch.setitem(
@@ -106,29 +147,62 @@ def test_apply_otel_capabilities_to_channel_enabled_otel_installed(monkeypatch):
         sys.modules, "opentelemetry.instrumentation.grpc", mock_otel_grpc
     )
 
-    result = _observability.apply_otel_capabilities_to_channel(mock_channel)
+    result = _observability._get_otel_interceptor(client_options=options)
+    assert result is mock_interceptor
+    mock_otel_grpc.client_interceptor.assert_called_once_with(
+        tracer_provider=mock_tracer_provider
+    )
+
 
-    assert result is mock_intercepted_channel
-    mock_otel_grpc.client_interceptor.assert_called_once_with(tracer_provider=None)
-    mock_otel_grpc.intercept_channel.assert_called_once_with(
-        mock_channel, mock_interceptor
+def test_get_otel_interceptor_async(monkeypatch):
+    mock_tracer_provider = object()
+    options = ClientOptions(tracer_provider=mock_tracer_provider)
+
+    mock_otel = mock.Mock()
+    mock_otel_grpc = mock_otel.instrumentation.grpc
+    mock_async_interceptors = [mock.Mock()]
+    mock_otel_grpc.aio_client_interceptors.return_value = mock_async_interceptors
+
+    monkeypatch.setitem(sys.modules, "opentelemetry", mock_otel)
+    monkeypatch.setitem(
+        sys.modules, "opentelemetry.instrumentation", mock_otel.instrumentation
     )
+    monkeypatch.setitem(
+        sys.modules, "opentelemetry.instrumentation.grpc", mock_otel_grpc
+    )
+
+    result = _observability._get_otel_interceptor(client_options=options, is_async=True)
+    assert result is mock_async_interceptors
+    mock_otel_grpc.aio_client_interceptors.assert_called_once_with(
+        tracer_provider=mock_tracer_provider
+    )
+
+
+def test_get_otel_channel_wrapper_disabled(monkeypatch):
+    monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", "false")
+    assert _observability.get_otel_channel_wrapper() is None
+
+
+def test_get_otel_channel_wrapper_otel_missing(monkeypatch):
+    monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", "true")
+    monkeypatch.setitem(sys.modules, "opentelemetry.instrumentation.grpc", None)
+    assert _observability.get_otel_channel_wrapper() is None
 
 
-def test_apply_otel_capabilities_to_channel_enabled_via_config(monkeypatch):
-    # Tracing enabled via config (tracer_provider is set)
+def test_get_otel_channel_wrapper_enabled(monkeypatch):
+    monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", "true")
     mock_tracer_provider = object()
     options = ClientOptions(tracer_provider=mock_tracer_provider)
 
-    mock_channel = mock.Mock()
-    mock_intercepted_channel = mock.Mock()
+    mock_raw_channel = mock.Mock(name="raw_channel")
+    mock_wrapped_channel = mock.Mock(name="wrapped_channel")
 
     mock_otel = mock.Mock()
     mock_otel_grpc = mock_otel.instrumentation.grpc
-    mock_interceptor = mock.Mock()
+    mock_interceptor = mock.Mock(name="otel_interceptor")
 
     mock_otel_grpc.client_interceptor.return_value = mock_interceptor
-    mock_otel_grpc.intercept_channel.return_value = mock_intercepted_channel
+    mock_otel_grpc.intercept_channel.return_value = mock_wrapped_channel
 
     monkeypatch.setitem(sys.modules, "opentelemetry", mock_otel)
     monkeypatch.setitem(
@@ -138,33 +212,38 @@ def test_apply_otel_capabilities_to_channel_enabled_via_config(monkeypatch):
         sys.modules, "opentelemetry.instrumentation.grpc", mock_otel_grpc
     )
 
-    result = _observability.apply_otel_capabilities_to_channel(
-        mock_channel, client_options=options
-    )
+    wrapper = _observability.get_otel_channel_wrapper(client_options=options)
+    assert callable(wrapper)
 
-    assert result is mock_intercepted_channel
     mock_otel_grpc.client_interceptor.assert_called_once_with(
         tracer_provider=mock_tracer_provider
     )
+
+    result = wrapper(mock_raw_channel)
+    assert result is mock_wrapped_channel
     mock_otel_grpc.intercept_channel.assert_called_once_with(
-        mock_channel, mock_interceptor
+        mock_raw_channel, mock_interceptor
     )
 
 
-def test_apply_otel_capabilities_to_channel_enabled_via_dict_config(monkeypatch):
-    # Tracing enabled via dict config
+def test_get_otel_channel_wrapper_with_apply_channel_wrappers(monkeypatch):
+    """Proves that get_otel_channel_wrapper integrates seamlessly into apply_channel_wrappers."""
+    pytest.importorskip("grpc")
+    from google.api_core import grpc_helpers
+
+    monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", "true")
     mock_tracer_provider = object()
-    options = {"tracer_provider": mock_tracer_provider}
+    options = ClientOptions(tracer_provider=mock_tracer_provider)
 
-    mock_channel = mock.Mock()
-    mock_intercepted_channel = mock.Mock()
+    mock_raw_channel = mock.Mock(name="raw_channel")
+    mock_wrapped_channel = mock.Mock(name="wrapped_channel")
 
     mock_otel = mock.Mock()
     mock_otel_grpc = mock_otel.instrumentation.grpc
-    mock_interceptor = mock.Mock()
+    mock_interceptor = mock.Mock(name="otel_interceptor")
 
     mock_otel_grpc.client_interceptor.return_value = mock_interceptor
-    mock_otel_grpc.intercept_channel.return_value = mock_intercepted_channel
+    mock_otel_grpc.intercept_channel.return_value = mock_wrapped_channel
 
     monkeypatch.setitem(sys.modules, "opentelemetry", mock_otel)
     monkeypatch.setitem(
@@ -174,14 +253,50 @@ def test_apply_otel_capabilities_to_channel_enabled_via_dict_config(monkeypatch)
         sys.modules, "opentelemetry.instrumentation.grpc", mock_otel_grpc
     )
 
-    result = _observability.apply_otel_capabilities_to_channel(
-        mock_channel, client_options=options
-    )
+    otel_wrapper = _observability.get_otel_channel_wrapper(client_options=options)
+    assert callable(otel_wrapper)
 
-    assert result is mock_intercepted_channel
-    mock_otel_grpc.client_interceptor.assert_called_once_with(
-        tracer_provider=mock_tracer_provider
+    result = grpc_helpers.apply_channel_wrappers(
+        mock_raw_channel, wrappers=[otel_wrapper]
     )
+    assert result is mock_wrapped_channel
     mock_otel_grpc.intercept_channel.assert_called_once_with(
-        mock_channel, mock_interceptor
+        mock_raw_channel, mock_interceptor
+    )
+
+
+def test_get_otel_async_interceptor_disabled(monkeypatch):
+    monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", "false")
+    assert _observability.get_otel_async_interceptor() is None
+
+
+def test_get_otel_async_interceptor_otel_missing(monkeypatch):
+    monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", "true")
+    monkeypatch.setitem(sys.modules, "opentelemetry.instrumentation.grpc", None)
+    assert _observability.get_otel_async_interceptor() is None
+
+
+def test_get_otel_async_interceptor_enabled(monkeypatch):
+    monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", "true")
+    mock_tracer_provider = object()
+    options = ClientOptions(tracer_provider=mock_tracer_provider)
+
+    mock_async_interceptors = [mock.Mock(name="otel_async_interceptor")]
+
+    mock_otel = mock.Mock()
+    mock_otel_grpc = mock_otel.instrumentation.grpc
+    mock_otel_grpc.aio_client_interceptors.return_value = mock_async_interceptors
+
+    monkeypatch.setitem(sys.modules, "opentelemetry", mock_otel)
+    monkeypatch.setitem(
+        sys.modules, "opentelemetry.instrumentation", mock_otel.instrumentation
+    )
+    monkeypatch.setitem(
+        sys.modules, "opentelemetry.instrumentation.grpc", mock_otel_grpc
+    )
+
+    result = _observability.get_otel_async_interceptor(client_options=options)
+    assert result is mock_async_interceptors
+    mock_otel_grpc.aio_client_interceptors.assert_called_once_with(
+        tracer_provider=mock_tracer_provider
     )

From b907213055b895316716cea02348f572cf17816b Mon Sep 17 00:00:00 2001
From: chalmer lowe <chalmerlowe@google.com>
Date: Tue, 1 Sep 2026 13:14:07 -0400
Subject: [PATCH 12/16] feat(api-core): rename channel wrapper terminology to
 interceptor

---
 .../google/api_core/_observability.py         | 22 +++----
 .../google/api_core/grpc_helpers.py           | 49 ++++++++--------
 .../tests/unit/test_grpc_helpers.py           | 57 ++++++++++---------
 .../tests/unit/test_observability.py          | 29 +++++-----
 4 files changed, 78 insertions(+), 79 deletions(-)

diff --git a/packages/google-api-core/google/api_core/_observability.py b/packages/google-api-core/google/api_core/_observability.py
index 16f9917e736d..414ef4c7208b 100644
--- a/packages/google-api-core/google/api_core/_observability.py
+++ b/packages/google-api-core/google/api_core/_observability.py
@@ -22,9 +22,9 @@
 from google.api_core.client_options import ClientOptions
 
 if TYPE_CHECKING:
-    from google.api_core.grpc_helpers import ChannelWrapperCallable
+    from google.api_core.grpc_helpers import ClientInterceptorCallable
 else:
-    ChannelWrapperCallable = Callable[[Any], Any]
+    ClientInterceptorCallable = Callable[[Any], Any]
 
 _TRACER_PROVIDER = "tracer_provider"
 
@@ -67,7 +67,7 @@ def _get_otel_interceptor(
 
     Args:
         client_options: The client options object or dictionary.
-        is_async: If True, returns an async interceptor (`aio_client_interceptor`),
+        is_async: If True, returns an async interceptor (`aio_client_interceptors`),
             otherwise returns a sync interceptor (`client_interceptor`).
 
     Returns:
@@ -86,17 +86,17 @@ def _get_otel_interceptor(
     return otel_grpc.client_interceptor(tracer_provider=tracer_provider)
 
 
-def get_otel_channel_wrapper(
+def get_otel_interceptor(
     client_options: ClientOptions | dict[str, Any] | None = None,
-) -> ChannelWrapperCallable | None:
-    """Returns a channel wrapper callable that wraps a sync gRPC channel with OpenTelemetry tracing.
+) -> ClientInterceptorCallable | None:
+    """Returns an interceptor callable that wraps a sync gRPC channel with OpenTelemetry tracing.
 
     Args:
         client_options: The client options object or dictionary used for feature gating
             and extracting the tracer provider.
 
     Returns:
-        Optional[ChannelWrapperCallable]: A channel-wrapping callable if OpenTelemetry
+        Optional[ClientInterceptorCallable]: An interceptor callable if OpenTelemetry
             tracing is enabled and installed, None otherwise.
     """
     if not is_otel_capabilities_enabled(client_options):
@@ -106,23 +106,23 @@ def get_otel_channel_wrapper(
 
     interceptor = _get_otel_interceptor(client_options, is_async=False)
 
-    def channel_wrapper(channel: Any) -> Any:
+    def otel_interceptor(channel: Any) -> Any:
         return otel_grpc.intercept_channel(channel, interceptor)
 
-    return channel_wrapper
+    return otel_interceptor
 
 
 def get_otel_async_interceptor(
     client_options: ClientOptions | dict[str, Any] | None = None,
 ) -> Any | None:
-    """Returns an async gRPC client interceptor for OpenTelemetry tracing.
+    """Returns async gRPC client interceptors for OpenTelemetry tracing.
 
     Args:
         client_options: The client options object or dictionary used for feature gating
             and extracting the tracer provider.
 
     Returns:
-        Optional[Any]: An instantiated OpenTelemetry async client interceptor
+        Optional[Any]: Instantiated OpenTelemetry async client interceptors
             if tracing is enabled and installed, None otherwise.
     """
     if not is_otel_capabilities_enabled(client_options):
diff --git a/packages/google-api-core/google/api_core/grpc_helpers.py b/packages/google-api-core/google/api_core/grpc_helpers.py
index 13f89c64a13f..11071b561bdb 100644
--- a/packages/google-api-core/google/api_core/grpc_helpers.py
+++ b/packages/google-api-core/google/api_core/grpc_helpers.py
@@ -35,7 +35,6 @@
 import google.auth.transport.requests
 import google.protobuf
 import grpc
-
 from google.api_core import exceptions, general_helpers
 
 # The list of gRPC Callable interfaces that return iterators.
@@ -55,11 +54,11 @@
 # Runtime tuple of gRPC client interceptor base classes for isinstance checks
 _CLIENT_INTERCEPTOR_CLASSES = get_args(ClientInterceptor)
 
-# Type alias representing a channel-wrapping callable
-ChannelWrapperCallable: TypeAlias = Callable[[grpc.Channel], grpc.Channel]
+# Type alias representing a channel-intercepting callable
+ClientInterceptorCallable: TypeAlias = Callable[[grpc.Channel], grpc.Channel]
 
-# Generic type alias representing any channel wrapper (interceptor or callable)
-ChannelWrapper: TypeAlias = ClientInterceptor | ChannelWrapperCallable
+# Generic type alias representing any client interceptor (standard interceptor or callable)
+ClientInterceptorType: TypeAlias = ClientInterceptor | ClientInterceptorCallable
 
 
 def _patch_callable_name(callable_):
@@ -446,43 +445,43 @@ def _modify_target_for_direct_path(target: str) -> str:
     return target
 
 
-def apply_channel_wrappers(
+def apply_channel_interceptors(
     channel: grpc.Channel,
-    wrappers: Sequence[ChannelWrapper] | None = None,
+    interceptors: Sequence[ClientInterceptorType] | None = None,
 ) -> grpc.Channel:
-    """Applies channel wrappers (client interceptors or channel-wrapping callables) to a gRPC channel.
+    """Applies client interceptors or channel-intercepting callables to a gRPC channel.
 
-    Executes in reverse order so the first wrapper in the sequence becomes the
-    outermost layer on outbound requests and the innermost layer on inbound responses.
+    Executes in reverse order so the first interceptor in the sequence becomes the
+    outermost layer on outbound requests and the innermost layer on inbound responses,
+    aligning with the behavior of ``grpc.intercept_channel``.
 
     Args:
-        channel (grpc.Channel): The channel to wrap.
-        wrappers (Optional[Sequence[ChannelWrapper]]):
-            An optional sequence of client interceptors or channel-wrapping
-            callables to apply.
+        channel (grpc.Channel): The channel to intercept.
+        interceptors (Optional[Sequence[Union[ClientInterceptor, Callable[[grpc.Channel], grpc.Channel]]]]):
+            Additional interceptors (or callables that apply interceptors) to apply to the gRPC channel.
 
     Returns:
-        grpc.Channel: The wrapped channel, or the original channel if no
-            wrappers were provided.
+        grpc.Channel: The intercepted channel, or the original channel if no
+            interceptors were provided.
 
     Raises:
-        TypeError: If an item in ``wrappers`` is neither a gRPC ClientInterceptor
+        TypeError: If an item in ``interceptors`` is neither a gRPC ClientInterceptor
             nor a Callable[[Channel], Channel].
     """
-    if not wrappers:
+    if not interceptors:
         return channel
 
     modified_channel = channel
     # Reverse the inputs to align with the behavior of grpc.create_channel(*interceptors)
-    for wrapper in reversed(list(wrappers)):
-        if isinstance(wrapper, _CLIENT_INTERCEPTOR_CLASSES):
-            modified_channel = grpc.intercept_channel(modified_channel, wrapper)
-        elif callable(wrapper):
-            wrapper_callable = cast(ChannelWrapperCallable, wrapper)
-            modified_channel = wrapper_callable(modified_channel)
+    for interceptor in reversed(list(interceptors)):
+        if isinstance(interceptor, _CLIENT_INTERCEPTOR_CLASSES):
+            modified_channel = grpc.intercept_channel(modified_channel, interceptor)
+        elif callable(interceptor):
+            interceptor_callable = cast(ClientInterceptorCallable, interceptor)
+            modified_channel = interceptor_callable(modified_channel)
         else:
             raise TypeError(
-                f"Expected ChannelWrapper (ClientInterceptor or Callable[[Channel], Channel]), got {type(wrapper).__name__}"
+                f"Expected ClientInterceptor or Callable[[Channel], Channel], got {type(interceptor).__name__}"
             )
 
     return modified_channel
diff --git a/packages/google-api-core/tests/unit/test_grpc_helpers.py b/packages/google-api-core/tests/unit/test_grpc_helpers.py
index 4367552651d7..f28ebe40090e 100644
--- a/packages/google-api-core/tests/unit/test_grpc_helpers.py
+++ b/packages/google-api-core/tests/unit/test_grpc_helpers.py
@@ -24,9 +24,8 @@
     pytest.skip("No GRPC", allow_module_level=True)
 
 import google.auth.credentials
-from google.longrunning import operations_pb2
-
 from google.api_core import exceptions, grpc_helpers
+from google.longrunning import operations_pb2
 
 
 def test__patch_callable_name():
@@ -934,15 +933,17 @@ def test_close(self):
         assert channel.close() is None
 
 
-@pytest.mark.parametrize("falsy_wrappers", [None, [], ()])
-def test_apply_channel_wrappers_passthrough(falsy_wrappers):
-    """Verify that falsy or empty wrapper sequences return the channel unmodified."""
+@pytest.mark.parametrize("falsy_interceptors", [None, [], ()])
+def test_apply_channel_interceptors_passthrough(falsy_interceptors):
+    """Verify that falsy or empty interceptor sequences return the channel unmodified."""
     mock_base_channel = mock.Mock(name="base_channel")
-    result = grpc_helpers.apply_channel_wrappers(mock_base_channel, falsy_wrappers)
+    result = grpc_helpers.apply_channel_interceptors(
+        mock_base_channel, falsy_interceptors
+    )
     assert result is mock_base_channel
 
 
-def test_apply_channel_wrappers_grpc_client_interceptors():
+def test_apply_channel_interceptors_grpc_client_interceptors():
     """Verify that standard gRPC ClientInterceptor instances are applied via grpc.intercept_channel."""
 
     class DummyUnaryInterceptor(grpc.UnaryUnaryClientInterceptor):
@@ -966,7 +967,7 @@ def intercept_stream_stream(
         "grpc.intercept_channel",
         side_effect=[mock_chan_after_i2, mock_chan_after_i1],
     ) as mock_intercept:
-        result = grpc_helpers.apply_channel_wrappers(
+        result = grpc_helpers.apply_channel_interceptors(
             mock_base_channel, [interceptor1, interceptor2]
         )
 
@@ -981,27 +982,27 @@ def intercept_stream_stream(
     )
 
 
-def test_apply_channel_wrappers_callables():
-    """Verify that channel-wrapping callables Callable[[Channel], Channel] are invoked in sequence."""
+def test_apply_channel_interceptors_callables():
+    """Verify that channel-intercepting callables Callable[[Channel], Channel] are invoked in sequence."""
     mock_base_channel = mock.Mock(name="base_channel")
     mock_chan_1 = mock.Mock(name="chan_1")
     mock_chan_2 = mock.Mock(name="chan_2")
 
-    wrapper1 = mock.Mock(side_effect=lambda ch: mock_chan_2)
-    wrapper2 = mock.Mock(side_effect=lambda ch: mock_chan_1)
+    interceptor1 = mock.Mock(side_effect=lambda ch: mock_chan_2)
+    interceptor2 = mock.Mock(side_effect=lambda ch: mock_chan_1)
 
-    result = grpc_helpers.apply_channel_wrappers(
-        mock_base_channel, [wrapper1, wrapper2]
+    result = grpc_helpers.apply_channel_interceptors(
+        mock_base_channel, [interceptor1, interceptor2]
     )
 
     assert result is mock_chan_2
-    # Executed in reverse order: wrapper2 runs first on base channel, then wrapper1
-    wrapper2.assert_called_once_with(mock_base_channel)
-    wrapper1.assert_called_once_with(mock_chan_1)
+    # Executed in reverse order: interceptor2 runs first on base channel, then interceptor1
+    interceptor2.assert_called_once_with(mock_base_channel)
+    interceptor1.assert_called_once_with(mock_chan_1)
 
 
-def test_apply_channel_wrappers_interspersed():
-    """Verify that a mixed sequence of gRPC interceptors and channel wrapper callables are applied."""
+def test_apply_channel_interceptors_interspersed():
+    """Verify that a mixed sequence of gRPC interceptors and interceptor callables are applied."""
 
     class DummyUnaryInterceptor(grpc.UnaryUnaryClientInterceptor):
         def intercept_unary_unary(self, continuation, client_call_details, request):
@@ -1009,25 +1010,25 @@ def intercept_unary_unary(self, continuation, client_call_details, request):
 
     interceptor = DummyUnaryInterceptor()
     mock_base_channel = mock.Mock(name="base_channel")
-    mock_chan_after_wrapper = mock.Mock(name="chan_after_wrapper")
+    mock_chan_after_callable = mock.Mock(name="chan_after_callable")
     mock_chan_after_interceptor = mock.Mock(name="chan_after_interceptor")
 
-    wrapper = mock.Mock(return_value=mock_chan_after_wrapper)
+    interceptor_callable = mock.Mock(return_value=mock_chan_after_callable)
 
     with mock.patch(
         "grpc.intercept_channel", return_value=mock_chan_after_interceptor
     ) as mock_intercept:
-        result = grpc_helpers.apply_channel_wrappers(
-            mock_base_channel, [interceptor, wrapper]
+        result = grpc_helpers.apply_channel_interceptors(
+            mock_base_channel, [interceptor, interceptor_callable]
         )
 
     assert result is mock_chan_after_interceptor
-    wrapper.assert_called_once_with(mock_base_channel)
-    mock_intercept.assert_called_once_with(mock_chan_after_wrapper, interceptor)
+    interceptor_callable.assert_called_once_with(mock_base_channel)
+    mock_intercept.assert_called_once_with(mock_chan_after_callable, interceptor)
 
 
-def test_apply_channel_wrappers_invalid_type_raises():
+def test_apply_channel_interceptors_invalid_type_raises():
     """Verify that passing an invalid object that is neither an interceptor nor callable raises TypeError."""
     mock_base_channel = mock.Mock(name="base_channel")
-    with pytest.raises(TypeError, match="Expected ChannelWrapper"):
-        grpc_helpers.apply_channel_wrappers(mock_base_channel, [12345])
+    with pytest.raises(TypeError, match="Expected ClientInterceptor or Callable"):
+        grpc_helpers.apply_channel_interceptors(mock_base_channel, [12345])
diff --git a/packages/google-api-core/tests/unit/test_observability.py b/packages/google-api-core/tests/unit/test_observability.py
index 6b2d2fc7005f..6f3eb34b7cc4 100644
--- a/packages/google-api-core/tests/unit/test_observability.py
+++ b/packages/google-api-core/tests/unit/test_observability.py
@@ -16,7 +16,6 @@
 from unittest import mock
 
 import pytest
-
 from google.api_core import _observability
 from google.api_core._feature_gating_helpers import FeatureGatingError
 from google.api_core.client_options import ClientOptions
@@ -178,18 +177,18 @@ def test_get_otel_interceptor_async(monkeypatch):
     )
 
 
-def test_get_otel_channel_wrapper_disabled(monkeypatch):
+def test_get_otel_interceptor_disabled(monkeypatch):
     monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", "false")
-    assert _observability.get_otel_channel_wrapper() is None
+    assert _observability.get_otel_interceptor() is None
 
 
-def test_get_otel_channel_wrapper_otel_missing(monkeypatch):
+def test_get_otel_interceptor_otel_missing(monkeypatch):
     monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", "true")
     monkeypatch.setitem(sys.modules, "opentelemetry.instrumentation.grpc", None)
-    assert _observability.get_otel_channel_wrapper() is None
+    assert _observability.get_otel_interceptor() is None
 
 
-def test_get_otel_channel_wrapper_enabled(monkeypatch):
+def test_get_otel_interceptor_enabled(monkeypatch):
     monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", "true")
     mock_tracer_provider = object()
     options = ClientOptions(tracer_provider=mock_tracer_provider)
@@ -212,22 +211,22 @@ def test_get_otel_channel_wrapper_enabled(monkeypatch):
         sys.modules, "opentelemetry.instrumentation.grpc", mock_otel_grpc
     )
 
-    wrapper = _observability.get_otel_channel_wrapper(client_options=options)
-    assert callable(wrapper)
+    interceptor = _observability.get_otel_interceptor(client_options=options)
+    assert callable(interceptor)
 
     mock_otel_grpc.client_interceptor.assert_called_once_with(
         tracer_provider=mock_tracer_provider
     )
 
-    result = wrapper(mock_raw_channel)
+    result = interceptor(mock_raw_channel)
     assert result is mock_wrapped_channel
     mock_otel_grpc.intercept_channel.assert_called_once_with(
         mock_raw_channel, mock_interceptor
     )
 
 
-def test_get_otel_channel_wrapper_with_apply_channel_wrappers(monkeypatch):
-    """Proves that get_otel_channel_wrapper integrates seamlessly into apply_channel_wrappers."""
+def test_get_otel_interceptor_with_apply_channel_interceptors(monkeypatch):
+    """Proves that get_otel_interceptor integrates seamlessly into apply_channel_interceptors."""
     pytest.importorskip("grpc")
     from google.api_core import grpc_helpers
 
@@ -253,11 +252,11 @@ def test_get_otel_channel_wrapper_with_apply_channel_wrappers(monkeypatch):
         sys.modules, "opentelemetry.instrumentation.grpc", mock_otel_grpc
     )
 
-    otel_wrapper = _observability.get_otel_channel_wrapper(client_options=options)
-    assert callable(otel_wrapper)
+    otel_interceptor = _observability.get_otel_interceptor(client_options=options)
+    assert callable(otel_interceptor)
 
-    result = grpc_helpers.apply_channel_wrappers(
-        mock_raw_channel, wrappers=[otel_wrapper]
+    result = grpc_helpers.apply_channel_interceptors(
+        mock_raw_channel, interceptors=[otel_interceptor]
     )
     assert result is mock_wrapped_channel
     mock_otel_grpc.intercept_channel.assert_called_once_with(

From ceb0be89a8402e493784b74b2cba6669f1701715 Mon Sep 17 00:00:00 2001
From: chalmer lowe <chalmerlowe@google.com>
Date: Tue, 1 Sep 2026 14:57:33 -0400
Subject: [PATCH 13/16] style(api-core): format imports with isort via ruff

---
 packages/google-api-core/google/api_core/grpc_helpers.py  | 1 +
 packages/google-api-core/tests/unit/test_grpc_helpers.py  | 3 ++-
 packages/google-api-core/tests/unit/test_observability.py | 1 +
 3 files changed, 4 insertions(+), 1 deletion(-)

diff --git a/packages/google-api-core/google/api_core/grpc_helpers.py b/packages/google-api-core/google/api_core/grpc_helpers.py
index 11071b561bdb..b1b4e95c2031 100644
--- a/packages/google-api-core/google/api_core/grpc_helpers.py
+++ b/packages/google-api-core/google/api_core/grpc_helpers.py
@@ -35,6 +35,7 @@
 import google.auth.transport.requests
 import google.protobuf
 import grpc
+
 from google.api_core import exceptions, general_helpers
 
 # The list of gRPC Callable interfaces that return iterators.
diff --git a/packages/google-api-core/tests/unit/test_grpc_helpers.py b/packages/google-api-core/tests/unit/test_grpc_helpers.py
index f28ebe40090e..1a1153a5ebd0 100644
--- a/packages/google-api-core/tests/unit/test_grpc_helpers.py
+++ b/packages/google-api-core/tests/unit/test_grpc_helpers.py
@@ -24,9 +24,10 @@
     pytest.skip("No GRPC", allow_module_level=True)
 
 import google.auth.credentials
-from google.api_core import exceptions, grpc_helpers
 from google.longrunning import operations_pb2
 
+from google.api_core import exceptions, grpc_helpers
+
 
 def test__patch_callable_name():
     callable = mock.Mock(spec=["__class__"])
diff --git a/packages/google-api-core/tests/unit/test_observability.py b/packages/google-api-core/tests/unit/test_observability.py
index 6f3eb34b7cc4..a42458604036 100644
--- a/packages/google-api-core/tests/unit/test_observability.py
+++ b/packages/google-api-core/tests/unit/test_observability.py
@@ -16,6 +16,7 @@
 from unittest import mock
 
 import pytest
+
 from google.api_core import _observability
 from google.api_core._feature_gating_helpers import FeatureGatingError
 from google.api_core.client_options import ClientOptions

From 49b2824d14d94266b7e2b6c47c032ed98c059dbb Mon Sep 17 00:00:00 2001
From: chalmer lowe <chalmerlowe@google.com>
Date: Wed, 2 Sep 2026 05:53:24 -0400
Subject: [PATCH 14/16] refactor(api-core): simplify interceptor type
 signatures and use concrete channel types

---
 .../google/api_core/_observability.py             | 13 +++++++------
 .../google/api_core/grpc_helpers.py               | 15 ++++++---------
 2 files changed, 13 insertions(+), 15 deletions(-)

diff --git a/packages/google-api-core/google/api_core/_observability.py b/packages/google-api-core/google/api_core/_observability.py
index 414ef4c7208b..4d53b70a9b2c 100644
--- a/packages/google-api-core/google/api_core/_observability.py
+++ b/packages/google-api-core/google/api_core/_observability.py
@@ -16,15 +16,16 @@
 
 """OpenTelemetry helpers for resolving and instantiating interceptors."""
 
+from __future__ import annotations
+
 from typing import TYPE_CHECKING, Any, Callable
 
 from google.api_core import _feature_gating_helpers
 from google.api_core.client_options import ClientOptions
 
 if TYPE_CHECKING:
-    from google.api_core.grpc_helpers import ClientInterceptorCallable
-else:
-    ClientInterceptorCallable = Callable[[Any], Any]
+    # flake8: grpc is imported only for static analysis and type annotations
+    import grpc  # noqa: F401
 
 _TRACER_PROVIDER = "tracer_provider"
 
@@ -88,7 +89,7 @@ def _get_otel_interceptor(
 
 def get_otel_interceptor(
     client_options: ClientOptions | dict[str, Any] | None = None,
-) -> ClientInterceptorCallable | None:
+) -> Callable[[grpc.Channel], grpc.Channel] | None:
     """Returns an interceptor callable that wraps a sync gRPC channel with OpenTelemetry tracing.
 
     Args:
@@ -96,7 +97,7 @@ def get_otel_interceptor(
             and extracting the tracer provider.
 
     Returns:
-        Optional[ClientInterceptorCallable]: An interceptor callable if OpenTelemetry
+        Optional[Callable[[grpc.Channel], grpc.Channel]]: An interceptor callable if OpenTelemetry
             tracing is enabled and installed, None otherwise.
     """
     if not is_otel_capabilities_enabled(client_options):
@@ -106,7 +107,7 @@ def get_otel_interceptor(
 
     interceptor = _get_otel_interceptor(client_options, is_async=False)
 
-    def otel_interceptor(channel: Any) -> Any:
+    def otel_interceptor(channel: grpc.Channel) -> grpc.Channel:
         return otel_grpc.intercept_channel(channel, interceptor)
 
     return otel_interceptor
diff --git a/packages/google-api-core/google/api_core/grpc_helpers.py b/packages/google-api-core/google/api_core/grpc_helpers.py
index b1b4e95c2031..9c1058409d5a 100644
--- a/packages/google-api-core/google/api_core/grpc_helpers.py
+++ b/packages/google-api-core/google/api_core/grpc_helpers.py
@@ -35,7 +35,6 @@
 import google.auth.transport.requests
 import google.protobuf
 import grpc
-
 from google.api_core import exceptions, general_helpers
 
 # The list of gRPC Callable interfaces that return iterators.
@@ -55,12 +54,6 @@
 # Runtime tuple of gRPC client interceptor base classes for isinstance checks
 _CLIENT_INTERCEPTOR_CLASSES = get_args(ClientInterceptor)
 
-# Type alias representing a channel-intercepting callable
-ClientInterceptorCallable: TypeAlias = Callable[[grpc.Channel], grpc.Channel]
-
-# Generic type alias representing any client interceptor (standard interceptor or callable)
-ClientInterceptorType: TypeAlias = ClientInterceptor | ClientInterceptorCallable
-
 
 def _patch_callable_name(callable_):
     """Fix-up gRPC callable attributes.
@@ -448,7 +441,9 @@ def _modify_target_for_direct_path(target: str) -> str:
 
 def apply_channel_interceptors(
     channel: grpc.Channel,
-    interceptors: Sequence[ClientInterceptorType] | None = None,
+    interceptors: (
+        Sequence[ClientInterceptor | Callable[[grpc.Channel], grpc.Channel]] | None
+    ) = None,
 ) -> grpc.Channel:
     """Applies client interceptors or channel-intercepting callables to a gRPC channel.
 
@@ -478,7 +473,9 @@ def apply_channel_interceptors(
         if isinstance(interceptor, _CLIENT_INTERCEPTOR_CLASSES):
             modified_channel = grpc.intercept_channel(modified_channel, interceptor)
         elif callable(interceptor):
-            interceptor_callable = cast(ClientInterceptorCallable, interceptor)
+            interceptor_callable = cast(
+                Callable[[grpc.Channel], grpc.Channel], interceptor
+            )
             modified_channel = interceptor_callable(modified_channel)
         else:
             raise TypeError(

From abd9847192ed6c554e4a8243fc40c69531b8e314 Mon Sep 17 00:00:00 2001
From: chalmer lowe <chalmerlowe@google.com>
Date: Wed, 2 Sep 2026 06:04:41 -0400
Subject: [PATCH 15/16] style(api-core): separate first-party imports in
 grpc_helpers

---
 packages/google-api-core/google/api_core/grpc_helpers.py | 1 +
 1 file changed, 1 insertion(+)

diff --git a/packages/google-api-core/google/api_core/grpc_helpers.py b/packages/google-api-core/google/api_core/grpc_helpers.py
index 9c1058409d5a..71ed528f2795 100644
--- a/packages/google-api-core/google/api_core/grpc_helpers.py
+++ b/packages/google-api-core/google/api_core/grpc_helpers.py
@@ -35,6 +35,7 @@
 import google.auth.transport.requests
 import google.protobuf
 import grpc
+
 from google.api_core import exceptions, general_helpers
 
 # The list of gRPC Callable interfaces that return iterators.

From a3b25e0f1519f4e238fbc9389b1ebb53a518b0f3 Mon Sep 17 00:00:00 2001
From: chalmer lowe <chalmerlowe@google.com>
Date: Wed, 2 Sep 2026 06:57:31 -0400
Subject: [PATCH 16/16] refactor(api-core): type get_otel_async_interceptor
 with grpc.aio.ClientInterceptor

---
 .../google-api-core/google/api_core/_observability.py     | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git a/packages/google-api-core/google/api_core/_observability.py b/packages/google-api-core/google/api_core/_observability.py
index 4d53b70a9b2c..4852e4de7b83 100644
--- a/packages/google-api-core/google/api_core/_observability.py
+++ b/packages/google-api-core/google/api_core/_observability.py
@@ -18,7 +18,7 @@
 
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Any, Callable
+from typing import TYPE_CHECKING, Any, Callable, Sequence
 
 from google.api_core import _feature_gating_helpers
 from google.api_core.client_options import ClientOptions
@@ -115,7 +115,7 @@ def otel_interceptor(channel: grpc.Channel) -> grpc.Channel:
 
 def get_otel_async_interceptor(
     client_options: ClientOptions | dict[str, Any] | None = None,
-) -> Any | None:
+) -> Sequence[grpc.aio.ClientInterceptor] | None:
     """Returns async gRPC client interceptors for OpenTelemetry tracing.
 
     Args:
@@ -123,8 +123,8 @@ def get_otel_async_interceptor(
             and extracting the tracer provider.
 
     Returns:
-        Optional[Any]: Instantiated OpenTelemetry async client interceptors
-            if tracing is enabled and installed, None otherwise.
+        Optional[Sequence[grpc.aio.ClientInterceptor]]: Instantiated OpenTelemetry async
+            client interceptors if tracing is enabled and installed, None otherwise.
     """
     if not is_otel_capabilities_enabled(client_options):
         return None