diff --git a/AGENTS.md b/AGENTS.md
index 3d161cbf..5b31af82 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -6,6 +6,10 @@ This document provides context for the Gemini CLI and Gemini Code Assist to unde
 
 The Agent Development Kit (ADK) is an open-source, code-first Python toolkit for building, evaluating, and deploying sophisticated AI agents with flexibility and control. While optimized for Gemini and the Google ecosystem, ADK is model-agnostic, deployment-agnostic, and is built for compatibility with other frameworks. ADK was designed to make agent development feel more like software development, to make it easier for developers to create, deploy, and orchestrate agentic architectures that range from simple tasks to complex workflows.
 
+## Project Architecture
+
+Please refer to [ADK Project Overview and Architecture](https://github.com/google/adk-python/blob/main/contributing/adk_project_overview_and_architecture.md) for details.
+
 ## ADK: Style Guides
 
 ### Python Style Guide
diff --git a/contributing/README.md b/contributing/README.md
index f5099b7b..62dba9df 100644
--- a/contributing/README.md
+++ b/contributing/README.md
@@ -7,3 +7,10 @@ This folder host resources for ADK contributors, for example, testing samples et
 Samples folder host samples to test different features. The samples are usually minimal and simplistic to test one or a few scenarios.
 
 **Note**: This is different from the [google/adk-samples](https://github.com/google/adk-samples) repo, which hosts more complex e2e samples for customers to use or modify directly.
+
+## ADK project and architecture overview
+
+The [adk_project_overview_and_architecture.md](adk_project_overview_and_architecture.md) describes the ADK project overview and its technical architecture from high-level.
+
+This is helpful for contributors to understand the project and design philosophy.
+ It can also be feed into LLMs for vibe-coding.
diff --git a/contributing/adk_project_overview_and_architecture.md b/contributing/adk_project_overview_and_architecture.md
new file mode 100644
index 00000000..106ecb03
--- /dev/null
+++ b/contributing/adk_project_overview_and_architecture.md
@@ -0,0 +1,111 @@
+# ADK Project Overview and Architecture
+
+Google Agent Development Kit (ADK) for Python
+
+## Core Philosophy & Architecture
+
+- Code-First: Everything is defined in Python code for versioning, testing, and IDE support. Avoid GUI-based logic.
+
+- Modularity & Composition: We build complex multi-agent systems by composing multiple, smaller, specialized agents.
+
+- Deployment-Agnostic: The agent's core logic is separate from its deployment environment. The same agent.py can be run locally for testing, served via an API, or deployed to the cloud.
+
+## Foundational Abstractions (Our Vocabulary)
+
+- Agent: The blueprint. It defines an agent's identity, instructions, and tools. It's a declarative configuration object.
+
+- Tool: A capability. A Python function an agent can call to interact with the world (e.g., search, API call).
+
+- Runner: The engine. It orchestrates the "Reason-Act" loop, manages LLM calls, and executes tools.
+
+- Session: The conversation state. It holds the history for a single, continuous dialogue.
+
+- Memory: Long-term recall across different sessions.
+
+- Artifact Service: Manages non-textual data like files.
+
+## Canonical Project Structure
+
+Adhere to this structure for compatibility with ADK tooling.
+
+my_adk_project/ \
+└── src/ \
+    └── my_app/ \
+        ├── agents/ \
+        │   ├── my_agent/ \
+        │   │   ├── __init__.py   # Must contain: from. import agent \
+        │   │   └── agent.py      # Must contain: root_agent = Agent(...) \
+        │   └── another_agent/ \
+        │       ├── __init__.py \
+        │       └── agent.py\
+
+agent.py: Must define the agent and assign it to a variable named root_agent. This is how ADK's tools find it.
+
+__init__.py: In each agent directory, it must contain from. import agent to make the agent discoverable.
+
+## Local Development & Debugging
+
+Interactive UI (adk web): This is our primary debugging tool. It's a decoupled system:
+
+Backend: A FastAPI server started with adk api_server.
+
+Frontend: An Angular app that connects to the backend.
+
+Use the "Events" tab to inspect the full execution trace (prompts, tool calls, responses).
+
+CLI (adk run): For quick, stateless functional checks in the terminal.
+
+Programmatic (pytest): For writing automated unit and integration tests.
+
+## The API Layer (FastAPI)
+
+We expose agents as production APIs using FastAPI.
+
+- get_fast_api_app: This is the key helper function from google.adk.cli.fast_api that creates a FastAPI app from our agent directory.
+
+- Standard Endpoints: The generated app includes standard routes like /list-apps and /run_sse for streaming responses. The wire format is camelCase.
+
+- Custom Endpoints: We can add our own routes (e.g., /health) to the app object returned by the helper.
+
+Python
+
+from google.adk.cli.fast_api import get_fast_api_app
+app = get_fast_api_app(agent_dir="./agents")
+
+@app.get("/health")
+async def health_check():
+    return {"status": "ok"}
+
+
+## Deployment to Production
+
+The adk cli provides the "adk deploy" command to deploy to Google Vertex Agent Engine, Google CloudRun, Google GKE.
+
+## Testing & Evaluation Strategy
+
+Testing is layered, like a pyramid.
+
+### Layer 1: Unit Tests (Base)
+
+What: Test individual Tool functions in isolation.
+
+How: Use pytest in tests/test_tools.py. Verify deterministic logic.
+
+### Layer 2: Integration Tests (Middle)
+
+What: Test the agent's internal logic and interaction with tools.
+
+How: Use pytest in tests/test_agent.py, often with mocked LLMs or services.
+
+### Layer 3: Evaluation Tests (Top)
+
+What: Assess end-to-end performance with a live LLM. This is about quality, not just pass/fail.
+
+How: Use the ADK Evaluation Framework.
+
+Test Cases: Create JSON files with input and a reference (expected tool calls and final response).
+
+Metrics: tool_trajectory_avg_score (does it use tools correctly?) and response_match_score (is the final answer good?).
+
+Run via: adk web (UI), pytest (for CI/CD), or adk eval (CLI).
+
diff --git a/contributing/samples/live_bidi_streaming_tools_agent/__init__.py b/contributing/samples/live_bidi_streaming_tools_agent/__init__.py
new file mode 100644
index 00000000..c48963cd
--- /dev/null
+++ b/contributing/samples/live_bidi_streaming_tools_agent/__init__.py
@@ -0,0 +1,15 @@
+# Copyright 2025 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_tools_agent/agent.py b/contributing/samples/live_bidi_streaming_tools_agent/agent.py
new file mode 100644
index 00000000..cdb09217
--- /dev/null
+++ b/contributing/samples/live_bidi_streaming_tools_agent/agent.py
@@ -0,0 +1,140 @@
+# Copyright 2025 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.
+
+import asyncio
+from typing import AsyncGenerator
+
+from google.adk.agents import LiveRequestQueue
+from google.adk.agents.llm_agent import Agent
+from google.adk.tools.function_tool import FunctionTool
+from google.genai import Client
+from google.genai import types as genai_types
+
+
+async def monitor_stock_price(stock_symbol: str) -> AsyncGenerator[str, None]:
+  """This function will monitor the price for the given stock_symbol in a continuous, streaming and asynchronously way."""
+  print(f"Start monitor stock price for {stock_symbol}!")
+
+  # Let's mock stock price change.
+  await asyncio.sleep(4)
+  price_alert1 = f"the price for {stock_symbol} is 300"
+  yield price_alert1
+  print(price_alert1)
+
+  await asyncio.sleep(4)
+  price_alert1 = f"the price for {stock_symbol} is 400"
+  yield price_alert1
+  print(price_alert1)
+
+  await asyncio.sleep(20)
+  price_alert1 = f"the price for {stock_symbol} is 900"
+  yield price_alert1
+  print(price_alert1)
+
+  await asyncio.sleep(20)
+  price_alert1 = f"the price for {stock_symbol} is 500"
+  yield price_alert1
+  print(price_alert1)
+
+
+# for video streaming, `input_stream: LiveRequestQueue` is required and reserved key parameter for ADK to pass the video streams in.
+async def monitor_video_stream(
+    input_stream: LiveRequestQueue,
+) -> AsyncGenerator[str, None]:
+  """Monitor how many people are in the video streams."""
+  print("start monitor_video_stream!")
+  client = Client(vertexai=False)
+  prompt_text = (
+      "Count the number of people in this image. Just respond with a numeric"
+      " number."
+  )
+  last_count = None
+  while True:
+    last_valid_req = None
+    print("Start monitoring loop")
+
+    # use this loop to pull the latest images and discard the old ones
+    while input_stream._queue.qsize() != 0:
+      live_req = await input_stream.get()
+
+      if live_req.blob is not None and live_req.blob.mime_type == "image/jpeg":
+        last_valid_req = live_req
+
+    # If we found a valid image, process it
+    if last_valid_req is not None:
+      print("Processing the most recent frame from the queue")
+
+      # Create an image part using the blob's data and mime type
+      image_part = genai_types.Part.from_bytes(
+          data=last_valid_req.blob.data, mime_type=last_valid_req.blob.mime_type
+      )
+
+      contents = genai_types.Content(
+          role="user",
+          parts=[image_part, genai_types.Part.from_text(text=prompt_text)],
+      )
+
+      # Call the model to generate content based on the provided image and prompt
+      response = client.models.generate_content(
+          model="gemini-2.0-flash-exp",
+          contents=contents,
+          config=genai_types.GenerateContentConfig(
+              system_instruction=(
+                  "You are a helpful video analysis assistant. You can count"
+                  " the number of people in this image or video. Just respond"
+                  " with a numeric number."
+              )
+          ),
+      )
+      if not last_count:
+        last_count = response.candidates[0].content.parts[0].text
+      elif last_count != response.candidates[0].content.parts[0].text:
+        last_count = response.candidates[0].content.parts[0].text
+        yield response
+        print("response:", response)
+
+    # Wait before checking for new images
+    await asyncio.sleep(0.5)
+
+
+# Use this exact function to help ADK stop your streaming tools when requested.
+# for example, if we want to stop `monitor_stock_price`, then the agent will
+# invoke this function with stop_streaming(function_name=monitor_stock_price).
+def stop_streaming(function_name: str):
+  """Stop the streaming
+
+  Args:
+    function_name: The name of the streaming function to stop.
+  """
+  pass
+
+
+root_agent = Agent(
+    model="gemini-live-2.5-flash-preview",
+    name="video_streaming_agent",
+    instruction="""
+      You are a monitoring agent. You can do video monitoring and stock price monitoring
+      using the provided tools/functions.
+      When users want to monitor a video stream,
+      You can use monitor_video_stream function to do that. When monitor_video_stream
+      returns the alert, you should tell the users.
+      When users want to monitor a stock price, you can use monitor_stock_price.
+      Don't ask too many questions. Don't be too talkative.
+    """,
+    tools=[
+        monitor_video_stream,
+        monitor_stock_price,
+        FunctionTool(stop_streaming),
+    ],
+)
diff --git a/contributing/samples/live_bidi_streaming_tools_agent/readme.md b/contributing/samples/live_bidi_streaming_tools_agent/readme.md
new file mode 100644
index 00000000..fe44fa39
--- /dev/null
+++ b/contributing/samples/live_bidi_streaming_tools_agent/readme.md
@@ -0,0 +1,19 @@
+ This is only supported in streaming(live) agents/api.
+
+Streaming tools allows tools(functions) to stream intermediate results back to agents and agents can respond to those intermediate results. 
+For example, we can use streaming tools to monitor the changes of the stock price and have the agent react to it. Another example is we can have the agent monitor the video stream, and when there is changes in video stream, the agent can report the changes.
+
+To define a streaming tool, you must adhere to the following:
+
+1.  **Asynchronous Function:** The tool must be an `async` Python function.
+2.  **AsyncGenerator Return Type:** The function must be typed to return an `AsyncGenerator`. The first type parameter to `AsyncGenerator` is the type of the data you `yield` (e.g., `str` for text messages, or a custom object for structured data). The second type parameter is typically `None` if the generator doesn't receive values via `send()`.
+
+
+We support two types of streaming tools:
+- Simple type. This is a one type of streaming tools that only take non video/audio streams(the streams that you feed to adk web or adk runner) as input.
+- Video streaming tools. This only works in video streaming and the video stream(the streams that you feed to adk web or adk runner) will be passed into this function.
+
+
+Here are some sample queries to test:
+- Help me monitor the stock price for $XYZ stock.
+- Help me monitor how many people are there in the video stream.
\ No newline at end of file
diff --git a/src/google/adk/artifacts/in_memory_artifact_service.py b/src/google/adk/artifacts/in_memory_artifact_service.py
index 1dd724bb..e4837695 100644
--- a/src/google/adk/artifacts/in_memory_artifact_service.py
+++ b/src/google/adk/artifacts/in_memory_artifact_service.py
@@ -11,8 +11,7 @@
 # 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.
-
-"""An in-memory implementation of the artifact service."""
+from __future__ import annotations
 
 import logging
 from typing import Optional
@@ -28,7 +27,11 @@ logger = logging.getLogger("google_adk." + __name__)
 
 
 class InMemoryArtifactService(BaseArtifactService, BaseModel):
-  """An in-memory implementation of the artifact service."""
+  """An in-memory implementation of the artifact service.
+
+  It is not suitable for multi-threaded production environments. Use it for
+  testing and development only.
+  """
 
   artifacts: dict[str, list[types.Part]] = Field(default_factory=dict)
 
diff --git a/src/google/adk/events/event.py b/src/google/adk/events/event.py
index 6dd617ff..48104fc4 100644
--- a/src/google/adk/events/event.py
+++ b/src/google/adk/events/event.py
@@ -14,9 +14,8 @@
 from __future__ import annotations
 
 from datetime import datetime
-import random
-import string
 from typing import Optional
+import uuid
 
 from google.genai import types
 from pydantic import alias_generators
@@ -132,5 +131,4 @@ class Event(LlmResponse):
 
   @staticmethod
   def new_id():
-    characters = string.ascii_letters + string.digits
-    return ''.join(random.choice(characters) for _ in range(8))
+    return str(uuid.uuid4())
diff --git a/src/google/adk/memory/in_memory_memory_service.py b/src/google/adk/memory/in_memory_memory_service.py
index a49aca5b..129ba5bc 100644
--- a/src/google/adk/memory/in_memory_memory_service.py
+++ b/src/google/adk/memory/in_memory_memory_service.py
@@ -11,8 +11,6 @@
 # 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 __future__ import annotations
 
 import re
@@ -43,6 +41,9 @@ class InMemoryMemoryService(BaseMemoryService):
   """An in-memory memory service for prototyping purpose only.
 
   Uses keyword matching instead of semantic search.
+
+  It is not suitable for multi-threaded production environments. Use it for
+  testing and development only.
   """
 
   def __init__(self):
diff --git a/src/google/adk/runners.py b/src/google/adk/runners.py
index 017997bb..a95f1081 100644
--- a/src/google/adk/runners.py
+++ b/src/google/adk/runners.py
@@ -307,27 +307,44 @@ class Runner:
     root_agent = self.agent
     invocation_context.agent = self._find_agent_to_run(session, root_agent)
 
+    # Pre-processing for live streaming tools
+    # Inspect the tool's parameters to find if it uses LiveRequestQueue
     invocation_context.active_streaming_tools = {}
     # TODO(hangfei): switch to use canonical_tools.
     # for shell agents, there is no tools associated with it so we should skip.
     if hasattr(invocation_context.agent, 'tools'):
-      for tool in invocation_context.agent.tools:
-        # replicate a LiveRequestQueue for streaming tools that relis on
-        # LiveRequestQueue
-        from typing import get_type_hints
+      import inspect
 
-        type_hints = get_type_hints(tool)
-        for arg_type in type_hints.values():
-          if arg_type is LiveRequestQueue:
+      for tool in invocation_context.agent.tools:
+        # We use `inspect.signature()` to examine the tool's underlying function (`tool.func`).
+        # This approach is deliberately chosen over `typing.get_type_hints()` for robustness.
+        #
+        # The Problem with `get_type_hints()`:
+        # `get_type_hints()` attempts to resolve forward-referenced (string-based) type
+        # annotations. This resolution can easily fail with a `NameError` (e.g., "Union not found")
+        # if the type isn't available in the scope where `get_type_hints()` is called.
+        # This is a common and brittle issue in framework code that inspects functions
+        # defined in separate user modules.
+        #
+        # Why `inspect.signature()` is Better Here:
+        # `inspect.signature()` does NOT resolve the annotations; it retrieves the raw
+        # annotation object as it was defined on the function. This allows us to
+        # perform a direct and reliable identity check (`param.annotation is LiveRequestQueue`)
+        # without risking a `NameError`.
+        callable_to_inspect = tool.func if hasattr(tool, 'func') else tool
+        # Ensure the target is actually callable before inspecting to avoid errors.
+        if not callable(callable_to_inspect):
+          continue
+        for param in inspect.signature(callable_to_inspect).parameters.values():
+          if param.annotation is LiveRequestQueue:
             if not invocation_context.active_streaming_tools:
               invocation_context.active_streaming_tools = {}
-            active_streaming_tools = ActiveStreamingTool(
+            active_streaming_tool = ActiveStreamingTool(
                 stream=LiveRequestQueue()
             )
             invocation_context.active_streaming_tools[tool.__name__] = (
-                active_streaming_tools
+                active_streaming_tool
             )
-
     async for event in invocation_context.agent.run_live(invocation_context):
       await self.session_service.append_event(session=session, event=event)
       yield event
diff --git a/src/google/adk/sessions/in_memory_session_service.py b/src/google/adk/sessions/in_memory_session_service.py
index b2a84eff..70e75411 100644
--- a/src/google/adk/sessions/in_memory_session_service.py
+++ b/src/google/adk/sessions/in_memory_session_service.py
@@ -33,7 +33,11 @@ logger = logging.getLogger('google_adk.' + __name__)
 
 
 class InMemorySessionService(BaseSessionService):
-  """An in-memory implementation of the session service."""
+  """An in-memory implementation of the session service.
+
+  It is not suitable for multi-threaded production environments. Use it for
+  testing and development only.
+  """
 
   def __init__(self):
     # A map from app name to a map from user ID to a map from session ID to
diff --git a/src/google/adk/sessions/vertex_ai_session_service.py b/src/google/adk/sessions/vertex_ai_session_service.py
index eac13367..ebd98e36 100644
--- a/src/google/adk/sessions/vertex_ai_session_service.py
+++ b/src/google/adk/sessions/vertex_ai_session_service.py
@@ -381,6 +381,7 @@ def _convert_event_to_json(event: Event) -> Dict[str, Any]:
       'turn_complete': event.turn_complete,
       'interrupted': event.interrupted,
       'branch': event.branch,
+      'custom_metadata': event.custom_metadata,
       'long_running_tool_ids': (
           list(event.long_running_tool_ids)
           if event.long_running_tool_ids
@@ -460,6 +461,9 @@ def _from_api_event(api_event: Dict[str, Any]) -> Event:
     event.turn_complete = api_event['eventMetadata'].get('turnComplete', None)
     event.interrupted = api_event['eventMetadata'].get('interrupted', None)
     event.branch = api_event['eventMetadata'].get('branch', None)
+    event.custom_metadata = api_event['eventMetadata'].get(
+        'customMetadata', None
+    )
     event.grounding_metadata = _session_util.decode_grounding_metadata(
         api_event['eventMetadata'].get('groundingMetadata', None)
     )
diff --git a/tests/unittests/streaming/test_streaming.py b/tests/unittests/streaming/test_streaming.py
index c1e1eaad..0754d3df 100644
--- a/tests/unittests/streaming/test_streaming.py
+++ b/tests/unittests/streaming/test_streaming.py
@@ -14,6 +14,7 @@
 
 from google.adk.agents import Agent
 from google.adk.agents import LiveRequestQueue
+from google.adk.agents.run_config import RunConfig
 from google.adk.models import LlmResponse
 from google.genai import types
 import pytest
@@ -47,3 +48,394 @@ def test_streaming():
   assert (
       len(res_events) > 0
   ), 'Expected at least one response, but got an empty list.'
+
+
+def test_streaming_with_output_audio_transcription():
+  """Test streaming with output audio transcription configuration."""
+  response1 = LlmResponse(
+      turn_complete=True,
+  )
+
+  mock_model = testing_utils.MockModel.create([response1])
+
+  root_agent = Agent(
+      name='root_agent',
+      model=mock_model,
+      tools=[],
+  )
+
+  runner = testing_utils.InMemoryRunner(
+      root_agent=root_agent, response_modalities=['AUDIO']
+  )
+
+  # Create run config with output audio transcription
+  run_config = RunConfig(
+      output_audio_transcription=types.AudioTranscriptionConfig()
+  )
+
+  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.'
+
+
+def test_streaming_with_input_audio_transcription():
+  """Test streaming with input audio transcription configuration."""
+  response1 = LlmResponse(
+      turn_complete=True,
+  )
+
+  mock_model = testing_utils.MockModel.create([response1])
+
+  root_agent = Agent(
+      name='root_agent',
+      model=mock_model,
+      tools=[],
+  )
+
+  runner = testing_utils.InMemoryRunner(
+      root_agent=root_agent, response_modalities=['AUDIO']
+  )
+
+  # Create run config with input audio transcription
+  run_config = RunConfig(
+      input_audio_transcription=types.AudioTranscriptionConfig()
+  )
+
+  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.'
+
+
+def test_streaming_with_realtime_input_config():
+  """Test streaming with realtime input configuration."""
+  response1 = LlmResponse(
+      turn_complete=True,
+  )
+
+  mock_model = testing_utils.MockModel.create([response1])
+
+  root_agent = Agent(
+      name='root_agent',
+      model=mock_model,
+      tools=[],
+  )
+
+  runner = testing_utils.InMemoryRunner(
+      root_agent=root_agent, response_modalities=['AUDIO']
+  )
+
+  # Create run config with realtime input config
+  run_config = RunConfig(
+      realtime_input_config=types.RealtimeInputConfig(
+          automatic_activity_detection=types.AutomaticActivityDetection(
+              disabled=True
+          )
+      )
+  )
+
+  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.'
+
+
+def test_streaming_with_realtime_input_config_vad_enabled():
+  """Test streaming with realtime input configuration with VAD enabled."""
+  response1 = LlmResponse(
+      turn_complete=True,
+  )
+
+  mock_model = testing_utils.MockModel.create([response1])
+
+  root_agent = Agent(
+      name='root_agent',
+      model=mock_model,
+      tools=[],
+  )
+
+  runner = testing_utils.InMemoryRunner(
+      root_agent=root_agent, response_modalities=['AUDIO']
+  )
+
+  # Create run config with realtime input config with VAD enabled
+  run_config = RunConfig(
+      realtime_input_config=types.RealtimeInputConfig(
+          automatic_activity_detection=types.AutomaticActivityDetection(
+              disabled=False
+          )
+      )
+  )
+
+  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.'
+
+
+def test_streaming_with_enable_affective_dialog_true():
+  """Test streaming with affective dialog enabled."""
+  response1 = LlmResponse(
+      turn_complete=True,
+  )
+
+  mock_model = testing_utils.MockModel.create([response1])
+
+  root_agent = Agent(
+      name='root_agent',
+      model=mock_model,
+      tools=[],
+  )
+
+  runner = testing_utils.InMemoryRunner(
+      root_agent=root_agent, response_modalities=['AUDIO']
+  )
+
+  # Create run config with affective dialog enabled
+  run_config = RunConfig(enable_affective_dialog=True)
+
+  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.'
+
+
+def test_streaming_with_enable_affective_dialog_false():
+  """Test streaming with affective dialog disabled."""
+  response1 = LlmResponse(
+      turn_complete=True,
+  )
+
+  mock_model = testing_utils.MockModel.create([response1])
+
+  root_agent = Agent(
+      name='root_agent',
+      model=mock_model,
+      tools=[],
+  )
+
+  runner = testing_utils.InMemoryRunner(
+      root_agent=root_agent, response_modalities=['AUDIO']
+  )
+
+  # Create run config with affective dialog disabled
+  run_config = RunConfig(enable_affective_dialog=False)
+
+  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.'
+
+
+def test_streaming_with_proactivity_config():
+  """Test streaming with proactivity configuration."""
+  response1 = LlmResponse(
+      turn_complete=True,
+  )
+
+  mock_model = testing_utils.MockModel.create([response1])
+
+  root_agent = Agent(
+      name='root_agent',
+      model=mock_model,
+      tools=[],
+  )
+
+  runner = testing_utils.InMemoryRunner(
+      root_agent=root_agent, response_modalities=['AUDIO']
+  )
+
+  # Create run config with proactivity config
+  run_config = RunConfig(proactivity=types.ProactivityConfig())
+
+  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.'
+
+
+def test_streaming_with_combined_audio_transcription_configs():
+  """Test streaming with both input and output audio transcription configurations."""
+  response1 = LlmResponse(
+      turn_complete=True,
+  )
+
+  mock_model = testing_utils.MockModel.create([response1])
+
+  root_agent = Agent(
+      name='root_agent',
+      model=mock_model,
+      tools=[],
+  )
+
+  runner = testing_utils.InMemoryRunner(
+      root_agent=root_agent, response_modalities=['AUDIO']
+  )
+
+  # Create run config with both input and output audio transcription
+  run_config = RunConfig(
+      input_audio_transcription=types.AudioTranscriptionConfig(),
+      output_audio_transcription=types.AudioTranscriptionConfig(),
+  )
+
+  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.'
+
+
+def test_streaming_with_all_configs_combined():
+  """Test streaming with all the new configurations combined."""
+  response1 = LlmResponse(
+      turn_complete=True,
+  )
+
+  mock_model = testing_utils.MockModel.create([response1])
+
+  root_agent = Agent(
+      name='root_agent',
+      model=mock_model,
+      tools=[],
+  )
+
+  runner = testing_utils.InMemoryRunner(
+      root_agent=root_agent, response_modalities=['AUDIO']
+  )
+
+  # Create run config with all configurations
+  run_config = RunConfig(
+      output_audio_transcription=types.AudioTranscriptionConfig(),
+      input_audio_transcription=types.AudioTranscriptionConfig(),
+      realtime_input_config=types.RealtimeInputConfig(
+          automatic_activity_detection=types.AutomaticActivityDetection(
+              disabled=True
+          )
+      ),
+      enable_affective_dialog=True,
+      proactivity=types.ProactivityConfig(),
+  )
+
+  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.'
+
+
+def test_streaming_config_validation():
+  """Test that run_config values are properly set and accessible."""
+  # Test that RunConfig properly validates and stores the configurations
+  run_config = RunConfig(
+      output_audio_transcription=types.AudioTranscriptionConfig(),
+      input_audio_transcription=types.AudioTranscriptionConfig(),
+      realtime_input_config=types.RealtimeInputConfig(
+          automatic_activity_detection=types.AutomaticActivityDetection(
+              disabled=False
+          )
+      ),
+      enable_affective_dialog=True,
+      proactivity=types.ProactivityConfig(),
+  )
+
+  # Verify configurations are properly set
+  assert run_config.output_audio_transcription is not None
+  assert run_config.input_audio_transcription is not None
+  assert run_config.realtime_input_config is not None
+  assert (
+      run_config.realtime_input_config.automatic_activity_detection.disabled
+      == False
+  )
+  assert run_config.enable_affective_dialog == True
+  assert run_config.proactivity is not None
+
+
+def test_streaming_with_multiple_audio_configs():
+  """Test streaming with multiple audio transcription configurations."""
+  response1 = LlmResponse(
+      turn_complete=True,
+  )
+
+  mock_model = testing_utils.MockModel.create([response1])
+
+  root_agent = Agent(
+      name='root_agent',
+      model=mock_model,
+      tools=[],
+  )
+
+  runner = testing_utils.InMemoryRunner(
+      root_agent=root_agent, response_modalities=['AUDIO']
+  )
+
+  # Create run config with multiple audio transcription configs
+  run_config = RunConfig(
+      input_audio_transcription=types.AudioTranscriptionConfig(),
+      output_audio_transcription=types.AudioTranscriptionConfig(),
+      enable_affective_dialog=True,
+  )
+
+  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.'