feat(live): update from StreamTranslationConfig to TranslationConfig

Author: allen-stephen

pyproject.toml

@@ -38,7 +38,7 @@ dependencies = [
   "click>=8.1.8,<9",
   "fastapi>=0.133,<1",
   "google-auth[pyopenssl]>=2.47",
-  "google-genai>=2.4,<3",
+  "google-genai>=2.8,<3",
   "graphviz>=0.20.2,<1",
   "httpx>=0.27,<1",
   "jsonschema>=4.23,<5",

src/google/adk/agents/run_config.py

@@ -239,8 +239,12 @@ class RunConfig(BaseModel):
   realtime_input_config: Optional[types.RealtimeInputConfig] = None
   """Realtime input config for live agents with audio input from user."""
 
-  translation_config: Optional[types.StreamTranslationConfig] = None
-  """Configures real-time speech-to-speech translation for live agents."""
+  translation_config: Optional[types.TranslationConfig] = None
+  """Configures real-time speech-to-speech translation.
+
+  Only supported by translation models such as
+  `gemini-3.5-live-translate-preview`.
+  """
 
   enable_affective_dialog: Optional[bool] = None
   """If enabled, the model will detect emotions and adapt its responses accordingly."""

src/google/adk/flows/llm_flows/basic.py

@@ -83,7 +83,7 @@ def _build_basic_request(
   llm_request.live_connect_config.realtime_input_config = (
       invocation_context.run_config.realtime_input_config
   )
-  llm_request.live_connect_config.stream_translation_config = (
+  llm_request.live_connect_config.translation_config = (
       invocation_context.run_config.translation_config
   )
   active_model_name = (

src/google/adk/models/gemini_llm_connection.py

@@ -53,6 +53,9 @@ def __init__(
     self._is_gemini_3_1_flash_live = model_name_utils.is_gemini_3_1_flash_live(
         model_version
     )
+    self._is_gemini_3_5_live_translate = (
+        model_name_utils.is_gemini_3_5_live_translate(model_version)
+    )
 
   async def send_history(self, history: list[types.Content]):
     """Sends the conversation history to the gemini model.
@@ -160,7 +163,7 @@ async def send_realtime(self, input: RealtimeInput):
     if isinstance(input, types.Blob):
       # The blob is binary and is very large. So let's not log it.
       logger.debug('Sending LLM Blob.')
-      if self._is_gemini_3_1_flash_live:
+      if self._is_gemini_3_1_flash_live or self._is_gemini_3_5_live_translate:
         if input.mime_type and input.mime_type.startswith('audio/'):
           await self._gemini_session.send_realtime_input(audio=input)
         elif input.mime_type and input.mime_type.startswith('image/'):

src/google/adk/utils/model_name_utils.py

@@ -174,3 +174,18 @@ def is_gemini_3_1_flash_live(model_string: Optional[str]) -> bool:
     return False
   model_name = extract_model_name(model_string)
   return model_name.startswith('gemini-3.1-flash-live')
+
+
+def is_gemini_3_5_live_translate(model_string: Optional[str]) -> bool:
+  """Check if the model is a Gemini 3.5 Live Translate model.
+
+  Args:
+    model_string: The model name
+
+  Returns:
+    True if it's a Gemini 3.5 Live Translate model, False otherwise
+  """
+  if not model_string:
+    return False
+  model_name = extract_model_name(model_string)
+  return model_name.startswith('gemini-3.5-live-translate')

tests/unittests/flows/llm_flows/test_basic_processor.py

@@ -264,7 +264,7 @@ async def test_sets_translation_config(self):
     )
     invocation_context = await _create_invocation_context(agent)
     invocation_context.run_config = RunConfig(
-        translation_config=types.StreamTranslationConfig(
+        translation_config=types.TranslationConfig(
             target_language_code='pl',
             echo_target_language=True,
         ),
@@ -275,11 +275,9 @@ async def test_sets_translation_config(self):
     async for _ in processor.run_async(invocation_context, llm_request):
       pass
 
-    stream_translation_config = (
-        llm_request.live_connect_config.stream_translation_config
-    )
-    assert stream_translation_config.target_language_code == 'pl'
-    assert stream_translation_config.echo_target_language is True
+    translation_config = llm_request.live_connect_config.translation_config
+    assert translation_config.target_language_code == 'pl'
+    assert translation_config.echo_target_language is True
 
   @pytest.mark.asyncio
   async def test_translation_config_defaults_to_none(self):
@@ -295,4 +293,4 @@ async def test_translation_config_defaults_to_none(self):
     async for _ in processor.run_async(invocation_context, llm_request):
       pass
 
-    assert llm_request.live_connect_config.stream_translation_config is None
+    assert llm_request.live_connect_config.translation_config is None

tests/unittests/models/test_gemini_llm_connection.py

@@ -71,6 +71,24 @@ async def test_send_realtime_default_behavior(
   mock_gemini_session.send.assert_not_called()
 
 
+@pytest.mark.asyncio
+async def test_send_realtime_audio_uses_audio_channel_for_live_translate(
+    mock_gemini_session, test_blob
+):
+  """Live Translate models stream audio via the dedicated `audio=` channel."""
+  connection = GeminiLlmConnection(
+      mock_gemini_session,
+      api_backend=GoogleLLMVariant.GEMINI_API,
+      model_version='gemini-3.5-live-translate-preview',
+  )
+
+  await connection.send_realtime(test_blob)
+
+  mock_gemini_session.send_realtime_input.assert_called_once_with(
+      audio=test_blob
+  )
+
+
 @pytest.mark.asyncio
 async def test_send_history(gemini_connection, mock_gemini_session):
   """Test send_history method."""

tests/unittests/utils/test_model_name_utils.py

@@ -17,6 +17,7 @@
 from google.adk.utils.model_name_utils import extract_model_name
 from google.adk.utils.model_name_utils import is_gemini_1_model
 from google.adk.utils.model_name_utils import is_gemini_3_1_flash_live
+from google.adk.utils.model_name_utils import is_gemini_3_5_live_translate
 from google.adk.utils.model_name_utils import is_gemini_eap_or_2_or_above
 from google.adk.utils.model_name_utils import is_gemini_model
 from google.adk.utils.model_name_utils import is_gemini_model_id_check_disabled
@@ -366,3 +367,22 @@ def test_is_gemini_3_1_flash_live_edge_cases(self):
     """Test edge cases."""
     assert is_gemini_3_1_flash_live(None) is False
     assert is_gemini_3_1_flash_live('') is False
+
+
+class TestIsGemini35LiveTranslate:
+  """Test the is_gemini_3_5_live_translate function."""
+
+  def test_is_gemini_3_5_live_translate_simple_name(self):
+    """Test with simple model name format."""
+    assert is_gemini_3_5_live_translate('gemini-3.5-live-translate') is True
+    assert is_gemini_3_5_live_translate('gemini-3.5-flash-live') is False
+
+  def test_is_gemini_3_5_live_translate_path_based_name(self):
+    """Test with path-based format (Vertex AI etc.)."""
+    vertex_path = 'projects/123/locations/us-central1/publishers/google/models/gemini-3.5-live-translate-preview'
+    assert is_gemini_3_5_live_translate(vertex_path) is True
+
+  def test_is_gemini_3_5_live_translate_edge_cases(self):
+    """Test edge cases."""
+    assert is_gemini_3_5_live_translate(None) is False
+    assert is_gemini_3_5_live_translate('') is False