Merge pull request #3208 from pipecat-ai/thor/add-client-identification

add Gemini client identification

changelog/3208.added.md

@@ -0,0 +1 @@
+- Added `x-goog-api-client` header with Pipecat's version to all Google services' requests.

src/pipecat/__init__.py

@@ -5,14 +5,20 @@
 #
 
 import sys
-from importlib.metadata import version
+from importlib.metadata import version as lib_version
 
 from loguru import logger
 
-__version__ = version("pipecat-ai")
+__version__ = lib_version("pipecat-ai")
 
 logger.info(f"ᓚᘏᗢ Pipecat {__version__} (Python {sys.version}) ᓚᘏᗢ")
 
+
+def version() -> str:
+    """Returns the Pipecat version."""
+    return __version__
+
+
 # We replace `asyncio.wait_for()` for `wait_for2.wait_for()` for Python < 3.12.
 #
 # In Python 3.12, `asyncio.wait_for()` is implemented in terms of

src/pipecat/services/google/gemini_live/llm.py

@@ -68,6 +68,7 @@ from pipecat.processors.aggregators.openai_llm_context import (
 )
 from pipecat.processors.frame_processor import FrameDirection
 from pipecat.services.google.frames import LLMSearchOrigin, LLMSearchResponseFrame, LLMSearchResult
+from pipecat.services.google.utils import update_google_client_http_options
 from pipecat.services.llm_service import FunctionCallFromLLM, LLMService
 from pipecat.services.openai.llm import (
     OpenAIAssistantContextAggregator,
@@ -681,7 +682,7 @@ class GeminiLiveLLMService(LLMService):
         self._video_input_paused = start_video_paused
         self._context = None
         self._api_key = api_key
-        self._http_options = http_options
+        self._http_options = update_google_client_http_options(http_options)
         self._session: AsyncSession = None
         self._connection_task = None
 

src/pipecat/services/google/gemini_live/llm_vertex.py

@@ -126,6 +126,7 @@ class GeminiLiveVertexLLMService(GeminiLiveLLMService):
             credentials=self._credentials,
             project=self._project_id,
             location=self._location,
+            http_options=self._http_options,
         )
 
     @property

src/pipecat/services/google/image.py

@@ -16,13 +16,14 @@ import os
 # Suppress gRPC fork warnings
 os.environ["GRPC_ENABLE_FORK_SUPPORT"] = "false"
 
-from typing import AsyncGenerator, Optional
+from typing import Any, AsyncGenerator, Optional
 
 from loguru import logger
 from PIL import Image
 from pydantic import BaseModel, Field
 
 from pipecat.frames.frames import ErrorFrame, Frame, URLImageRawFrame
+from pipecat.services.google.utils import update_google_client_http_options
 from pipecat.services.image_service import ImageGenService
 
 try:
@@ -60,6 +61,7 @@ class GoogleImageGenService(ImageGenService):
         *,
         api_key: str,
         params: Optional[InputParams] = None,
+        http_options: Optional[Any] = None,
         **kwargs,
     ):
         """Initialize the GoogleImageGenService with API key and parameters.
@@ -67,11 +69,16 @@ class GoogleImageGenService(ImageGenService):
         Args:
             api_key: Google AI API key for authentication.
             params: Configuration parameters for image generation. Defaults to InputParams().
+            http_options: HTTP options for the client.
             **kwargs: Additional arguments passed to the parent ImageGenService.
         """
         super().__init__(**kwargs)
         self._params = params or GoogleImageGenService.InputParams()
-        self._client = genai.Client(api_key=api_key)
+
+        # Add client header
+        http_options = update_google_client_http_options(http_options)
+
+        self._client = genai.Client(api_key=api_key, http_options=http_options)
         self.set_model_name(self._params.model)
 
     def can_generate_metrics(self) -> bool:

src/pipecat/services/google/llm.py

@@ -50,6 +50,7 @@ from pipecat.processors.aggregators.openai_llm_context import (
 )
 from pipecat.processors.frame_processor import FrameDirection
 from pipecat.services.google.frames import LLMSearchResponseFrame
+from pipecat.services.google.utils import update_google_client_http_options
 from pipecat.services.llm_service import FunctionCallFromLLM, LLMService
 from pipecat.services.openai.llm import (
     OpenAIAssistantContextAggregator,
@@ -713,7 +714,7 @@ class GoogleLLMService(LLMService):
         self.set_model_name(model)
         self._api_key = api_key
         self._system_instruction = system_instruction
-        self._http_options = http_options
+        self._http_options = update_google_client_http_options(http_options)
 
         self._settings = {
             "max_tokens": params.max_tokens,

src/pipecat/services/google/utils.py

@@ -0,0 +1,43 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""Utility functions for Google services."""
+
+from typing import Any, Dict, Optional, Union
+
+from pipecat import version as pipecat_version
+
+
+def update_google_client_http_options(http_options: Optional[Union[Dict[str, Any], Any]]) -> Any:
+    """Updates http_options with the x-goog-api-client header.
+
+    Args:
+        http_options: The existing http_options, which can be None, a dictionary,
+                      or an object with a 'headers' attribute.
+
+    Returns:
+        The updated http_options.
+    """
+    client_header = {"x-goog-api-client": f"pipecat/{pipecat_version()}"}
+
+    if http_options is None:
+        http_options = {"headers": client_header}
+    elif isinstance(http_options, dict):
+        # Create a copy to avoid modifying the original dictionary if it's reused elsewhere
+        http_options = http_options.copy()
+        if "headers" in http_options:
+            http_options["headers"].update(client_header)
+        else:
+            http_options["headers"] = client_header
+    elif hasattr(http_options, "headers"):
+        # We can't easily copy an arbitrary object, so we modify it in place.
+        # This assumes the object is mutable and it's safe to do so.
+        if http_options.headers is None:
+            http_options.headers = client_header
+        else:
+            http_options.headers.update(client_header)
+
+    return http_options

tests/test_google_utils.py

@@ -0,0 +1,55 @@
+#
+# Copyright (c) 2024-2025 Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import unittest
+
+import pipecat.services.google.utils
+from pipecat.services.google.utils import update_google_client_http_options
+
+MOCKED_VERSION = "0.0.0-test"
+
+pipecat.services.google.utils.pipecat_version = lambda: MOCKED_VERSION
+
+
+class TestGoogleUtils(unittest.TestCase):
+    def test_update_google_client_http_options_none(self):
+        options = update_google_client_http_options(None)
+        self.assertEqual(options, {"headers": {"x-goog-api-client": f"pipecat/{MOCKED_VERSION}"}})
+
+    def test_update_google_client_http_options_dict_empty(self):
+        options = update_google_client_http_options({})
+        self.assertEqual(options, {"headers": {"x-goog-api-client": f"pipecat/{MOCKED_VERSION}"}})
+
+    def test_update_google_client_http_options_dict_existing_headers(self):
+        initial_options = {"headers": {"Authorization": "Bearer token"}}
+        options = update_google_client_http_options(initial_options)
+        self.assertEqual(options["headers"]["Authorization"], "Bearer token")
+        self.assertEqual(options["headers"]["x-goog-api-client"], f"pipecat/{MOCKED_VERSION}")
+
+    def test_update_google_client_http_options_object(self):
+        class HttpOptions:
+            def __init__(self):
+                self.headers = None
+
+        http_options = HttpOptions()
+        updated_options = update_google_client_http_options(http_options)
+        self.assertEqual(
+            updated_options.headers, {"x-goog-api-client": f"pipecat/{MOCKED_VERSION}"}
+        )
+
+    def test_update_google_client_http_options_object_existing_headers(self):
+        class HttpOptions:
+            def __init__(self):
+                self.headers = {"Authorization": "Bearer token"}
+
+        http_options = HttpOptions()
+        updated_options = update_google_client_http_options(http_options)
+        self.assertEqual(updated_options.headers["Authorization"], "Bearer token")
+        self.assertEqual(updated_options.headers["x-goog-api-client"], f"pipecat/{MOCKED_VERSION}")
+
+
+if __name__ == "__main__":
+    unittest.main()