feat: Add support for openai realtime api in adk

src/google/adk/agents/base_agent.py

@@ -283,6 +283,41 @@ async def _run_with_trace() -> AsyncGenerator[Event, None]:
       async for event in agen:
         yield event
 
+  @final
+  async def run_realtime(
+      self,
+      parent_context: InvocationContext,
+  ) -> AsyncGenerator[Event, None]:
+    """Entry method to run an agent via video/audio-based conversation.
+
+    Args:
+      parent_context: InvocationContext, the invocation context of the parent
+        agent.
+
+    Yields:
+      Event: the events generated by the agent.
+    """
+
+    async def _run_with_trace() -> AsyncGenerator[Event, None]:
+      with tracer.start_as_current_span(f'agent_run [{self.name}]'):
+        ctx = self._create_invocation_context(parent_context)
+
+        if event := await self.__handle_before_agent_callback(ctx):
+          yield event
+        if ctx.end_invocation:
+          return
+
+        async with Aclosing(self._run_realtime_impl(ctx)) as agen:
+          async for event in agen:
+            yield event
+
+        if event := await self.__handle_after_agent_callback(ctx):
+          yield event
+
+    async with Aclosing(_run_with_trace()) as agen:
+      async for event in agen:
+        yield event
+
   async def _run_async_impl(
       self, ctx: InvocationContext
   ) -> AsyncGenerator[Event, None]:
@@ -315,6 +350,22 @@ async def _run_live_impl(
     )
     yield  # AsyncGenerator requires having at least one yield statement
 
+  async def _run_realtime_impl(
+      self, ctx: InvocationContext
+  ) -> AsyncGenerator[Event, None]:
+    """Core logic to run this agent via video/audio-based conversation.
+
+    Args:
+      ctx: InvocationContext, the invocation context for this agent.
+
+    Yields:
+      Event: the events generated by the agent.
+    """
+    raise NotImplementedError(
+        f'_run_live_impl for {type(self)} is not implemented.'
+    )
+    yield  # AsyncGenerator requires having at least one yield statement
+
   @property
   def root_agent(self) -> BaseAgent:
     """Gets the root agent of this agent."""

src/google/adk/agents/llm_agent.py

@@ -40,10 +40,14 @@
 from ..events.event import Event
 from ..flows.llm_flows.auto_flow import AutoFlow
 from ..flows.llm_flows.base_llm_flow import BaseLlmFlow
+from ..flows.llm_flows.openai_llm_flow import OpenAILlmFlow
+from ..flows.llm_flows.openai_llm_flow import OpenAutoFlow
+from ..flows.llm_flows.openai_llm_flow import OpenSingleFlow
 from ..flows.llm_flows.single_flow import SingleFlow
 from ..models.base_llm import BaseLlm
 from ..models.llm_request import LlmRequest
 from ..models.llm_response import LlmResponse
+from ..models.openai_llm import OpenAIRealtime
 from ..models.registry import LLMRegistry
 from ..planners.base_planner import BasePlanner
 from ..tools.base_tool import BaseTool
@@ -300,6 +304,17 @@ async def _run_live_impl(
       if ctx.end_invocation:
         return
 
+  @override
+  async def _run_realtime_impl(
+      self, ctx: InvocationContext
+  ) -> AsyncGenerator[Event, None]:
+    async with Aclosing(self._open_llm_flow.run_live(ctx)) as agen:
+      async for event in agen:
+        self.__maybe_save_output_to_state(event)
+        yield event
+      if ctx.end_invocation:
+        return
+
   @property
   def canonical_model(self) -> BaseLlm:
     """The resolved self.model field as BaseLlm.
@@ -443,6 +458,17 @@ def _llm_flow(self) -> BaseLlmFlow:
     else:
       return AutoFlow()
 
+  @property
+  def _open_llm_flow(self) -> OpenAILlmFlow:
+    if (
+        self.disallow_transfer_to_parent
+        and self.disallow_transfer_to_peers
+        and not self.sub_agents
+    ):
+      return OpenSingleFlow()
+    else:
+      return OpenAutoFlow()
+
   def __maybe_save_output_to_state(self, event: Event):
     """Saves the model output to state if needed."""
     # skip if the event was authored by some other agent (e.g. current agent

src/google/adk/agents/run_config.py

@@ -17,6 +17,7 @@
 from enum import Enum
 import logging
 import sys
+from typing import Any
 from typing import Optional
 
 from google.genai import types
@@ -107,3 +108,30 @@ def validate_max_llm_calls(cls, value: int) -> int:
       )
 
     return value
+
+  # ---------------- OpenAI Realtime (vendor-specific) ----------------
+  openai_realtime_session: Optional[Any] = None
+  """Optional raw OpenAI Realtime session settings.
+
+  These will be forwarded into the OpenAI Realtime `session.update` payload.
+  Examples:
+    {"modalities": ["text"], "turn_detection": {"type": "none"}, "voice": "ash"}
+
+  Accepts either a Python dict or a JSON string. Prefer generic fields
+  (response_modalities, realtime_input_config) when possible. Use this for
+  OpenAI-specific knobs not covered by the generic config.
+
+  Example (dict) with audio configuration:
+    {
+      "modalities": ["audio", "text"],
+      "voice": "ash",
+      "input_audio_format": "pcm16",
+      "output_audio_format": "pcm16",
+      "input_audio_transcription": {"model": "gpt-4o-mini-transcribe", "language": "fr"},
+      "turn_detection": {"type": "semantic_vad"}
+    }
+
+  Notes:
+    - If you pass a JSON string instead of a dict, it will be parsed the same
+      way (single quotes are not valid JSON; use double quotes).
+  """

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

@@ -82,6 +82,24 @@ async def run_async(
         invocation_context.run_config.session_resumption
     )
 
+    # Vendor-specific: OpenAI Realtime session settings passthrough
+    # These will be consumed by the OpenAI provider when building session.update
+    openai_session = getattr(
+        invocation_context.run_config, 'openai_realtime_session', None
+    )
+    if openai_session is not None:
+      if llm_request.config.labels is None:
+        llm_request.config.labels = {}
+      # Prefer dict passthrough; if a JSON string is provided, pass as-is.
+      payload = openai_session
+      try:
+        # Normalize Pydantic/GenAI types to plain JSON if available
+        if hasattr(types, 'any_serialize'):
+          payload = types.any_serialize(openai_session)
+      except Exception:
+        pass
+      llm_request.config.labels['adk_openai_session_json'] = payload
+
     # TODO: handle tool append here, instead of in BaseTool.process_llm_request.
 
     return