[DRAFT] fix: deduplicate x-goog-api-client headers

packages/google-api-core/google/api_core/gapic_v1/method.py

@@ -57,6 +57,35 @@ def _apply_decorators(func, decorators):
     return func
 
 
+def _extract_metrics_header(metadata):
+    """Extract x-google-api-client header from metadata list.
+
+    Args:
+        metadata (Sequence[Tuple[str, str]]): The metadata to extract from.
+
+    Returns:
+        Tuple[List[Tuple[str, str]], List[str]]: A tuple containing:
+            - A list of remaining metadata tuples.
+            - A list of metrics header values found.
+    """
+    if not metadata:
+        return [], []
+
+    for i, (key, val) in enumerate(metadata):
+        if key == client_info.METRICS_METADATA_KEY:
+            # Key located. Check the rest of the list for duplicate entries
+            arbitrary_metadata = list(metadata[:i])
+            metric_values = [val]
+            for k, v in metadata[i + 1 :]:
+                if k == client_info.METRICS_METADATA_KEY:
+                    metric_values.append(v)
+                else:
+                    arbitrary_metadata.append((k, v))
+            return arbitrary_metadata, metric_values
+    # No key found
+    return list(metadata), []
+
+
 class _GapicCallable(object):
     """Callable that applies retry, timeout, and metadata logic.
 
@@ -91,6 +120,9 @@ def __init__(
         self._timeout = timeout
         self._compression = compression
         self._metadata = metadata
+        # Pre-extract the x-goog-api-client header from the initialized metadata.
+        self._arbitrary_metadata, metric_values = _extract_metrics_header(metadata)
+        self._metrics_values = " ".join(metric_values) if metric_values else ""
 
     def __call__(
         self, *args, timeout=DEFAULT, retry=DEFAULT, compression=DEFAULT, **kwargs
@@ -114,14 +146,26 @@ def __call__(
 
         # Add the user agent metadata to the call.
         if self._metadata is not None:
-            metadata = kwargs.get("metadata", [])
-            # Due to the nature of invocation, None should be treated the same
-            # as not specified.
-            if metadata is None:
-                metadata = []
-            metadata = list(metadata)
-            metadata.extend(self._metadata)
-            kwargs["metadata"] = metadata
+            metadata = kwargs.get("metadata")
+            if not metadata:
+                if self._metrics_values:
+                    kwargs["metadata"] = [
+                        (client_info.METRICS_METADATA_KEY, self._metrics_values)
+                    ] + self._arbitrary_metadata
+                else:
+                    kwargs["metadata"] = self._arbitrary_metadata
+            else:
+                # Merge user-supplied metadata with library-supplied metadata.
+                merged_metadata, api_client_values = _extract_metrics_header(metadata)
+                if self._metrics_values:
+                    api_client_values.append(self._metrics_values)
+                if api_client_values:
+                    merged_metadata.append(
+                        (client_info.METRICS_METADATA_KEY, " ".join(api_client_values))
+                    )
+                merged_metadata.extend(self._arbitrary_metadata)
+                kwargs["metadata"] = merged_metadata
+
         if self._compression is not None:
             kwargs["compression"] = compression
 

packages/google-api-core/google/api_core/grpc_helpers.py

@@ -254,11 +254,20 @@ def _create_composite_credentials(
     request = google.auth.transport.requests.Request()
 
     # Create the metadata plugin for inserting the authorization header.
-    metadata_plugin = google.auth.transport.grpc.AuthMetadataPlugin(
-        credentials,
-        request,
-        default_host=default_host,
-    )
+    try:
+        metadata_plugin = google.auth.transport.grpc.AuthMetadataPlugin(
+            credentials,
+            request,
+            default_host=default_host,
+            suppress_metrics_header=True,
+        )
+    except TypeError:
+        # Support older versions of google-auth that do not accept suppress_metrics_header
+        metadata_plugin = google.auth.transport.grpc.AuthMetadataPlugin(
+            credentials,
+            request,
+            default_host=default_host,
+        )
 
     # Create a set of grpc.CallCredentials using the metadata plugin.
     google_auth_credentials = grpc.metadata_call_credentials(metadata_plugin)

packages/google-api-core/tests/unit/gapic/test_method.py

@@ -121,6 +121,47 @@ def test_invoke_wrapped_method_with_metadata_as_none():
     assert len(metadata) == 1
 
 
+def test_invoke_wrapped_method_with_duplicate_x_goog_api_client_metadata():
+    method = mock.Mock(spec=["__call__"])
+
+    # Create a custom ClientInfo with defined properties so we know exactly what is returned
+    client_info = google.api_core.gapic_v1.client_info.ClientInfo(
+        user_agent="custom-user-agent/1.0",
+        python_version="3.14.0",
+        grpc_version="1.76.0",
+        api_core_version="2.29.0",
+    )
+
+    wrapped_method = google.api_core.gapic_v1.method.wrap_method(
+        method, client_info=client_info
+    )
+
+    # Invoke the wrapped method with an explicit user-provided custom header
+    wrapped_method(
+        mock.sentinel.request,
+        metadata=[
+            ("x-goog-api-client", "override-client/2.0"),
+            ("other-header", "value"),
+        ],
+    )
+
+    method.assert_called_once_with(mock.sentinel.request, metadata=mock.ANY)
+    metadata = method.call_args[1]["metadata"]
+
+    # There should only be one "x-goog-api-client" header, containing both values joined by space,
+    # plus the other-header.
+    assert len(metadata) == 2
+    metadata_dict = dict(metadata)
+    assert "other-header" in metadata_dict
+    assert metadata_dict["other-header"] == "value"
+    assert "x-goog-api-client" in metadata_dict
+    # Verify both the user-provided override value and the library system telemetry are merged explicitly
+    assert (
+        metadata_dict["x-goog-api-client"]
+        == "override-client/2.0 custom-user-agent/1.0 gl-python/3.14.0 grpc/1.76.0 gax/2.29.0"
+    )
+
+
 @mock.patch("time.sleep")
 def test_wrap_method_with_default_retry_and_timeout_and_compression(unused_sleep):
     method = mock.Mock(

packages/google-auth/google/auth/transport/grpc.py

@@ -47,16 +47,21 @@ class AuthMetadataPlugin(grpc.AuthMetadataPlugin):
         default_host (Optional[str]): A host like "pubsub.googleapis.com".
             This is used when a self-signed JWT is created from service
             account credentials.
+        suppress_metrics_header (bool): When enabled, ``x-goog-api-client``
+            will be stripped from authorization headers.
     """
 
-    def __init__(self, credentials, request, default_host=None):
+    def __init__(
+        self, credentials, request, default_host=None, *, suppress_metrics_header=False
+    ):
         # pylint: disable=no-value-for-parameter
         # pylint doesn't realize that the super method takes no arguments
         # because this class is the same name as the superclass.
         super(AuthMetadataPlugin, self).__init__()
         self._credentials = credentials
         self._request = request
         self._default_host = default_host
+        self._suppress_metrics_header = suppress_metrics_header
 
     def _get_authorization_headers(self, context):
         """Gets the authorization headers for a request.
@@ -80,6 +85,9 @@ def _get_authorization_headers(self, context):
             self._request, context.method_name, context.service_url, headers
         )
 
+        if self._suppress_metrics_header and "x-goog-api-client" in headers:
+            del headers["x-goog-api-client"]
+
         return list(headers.items())
 
     def __call__(self, context, callback):

packages/google-auth/tests/transport/test_grpc.py

@@ -132,6 +132,35 @@ def test__get_authorization_headers_with_service_account_and_default_host(self):
             "https://{}/".format(default_host)
         )
 
+    def test_suppress_metrics_header(self):
+        credentials = mock.create_autospec(service_account.Credentials)
+
+        # Mock credentials before_request that adds metric and authorization
+        def mock_before_request(request, method, url, headers):
+            headers["x-goog-api-client"] = "foo"
+            headers["authorization"] = "Bearer token"
+
+        credentials.before_request.side_effect = mock_before_request
+        request = mock.create_autospec(transport.Request)
+
+        # By default, suppress_metrics_header=False
+        plugin = google.auth.transport.grpc.AuthMetadataPlugin(credentials, request)
+        context = mock.create_autospec(grpc.AuthMetadataContext, instance=True)
+        context.method_name = "methodName"
+        context.service_url = "https://pubsub.googleapis.com/methodName"
+
+        headers = dict(plugin._get_authorization_headers(context))
+        assert "x-goog-api-client" in headers
+        assert headers["x-goog-api-client"] == "foo"
+
+        # With suppress_metrics_header=True
+        plugin_suppressed = google.auth.transport.grpc.AuthMetadataPlugin(
+            credentials, request, suppress_metrics_header=True
+        )
+        headers_suppressed = dict(plugin_suppressed._get_authorization_headers(context))
+        assert "x-goog-api-client" not in headers_suppressed
+        assert headers_suppressed["authorization"] == "Bearer token"
+
 
 @mock.patch(
     "google.auth.transport._mtls_helper.get_client_ssl_credentials", autospec=True