Python: Add User-Agent to google-genai (#13703)

### Motivation and Context

Currently, the Python connectors for Google AI and Vertex AI don't pass
the standard semantic-kernel user agent string. This PR adds the user
agent header so that the Google API calls properly identify Semantic
Kernel.

This aligns with the dotnet implementation, and our (Google DeepMind's)
[best
practices](https://ai.google.dev/gemini-api/docs/partner-integration).

### Description

Sends the `User-Agent` header conditionally in `GoogleAIChatCompletion`,
`GoogleAITextCompletion`, and `GoogleAITextEmbedding`, matching the
OpenAI implementation as much as possible (gated behind telemetry flag,
using `SEMANTIC_KERNEL_USER_AGENT`).

### Contribution Checklist

- [x] The code builds clean without any errors or warnings
- [x] The PR follows the [SK Contribution
Guidelines](https://github.com/microsoft/semantic-kernel/blob/main/CONTRIBUTING.md)
and the [pre-submission formatting
script](https://github.com/microsoft/semantic-kernel/blob/main/CONTRIBUTING.md#development-scripts)
raises no violations
- [x] All unit tests pass, and I have added new tests where possible
- [x] I didn't break anyone 😄.

python/semantic_kernel/connectors/ai/google/google_ai/services/google_ai_base.py

@@ -3,10 +3,12 @@
 from abc import ABC
 from typing import ClassVar
 
-from google.genai import Client
+from google.genai import Client, types
 
 from semantic_kernel.connectors.ai.google.google_ai.google_ai_settings import GoogleAISettings
+from semantic_kernel.const import USER_AGENT
 from semantic_kernel.kernel_pydantic import KernelBaseModel
+from semantic_kernel.utils.telemetry.user_agent import IS_TELEMETRY_ENABLED, SEMANTIC_KERNEL_USER_AGENT
 
 
 class GoogleAIBase(KernelBaseModel, ABC):
@@ -17,3 +19,14 @@ class GoogleAIBase(KernelBaseModel, ABC):
     service_settings: GoogleAISettings
 
     client: Client | None = None
+
+    def _get_http_options(self) -> types.HttpOptionsDict | None:
+        """Get the HTTP options for the Google AI client.
+
+        Returns:
+            The HTTP options dictionary, or None if telemetry is disabled.
+        """
+        if not IS_TELEMETRY_ENABLED:
+            return None
+
+        return types.HttpOptionsDict(headers={USER_AGENT: SEMANTIC_KERNEL_USER_AGENT})

python/semantic_kernel/connectors/ai/google/google_ai/services/google_ai_chat_completion.py

@@ -171,10 +171,14 @@ class GoogleAIChatCompletion(GoogleAIBase, ChatCompletionClientBase):
                 vertexai=True,
                 project=self.service_settings.cloud_project_id,
                 location=self.service_settings.cloud_region,
+                http_options=self._get_http_options(),
             ) as client:
                 response: GenerateContentResponse = await _generate_content(client)  # type: ignore[no-redef]
         else:
-            with Client(api_key=self.service_settings.api_key.get_secret_value()) as client:  # type: ignore[union-attr]
+            with Client(
+                api_key=self.service_settings.api_key.get_secret_value(),  # type: ignore[union-attr]
+                http_options=self._get_http_options(),
+            ) as client:
                 response: GenerateContentResponse = await _generate_content(client)  # type: ignore[no-redef]
 
         return [self._create_chat_message_content(response, candidate) for candidate in response.candidates]  # type: ignore
@@ -218,6 +222,7 @@ class GoogleAIChatCompletion(GoogleAIBase, ChatCompletionClientBase):
                 vertexai=True,
                 project=self.service_settings.cloud_project_id,
                 location=self.service_settings.cloud_region,
+                http_options=self._get_http_options(),
             ) as client:
                 async for chunk in _generate_content_stream(client):
                     yield [
@@ -225,7 +230,10 @@ class GoogleAIChatCompletion(GoogleAIBase, ChatCompletionClientBase):
                         for candidate in chunk.candidates  # type: ignore
                     ]
         else:
-            with Client(api_key=self.service_settings.api_key.get_secret_value()) as client:  # type: ignore[union-attr]
+            with Client(
+                api_key=self.service_settings.api_key.get_secret_value(),  # type: ignore[union-attr]
+                http_options=self._get_http_options(),
+            ) as client:
                 async for chunk in _generate_content_stream(client):
                     yield [
                         self._create_streaming_chat_message_content(chunk, candidate, function_invoke_attempt)

python/semantic_kernel/connectors/ai/google/google_ai/services/google_ai_text_completion.py

@@ -137,10 +137,14 @@ class GoogleAITextCompletion(GoogleAIBase, TextCompletionClientBase):
                 vertexai=True,
                 project=self.service_settings.cloud_project_id,
                 location=self.service_settings.cloud_region,
+                http_options=self._get_http_options(),
             ) as client:
                 response: GenerateContentResponse = await _generate_content(client)  # type: ignore[no-redef]
         else:
-            with Client(api_key=self.service_settings.api_key.get_secret_value()) as client:  # type: ignore[union-attr]
+            with Client(
+                api_key=self.service_settings.api_key.get_secret_value(),  # type: ignore[union-attr]
+                http_options=self._get_http_options(),
+            ) as client:
                 response: GenerateContentResponse = await _generate_content(client)  # type: ignore[no-redef]
 
         return [self._create_text_content(response, candidate) for candidate in response.candidates]  # type: ignore
@@ -175,11 +179,15 @@ class GoogleAITextCompletion(GoogleAIBase, TextCompletionClientBase):
                 vertexai=True,
                 project=self.service_settings.cloud_project_id,
                 location=self.service_settings.cloud_region,
+                http_options=self._get_http_options(),
             ) as client:
                 async for chunk in _generate_content_stream(client):
                     yield [self._create_streaming_text_content(chunk, candidate) for candidate in chunk.candidates]  # type: ignore
         else:
-            with Client(api_key=self.service_settings.api_key.get_secret_value()) as client:  # type: ignore[union-attr]
+            with Client(
+                api_key=self.service_settings.api_key.get_secret_value(),  # type: ignore[union-attr]
+                http_options=self._get_http_options(),
+            ) as client:
                 async for chunk in _generate_content_stream(client):
                     yield [self._create_streaming_text_content(chunk, candidate) for candidate in chunk.candidates]  # type: ignore
 

python/semantic_kernel/connectors/ai/google/google_ai/services/google_ai_text_embedding.py

@@ -133,10 +133,14 @@ class GoogleAITextEmbedding(GoogleAIBase, EmbeddingGeneratorBase):
                 vertexai=True,
                 project=self.service_settings.cloud_project_id,
                 location=self.service_settings.cloud_region,
+                http_options=self._get_http_options(),
             ) as client:
                 response: EmbedContentResponse = await _embed_content(client)  # type: ignore[no-redef]
         else:
-            with Client(api_key=self.service_settings.api_key.get_secret_value()) as client:  # type: ignore[union-attr]
+            with Client(
+                api_key=self.service_settings.api_key.get_secret_value(),  # type: ignore[union-attr]
+                http_options=self._get_http_options(),
+            ) as client:
                 response: EmbedContentResponse = await _embed_content(client)  # type: ignore[no-redef]
 
         return [embedding.values for embedding in response.embeddings]  # type: ignore

python/tests/unit/connectors/ai/google/google_ai/services/test_google_ai_user_agent.py

@@ -0,0 +1,172 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+from unittest.mock import AsyncMock, patch
+
+from semantic_kernel.connectors.ai.google.google_ai.google_ai_prompt_execution_settings import (
+    GoogleAIChatPromptExecutionSettings,
+    GoogleAIEmbeddingPromptExecutionSettings,
+    GoogleAITextPromptExecutionSettings,
+)
+from semantic_kernel.connectors.ai.google.google_ai.services.google_ai_chat_completion import GoogleAIChatCompletion
+from semantic_kernel.connectors.ai.google.google_ai.services.google_ai_text_completion import GoogleAITextCompletion
+from semantic_kernel.connectors.ai.google.google_ai.services.google_ai_text_embedding import GoogleAITextEmbedding
+from semantic_kernel.const import USER_AGENT
+from semantic_kernel.contents.chat_history import ChatHistory
+
+
+async def test_google_ai_chat_completion_user_agent(google_ai_unit_test_env):
+    """Test that GoogleAIChatCompletion sends the User-Agent header."""
+    chat_history = ChatHistory()
+    chat_history.add_user_message("hi")
+
+    with patch(
+        "semantic_kernel.connectors.ai.google.google_ai.services.google_ai_chat_completion.Client"
+    ) as mock_client:
+        mock_instance = mock_client.return_value.__enter__.return_value
+        mock_instance.aio.models.generate_content = AsyncMock()
+
+        service = GoogleAIChatCompletion(gemini_model_id="gemini-3-flash-preview", api_key="test-api-key")
+
+        await service.get_chat_message_contents(
+            chat_history=chat_history, settings=GoogleAIChatPromptExecutionSettings()
+        )
+
+        _, kwargs = mock_client.call_args
+        assert "http_options" in kwargs
+        assert kwargs["http_options"] is not None
+        assert "headers" in kwargs["http_options"]
+        assert USER_AGENT in kwargs["http_options"]["headers"]
+        assert "semantic-kernel-python" in kwargs["http_options"]["headers"][USER_AGENT]
+
+
+async def test_google_ai_chat_completion_vertex_ai_user_agent(google_ai_unit_test_env):
+    """Test that GoogleAIChatCompletion sends the User-Agent header with Vertex AI."""
+    chat_history = ChatHistory()
+    chat_history.add_user_message("hi")
+
+    with patch(
+        "semantic_kernel.connectors.ai.google.google_ai.services.google_ai_chat_completion.Client"
+    ) as mock_client:
+        mock_instance = mock_client.return_value.__enter__.return_value
+        mock_instance.aio.models.generate_content = AsyncMock()
+
+        service = GoogleAIChatCompletion(
+            gemini_model_id="gemini-3-flash-preview",
+            use_vertexai=True,
+            project_id="test-project",
+            region="test-region",
+        )
+
+        await service.get_chat_message_contents(
+            chat_history=chat_history, settings=GoogleAIChatPromptExecutionSettings()
+        )
+
+        _, kwargs = mock_client.call_args
+        assert "http_options" in kwargs
+        assert kwargs["http_options"] is not None
+        assert "headers" in kwargs["http_options"]
+        assert USER_AGENT in kwargs["http_options"]["headers"]
+        assert "semantic-kernel-python" in kwargs["http_options"]["headers"][USER_AGENT]
+
+
+async def test_google_ai_chat_completion_no_telemetry(google_ai_unit_test_env):
+    """Test that GoogleAIChatCompletion does not send the User-Agent header when telemetry is disabled."""
+    chat_history = ChatHistory()
+    chat_history.add_user_message("hi")
+
+    with (
+        patch("semantic_kernel.connectors.ai.google.google_ai.services.google_ai_base.IS_TELEMETRY_ENABLED", False),
+        patch(
+            "semantic_kernel.connectors.ai.google.google_ai.services.google_ai_chat_completion.Client"
+        ) as mock_client,
+    ):
+        mock_instance = mock_client.return_value.__enter__.return_value
+        mock_instance.aio.models.generate_content = AsyncMock()
+
+        service = GoogleAIChatCompletion(gemini_model_id="gemini-3-flash-preview", api_key="test-api-key")
+
+        await service.get_chat_message_contents(
+            chat_history=chat_history, settings=GoogleAIChatPromptExecutionSettings()
+        )
+
+        _, kwargs = mock_client.call_args
+        assert "http_options" in kwargs
+        assert kwargs["http_options"] is None
+
+
+async def test_google_ai_text_completion_user_agent(google_ai_unit_test_env):
+    """Test that GoogleAITextCompletion sends the User-Agent header."""
+    with patch(
+        "semantic_kernel.connectors.ai.google.google_ai.services.google_ai_text_completion.Client"
+    ) as mock_client:
+        mock_instance = mock_client.return_value.__enter__.return_value
+        mock_instance.aio.models.generate_content = AsyncMock()
+
+        service = GoogleAITextCompletion(gemini_model_id="gemini-3-flash-preview", api_key="test-api-key")
+
+        await service.get_text_contents(prompt="hi", settings=GoogleAITextPromptExecutionSettings())
+
+        _, kwargs = mock_client.call_args
+        assert "http_options" in kwargs
+        assert kwargs["http_options"] is not None
+        assert "headers" in kwargs["http_options"]
+        assert USER_AGENT in kwargs["http_options"]["headers"]
+        assert "semantic-kernel-python" in kwargs["http_options"]["headers"][USER_AGENT]
+
+
+async def test_google_ai_text_completion_no_telemetry(google_ai_unit_test_env):
+    """Test that GoogleAITextCompletion does not send the User-Agent header when telemetry is disabled."""
+    with (
+        patch("semantic_kernel.connectors.ai.google.google_ai.services.google_ai_base.IS_TELEMETRY_ENABLED", False),
+        patch(
+            "semantic_kernel.connectors.ai.google.google_ai.services.google_ai_text_completion.Client"
+        ) as mock_client,
+    ):
+        mock_instance = mock_client.return_value.__enter__.return_value
+        mock_instance.aio.models.generate_content = AsyncMock()
+
+        service = GoogleAITextCompletion(gemini_model_id="gemini-3-flash-preview", api_key="test-api-key")
+
+        await service.get_text_contents(prompt="hi", settings=GoogleAITextPromptExecutionSettings())
+
+        _, kwargs = mock_client.call_args
+        assert "http_options" in kwargs
+        assert kwargs["http_options"] is None
+
+
+async def test_google_ai_text_embedding_user_agent(google_ai_unit_test_env):
+    """Test that GoogleAITextEmbedding sends the User-Agent header."""
+    with patch(
+        "semantic_kernel.connectors.ai.google.google_ai.services.google_ai_text_embedding.Client"
+    ) as mock_client:
+        mock_instance = mock_client.return_value.__enter__.return_value
+        mock_instance.aio.models.embed_content = AsyncMock()
+
+        service = GoogleAITextEmbedding(embedding_model_id="gemini-embedding-2-preview", api_key="test-api-key")
+
+        await service.generate_embeddings(texts=["hi"], settings=GoogleAIEmbeddingPromptExecutionSettings())
+
+        _, kwargs = mock_client.call_args
+        assert "http_options" in kwargs
+        assert kwargs["http_options"] is not None
+        assert "headers" in kwargs["http_options"]
+        assert USER_AGENT in kwargs["http_options"]["headers"]
+        assert "semantic-kernel-python" in kwargs["http_options"]["headers"][USER_AGENT]
+
+
+async def test_google_ai_text_embedding_no_telemetry(google_ai_unit_test_env):
+    """Test that GoogleAITextEmbedding does not send the User-Agent header when telemetry is disabled."""
+    with (
+        patch("semantic_kernel.connectors.ai.google.google_ai.services.google_ai_base.IS_TELEMETRY_ENABLED", False),
+        patch("semantic_kernel.connectors.ai.google.google_ai.services.google_ai_text_embedding.Client") as mock_client,
+    ):
+        mock_instance = mock_client.return_value.__enter__.return_value
+        mock_instance.aio.models.embed_content = AsyncMock()
+
+        service = GoogleAITextEmbedding(embedding_model_id="gemini-embedding-2-preview", api_key="test-api-key")
+
+        await service.generate_embeddings(texts=["hi"], settings=GoogleAIEmbeddingPromptExecutionSettings())
+
+        _, kwargs = mock_client.call_args
+        assert "http_options" in kwargs
+        assert kwargs["http_options"] is None