feat: Live avatar support in ADK

Testing plan: Added new unit tests
- `test_avatar_config_initialization`
- `test_avatar_config_with_name`
- `test_receive_video_content`
- `test_streaming_with_avatar_config`

PiperOrigin-RevId: 899193911

Author: google-genai-bot

pyproject.toml

@@ -44,7 +44,7 @@ dependencies = [
   "google-cloud-spanner>=3.56.0, <4.0.0",                    # For Spanner database
   "google-cloud-speech>=2.30.0, <3.0.0",                     # For Audio Transcription
   "google-cloud-storage>=2.18.0, <4.0.0",                    # For GCS Artifact service
-  "google-genai>=1.64.0, <2.0.0",                            # Google GenAI SDK
+  "google-genai>=1.72.0, <2.0.0",                            # Google GenAI SDK
   "graphviz>=0.20.2, <1.0.0",                                # Graphviz for graph rendering
   "httpx>=0.27.0, <1.0.0",                                   # HTTP client library
   "jsonschema>=4.23.0, <5.0.0",                              # Agent Builder config validation

src/google/adk/agents/run_config.py

@@ -198,6 +198,9 @@ class RunConfig(BaseModel):
   response_modalities: Optional[list[str]] = None
   """The output modalities. If not set, it's default to AUDIO."""
 
+  avatar_config: Optional[types.AvatarConfig] = None
+  """Avatar configuration for the live agent."""
+
   save_input_blobs_as_artifacts: bool = Field(
       default=False,
       deprecated=True,

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

@@ -90,6 +90,9 @@ def _build_basic_request(
   llm_request.live_connect_config.context_window_compression = (
       invocation_context.run_config.context_window_compression
   )
+  llm_request.live_connect_config.avatar_config = (
+      invocation_context.run_config.avatar_config
+  )
 
 
 class _BasicLlmRequestProcessor(BaseLlmRequestProcessor):

src/google/adk/models/gemini_llm_connection.py

@@ -115,16 +115,7 @@ async def send_content(self, content: types.Content):
       is_gemini_31 = model_name_utils.is_gemini_3_1_flash_live(
           self._model_version
       )
-      is_gemini_api = self._api_backend == GoogleLLMVariant.GEMINI_API
-
-      # As of now, Gemini 3.1 Flash Live is only available in Gemini API, not
-      # Vertex AI.
-      if (
-          is_gemini_31
-          and is_gemini_api
-          and len(content.parts) == 1
-          and content.parts[0].text
-      ):
+      if is_gemini_31 and len(content.parts) == 1 and content.parts[0].text:
         logger.debug('Using send_realtime_input for Gemini 3.1 text input')
         await self._gemini_session.send_realtime_input(
             text=content.parts[0].text
@@ -149,11 +140,7 @@ async def send_realtime(self, input: RealtimeInput):
       is_gemini_31 = model_name_utils.is_gemini_3_1_flash_live(
           self._model_version
       )
-      is_gemini_api = self._api_backend == GoogleLLMVariant.GEMINI_API
-
-      # As of now, Gemini 3.1 Flash Live is only available in Gemini API, not
-      # Vertex AI.
-      if is_gemini_31 and is_gemini_api:
+      if is_gemini_31:
         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

@@ -130,9 +130,6 @@ def is_gemini_2_or_above(model_string: Optional[str]) -> bool:
 def is_gemini_3_1_flash_live(model_string: Optional[str]) -> bool:
   """Check if the model is a Gemini 3.1 Flash Live model.
 
-  Note: This is a very specific model name for live bidi streaming, so we check
-  for exact match.
-
   Args:
     model_string: The model name
 
@@ -141,5 +138,4 @@ def is_gemini_3_1_flash_live(model_string: Optional[str]) -> bool:
   """
   if not model_string:
     return False
-
-  return model_string == 'gemini-3.1-flash-live-preview'
+  return model_string.startswith('gemini-3.1-flash-live')

tests/unittests/agents/test_run_config.py

@@ -17,6 +17,7 @@
 from unittest.mock import patch
 
 from google.adk.agents.run_config import RunConfig
+from google.genai import types
 import pytest
 
 
@@ -64,3 +65,35 @@ def test_audio_transcription_configs_are_not_shared_between_instances():
   assert (
       config1.input_audio_transcription is not config2.input_audio_transcription
   )
+
+
+def test_avatar_config_initialization():
+  custom_avatar = types.CustomizedAvatar(
+      image_mime_type="image/jpeg", image_data=b"image_bytes"
+  )
+  avatar_config = types.AvatarConfig(
+      audio_bitrate_bps=128000,
+      video_bitrate_bps=1000000,
+      customized_avatar=custom_avatar,
+  )
+  run_config = RunConfig(avatar_config=avatar_config)
+
+  assert run_config.avatar_config == avatar_config
+  assert run_config.avatar_config.customized_avatar == custom_avatar
+  assert (
+      run_config.avatar_config.customized_avatar.image_mime_type == "image/jpeg"
+  )
+  assert run_config.avatar_config.customized_avatar.image_data == b"image_bytes"
+
+
+def test_avatar_config_with_name():
+  avatar_config = types.AvatarConfig(
+      audio_bitrate_bps=128000,
+      video_bitrate_bps=1000000,
+      avatar_name="test_avatar",
+  )
+  run_config = RunConfig(avatar_config=avatar_config)
+
+  assert run_config.avatar_config == avatar_config
+  assert run_config.avatar_config.avatar_name == "test_avatar"
+  assert run_config.avatar_config.customized_avatar is None

tests/unittests/models/test_gemini_llm_connection.py

@@ -1120,3 +1120,43 @@ async def mock_receive_generator():
 
   assert len(responses) == 1
   assert responses[0].go_away == mock_go_away
+
+
+@pytest.mark.asyncio
+async def test_receive_video_content(gemini_connection, mock_gemini_session):
+  """Test receive with video content."""
+  mock_content = types.Content(
+      role='model',
+      parts=[
+          types.Part(
+              inline_data=types.Blob(data=b'video_data', mime_type='video/mp4')
+          )
+      ],
+  )
+  mock_server_content = mock.Mock()
+  mock_server_content.model_turn = mock_content
+  mock_server_content.interrupted = False
+  mock_server_content.input_transcription = None
+  mock_server_content.output_transcription = None
+  mock_server_content.turn_complete = False
+  mock_server_content.grounding_metadata = None
+
+  mock_message = mock.AsyncMock()
+  mock_message.usage_metadata = None
+  mock_message.server_content = mock_server_content
+  mock_message.tool_call = None
+  mock_message.session_resumption_update = None
+  mock_message.go_away = None
+
+  async def mock_receive_generator():
+    yield mock_message
+
+  receive_mock = mock.Mock(return_value=mock_receive_generator())
+  mock_gemini_session.receive = receive_mock
+
+  responses = [resp async for resp in gemini_connection.receive()]
+
+  assert responses
+  content_response = next((r for r in responses if r.content), None)
+  assert content_response is not None
+  assert content_response.content == mock_content

tests/unittests/streaming/test_live_streaming_configs.py

@@ -642,3 +642,90 @@ def test_streaming_with_context_window_compression_config():
       llm_request_sent_to_mock.live_connect_config.context_window_compression.sliding_window.target_tokens
       == 500
   )
+
+
+def test_streaming_with_avatar_config():
+  """Test avatar_config propagation and video content through run_live.
+
+  Verifies:
+    1. avatar_config from RunConfig is propagated to live_connect_config.
+    2. Video inline_data from the model flows through events correctly.
+  """
+  # Mock model returns video content followed by turn_complete.
+  video_response = LlmResponse(
+      content=types.Content(
+          role='model',
+          parts=[
+              types.Part(
+                  inline_data=types.Blob(
+                      data=b'video_data', mime_type='video/mp4'
+                  )
+              )
+          ],
+      ),
+  )
+  turn_complete_response = LlmResponse(
+      turn_complete=True,
+  )
+
+  mock_model = testing_utils.MockModel.create(
+      [video_response, turn_complete_response]
+  )
+
+  root_agent = Agent(
+      name='root_agent',
+      model=mock_model,
+      tools=[],
+  )
+
+  runner = testing_utils.InMemoryRunner(
+      root_agent=root_agent, response_modalities=['VIDEO']
+  )
+
+  run_config = RunConfig(
+      response_modalities=['VIDEO'],
+      avatar_config=types.AvatarConfig(avatar_name='Kai'),
+  )
+
+  live_request_queue = LiveRequestQueue()
+  live_request_queue.send_realtime(
+      blob=types.Blob(data=b'\x00\xFF', mime_type='audio/pcm')
+  )
+  res_events = runner.run_live(live_request_queue, run_config)
+
+  assert res_events is not None, 'Expected a list of events, got None.'
+  assert (
+      len(res_events) > 0
+  ), 'Expected at least one response, but got an empty list.'
+  assert len(mock_model.requests) == 1
+
+  # 1. Verify avatar_config was propagated to the live_connect_config.
+  llm_request_sent_to_mock = mock_model.requests[0]
+  assert llm_request_sent_to_mock.live_connect_config is not None
+  assert llm_request_sent_to_mock.live_connect_config.avatar_config is not None
+  assert (
+      llm_request_sent_to_mock.live_connect_config.avatar_config.avatar_name
+      == 'Kai'
+  )
+
+  # 2. Verify video content flows through events.
+  video_events = [
+      e
+      for e in res_events
+      if e.content
+      and e.content.parts
+      and any(
+          p.inline_data
+          and p.inline_data.mime_type
+          and p.inline_data.mime_type.startswith('video/')
+          for p in e.content.parts
+      )
+  ]
+  assert video_events, 'Expected at least one event with video inline_data.'
+
+  video_event = video_events[0]
+  assert video_event.content.role == 'model'
+  video_part = video_event.content.parts[0]
+  assert video_part.inline_data is not None
+  assert video_part.inline_data.data == b'video_data'
+  assert video_part.inline_data.mime_type == 'video/mp4'