diff --git a/.github/.release-please-manifest.json b/.github/.release-please-manifest.json
index bac4ebcfdc..b0daa7d753 100644
--- a/.github/.release-please-manifest.json
+++ b/.github/.release-please-manifest.json
@@ -1,3 +1,3 @@
 {
-  ".": "1.28.0"
+  ".": "1.28.1"
 }
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 8a1fabb0da..34048e1e98 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,18 @@
 # Changelog
 
+## [1.28.1](https://github.com/google/adk-python/compare/v1.28.0...v1.28.1) (2026-04-02)
+
+
+### Features
+
+* **live:** support live for `gemini-3.1-flash-live-preview` model ([ee69661](https://github.com/google/adk-python/commit/ee69661a616056fa89e0ec2188aaa59bd714d8c9))
+
+
+### Bug Fixes
+
+* Disallow args on /builder and Add warning about Web UI usage to CLI help ([f037f68](https://github.com/google/adk-python/commit/f037f68d67ae1bd16b00df0c7523fb67cbd1e911))
+* **live:** Buffer tool calls and emit them together upon turn completion ([081adbd](https://github.com/google/adk-python/commit/081adbdfa41490e4868b028a1cdabceb811a7505))
+
 ## [1.28.0](https://github.com/google/adk-python/compare/v1.27.5...v1.28.0) (2026-03-26)
 
 
diff --git a/contributing/samples/live_bidi_streaming_parallel_tools_agent/README.md b/contributing/samples/live_bidi_streaming_parallel_tools_agent/README.md
new file mode 100644
index 0000000000..cc96819c38
--- /dev/null
+++ b/contributing/samples/live_bidi_streaming_parallel_tools_agent/README.md
@@ -0,0 +1,38 @@
+# Simple Live (Bidi-Streaming) Agent with Parallel Tools
+This project provides a basic example of a live, [bidirectional streaming](https://google.github.io/adk-docs/streaming/) agent that demonstrates parallel tool execution.
+
+## Getting Started
+
+Follow these steps to get the agent up and running:
+
+1.  **Start the ADK Web Server**
+    Open your terminal, navigate to the root directory that contains the
+    `live_bidi_streaming_parallel_tools_agent` folder, and execute the following
+    command:
+    ```bash
+    adk web
+    ```
+
+2.  **Access the ADK Web UI**
+    Once the server is running, open your web browser and navigate to the URL 
+    provided in the terminal (it will typically be `http://localhost:8000`).
+
+3.  **Select the Agent**
+    In the top-left corner of the ADK Web UI, use the dropdown menu to select 
+    this agent (`live_bidi_streaming_parallel_tools_agent`).
+
+4.  **Start Streaming**
+    Click on the **Audio** icon located near the chat input 
+    box to begin the streaming session.
+
+5.  **Interact with the Agent**
+    You can now begin talking to the agent, and it will respond in real-time.
+    Try asking it to perform multiple actions at once, for example: "Turn on the
+    lights and the TV at the same time." The agent will be able to invoke both
+    `turn_on_lights` and `turn_on_tv` tools in parallel.
+
+## Usage Notes
+
+* You only need to click the **Audio** button once to initiate the
+ stream. The current version does not support stopping and restarting the stream
+  by clicking the button again during a session.
diff --git a/contributing/samples/live_bidi_streaming_parallel_tools_agent/__init__.py b/contributing/samples/live_bidi_streaming_parallel_tools_agent/__init__.py
new file mode 100644
index 0000000000..4015e47d6e
--- /dev/null
+++ b/contributing/samples/live_bidi_streaming_parallel_tools_agent/__init__.py
@@ -0,0 +1,15 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from . import agent
diff --git a/contributing/samples/live_bidi_streaming_parallel_tools_agent/agent.py b/contributing/samples/live_bidi_streaming_parallel_tools_agent/agent.py
new file mode 100644
index 0000000000..519c31a61b
--- /dev/null
+++ b/contributing/samples/live_bidi_streaming_parallel_tools_agent/agent.py
@@ -0,0 +1,36 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+from google.adk.agents.llm_agent import Agent
+
+
+def turn_on_lights():
+  """Turn on the lights."""
+  print("turn_on_lights")
+  return {"status": "OK"}
+
+
+def turn_on_tv():
+  """Turn on the tv."""
+  print("turn_on_tv")
+  return {"status": "OK"}
+
+
+root_agent = Agent(
+    model="gemini-live-2.5-flash-native-audio",
+    name="Home_helper",
+    instruction="Be polite and answer all user's questions.",
+    tools=[turn_on_lights, turn_on_tv],
+)
diff --git a/src/google/adk/cli/cli_tools_click.py b/src/google/adk/cli/cli_tools_click.py
index b9925724bd..171a63b425 100644
--- a/src/google/adk/cli/cli_tools_click.py
+++ b/src/google/adk/cli/cli_tools_click.py
@@ -1713,7 +1713,8 @@ def cli_api_server(
     default=False,
     help=(
         "Optional. Deploy ADK Web UI if set. (default: deploy ADK API server"
-        " only)"
+        " only). WARNING: The web UI is for development and testing only — do"
+        " not use in production."
     ),
 )
 @click.option(
@@ -2229,7 +2230,8 @@ def cli_deploy_agent_engine(
     default=False,
     help=(
         "Optional. Deploy ADK Web UI if set. (default: deploy ADK API server"
-        " only)"
+        " only). WARNING: The web UI is for development and testing only — do"
+        " not use in production."
     ),
 )
 @click.option(
diff --git a/src/google/adk/cli/fast_api.py b/src/google/adk/cli/fast_api.py
index 0b6f3fb6fe..4b207b4b7d 100644
--- a/src/google/adk/cli/fast_api.py
+++ b/src/google/adk/cli/fast_api.py
@@ -27,6 +27,7 @@
 
 import click
 from fastapi import FastAPI
+from fastapi import HTTPException
 from fastapi import UploadFile
 from fastapi.responses import FileResponse
 from fastapi.responses import PlainTextResponse
@@ -293,6 +294,39 @@ def _has_parent_reference(path: str) -> bool:
 
     _ALLOWED_EXTENSIONS = frozenset({".yaml", ".yml"})
 
+    # --- YAML content security ---
+    # The `args` key in agent YAML configs (CodeConfig.args, ToolConfig.args)
+    # allows callers to pass arbitrary arguments to Python constructors and
+    # functions, which is an RCE vector when exposed through the builder UI.
+    # Block any upload that contains an `args` key anywhere in the document.
+    _BLOCKED_YAML_KEYS = frozenset({"args"})
+
+    def _check_yaml_for_blocked_keys(content: bytes, filename: str) -> None:
+      """Raise if the YAML document contains any blocked keys."""
+      import yaml
+
+      try:
+        docs = list(yaml.safe_load_all(content))
+      except yaml.YAMLError as exc:
+        raise ValueError(f"Invalid YAML in {filename!r}: {exc}") from exc
+
+      def _walk(node: Any) -> None:
+        if isinstance(node, dict):
+          for key, value in node.items():
+            if key in _BLOCKED_YAML_KEYS:
+              raise ValueError(
+                  f"Blocked key {key!r} found in {filename!r}. "
+                  f"The '{key}' field is not allowed in builder uploads "
+                  "because it can execute arbitrary code."
+              )
+            _walk(value)
+        elif isinstance(node, list):
+          for item in node:
+            _walk(item)
+
+      for doc in docs:
+        _walk(doc)
+
     def _parse_upload_filename(filename: Optional[str]) -> tuple[str, str]:
       if not filename:
         raise ValueError("Upload filename is missing.")
@@ -430,40 +464,14 @@ async def builder_build(
         files: list[UploadFile], tmp: Optional[bool] = False
     ) -> bool:
       try:
-        if tmp:
-          app_names = set()
-          uploads = []
-          for file in files:
-            app_name, rel_path = _parse_upload_filename(file.filename)
-            app_names.add(app_name)
-            uploads.append((rel_path, file))
-
-          if len(app_names) != 1:
-            logger.error(
-                "Exactly one app name is required, found: %s",
-                sorted(app_names),
-            )
-            return False
-
-          app_name = next(iter(app_names))
-          app_root = _get_app_root(app_name)
-          tmp_agent_root = _get_tmp_agent_root(app_root, app_name)
-          tmp_agent_root.mkdir(parents=True, exist_ok=True)
-
-          for rel_path, file in uploads:
-            destination_path = _resolve_under_dir(tmp_agent_root, rel_path)
-            destination_path.parent.mkdir(parents=True, exist_ok=True)
-            with destination_path.open("wb") as buffer:
-              shutil.copyfileobj(file.file, buffer)
-
-          return True
-
-        app_names = set()
-        uploads = []
+        # Phase 1: parse filenames and read content into memory.
+        app_names: set[str] = set()
+        uploads: list[tuple[str, bytes]] = []
         for file in files:
           app_name, rel_path = _parse_upload_filename(file.filename)
           app_names.add(app_name)
-          uploads.append((rel_path, file))
+          content = await file.read()
+          uploads.append((rel_path, content))
 
         if len(app_names) != 1:
           logger.error(
@@ -473,6 +481,24 @@ async def builder_build(
           return False
 
         app_name = next(iter(app_names))
+
+        # Phase 2: validate every file *before* writing anything to disk.
+        for rel_path, content in uploads:
+          _check_yaml_for_blocked_keys(content, f"{app_name}/{rel_path}")
+
+        # Phase 3: write validated files to disk.
+        if tmp:
+          app_root = _get_app_root(app_name)
+          tmp_agent_root = _get_tmp_agent_root(app_root, app_name)
+          tmp_agent_root.mkdir(parents=True, exist_ok=True)
+
+          for rel_path, content in uploads:
+            destination_path = _resolve_under_dir(tmp_agent_root, rel_path)
+            destination_path.parent.mkdir(parents=True, exist_ok=True)
+            destination_path.write_bytes(content)
+
+          return True
+
         app_root = _get_app_root(app_name)
         app_root.mkdir(parents=True, exist_ok=True)
 
@@ -480,16 +506,15 @@ async def builder_build(
         if tmp_agent_root.is_dir():
           copy_dir_contents(tmp_agent_root, app_root)
 
-        for rel_path, file in uploads:
+        for rel_path, content in uploads:
           destination_path = _resolve_under_dir(app_root, rel_path)
           destination_path.parent.mkdir(parents=True, exist_ok=True)
-          with destination_path.open("wb") as buffer:
-            shutil.copyfileobj(file.file, buffer)
+          destination_path.write_bytes(content)
 
         return cleanup_tmp(app_name)
       except ValueError as exc:
         logger.exception("Error in builder_build: %s", exc)
-        return False
+        raise HTTPException(status_code=400, detail=str(exc))
       except OSError as exc:
         logger.exception("Error in builder_build: %s", exc)
         return False
diff --git a/src/google/adk/models/gemini_llm_connection.py b/src/google/adk/models/gemini_llm_connection.py
index da508891d4..7fc39748ec 100644
--- a/src/google/adk/models/gemini_llm_connection.py
+++ b/src/google/adk/models/gemini_llm_connection.py
@@ -20,6 +20,7 @@
 
 from google.genai import types
 
+from ..utils import model_name_utils
 from ..utils.content_utils import filter_audio_parts
 from ..utils.context_utils import Aclosing
 from ..utils.variant_utils import GoogleLLMVariant
@@ -99,7 +100,6 @@ async def send_content(self, content: types.Content):
     Args:
       content: The content to send to the model.
     """
-
     assert content.parts
     if content.parts[0].function_response:
       # All parts have to be function responses.
@@ -112,12 +112,30 @@ async def send_content(self, content: types.Content):
       )
     else:
       logger.debug('Sending LLM new content %s', content)
-      await self._gemini_session.send(
-          input=types.LiveClientContent(
-              turns=[content],
-              turn_complete=True,
-          )
+      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
+      ):
+        logger.debug('Using send_realtime_input for Gemini 3.1 text input')
+        await self._gemini_session.send_realtime_input(
+            text=content.parts[0].text
+        )
+      else:
+        await self._gemini_session.send(
+            input=types.LiveClientContent(
+                turns=[content],
+                turn_complete=True,
+            )
+        )
 
   async def send_realtime(self, input: RealtimeInput):
     """Sends a chunk of audio or a frame of video to the model in realtime.
@@ -128,7 +146,26 @@ 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.')
-      await self._gemini_session.send_realtime_input(media=input)
+      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 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/'):
+          await self._gemini_session.send_realtime_input(video=input)
+        else:
+          logger.warning(
+              'Blob not sent. Unknown or empty mime type for'
+              ' send_realtime_input: %s',
+              input.mime_type,
+          )
+      else:
+        await self._gemini_session.send_realtime_input(media=input)
 
     elif isinstance(input, types.ActivityStart):
       logger.debug('Sending LLM activity start signal.')
@@ -166,6 +203,7 @@ async def receive(self) -> AsyncGenerator[LlmResponse, None]:
     """
 
     text = ''
+    tool_call_parts = []
     async with Aclosing(self._gemini_session.receive()) as agen:
       # TODO(b/440101573): Reuse StreamingResponseAggregator to accumulate
       # partial content and emit responses as needed.
@@ -295,6 +333,13 @@ async def receive(self) -> AsyncGenerator[LlmResponse, None]:
             if text:
               yield self.__build_full_text_response(text)
               text = ''
+            if tool_call_parts:
+              logger.debug('Returning aggregated tool_call_parts')
+              yield LlmResponse(
+                  content=types.Content(role='model', parts=tool_call_parts),
+                  model_version=self._model_version,
+              )
+              tool_call_parts = []
             yield LlmResponse(
                 turn_complete=True,
                 interrupted=message.server_content.interrupted,
@@ -316,17 +361,14 @@ async def receive(self) -> AsyncGenerator[LlmResponse, None]:
                   model_version=self._model_version,
               )
         if message.tool_call:
+          logger.debug('Received tool call: %s', message.tool_call)
           if text:
             yield self.__build_full_text_response(text)
             text = ''
-          parts = [
+          tool_call_parts.extend([
               types.Part(function_call=function_call)
               for function_call in message.tool_call.function_calls
-          ]
-          yield LlmResponse(
-              content=types.Content(role='model', parts=parts),
-              model_version=self._model_version,
-          )
+          ])
         if message.session_resumption_update:
           logger.debug('Received session resumption message: %s', message)
           yield (
@@ -335,6 +377,12 @@ async def receive(self) -> AsyncGenerator[LlmResponse, None]:
                   model_version=self._model_version,
               )
           )
+      if tool_call_parts:
+        logger.debug('Exited loop with pending tool_call_parts')
+        yield LlmResponse(
+            content=types.Content(role='model', parts=tool_call_parts),
+            model_version=self._model_version,
+        )
 
   async def close(self):
     """Closes the llm server connection."""
diff --git a/src/google/adk/utils/model_name_utils.py b/src/google/adk/utils/model_name_utils.py
index 57103fb2c7..86fd79ab64 100644
--- a/src/google/adk/utils/model_name_utils.py
+++ b/src/google/adk/utils/model_name_utils.py
@@ -125,3 +125,21 @@ def is_gemini_2_or_above(model_string: Optional[str]) -> bool:
     return False
 
   return parsed_version.major >= 2
+
+
+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
+
+  Returns:
+    True if it's a Gemini 3.1 Flash Live model, False otherwise
+  """
+  if not model_string:
+    return False
+
+  return model_string == 'gemini-3.1-flash-live-preview'
diff --git a/src/google/adk/version.py b/src/google/adk/version.py
index a020c76372..ed58e782e7 100644
--- a/src/google/adk/version.py
+++ b/src/google/adk/version.py
@@ -13,4 +13,4 @@
 # limitations under the License.
 
 # version: major.minor.patch
-__version__ = "1.28.0"
+__version__ = "1.28.1"
diff --git a/tests/unittests/cli/test_fast_api.py b/tests/unittests/cli/test_fast_api.py
index 15bc908ddb..95affeeb3e 100755
--- a/tests/unittests/cli/test_fast_api.py
+++ b/tests/unittests/cli/test_fast_api.py
@@ -1694,8 +1694,7 @@ def test_builder_save_rejects_traversal(builder_test_client, tmp_path):
           ("app/../escape.yaml", b"nope\n", "application/x-yaml"),
       )],
   )
-  assert response.status_code == 200
-  assert response.json() is False
+  assert response.status_code == 400
   assert not (tmp_path / "escape.yaml").exists()
   assert not (tmp_path / "app" / "tmp" / "escape.yaml").exists()
 
@@ -1709,8 +1708,7 @@ def test_builder_save_rejects_py_files(builder_test_client, tmp_path):
           ("app/agent.py", b"import os\nos.system('id')\n", "text/plain"),
       )],
   )
-  assert response.status_code == 200
-  assert response.json() is False
+  assert response.status_code == 400
   assert not (tmp_path / "app" / "tmp" / "app" / "agent.py").exists()
 
 
@@ -1732,8 +1730,7 @@ def test_builder_save_rejects_non_yaml_extensions(
             (f"app/file{ext}", content, "application/octet-stream"),
         )],
     )
-    assert response.status_code == 200, f"Expected 200 for {ext}"
-    assert response.json() is False, f"Expected False for {ext}"
+    assert response.status_code == 400, f"Expected 400 for {ext}"
 
 
 def test_builder_save_allows_yaml_files(builder_test_client, tmp_path):
@@ -1759,6 +1756,44 @@ def test_builder_save_allows_yaml_files(builder_test_client, tmp_path):
   assert response.json() is True
 
 
+def test_builder_save_rejects_args_key(builder_test_client, tmp_path):
+  """Uploading YAML with an `args` key is rejected (RCE prevention)."""
+  yaml_with_args = b"""\
+name: my_tool
+args:
+  key: value
+"""
+  response = builder_test_client.post(
+      "/builder/save?tmp=true",
+      files=[(
+          "files",
+          ("app/root_agent.yaml", yaml_with_args, "application/x-yaml"),
+      )],
+  )
+  assert response.status_code == 400
+  assert "args" in response.json()["detail"]
+  assert not (tmp_path / "app" / "tmp" / "app" / "root_agent.yaml").exists()
+
+
+def test_builder_save_rejects_nested_args_key(builder_test_client, tmp_path):
+  """Uploading YAML with a nested `args` key is rejected."""
+  yaml_with_nested_args = b"""\
+tools:
+  - name: some_tool
+    args:
+      param: value
+"""
+  response = builder_test_client.post(
+      "/builder/save?tmp=true",
+      files=[(
+          "files",
+          ("app/root_agent.yaml", yaml_with_nested_args, "application/x-yaml"),
+      )],
+  )
+  assert response.status_code == 400
+  assert "args" in response.json()["detail"]
+
+
 def test_builder_get_rejects_non_yaml_file_paths(builder_test_client, tmp_path):
   """GET /builder/app/{app_name}?file_path=... rejects non-YAML extensions."""
   app_root = tmp_path / "app"
diff --git a/tests/unittests/models/test_gemini_llm_connection.py b/tests/unittests/models/test_gemini_llm_connection.py
index 7b580c6fc0..09bd537d8e 100644
--- a/tests/unittests/models/test_gemini_llm_connection.py
+++ b/tests/unittests/models/test_gemini_llm_connection.py
@@ -933,33 +933,142 @@ async def test_receive_tool_call_and_grounding_metadata_with_native_audio(
   mock_metadata_msg.tool_call = None
   mock_metadata_msg.session_resumption_update = None
 
+  # 3. Message with turn_complete
+  mock_turn_complete_content = mock.create_autospec(
+      types.LiveServerContent, instance=True
+  )
+  mock_turn_complete_content.model_turn = None
+  mock_turn_complete_content.grounding_metadata = None
+  mock_turn_complete_content.turn_complete = True
+  mock_turn_complete_content.interrupted = False
+  mock_turn_complete_content.input_transcription = None
+  mock_turn_complete_content.output_transcription = None
+
+  mock_turn_complete_msg = mock.create_autospec(
+      types.LiveServerMessage, instance=True
+  )
+  mock_turn_complete_msg.usage_metadata = None
+  mock_turn_complete_msg.server_content = mock_turn_complete_content
+  mock_turn_complete_msg.tool_call = None
+  mock_turn_complete_msg.session_resumption_update = None
+
   async def mock_receive_generator():
     yield mock_tool_call_msg
     yield mock_metadata_msg
+    yield mock_turn_complete_msg
 
   receive_mock = mock.Mock(return_value=mock_receive_generator())
   mock_gemini_session.receive = receive_mock
 
   responses = [resp async for resp in connection.receive()]
 
-  assert len(responses) == 2
+  assert len(responses) == 3
 
-  # First response: the tool call
+  # First response: the audio content and grounding metadata
+  assert responses[0].grounding_metadata == grounding_metadata
+  assert responses[0].content == mock_content
   assert responses[0].content is not None
   assert responses[0].content.parts is not None
-  assert responses[0].content.parts[0].function_call is not None
+  assert responses[0].content.parts[0].inline_data == audio_blob
+
+  # Second response: the tool call, buffered until turn_complete
+  assert responses[1].content is not None
+  assert responses[1].content.parts is not None
+  assert responses[1].content.parts[0].function_call is not None
   assert (
-      responses[0].content.parts[0].function_call.name
+      responses[1].content.parts[0].function_call.name
       == 'enterprise_web_search'
   )
-  assert responses[0].content.parts[0].function_call.args == {
+  assert responses[1].content.parts[0].function_call.args == {
       'query': 'Google stock price today'
   }
-  assert responses[0].grounding_metadata is None
+  assert responses[1].grounding_metadata is None
 
-  # Second response: the audio content and grounding metadata
-  assert responses[1].grounding_metadata == grounding_metadata
-  assert responses[1].content == mock_content
-  assert responses[1].content is not None
-  assert responses[1].content.parts is not None
-  assert responses[1].content.parts[0].inline_data == audio_blob
+  # Third response: the turn_complete
+  assert responses[2].turn_complete is True
+
+
+@pytest.mark.asyncio
+async def test_receive_multiple_tool_calls_buffered_until_turn_complete(
+    gemini_connection, mock_gemini_session
+):
+  """Test receive buffers multiple tool call messages until turn complete."""
+  # First tool call message
+  mock_tool_call_msg1 = mock.create_autospec(
+      types.LiveServerMessage, instance=True
+  )
+  mock_tool_call_msg1.usage_metadata = None
+  mock_tool_call_msg1.server_content = None
+  mock_tool_call_msg1.session_resumption_update = None
+
+  function_call1 = types.FunctionCall(
+      name='tool_1',
+      args={'arg': 'value1'},
+  )
+  mock_tool_call1 = mock.create_autospec(
+      types.LiveServerToolCall, instance=True
+  )
+  mock_tool_call1.function_calls = [function_call1]
+  mock_tool_call_msg1.tool_call = mock_tool_call1
+
+  # Second tool call message
+  mock_tool_call_msg2 = mock.create_autospec(
+      types.LiveServerMessage, instance=True
+  )
+  mock_tool_call_msg2.usage_metadata = None
+  mock_tool_call_msg2.server_content = None
+  mock_tool_call_msg2.session_resumption_update = None
+
+  function_call2 = types.FunctionCall(
+      name='tool_2',
+      args={'arg': 'value2'},
+  )
+  mock_tool_call2 = mock.create_autospec(
+      types.LiveServerToolCall, instance=True
+  )
+  mock_tool_call2.function_calls = [function_call2]
+  mock_tool_call_msg2.tool_call = mock_tool_call2
+
+  # Turn complete message
+  mock_turn_complete_content = mock.create_autospec(
+      types.LiveServerContent, instance=True
+  )
+  mock_turn_complete_content.model_turn = None
+  mock_turn_complete_content.grounding_metadata = None
+  mock_turn_complete_content.turn_complete = True
+  mock_turn_complete_content.interrupted = False
+  mock_turn_complete_content.input_transcription = None
+  mock_turn_complete_content.output_transcription = None
+
+  mock_turn_complete_msg = mock.create_autospec(
+      types.LiveServerMessage, instance=True
+  )
+  mock_turn_complete_msg.usage_metadata = None
+  mock_turn_complete_msg.server_content = mock_turn_complete_content
+  mock_turn_complete_msg.tool_call = None
+  mock_turn_complete_msg.session_resumption_update = None
+
+  async def mock_receive_generator():
+    yield mock_tool_call_msg1
+    yield mock_tool_call_msg2
+    yield mock_turn_complete_msg
+
+  receive_mock = mock.Mock(return_value=mock_receive_generator())
+  mock_gemini_session.receive = receive_mock
+
+  responses = [resp async for resp in gemini_connection.receive()]
+
+  # Expected: One LlmResponse with both tool calls, then one with turn_complete
+  assert len(responses) == 2
+
+  # First response: single LlmResponse carrying both function calls
+  assert responses[0].content is not None
+  parts = responses[0].content.parts
+  assert len(parts) == 2
+  assert parts[0].function_call.name == 'tool_1'
+  assert parts[0].function_call.args == {'arg': 'value1'}
+  assert parts[1].function_call.name == 'tool_2'
+  assert parts[1].function_call.args == {'arg': 'value2'}
+
+  # Second response: turn_complete True
+  assert responses[1].turn_complete is True