chore: move into services file.

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

@@ -24,7 +24,7 @@ from loguru import logger
 from PIL import Image
 from pydantic import BaseModel, Field
 
-from pipecat import __version__
+from pipecat.services.google.utils import update_google_client_http_options
 from pipecat.adapters.schemas.tools_schema import ToolsSchema
 from pipecat.adapters.services.gemini_adapter import GeminiLLMAdapter
 from pipecat.frames.frames import (
@@ -682,22 +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
-
-        # Add client header
-        client_header = {"x-goog-api-client": f"pipecat/{__version__}"}
-        if self._http_options is None:
-            self._http_options = {"headers": client_header}
-        elif isinstance(self._http_options, dict):
-            if "headers" in self._http_options:
-                self._http_options["headers"].update(client_header)
-            else:
-                self._http_options["headers"] = client_header
-        elif hasattr(self._http_options, "headers"):
-            if self._http_options.headers is None:
-                self._http_options.headers = client_header
-            else:
-                self._http_options.headers.update(client_header)
+        self._http_options = update_google_client_http_options(http_options)
         self._session: AsyncSession = None
         self._connection_task = None
 

src/pipecat/services/google/image.py

@@ -22,7 +22,8 @@ from loguru import logger
 from PIL import Image
 from pydantic import BaseModel, Field
 
-from pipecat import __version__
+
+from pipecat.services.google.utils import update_google_client_http_options
 from pipecat.frames.frames import ErrorFrame, Frame, URLImageRawFrame
 from pipecat.services.image_service import ImageGenService
 
@@ -76,19 +77,7 @@ class GoogleImageGenService(ImageGenService):
         self._params = params or GoogleImageGenService.InputParams()
 
         # Add client header
-        client_header = {"x-goog-api-client": f"pipecat/{__version__}"}
-        if http_options is None:
-            http_options = {"headers": client_header}
-        elif isinstance(http_options, dict):
-            if "headers" in http_options:
-                http_options["headers"].update(client_header)
-            else:
-                http_options["headers"] = client_header
-        elif hasattr(http_options, "headers"):
-            if http_options.headers is None:
-                http_options.headers = client_header
-            else:
-                http_options.headers.update(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)

src/pipecat/services/google/llm.py

@@ -22,7 +22,7 @@ from loguru import logger
 from PIL import Image
 from pydantic import BaseModel, Field
 
-from pipecat import __version__
+from pipecat.services.google.utils import update_google_client_http_options
 from pipecat.adapters.services.gemini_adapter import GeminiLLMAdapter, GeminiLLMInvocationParams
 from pipecat.frames.frames import (
     AudioRawFrame,
@@ -714,22 +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
-        
-        # Add client header
-        client_header = {"x-goog-api-client": f"pipecat/{__version__}"}
-        if self._http_options is None:
-            self._http_options = {"headers": client_header}
-        elif isinstance(self._http_options, dict):
-            if "headers" in self._http_options:
-                self._http_options["headers"].update(client_header)
-            else:
-                self._http_options["headers"] = client_header
-        elif hasattr(self._http_options, "headers"):
-            if self._http_options.headers is None:
-                self._http_options.headers = client_header
-            else:
-                self._http_options.headers.update(client_header)
+        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,41 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+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