chalmerlowe · Nov 3, 2026 · Oct 27, 2026 · Oct 30, 2026 · Oct 30, 2026 · Oct 30, 2026
@@ -13,6 +13,7 @@ falling back to the local implementation if not present. #}
 {# TODO(https://github.com/googleapis/google-cloud-python/issues/17883):
 Backfill compatibility functions being removed from the client layer. #}
 
+import inspect
 import os
 import json
 {% if has_auto_populated_fields %}
@@ -99,6 +100,14 @@ ASYNC_WRAP_METHOD_SUPPORTS_KIND = WRAP_METHOD_SUPPORTS_TRACING or hasattr(
     gapic_v1.method_async, "_DEFAULT_ASYNC_TRANSPORT_KIND"
 )
 
+# The `default_host` parameter in gapic_v1.method.wrap_method,
+# gapic_v1.method_async.wrap_method, _observability.get_otel_interceptor, and
+# _observability.apply_otel_async_channel_interceptors allows transports to pass
+# their resolved default host without mutating caller-owned client_options.
+SUPPORTS_DEFAULT_HOST = (
+    "default_host" in inspect.signature(gapic_v1.method.wrap_method).parameters
+)
+
 try:
     from google.api_core import grpc_helpers
 except ImportError:  # pragma: NO COVER

@@ -9,6 +9,7 @@ from typing import {% if service.any_extended_operations_methods %}Any, {% endif
 from {{package_path}} import gapic_version as package_version
 from {{package_path}}._compat import (
     ASYNC_WRAP_METHOD_SUPPORTS_KIND as _ASYNC_WRAP_METHOD_SUPPORTS_KIND,
+    SUPPORTS_DEFAULT_HOST as _SUPPORTS_DEFAULT_HOST,
     WRAP_METHOD_SUPPORTS_TRACING as _WRAP_METHOD_SUPPORTS_TRACING,
 )
 _ASYNC_WRAP_METHOD_SUPPORTS_TRACING = _WRAP_METHOD_SUPPORTS_TRACING
@@ -128,6 +129,8 @@ class {{ service.name }}Transport(abc.ABC):
         self._scopes = scopes
         if not hasattr(self, "_ignore_credentials"):
             self._ignore_credentials: bool = False
+        if not hasattr(self, "_user_provided_channel"):
+            self._user_provided_channel: bool = False
 
         # If no credentials are provided, then determine the appropriate
         # defaults.
@@ -163,6 +166,12 @@ class {{ service.name }}Transport(abc.ABC):
         self._host = host
 
         self._client_options = client_options
+        # If a pre-constructed gRPC channel was provided to the subclass before
+        # super().__init__, self._host is not used to create the channel and
+        # cannot be trusted as the default span endpoint.
+        self._default_host: Optional[str] = (
+            None if self._user_provided_channel else self._host
+        )
         self._wrapped_methods: Dict[Callable, Callable] = {}
 
     @property
@@ -181,6 +190,8 @@ class {{ service.name }}Transport(abc.ABC):
         """
         if _WRAP_METHOD_SUPPORTS_TRACING:
             kwargs["client_options"] = self._client_options
+            if _SUPPORTS_DEFAULT_HOST:
+                kwargs["default_host"] = self._default_host
             try:
                 kind = self.kind
             except NotImplementedError:
@@ -193,7 +204,7 @@ class {{ service.name }}Transport(abc.ABC):
         # - Era 2 (>= 2.29.0, < 2.36.0): `kind` is supported (prevents REST from
         #   falling back to gRPC wrapping), but tracing args are not.
         # (These fallbacks can be removed once google-api-core < 2.36.0 is no longer supported.)
-        for k in ["client_options", "method_name", "is_streaming"]:
+        for k in ["client_options", "method_name", "is_streaming", "default_host"]:
             kwargs.pop(k, None)
 
         try:
@@ -220,6 +231,8 @@ class {{ service.name }}Transport(abc.ABC):
         """
         if _ASYNC_WRAP_METHOD_SUPPORTS_TRACING:
             kwargs["client_options"] = self._client_options
+            if _SUPPORTS_DEFAULT_HOST:
+                kwargs["default_host"] = self._default_host
             try:
                 kind = self.kind
             except NotImplementedError:
@@ -232,7 +245,7 @@ class {{ service.name }}Transport(abc.ABC):
         # - Era 2 (>= 2.29.0, < 2.36.0): `kind` is supported (prevents REST from
         #   falling back to gRPC wrapping), but tracing args are not.
         # (These fallbacks can be removed once google-api-core < 2.36.0 is no longer supported.)
-        for k in ["client_options", "method_name", "is_streaming"]:
+        for k in ["client_options", "method_name", "is_streaming", "default_host"]:
             kwargs.pop(k, None)
 
         try:

@@ -12,7 +12,11 @@ from typing import Callable, Dict, Optional, Sequence, Tuple, Union
 from typing import TYPE_CHECKING
 
 {% set package_path = api.naming.module_namespace|join('.') + "." + api.naming.versioned_module_name %}
-from {{package_path}}._compat import _observability, apply_channel_interceptors
+from {{package_path}}._compat import (
+    _observability,
+    SUPPORTS_DEFAULT_HOST as _SUPPORTS_DEFAULT_HOST,
+    apply_channel_interceptors,
+)
 
 from google.api_core import client_options as client_options_lib
 {% if service.has_resumable_upload_methods %}
@@ -231,6 +235,7 @@ class {{ service.name }}GrpcTransport({{ service.name }}Transport):
             self._ignore_credentials = True
             # If a channel was explicitly provided, set it.
             self._grpc_channel = channel
+            self._user_provided_channel = True
             self._ssl_channel_credentials = None
 
         else:
@@ -287,9 +292,12 @@ class {{ service.name }}GrpcTransport({{ service.name }}Transport):
             )
 
         channel_interceptors = list(interceptors) if interceptors else []
+        _otel_interceptor_kwargs = (
+            {"default_host": self._default_host} if _SUPPORTS_DEFAULT_HOST else {}
+        )
         if (
             _observability is not None
-            and (otel_interceptor := _observability.get_otel_interceptor(self._client_options)) is not None
+            and (otel_interceptor := _observability.get_otel_interceptor(self._client_options, **_otel_interceptor_kwargs)) is not None
             and otel_interceptor not in channel_interceptors
             and not any(getattr(i, "_is_otel_interceptor", None) is True for i in channel_interceptors)
         ):

@@ -13,7 +13,11 @@ from typing import {% if service.has_resumable_upload_methods %}Any, {% endif %}
 from typing import TYPE_CHECKING
 
 {% set package_path = api.naming.module_namespace|join('.') + "." + api.naming.versioned_module_name %}
-from {{package_path}}._compat import _observability, apply_async_channel_interceptors
+from {{package_path}}._compat import (
+    _observability,
+    SUPPORTS_DEFAULT_HOST as _SUPPORTS_DEFAULT_HOST,
+    apply_async_channel_interceptors,
+)
 
 from google.api_core import client_options as client_options_lib
 from google.api_core import gapic_v1
@@ -279,6 +283,7 @@ class {{ service.grpc_asyncio_transport_name }}({{ service.name }}Transport):
             self._ignore_credentials = True
             # If a channel was explicitly provided, set it.
             self._grpc_channel = channel
+            self._user_provided_channel = True
             self._ssl_channel_credentials = None
         else:
             if api_mtls_endpoint:
@@ -349,8 +354,11 @@ class {{ service.grpc_asyncio_transport_name }}({{ service.name }}Transport):
             # NOTE: Coverage tool ignores async interceptors in environments running
             # legacy google-api-core (< 2.36.0) where OpenTelemetry is unavailable.
             # TODO (Issue #18657): Remove once lowest constraints require google-api-core >= v2.43.0.
+            _otel_interceptor_kwargs = (
+                {"default_host": self._default_host} if _SUPPORTS_DEFAULT_HOST else {}
+            )  # pragma: NO COVER
             self._grpc_channel = _observability.apply_otel_async_channel_interceptors(
-                self._grpc_channel, self._client_options
+                self._grpc_channel, self._client_options, **_otel_interceptor_kwargs
             )  # pragma: NO COVER
 
         self._logged_channel = self._grpc_channel