diff --git a/CHANGELOG.md b/CHANGELOG.md
index daeae5d61..24484687f 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -15,8 +15,155 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   and `emotion` (60+ options) parameters for fine-grained speech generation
   control.
 
+- Expanded support for univeral `LLMContext` to `OpenAIRealtimeLLMService`.
+  As a reminder, the context-setup pattern when using `LLMContext` is:
+
+  ```python
+  context = LLMContext(messages, tools)
+  context_aggregator = LLMContextAggregatorPair(
+    context,
+    # This part is `OpenAIRealtimeLLMService`-specific.
+    # `expect_stripped_words=False` needed when OpenAI Realtime used with
+    # "audio" modality (the default).
+    assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False),
+  )
+  ```
+
+  (Note that even though `OpenAIRealtimeLLMService` now supports the universal
+  `LLMContext`, it is not meant to be swapped out for another LLM service at
+  runtime with `LLMSwitcher`.)
+
+  Note: `TranscriptionFrame`s and `InterimTranscriptionFrame`s now go upstream
+  from `OpenAIRealtimeLLMService`, so if you're using `TranscriptProcessor`,
+  say, you'll want to adjust accordingly:
+
+  ```python
+  pipeline = Pipeline(
+    [
+      transport.input(),
+      context_aggregator.user(),
+
+      # BEFORE
+      llm,
+      transcript.user(),
+
+      # AFTER
+      transcript.user(),
+      llm,
+
+      transport.output(),
+      transcript.assistant(),
+      context_aggregator.assistant(),
+    ]
+  )
+  ```
+
+  Also worth noting: whether or not you use the new context-setup pattern with
+  `OpenAIRealtimeLLMService`, some types have changed under the hood:
+
+  ```python
+  ## BEFORE:
+
+  # Context aggregator type
+  context_aggregator: OpenAIContextAggregatorPair
+
+  # Context frame type
+  frame: OpenAILLMContextFrame
+
+  # Context type
+  context: OpenAIRealtimeLLMContext
+  # or
+  context: OpenAILLMContext
+
+  ## AFTER:
+
+  # Context aggregator type
+  context_aggregator: LLMContextAggregatorPair
+
+  # Context frame type
+  frame: LLMContextFrame
+
+  # Context type
+  context: LLMContext
+  ```
+
+  Also note that `RealtimeMessagesUpdateFrame` and
+  `RealtimeFunctionCallResultFrame` have been deprecated, since they're no
+  longer used by `OpenAIRealtimeLLMService`. OpenAI Realtime now works more
+  like other LLM services in Pipecat, relying on updates to its context, pushed
+  by context aggregators, to update its internal state. Listen for
+  `LLMContextFrame`s for context updates.
+
+  Finally, `LLMTextFrame`s are no longer pushed from `OpenAIRealtimeLLMService`
+  when it's configured with `output_modalities=['audio']`. If you need
+  to process its output, listen for `TTSTextFrame`s instead.
+
+- Expanded support for universal `LLMContext` to `GeminiLiveLLMService`.
+  As a reminder, the context-setup pattern when using `LLMContext` is:
+
+  ```python
+  context = LLMContext(messages, tools)
+  context_aggregator = LLMContextAggregatorPair(
+    context,
+    # This part is `GeminiLiveLLMService`-specific.
+    # `expect_stripped_words=False` needed when Gemini Live used with AUDIO
+    #  modality (the default).
+    assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False),
+  )
+  ```
+
+  (Note that even though `GeminiLiveLLMService` now supports the universal
+  `LLMContext`, it is not meant to be swapped out for another LLM service at
+  runtime with `LLMSwitcher`.)
+
+  Worth noting: whether or not you use the new context-setup pattern with
+  `GeminiLiveLLMService`, some types have changed under the hood:
+
+  ```python
+  ## BEFORE:
+
+  # Context aggregator type
+  context_aggregator: GeminiLiveContextAggregatorPair
+
+  # Context frame type
+  frame: OpenAILLMContextFrame
+
+  # Context type
+  context: GeminiLiveLLMContext
+  # or
+  context: OpenAILLMContext
+
+  ## AFTER:
+
+  # Context aggregator type
+  context_aggregator: LLMContextAggregatorPair
+
+  # Context frame type
+  frame: LLMContextFrame
+
+  # Context type
+  context: LLMContext
+  ```
+
+  Also note that `LLMTextFrame`s are no longer pushed from `GeminiLiveLLMService`
+  when it's configured with `modalities=GeminiModalities.AUDIO`. If you need
+  to process its output, listen for `TTSTextFrame`s instead.
+
 ### Changed
 
+- `DailyTransport` triggers `on_error` event if transcription can't be started
+  or stopped.
+
+- `DailyTransport` updates: `start_dialout()` now returns two values:
+  `session_id` and `error`. `start_recording()` now returns two values:
+  `stream_id` and `error`.
+
+- Updated `daily-python` to 0.21.0.
+
+- `SimliVideoService` now accepts `api_key` and `face_id` parameters directly,
+  with optional `params` for `max_session_length` and `max_idle_time`
+  configuration, aligning with other Pipecat service patterns.
+
 - Updated the default model to `sonic-3` for `CartesiaTTSService` and
   `CartesiaHttpTTSService`.
 
@@ -26,6 +173,19 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Upgraded `aws_sdk_bedrock_runtime` to v0.1.1 to resolve potential CPU issues
   when running `AWSNovaSonicLLMService`.
 
+### Deprecated
+
+- The `send_transcription_frames` argument to `OpenAIRealtimeLLMService` is
+  deprecated. Transcription frames are now always sent. They go upstream, to be
+  handled by the user context aggregator. See "Added" section for details.
+
+- Types in `pipecat.services.openai.realtime.context` and
+  `pipecat.services.openai.realtime.frames` are deprecated, as they're no
+  longer used by `OpenAIRealtimeLLMService`. See "Added" section for details.
+
+- `SimliVideoService` `simli_config` parameter is deprecated. Use `api_key` and
+  `face_id` parameters instead.
+
 ### Removed
 
 - Removed the `aiohttp_session` arg from `SarvamTTSService` as it's no longer
@@ -42,6 +202,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Fixed an issue in `ServiceSwitcher` where the `STTService`s would result in
   all STT services producing `TranscriptionFrame`s.
 
+- Fixed an issue in `HumeTTSService` that was only using Octave 2, which does not support the `description` field. Now, if a description is provided, it switches to Octave 1.
+
 ## [0.0.91] - 2025-10-21
 
 ### Added
@@ -63,7 +225,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
   (Note that even though `AWSNovaSonicLLMService` now supports the universal
   `LLMContext`, it is not meant to be swapped out for another LLM service at
-  runtime.)
+  runtime with `LLMSwitcher`.)
 
   Worth noting: whether or not you use the new context-setup pattern with
   `AWSNovaSonicLLMService`, some types have changed under the hood:
@@ -142,8 +304,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   deprecated. Transcription frames are now always sent. They go upstream, to be
   handled by the user context aggregator. See "Added" section for details.
 
-- Types in `pipecat.services.aws.nova_sonic.context` have been deprecated due
-  to changes to support `LLMContext`. See "Changed" section for details.
+- Types in `pipecat.services.aws.nova_sonic.context` are deprecated, as they're
+  no longer used by `AWSNovaSonicLLMService`. See "Added" section for
+  details.
 
 ### Fixed
 
diff --git a/examples/foundational/19-openai-realtime.py b/examples/foundational/19-openai-realtime.py
index f182d7c8c..6907ec196 100644
--- a/examples/foundational/19-openai-realtime.py
+++ b/examples/foundational/19-openai-realtime.py
@@ -5,6 +5,7 @@
 #
 
 
+import asyncio
 import os
 from datetime import datetime
 
@@ -14,12 +15,14 @@ from loguru import logger
 from pipecat.adapters.schemas.function_schema import FunctionSchema
 from pipecat.adapters.schemas.tools_schema import ToolsSchema
 from pipecat.audio.vad.silero import SileroVADAnalyzer
-from pipecat.frames.frames import LLMRunFrame, TranscriptionMessage
+from pipecat.frames.frames import LLMRunFrame, LLMSetToolsFrame, TranscriptionMessage
 from pipecat.observers.loggers.transcription_log_observer import TranscriptionLogObserver
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
-from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.processors.aggregators.llm_response import LLMAssistantAggregatorParams
+from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
 from pipecat.processors.transcript_processor import TranscriptProcessor
 from pipecat.runner.types import RunnerArguments
 from pipecat.runner.utils import create_transport
@@ -52,6 +55,18 @@ async def fetch_weather_from_api(params: FunctionCallParams):
     )
 
 
+async def get_news(params: FunctionCallParams):
+    await params.result_callback(
+        {
+            "news": [
+                "Massive UFO currently hovering above New York City",
+                "Stock markets reach all-time highs",
+                "Living dinosaur species discovered in the Amazon rainforest",
+            ],
+        }
+    )
+
+
 async def fetch_restaurant_recommendation(params: FunctionCallParams):
     await params.result_callback({"name": "The Golden Dragon"})
 
@@ -73,6 +88,13 @@ weather_function = FunctionSchema(
     required=["location", "format"],
 )
 
+get_news_function = FunctionSchema(
+    name="get_news",
+    description="Get the current news.",
+    properties={},
+    required=[],
+)
+
 restaurant_function = FunctionSchema(
     name="get_restaurant_recommendation",
     description="Get a restaurant recommendation",
@@ -140,10 +162,6 @@ even if you're asked about them.
 You are participating in a voice conversation. Keep your responses concise, short, and to the point
 unless specifically asked to elaborate on a topic.
 
-You have access to the following tools:
-- get_current_weather: Get the current weather for a given location.
-- get_restaurant_recommendation: Get a restaurant recommendation for a given location.
-
 Remember, your responses should be short. Just one or two sentences, usually. Respond in English.""",
     )
 
@@ -157,25 +175,31 @@ Remember, your responses should be short. Just one or two sentences, usually. Re
     # llm.register_function(None, fetch_weather_from_api)
     llm.register_function("get_current_weather", fetch_weather_from_api)
     llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation)
+    llm.register_function("get_news", get_news)
 
     transcript = TranscriptProcessor()
 
     # Create a standard OpenAI LLM context object using the normal messages format. The
     # OpenAIRealtimeLLMService will convert this internally to messages that the
     # openai WebSocket API can understand.
-    context = OpenAILLMContext(
+    context = LLMContext(
         [{"role": "user", "content": "Say hello!"}],
         tools,
     )
 
-    context_aggregator = llm.create_context_aggregator(context)
+    context_aggregator = LLMContextAggregatorPair(
+        context,
+        # `expect_stripped_words=False` needed when OpenAI Realtime used with
+        # "audio" modality (the default)
+        assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False),
+    )
 
     pipeline = Pipeline(
         [
             transport.input(),  # Transport user input
             context_aggregator.user(),
+            transcript.user(),  # LLM pushes TranscriptionFrames upstream
             llm,  # LLM
-            transcript.user(),  # Placed after the LLM, as LLM pushes TranscriptionFrames downstream
             transport.output(),  # Transport bot output
             transcript.assistant(),  # After the transcript output, to time with the audio output
             context_aggregator.assistant(),
@@ -198,6 +222,13 @@ Remember, your responses should be short. Just one or two sentences, usually. Re
         # Kick off the conversation.
         await task.queue_frames([LLMRunFrame()])
 
+        # Add a new tool at runtime after a delay.
+        await asyncio.sleep(15)
+        new_tools = ToolsSchema(
+            standard_tools=[weather_function, restaurant_function, get_news_function]
+        )
+        await task.queue_frames([LLMSetToolsFrame(tools=new_tools)])
+
     @transport.event_handler("on_client_disconnected")
     async def on_client_disconnected(transport, client):
         logger.info(f"Client disconnected")
diff --git a/examples/foundational/19a-azure-realtime.py b/examples/foundational/19a-azure-realtime.py
index c4b0fc02a..7d9cf1b4b 100644
--- a/examples/foundational/19a-azure-realtime.py
+++ b/examples/foundational/19a-azure-realtime.py
@@ -18,7 +18,9 @@ from pipecat.frames.frames import LLMRunFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
-from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.processors.aggregators.llm_response import LLMAssistantAggregatorParams
+from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
 from pipecat.runner.types import RunnerArguments
 from pipecat.runner.utils import create_transport
 from pipecat.services.azure.realtime.llm import AzureRealtimeLLMService
@@ -155,10 +157,10 @@ Remember, your responses should be short. Just one or two sentences, usually. Re
     llm.register_function("get_current_weather", fetch_weather_from_api)
     llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation)
 
-    # Create a standard OpenAI LLM context object using the normal messages format. The
+    # Create a standard LLM context object using the normal messages format. The
     # OpenAIRealtimeBetaLLMService will convert this internally to messages that the
     # openai WebSocket API can understand.
-    context = OpenAILLMContext(
+    context = LLMContext(
         [{"role": "user", "content": "Say hello!"}],
         # [{"role": "user", "content": [{"type": "text", "text": "Say hello!"}]}],
         #     [
@@ -173,7 +175,12 @@ Remember, your responses should be short. Just one or two sentences, usually. Re
         tools,
     )
 
-    context_aggregator = llm.create_context_aggregator(context)
+    context_aggregator = LLMContextAggregatorPair(
+        context,
+        # `expect_stripped_words=False` needed when OpenAI Realtime used with
+        # "audio" modality (the default)
+        assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False),
+    )
 
     pipeline = Pipeline(
         [
diff --git a/examples/foundational/19b-openai-realtime-text.py b/examples/foundational/19b-openai-realtime-text.py
index bb63a4814..c1f33b7bf 100644
--- a/examples/foundational/19b-openai-realtime-text.py
+++ b/examples/foundational/19b-openai-realtime-text.py
@@ -18,7 +18,8 @@ from pipecat.frames.frames import LLMRunFrame, TranscriptionMessage
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
-from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
 from pipecat.processors.transcript_processor import TranscriptProcessor
 from pipecat.runner.types import RunnerArguments
 from pipecat.runner.utils import create_transport
@@ -169,20 +170,20 @@ Remember, your responses should be short. Just one or two sentences, usually. Re
     # Create a standard OpenAI LLM context object using the normal messages format. The
     # OpenAIRealtimeLLMService will convert this internally to messages that the
     # openai WebSocket API can understand.
-    context = OpenAILLMContext(
+    context = LLMContext(
         [{"role": "user", "content": "Say hello!"}],
         tools,
     )
 
-    context_aggregator = llm.create_context_aggregator(context)
+    context_aggregator = LLMContextAggregatorPair(context)
 
     pipeline = Pipeline(
         [
             transport.input(),  # Transport user input
             context_aggregator.user(),
+            transcript.user(),  # LLM pushes TranscriptionFrames upstream
             llm,  # LLM
             tts,  # TTS
-            transcript.user(),  # Placed after the LLM, as LLM pushes TranscriptionFrames downstream
             transport.output(),  # Transport bot output
             transcript.assistant(),  # After the transcript output, to time with the audio output
             context_aggregator.assistant(),
diff --git a/examples/foundational/20b-persistent-context-openai-realtime.py b/examples/foundational/20b-persistent-context-openai-realtime.py
index 629a17c67..e3f018c16 100644
--- a/examples/foundational/20b-persistent-context-openai-realtime.py
+++ b/examples/foundational/20b-persistent-context-openai-realtime.py
@@ -13,14 +13,15 @@ from datetime import datetime
 from dotenv import load_dotenv
 from loguru import logger
 
+from pipecat.adapters.schemas.function_schema import FunctionSchema
+from pipecat.adapters.schemas.tools_schema import ToolsSchema
 from pipecat.audio.vad.silero import SileroVADAnalyzer
 from pipecat.frames.frames import LLMRunFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
-from pipecat.processors.aggregators.openai_llm_context import (
-    OpenAILLMContext,
-)
+from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
 from pipecat.runner.types import RunnerArguments
 from pipecat.runner.utils import create_transport
 from pipecat.services.deepgram.stt import DeepgramSTTService
@@ -69,11 +70,11 @@ async def save_conversation(params: FunctionCallParams):
     timestamp = datetime.now().strftime("%Y-%m-%d_%H:%M:%S")
     filename = f"{BASE_FILENAME}{timestamp}.json"
     logger.debug(
-        f"writing conversation to {filename}\n{json.dumps(params.context.messages, indent=4)}"
+        f"writing conversation to {filename}\n{json.dumps(params.context.get_messages(), indent=4)}"
     )
     try:
         with open(filename, "w") as file:
-            messages = params.context.get_messages_for_persistent_storage()
+            messages = params.context.get_messages()
             # remove the last message, which is the instruction we just gave to save the conversation
             messages.pop()
             json.dump(messages, file, indent=2)
@@ -90,6 +91,10 @@ async def load_conversation(params: FunctionCallParams):
             with open(filename, "r") as file:
                 params.context.set_messages(json.load(file))
                 await params.llm.reset_conversation()
+                # NOTE: we manually create a response here rather than relying
+                # on the function callback to trigger one since we've reset the
+                # conversation so the remote service doesn't know about the
+                # in-progress tool call.
                 await params.llm._create_response()
         except Exception as e:
             await params.result_callback({"success": False, "error": str(e)})
@@ -97,14 +102,12 @@ async def load_conversation(params: FunctionCallParams):
     asyncio.create_task(_reset())
 
 
-tools = [
-    {
-        "type": "function",
-        "name": "get_current_weather",
-        "description": "Get the current weather",
-        "parameters": {
-            "type": "object",
-            "properties": {
+tools = ToolsSchema(
+    standard_tools=[
+        FunctionSchema(
+            name="get_current_weather",
+            description="Get the current weather",
+            properties={
                 "location": {
                     "type": "string",
                     "description": "The city and state, e.g. San Francisco, CA",
@@ -115,45 +118,33 @@ tools = [
                     "description": "The temperature unit to use. Infer this from the users location.",
                 },
             },
-            "required": ["location", "format"],
-        },
-    },
-    {
-        "type": "function",
-        "name": "save_conversation",
-        "description": "Save the current conversatione. Use this function to persist the current conversation to external storage.",
-        "parameters": {
-            "type": "object",
-            "properties": {},
-            "required": [],
-        },
-    },
-    {
-        "type": "function",
-        "name": "get_saved_conversation_filenames",
-        "description": "Get a list of saved conversation histories. Returns a list of filenames. Each filename includes a date and timestamp. Each file is conversation history that can be loaded into this session.",
-        "parameters": {
-            "type": "object",
-            "properties": {},
-            "required": [],
-        },
-    },
-    {
-        "type": "function",
-        "name": "load_conversation",
-        "description": "Load a conversation history. Use this function to load a conversation history into the current session.",
-        "parameters": {
-            "type": "object",
-            "properties": {
+            required=["location", "format"],
+        ),
+        FunctionSchema(
+            name="save_conversation",
+            description="Save the current conversatione. Use this function to persist the current conversation to external storage.",
+            properties={},
+            required=[],
+        ),
+        FunctionSchema(
+            name="get_saved_conversation_filenames",
+            description="Get a list of saved conversation histories. Returns a list of filenames. Each filename includes a date and timestamp. Each file is conversation history that can be loaded into this session.",
+            properties={},
+            required=[],
+        ),
+        FunctionSchema(
+            name="load_conversation",
+            description="Load a conversation history. Use this function to load a conversation history into the current session.",
+            properties={
                 "filename": {
                     "type": "string",
                     "description": "The filename of the conversation history to load.",
                 }
             },
-            "required": ["filename"],
-        },
-    },
-]
+            required=["filename"],
+        ),
+    ]
+)
 
 
 # We store functions so objects (e.g. SileroVADAnalyzer) don't get
@@ -224,8 +215,8 @@ Remember, your responses should be short. Just one or two sentences, usually."""
     llm.register_function("get_saved_conversation_filenames", get_saved_conversation_filenames)
     llm.register_function("load_conversation", load_conversation)
 
-    context = OpenAILLMContext([], tools)
-    context_aggregator = llm.create_context_aggregator(context)
+    context = LLMContext([{"role": "user", "content": "Say hello!"}], tools)
+    context_aggregator = LLMContextAggregatorPair(context)
 
     pipeline = Pipeline(
         [
diff --git a/examples/foundational/26a-gemini-live-transcription.py b/examples/foundational/26a-gemini-live-transcription.py
index de277156b..9ac22a814 100644
--- a/examples/foundational/26a-gemini-live-transcription.py
+++ b/examples/foundational/26a-gemini-live-transcription.py
@@ -16,7 +16,9 @@ from pipecat.frames.frames import LLMRunFrame, TranscriptionMessage
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
-from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.processors.aggregators.llm_response import LLMAssistantAggregatorParams
+from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
 from pipecat.processors.transcript_processor import TranscriptProcessor
 from pipecat.runner.types import RunnerArguments
 from pipecat.runner.utils import create_transport
@@ -72,7 +74,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
         # inference_on_context_initialization=False,
     )
 
-    context = OpenAILLMContext(
+    context = LLMContext(
         [
             {
                 "role": "user",
@@ -90,7 +92,12 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
             #     },
         ],
     )
-    context_aggregator = llm.create_context_aggregator(context)
+    context_aggregator = LLMContextAggregatorPair(
+        context,
+        # `expect_stripped_words=False` needed when Gemini Live used with AUDIO
+        #  modality (the default)
+        assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False),
+    )
 
     transcript = TranscriptProcessor()
 
diff --git a/examples/foundational/26b-gemini-live-function-calling.py b/examples/foundational/26b-gemini-live-function-calling.py
index 65d159bb0..fd2b36ab1 100644
--- a/examples/foundational/26b-gemini-live-function-calling.py
+++ b/examples/foundational/26b-gemini-live-function-calling.py
@@ -19,7 +19,9 @@ from pipecat.frames.frames import LLMRunFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
-from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.processors.aggregators.llm_response import LLMAssistantAggregatorParams
+from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
 from pipecat.runner.types import RunnerArguments
 from pipecat.runner.utils import create_transport
 from pipecat.services.google.gemini_live.llm import GeminiLiveLLMService
@@ -139,10 +141,15 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
     llm.register_function("get_current_weather", fetch_weather_from_api)
     llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation)
 
-    context = OpenAILLMContext(
+    context = LLMContext(
         [{"role": "user", "content": "Say hello."}],
     )
-    context_aggregator = llm.create_context_aggregator(context)
+    context_aggregator = LLMContextAggregatorPair(
+        context,
+        # `expect_stripped_words=False` needed when Gemini Live used with AUDIO
+        #  modality (the default)
+        assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False),
+    )
 
     pipeline = Pipeline(
         [
diff --git a/examples/foundational/26c-gemini-live-video.py b/examples/foundational/26c-gemini-live-video.py
index 7b765075e..be036a557 100644
--- a/examples/foundational/26c-gemini-live-video.py
+++ b/examples/foundational/26c-gemini-live-video.py
@@ -17,7 +17,9 @@ from pipecat.frames.frames import LLMRunFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
-from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.processors.aggregators.llm_response import LLMAssistantAggregatorParams
+from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
 from pipecat.runner.types import RunnerArguments
 from pipecat.runner.utils import (
     create_transport,
@@ -65,7 +67,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
         # inference_on_context_initialization=False,
     )
 
-    context = OpenAILLMContext(
+    context = LLMContext(
         [
             {
                 "role": "user",
@@ -73,7 +75,12 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
             },
         ],
     )
-    context_aggregator = llm.create_context_aggregator(context)
+    context_aggregator = LLMContextAggregatorPair(
+        context,
+        # `expect_stripped_words=False` needed when Gemini Live used with AUDIO
+        #  modality (the default)
+        assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False),
+    )
 
     pipeline = Pipeline(
         [
diff --git a/examples/foundational/26d-gemini-live-text.py b/examples/foundational/26d-gemini-live-text.py
index 062c0231b..fc9f68bcb 100644
--- a/examples/foundational/26d-gemini-live-text.py
+++ b/examples/foundational/26d-gemini-live-text.py
@@ -16,7 +16,8 @@ from pipecat.frames.frames import LLMRunFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
-from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
 from pipecat.runner.types import RunnerArguments
 from pipecat.runner.utils import create_transport
 from pipecat.services.cartesia.tts import CartesiaTTSService
@@ -109,8 +110,8 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
 
     # Set up conversation context and management
     # The context_aggregator will automatically collect conversation context
-    context = OpenAILLMContext(messages)
-    context_aggregator = llm.create_context_aggregator(context)
+    context = LLMContext(messages)
+    context_aggregator = LLMContextAggregatorPair(context)
 
     pipeline = Pipeline(
         [
diff --git a/examples/foundational/26e-gemini-live-google-search.py b/examples/foundational/26e-gemini-live-google-search.py
index 178fdd282..e80ed4536 100644
--- a/examples/foundational/26e-gemini-live-google-search.py
+++ b/examples/foundational/26e-gemini-live-google-search.py
@@ -16,7 +16,9 @@ from pipecat.frames.frames import LLMRunFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
-from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.processors.aggregators.llm_response import LLMAssistantAggregatorParams
+from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
 from pipecat.runner.types import RunnerArguments
 from pipecat.runner.utils import create_transport
 from pipecat.services.google.gemini_live.llm import GeminiLiveLLMService
@@ -90,7 +92,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
         tools=tools,
     )
 
-    context = OpenAILLMContext(
+    context = LLMContext(
         [
             {
                 "role": "user",
@@ -98,7 +100,12 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
             }
         ],
     )
-    context_aggregator = llm.create_context_aggregator(context)
+    context_aggregator = LLMContextAggregatorPair(
+        context,
+        # `expect_stripped_words=False` needed when Gemini Live used with AUDIO
+        #  modality (the default)
+        assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False),
+    )
 
     pipeline = Pipeline(
         [
diff --git a/examples/foundational/26f-gemini-live-files-api.py b/examples/foundational/26f-gemini-live-files-api.py
index eeda16f52..0091c01e6 100644
--- a/examples/foundational/26f-gemini-live-files-api.py
+++ b/examples/foundational/26f-gemini-live-files-api.py
@@ -16,7 +16,9 @@ from pipecat.frames.frames import LLMRunFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
-from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.processors.aggregators.llm_response import LLMAssistantAggregatorParams
+from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
 from pipecat.runner.types import RunnerArguments
 from pipecat.runner.utils import create_transport
 from pipecat.services.google.gemini_live.llm import GeminiLiveLLMService
@@ -129,7 +131,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
         mime_type = "text/plain"
 
         # Create context with file reference
-        context = OpenAILLMContext(
+        context = LLMContext(
             [
                 {
                     "role": "user",
@@ -152,7 +154,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
     except Exception as e:
         logger.error(f"Error uploading file: {e}")
         # Continue with a basic context if file upload fails
-        context = OpenAILLMContext(
+        context = LLMContext(
             [
                 {
                     "role": "user",
@@ -162,7 +164,12 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
         )
 
     # Create context aggregator
-    context_aggregator = llm.create_context_aggregator(context)
+    context_aggregator = LLMContextAggregatorPair(
+        context,
+        # `expect_stripped_words=False` needed when Gemini Live used with AUDIO
+        #  modality (the default)
+        assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False),
+    )
 
     # Build the pipeline
     pipeline = Pipeline(
diff --git a/examples/foundational/26g-gemini-live-groundingMetadata.py b/examples/foundational/26g-gemini-live-groundingMetadata.py
index bea1756b2..df86094da 100644
--- a/examples/foundational/26g-gemini-live-groundingMetadata.py
+++ b/examples/foundational/26g-gemini-live-groundingMetadata.py
@@ -10,7 +10,9 @@ from pipecat.frames.frames import Frame, LLMRunFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineTask
-from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.processors.aggregators.llm_response import LLMAssistantAggregatorParams
+from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 from pipecat.runner.types import RunnerArguments
 from pipecat.runner.utils import create_transport
@@ -124,8 +126,13 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
     ]
 
     # Set up conversation context and management
-    context = OpenAILLMContext(messages)
-    context_aggregator = llm.create_context_aggregator(context)
+    context = LLMContext(messages)
+    context_aggregator = LLMContextAggregatorPair(
+        context,
+        # `expect_stripped_words=False` needed when Gemini Live used with AUDIO
+        #  modality (the default)
+        assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False),
+    )
 
     pipeline = Pipeline(
         [
diff --git a/examples/foundational/26h-gemini-live-vertex-function-calling.py b/examples/foundational/26h-gemini-live-vertex-function-calling.py
index c0344a052..126d85ad7 100644
--- a/examples/foundational/26h-gemini-live-vertex-function-calling.py
+++ b/examples/foundational/26h-gemini-live-vertex-function-calling.py
@@ -9,21 +9,21 @@ import os
 from datetime import datetime
 
 from dotenv import load_dotenv
-from google.genai.types import HttpOptions
 from loguru import logger
 
 from pipecat.adapters.schemas.function_schema import FunctionSchema
-from pipecat.adapters.schemas.tools_schema import AdapterType, ToolsSchema
+from pipecat.adapters.schemas.tools_schema import ToolsSchema
 from pipecat.audio.vad.silero import SileroVADAnalyzer
 from pipecat.audio.vad.vad_analyzer import VADParams
 from pipecat.frames.frames import LLMRunFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
-from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.processors.aggregators.llm_response import LLMAssistantAggregatorParams
+from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
 from pipecat.runner.types import RunnerArguments
 from pipecat.runner.utils import create_transport
-from pipecat.services.google.gemini_live.llm import GeminiLiveLLMService
 from pipecat.services.google.gemini_live.llm_vertex import GeminiLiveVertexLLMService
 from pipecat.services.llm_service import FunctionCallParams
 from pipecat.transports.base_transport import BaseTransport, TransportParams
@@ -139,10 +139,13 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
     llm.register_function("get_current_weather", fetch_weather_from_api)
     llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation)
 
-    context = OpenAILLMContext(
-        [{"role": "user", "content": "Say hello."}],
+    context = LLMContext([{"role": "user", "content": "Say hello."}])
+    context_aggregator = LLMContextAggregatorPair(
+        context,
+        # `expect_stripped_words=False` needed when Gemini Live used with AUDIO
+        #  modality (the default)
+        assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False),
     )
-    context_aggregator = llm.create_context_aggregator(context)
 
     pipeline = Pipeline(
         [
diff --git a/examples/foundational/26i-gemini-live-graceful-end.py b/examples/foundational/26i-gemini-live-graceful-end.py
index e51bbb032..2865dbed4 100644
--- a/examples/foundational/26i-gemini-live-graceful-end.py
+++ b/examples/foundational/26i-gemini-live-graceful-end.py
@@ -18,7 +18,9 @@ from pipecat.frames.frames import EndTaskFrame, LLMRunFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
-from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.processors.aggregators.llm_response import LLMAssistantAggregatorParams
+from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
 from pipecat.processors.frame_processor import FrameDirection
 from pipecat.runner.types import RunnerArguments
 from pipecat.runner.utils import create_transport
@@ -62,7 +64,7 @@ You have three tools available to you:
 
 After you've responded to the user three times, do two things, in order:
 1. Politely let them know that that's all the time you have today and say goodbye.
-2. Call the end_conversation tool to gracefully end the conversation.
+2. *WITHOUT WAITING FOR THE USER TO RESPOND*, call the end_conversation tool to gracefully end the conversation.
 """
 
 
@@ -152,10 +154,15 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
     llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation)
     llm.register_function("end_conversation", end_conversation)
 
-    context = OpenAILLMContext(
+    context = LLMContext(
         [{"role": "user", "content": "Say hello."}],
     )
-    context_aggregator = llm.create_context_aggregator(context)
+    context_aggregator = LLMContextAggregatorPair(
+        context,
+        # `expect_stripped_words=False` needed when Gemini Live used with AUDIO
+        #  modality (the default)
+        assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False),
+    )
 
     pipeline = Pipeline(
         [
diff --git a/examples/foundational/27-simli-layer.py b/examples/foundational/27-simli-layer.py
index 9479632b5..348cf117b 100644
--- a/examples/foundational/27-simli-layer.py
+++ b/examples/foundational/27-simli-layer.py
@@ -9,7 +9,6 @@ import os
 
 from dotenv import load_dotenv
 from loguru import logger
-from simli import SimliConfig
 
 from pipecat.audio.turn.smart_turn.base_smart_turn import SmartTurnParams
 from pipecat.audio.turn.smart_turn.local_smart_turn_v3 import LocalSmartTurnAnalyzerV3
@@ -66,11 +65,12 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
 
     tts = CartesiaTTSService(
         api_key=os.getenv("CARTESIA_API_KEY"),
-        voice_id="a167e0f3-df7e-4d52-a9c3-f949145efdab",
+        voice_id="71a7ad14-091c-4e8e-a314-022ece01c121",
     )
 
     simli_ai = SimliVideoService(
-        SimliConfig(os.getenv("SIMLI_API_KEY"), os.getenv("SIMLI_FACE_ID")),
+        api_key=os.getenv("SIMLI_API_KEY"),
+        face_id="cace3ef7-a4c4-425d-a8cf-a5358eb0c427",
     )
 
     llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o-mini")
diff --git a/examples/foundational/46-video-processing.py b/examples/foundational/46-video-processing.py
index 62ea5debe..5f92139bc 100644
--- a/examples/foundational/46-video-processing.py
+++ b/examples/foundational/46-video-processing.py
@@ -15,7 +15,9 @@ from pipecat.frames.frames import Frame, InputImageRawFrame, LLMRunFrame, Output
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
-from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.processors.aggregators.llm_response import LLMAssistantAggregatorParams
+from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 from pipecat.processors.frameworks.rtvi import RTVIObserver, RTVIProcessor
 from pipecat.runner.types import RunnerArguments
@@ -108,8 +110,13 @@ async def run_bot(pipecat_transport):
         }
     ]
 
-    context = OpenAILLMContext(messages)
-    context_aggregator = llm.create_context_aggregator(context)
+    context = LLMContext(messages)
+    context_aggregator = LLMContextAggregatorPair(
+        context,
+        # `expect_stripped_words=False` needed when Gemini Live used with AUDIO
+        #  modality (the default)
+        assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False),
+    )
 
     # RTVI events for Pipecat client UI
     rtvi = RTVIProcessor()
diff --git a/pyproject.toml b/pyproject.toml
index fd992fa86..5760361b9 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -55,7 +55,7 @@ azure = [ "azure-cognitiveservices-speech~=1.42.0"]
 cartesia = [ "cartesia~=2.0.3", "pipecat-ai[websockets-base]" ]
 cerebras = []
 deepseek = []
-daily = [ "daily-python~=0.20.0" ]
+daily = [ "daily-python~=0.21.0" ]
 deepgram = [ "deepgram-sdk~=4.7.0" ]
 elevenlabs = [ "pipecat-ai[websockets-base]" ]
 fal = [ "fal-client~=0.5.9" ]
diff --git a/src/pipecat/adapters/schemas/tools_schema.py b/src/pipecat/adapters/schemas/tools_schema.py
index 05710616d..d0d798569 100644
--- a/src/pipecat/adapters/schemas/tools_schema.py
+++ b/src/pipecat/adapters/schemas/tools_schema.py
@@ -22,9 +22,12 @@ class AdapterType(Enum):
 
     Parameters:
         GEMINI: Google Gemini adapter - currently the only service supporting custom tools.
+        SHIM: Backward compatibility shim for creating ToolsSchemas from lists of tools in
+              any format, used by LLMContext.from_openai_context.
     """
 
     GEMINI = "gemini"  # that is the only service where we are able to add custom tools for now
+    SHIM = "shim"  # for use as backward compatibility shim for creating ToolsSchemas from list of tools in any format
 
 
 class ToolsSchema:
diff --git a/src/pipecat/adapters/services/aws_nova_sonic_adapter.py b/src/pipecat/adapters/services/aws_nova_sonic_adapter.py
index 60f12798b..dcc42ba68 100644
--- a/src/pipecat/adapters/services/aws_nova_sonic_adapter.py
+++ b/src/pipecat/adapters/services/aws_nova_sonic_adapter.py
@@ -16,7 +16,7 @@ from loguru import logger
 
 from pipecat.adapters.base_llm_adapter import BaseLLMAdapter
 from pipecat.adapters.schemas.function_schema import FunctionSchema
-from pipecat.adapters.schemas.tools_schema import ToolsSchema
+from pipecat.adapters.schemas.tools_schema import AdapterType, ToolsSchema
 from pipecat.processors.aggregators.llm_context import LLMContext, LLMContextMessage
 
 
@@ -210,4 +210,18 @@ class AWSNovaSonicLLMAdapter(BaseLLMAdapter[AWSNovaSonicLLMInvocationParams]):
             List of dictionaries in AWS Nova Sonic function format.
         """
         functions_schema = tools_schema.standard_tools
-        return [self._to_aws_nova_sonic_function_format(func) for func in functions_schema]
+        standard_tools = [
+            self._to_aws_nova_sonic_function_format(func) for func in functions_schema
+        ]
+
+        # For backward compatibility, AWS Nova Sonic can still be used with
+        # tools in dict format, even though it always uses `LLMContext` under
+        # the hood (via `LLMContext.from_openai_context()`).
+        # To support this behavior, we use "shimmed" custom tools here.
+        # (We maintain this backward compatibility because users aren't
+        # *knowingly* opting into the new `LLMContext`.)
+        shimmed_tools = []
+        if tools_schema.custom_tools:
+            shimmed_tools = tools_schema.custom_tools.get(AdapterType.SHIM, [])
+
+        return standard_tools + shimmed_tools
diff --git a/src/pipecat/adapters/services/gemini_adapter.py b/src/pipecat/adapters/services/gemini_adapter.py
index 26f037127..1fa8d9e6f 100644
--- a/src/pipecat/adapters/services/gemini_adapter.py
+++ b/src/pipecat/adapters/services/gemini_adapter.py
@@ -24,13 +24,7 @@ from pipecat.processors.aggregators.llm_context import (
 )
 
 try:
-    from google.genai.types import (
-        Blob,
-        Content,
-        FunctionCall,
-        FunctionResponse,
-        Part,
-    )
+    from google.genai.types import Blob, Content, FileData, FunctionCall, FunctionResponse, Part
 except ModuleNotFoundError as e:
     logger.error(f"Exception: {e}")
     logger.error("In order to use Google AI, you need to `pip install pipecat-ai[google]`.")
@@ -309,6 +303,7 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
                 parts.append(
                     Part(
                         function_call=FunctionCall(
+                            id=id,
                             name=name,
                             args=json.loads(tc["function"]["arguments"]),
                         )
@@ -334,9 +329,12 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
                 function_name = params.tool_call_id_to_name_mapping[tool_call_id]
 
             parts.append(
-                Part.from_function_response(
-                    name=function_name,
-                    response=response_dict,
+                Part(
+                    function_response=FunctionResponse(
+                        id=tool_call_id,
+                        name=function_name,
+                        response=response_dict,
+                    )
                 )
             )
         elif isinstance(content, str):
@@ -358,6 +356,16 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
                     input_audio = c["input_audio"]
                     audio_bytes = base64.b64decode(input_audio["data"])
                     parts.append(Part(inline_data=Blob(mime_type="audio/wav", data=audio_bytes)))
+                elif c["type"] == "file_data":
+                    file_data = c["file_data"]
+                    parts.append(
+                        Part(
+                            file_data=FileData(
+                                mime_type=file_data.get("mime_type"),
+                                file_uri=file_data.get("file_uri"),
+                            )
+                        )
+                    )
 
         return self.MessageConversionResult(
             content=Content(role=role, parts=parts),
diff --git a/src/pipecat/adapters/services/open_ai_realtime_adapter.py b/src/pipecat/adapters/services/open_ai_realtime_adapter.py
index 2ff629e2e..3d3650633 100644
--- a/src/pipecat/adapters/services/open_ai_realtime_adapter.py
+++ b/src/pipecat/adapters/services/open_ai_realtime_adapter.py
@@ -6,12 +6,18 @@
 
 """OpenAI Realtime LLM adapter for Pipecat."""
 
-from typing import Any, Dict, List, TypedDict
+import copy
+import json
+from dataclasses import dataclass
+from typing import Any, Dict, List, Optional, TypedDict
+
+from loguru import logger
 
 from pipecat.adapters.base_llm_adapter import BaseLLMAdapter
 from pipecat.adapters.schemas.function_schema import FunctionSchema
-from pipecat.adapters.schemas.tools_schema import ToolsSchema
-from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.adapters.schemas.tools_schema import AdapterType, ToolsSchema
+from pipecat.processors.aggregators.llm_context import LLMContext, LLMContextMessage
+from pipecat.services.openai.realtime import events
 
 
 class OpenAIRealtimeLLMInvocationParams(TypedDict):
@@ -20,7 +26,9 @@ class OpenAIRealtimeLLMInvocationParams(TypedDict):
     This is a placeholder until support for universal LLMContext machinery is added for OpenAI Realtime.
     """
 
-    pass
+    system_instruction: Optional[str]
+    messages: List[events.ConversationItem]
+    tools: List[Dict[str, Any]]
 
 
 class OpenAIRealtimeLLMAdapter(BaseLLMAdapter):
@@ -33,7 +41,7 @@ class OpenAIRealtimeLLMAdapter(BaseLLMAdapter):
     @property
     def id_for_llm_specific_messages(self) -> str:
         """Get the identifier used in LLMSpecificMessage instances for OpenAI Realtime."""
-        raise NotImplementedError("Universal LLMContext is not yet supported for OpenAI Realtime.")
+        return "openai-realtime"
 
     def get_llm_invocation_params(self, context: LLMContext) -> OpenAIRealtimeLLMInvocationParams:
         """Get OpenAI Realtime-specific LLM invocation parameters from a universal LLM context.
@@ -46,7 +54,13 @@ class OpenAIRealtimeLLMAdapter(BaseLLMAdapter):
         Returns:
             Dictionary of parameters for invoking OpenAI Realtime's API.
         """
-        raise NotImplementedError("Universal LLMContext is not yet supported for OpenAI Realtime.")
+        messages = self._from_universal_context_messages(self.get_messages(context))
+        return {
+            "system_instruction": messages.system_instruction,
+            "messages": messages.messages,
+            # NOTE: LLMContext's tools are guaranteed to be a ToolsSchema (or NOT_GIVEN)
+            "tools": self.from_standard_tools(context.tools) or [],
+        }
 
     def get_messages_for_logging(self, context) -> List[Dict[str, Any]]:
         """Get messages from a universal LLM context in a format ready for logging about OpenAI Realtime.
@@ -61,7 +75,124 @@ class OpenAIRealtimeLLMAdapter(BaseLLMAdapter):
         Returns:
             List of messages in a format ready for logging about OpenAI Realtime.
         """
-        raise NotImplementedError("Universal LLMContext is not yet supported for OpenAI Realtime.")
+        # NOTE: this is the same as in OpenAIAdapter, as that's what it was
+        # prior to a refactor. Worth noting that for OpenAI Realtime
+        # specifically, not everything handled here is necessarily supported
+        # (or supported yet).
+        msgs = []
+        for message in self.get_messages(context):
+            msg = copy.deepcopy(message)
+            if "content" in msg:
+                if isinstance(msg["content"], list):
+                    for item in msg["content"]:
+                        if item["type"] == "image_url":
+                            if item["image_url"]["url"].startswith("data:image/"):
+                                item["image_url"]["url"] = "data:image/..."
+                        if item["type"] == "input_audio":
+                            item["input_audio"]["data"] = "..."
+            if "mime_type" in msg and msg["mime_type"].startswith("image/"):
+                msg["data"] = "..."
+            msgs.append(msg)
+        return msgs
+
+    @dataclass
+    class ConvertedMessages:
+        """Container for OpenAI-formatted messages converted from universal context."""
+
+        messages: List[events.ConversationItem]
+        system_instruction: Optional[str] = None
+
+    def _from_universal_context_messages(
+        self, universal_context_messages: List[LLMContextMessage]
+    ) -> ConvertedMessages:
+        # We can't load a long conversation history into the openai realtime api yet. (The API/model
+        # forgets that it can do audio, if you do a series of `conversation.item.create` calls.) So
+        # our general strategy until this is fixed is just to put everything into a first "user"
+        # message as a single input.
+
+        if not universal_context_messages:
+            return self.ConvertedMessages(messages=[])
+
+        messages = copy.deepcopy(universal_context_messages)
+        system_instruction = None
+
+        # If we have a "system" message as our first message, let's pull that out into session
+        # "instructions"
+        if messages[0].get("role") == "system":
+            system = messages.pop(0)
+            content = system.get("content")
+            if isinstance(content, str):
+                system_instruction = content
+            elif isinstance(content, list):
+                system_instruction = content[0].get("text")
+            if not messages:
+                return self.ConvertedMessages(messages=[], system_instruction=system_instruction)
+
+        # If we have just a single "user" item, we can just send it normally
+        if len(messages) == 1 and messages[0].get("role") == "user":
+            return self.ConvertedMessages(
+                messages=[self._from_universal_context_message(messages[0])],
+                system_instruction=system_instruction,
+            )
+
+        # Otherwise, let's pack everything into a single "user" message with a bit of
+        # explanation for the LLM
+        intro_text = """
+        This is a previously saved conversation. Please treat this conversation history as a
+        starting point for the current conversation."""
+
+        trailing_text = """
+        This is the end of the previously saved conversation. Please continue the conversation
+        from here. If the last message is a user instruction or question, act on that instruction
+        or answer the question. If the last message is an assistant response, simple say that you
+        are ready to continue the conversation."""
+
+        return self.ConvertedMessages(
+            messages=[
+                {
+                    "role": "user",
+                    "type": "message",
+                    "content": [
+                        {
+                            "type": "input_text",
+                            "text": "\n\n".join(
+                                [intro_text, json.dumps(messages, indent=2), trailing_text]
+                            ),
+                        }
+                    ],
+                }
+            ],
+            system_instruction=system_instruction,
+        )
+
+    def _from_universal_context_message(
+        self, message: LLMContextMessage
+    ) -> events.ConversationItem:
+        if message.get("role") == "user":
+            content = message.get("content")
+            if isinstance(message.get("content"), list):
+                content = ""
+                for c in message.get("content"):
+                    if c.get("type") == "text":
+                        content += " " + c.get("text")
+                    else:
+                        logger.error(
+                            f"Unhandled content type in context message: {c.get('type')} - {message}"
+                        )
+            return events.ConversationItem(
+                role="user",
+                type="message",
+                content=[events.ItemContent(type="input_text", text=content)],
+            )
+        if message.get("role") == "assistant" and message.get("tool_calls"):
+            tc = message.get("tool_calls")[0]
+            return events.ConversationItem(
+                type="function_call",
+                call_id=tc["id"],
+                name=tc["function"]["name"],
+                arguments=tc["function"]["arguments"],
+            )
+        logger.error(f"Unhandled message type in _from_universal_context_message: {message}")
 
     @staticmethod
     def _to_openai_realtime_function_format(function: FunctionSchema) -> Dict[str, Any]:
@@ -94,4 +225,18 @@ class OpenAIRealtimeLLMAdapter(BaseLLMAdapter):
             List of function definitions in OpenAI Realtime format.
         """
         functions_schema = tools_schema.standard_tools
-        return [self._to_openai_realtime_function_format(func) for func in functions_schema]
+        standard_tools = [
+            self._to_openai_realtime_function_format(func) for func in functions_schema
+        ]
+
+        # For backward compatibility, OpenAI Realtime can still be used with
+        # tools in dict format, even though it always uses `LLMContext` under
+        # the hood (via `LLMContext.from_openai_context()`).
+        # To support this behavior, we use "shimmed" custom tools here.
+        # (We maintain this backward compatibility because users aren't
+        # *knowingly* opting into the new `LLMContext`.)
+        shimmed_tools = []
+        if tools_schema.custom_tools:
+            shimmed_tools = tools_schema.custom_tools.get(AdapterType.SHIM, [])
+
+        return standard_tools + shimmed_tools
diff --git a/src/pipecat/processors/aggregators/llm_context.py b/src/pipecat/processors/aggregators/llm_context.py
index 913566909..8dc79fb50 100644
--- a/src/pipecat/processors/aggregators/llm_context.py
+++ b/src/pipecat/processors/aggregators/llm_context.py
@@ -15,7 +15,6 @@ service-specific adapter.
 """
 
 import base64
-import copy
 import io
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Any, List, Optional, TypeAlias, Union
@@ -29,7 +28,7 @@ from openai.types.chat import (
 )
 from PIL import Image
 
-from pipecat.adapters.schemas.tools_schema import ToolsSchema
+from pipecat.adapters.schemas.tools_schema import AdapterType, ToolsSchema
 from pipecat.frames.frames import AudioRawFrame
 
 if TYPE_CHECKING:
@@ -83,9 +82,17 @@ class LLMContext:
         Returns:
             New LLMContext instance with converted messages and settings.
         """
+        # Convert tools to ToolsSchema if needed.
+        # If the tools are already a ToolsSchema, this is a no-op.
+        # Otherwise, we wrap them in a shim ToolsSchema.
+        converted_tools = openai_context.tools
+        if isinstance(converted_tools, list):
+            converted_tools = ToolsSchema(
+                standard_tools=[], custom_tools={AdapterType.SHIM: converted_tools}
+            )
         return LLMContext(
             messages=openai_context.get_messages(),
-            tools=openai_context.tools,
+            tools=converted_tools,
             tool_choice=openai_context.tool_choice,
         )
 
@@ -119,6 +126,33 @@ class LLMContext:
         """
         return self.get_messages()
 
+    def get_messages_for_persistent_storage(self) -> List[LLMContextMessage]:
+        """Get messages suitable for persistent storage.
+
+        NOTE: the only reason this method exists is because we're "silently"
+        switching from OpenAILLMContext to LLMContext under the hood in some
+        services and don't want to trip up users who may have been relying on
+        this method, which is part of the public API of OpenAILLMContext but
+        doesn't need to be for LLMContext.
+
+        .. deprecated::
+            Use `get_messages()` instead.
+
+        Returns:
+            List of conversation messages.
+        """
+        import warnings
+
+        with warnings.catch_warnings():
+            warnings.simplefilter("always")
+            warnings.warn(
+                "get_messages_for_persistent_storage() is deprecated, use get_messages() instead.",
+                DeprecationWarning,
+                stacklevel=2,
+            )
+
+        return self.get_messages()
+
     def get_messages(self, llm_specific_filter: Optional[str] = None) -> List[LLMContextMessage]:
         """Get the current messages list.
 
diff --git a/src/pipecat/processors/aggregators/llm_response_universal.py b/src/pipecat/processors/aggregators/llm_response_universal.py
index 69a8dd280..9f1e04fe0 100644
--- a/src/pipecat/processors/aggregators/llm_response_universal.py
+++ b/src/pipecat/processors/aggregators/llm_response_universal.py
@@ -290,6 +290,12 @@ class LLMUserAggregator(LLMContextAggregator):
             await self._handle_llm_messages_update(frame)
         elif isinstance(frame, LLMSetToolsFrame):
             self.set_tools(frame.tools)
+            # Push the LLMSetToolsFrame as well, since speech-to-speech LLM
+            # services (like OpenAI Realtime) may need to know about tool
+            # changes; unlike text-based LLM services they won't just "pick up
+            # the change" on the next LLM run, as the LLM is continuously
+            # running.
+            await self.push_frame(frame, direction)
         elif isinstance(frame, LLMSetToolChoiceFrame):
             self.set_tool_choice(frame.tool_choice)
         elif isinstance(frame, SpeechControlParamsFrame):
diff --git a/src/pipecat/services/azure/realtime/llm.py b/src/pipecat/services/azure/realtime/llm.py
index 1193b82d4..66ba95eea 100644
--- a/src/pipecat/services/azure/realtime/llm.py
+++ b/src/pipecat/services/azure/realtime/llm.py
@@ -38,7 +38,7 @@ class AzureRealtimeLLMService(OpenAIRealtimeLLMService):
         Args:
             api_key: The API key for the Azure OpenAI service.
             base_url: The full Azure WebSocket endpoint URL including api-version and deployment.
-                Example: "wss://my-project.openai.azure.com/openai/realtime?api-version=2024-10-01-preview&deployment=my-realtime-deployment"
+                Example: "wss://my-project.openai.azure.com/openai/realtime?api-version=2025-04-01-preview&deployment=my-realtime-deployment"
             **kwargs: Additional arguments passed to parent OpenAIRealtimeLLMService.
         """
         super().__init__(base_url=base_url, api_key=api_key, **kwargs)
diff --git a/src/pipecat/services/google/gemini_live/llm.py b/src/pipecat/services/google/gemini_live/llm.py
index 4b5f05209..e42913771 100644
--- a/src/pipecat/services/google/gemini_live/llm.py
+++ b/src/pipecat/services/google/gemini_live/llm.py
@@ -17,6 +17,7 @@ import json
 import random
 import time
 import uuid
+import warnings
 from dataclasses import dataclass
 from enum import Enum
 from typing import Any, Dict, List, Optional, Union
@@ -56,10 +57,12 @@ from pipecat.frames.frames import (
     UserStoppedSpeakingFrame,
 )
 from pipecat.metrics.metrics import LLMTokenUsage
+from pipecat.processors.aggregators.llm_context import LLMContext
 from pipecat.processors.aggregators.llm_response import (
     LLMAssistantAggregatorParams,
     LLMUserAggregatorParams,
 )
+from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
 from pipecat.processors.aggregators.openai_llm_context import (
     OpenAILLMContext,
     OpenAILLMContextFrame,
@@ -219,6 +222,10 @@ class GeminiLiveContext(OpenAILLMContext):
 
     Provides Gemini-specific context management including system instruction
     extraction and message format conversion for the Live API.
+
+    .. deprecated:: 0.0.92
+        Gemini Live no longer uses `GeminiLiveContext` under the hood.
+        It now uses `LLMContext`.
     """
 
     @staticmethod
@@ -231,6 +238,22 @@ class GeminiLiveContext(OpenAILLMContext):
         Returns:
             The upgraded Gemini context instance.
         """
+        # This warning is here rather than `__init__` since `upgrade()` was the
+        # "main" way that GeminiLiveContext instances were created.
+        # Almost no users should be seeing this message anyway, as
+        # GeminiLiveContext instances were typically created under the hood:
+        # the user would pass an OpenAILLMContext instance, which would be
+        # upgraded without them necessarily knowing.
+        with warnings.catch_warnings():
+            warnings.simplefilter("always")
+            warnings.warn(
+                "GeminiLiveContext is deprecated. "
+                "Gemini Live no longer uses GeminiLiveContext under the hood. "
+                "It now uses LLMContext.",
+                DeprecationWarning,
+                stacklevel=2,
+            )
+
         if isinstance(obj, OpenAILLMContext) and not isinstance(obj, GeminiLiveContext):
             logger.debug(f"Upgrading to Gemini Live Context: {obj}")
             obj.__class__ = GeminiLiveContext
@@ -328,8 +351,28 @@ class GeminiLiveUserContextAggregator(OpenAIUserContextAggregator):
 
     Extends OpenAI user aggregator to handle Gemini-specific message passing
     while maintaining compatibility with the standard aggregation pipeline.
+
+    .. deprecated:: 0.0.92
+        Gemini Live no longer expects a `GeminiLiveUserContextAggregator`.
+        It now expects a `LLMUserAggregator`.
     """
 
+    def __init__(self, *args, **kwargs):
+        """Initialize Gemini Live user context aggregator."""
+        # Almost no users should be seeing this message, as
+        # `GeminiLiveUserContextAggregator`` instances were typically created
+        # under the hood, as part of `llm.create_context_aggregator()`.
+        with warnings.catch_warnings():
+            warnings.simplefilter("always")
+            warnings.warn(
+                "GeminiLiveUserContextAggregator is deprecated. "
+                "Gemini Live no longer expects a GeminiLiveUserContextAggregator. "
+                "It now expects a LLMUserAggregator.",
+                DeprecationWarning,
+                stacklevel=2,
+            )
+        super().__init__(*args, **kwargs)
+
     async def process_frame(self, frame, direction):
         """Process incoming frames for user context aggregation.
 
@@ -349,8 +392,28 @@ class GeminiLiveAssistantContextAggregator(OpenAIAssistantContextAggregator):
     Handles assistant response aggregation while filtering out LLMTextFrames
     to prevent duplicate context entries, as Gemini Live pushes both
     LLMTextFrames and TTSTextFrames.
+
+    .. deprecated:: 0.0.92
+        Gemini Live no longer uses `GeminiLiveAssistantContextAggregator` under the hood.
+        It now uses `LLMAssistantAggregator`.
     """
 
+    def __init__(self, *args, **kwargs):
+        """Initialize Gemini Live assistant context aggregator."""
+        # Almost no users should be seeing this message, as
+        # `GeminiLiveAssistantContextAggregator` instances were typically
+        # created under the hood, as part of `llm.create_context_aggregator()`.
+        with warnings.catch_warnings():
+            warnings.simplefilter("always")
+            warnings.warn(
+                "GeminiLiveAssistantContextAggregator is deprecated. "
+                "Gemini Live no longer uses GeminiLiveAssistantContextAggregator under the hood. "
+                "It now uses LLMAssistantAggregator.",
+                DeprecationWarning,
+                stacklevel=2,
+            )
+        super().__init__(*args, **kwargs)
+
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         """Process incoming frames for assistant context aggregation.
 
@@ -380,6 +443,10 @@ class GeminiLiveAssistantContextAggregator(OpenAIAssistantContextAggregator):
 class GeminiLiveContextAggregatorPair:
     """Pair of user and assistant context aggregators for Gemini Live.
 
+    .. deprecated:: 0.0.92
+        `GeminiLiveContextAggregatorPair` is deprecated.
+        Use `LLMContextAggregatorPair` instead.
+
     Parameters:
         _user: The user context aggregator instance.
         _assistant: The assistant context aggregator instance.
@@ -388,6 +455,19 @@ class GeminiLiveContextAggregatorPair:
     _user: GeminiLiveUserContextAggregator
     _assistant: GeminiLiveAssistantContextAggregator
 
+    def __post_init__(self):
+        # Almost no users should be seeing this message, as
+        # `GeminiLiveContextAggregatorPair` instances were typically created
+        # under the hood, with `llm.create_context_aggregator()`.
+        with warnings.catch_warnings():
+            warnings.simplefilter("always")
+            warnings.warn(
+                "GeminiLiveContextAggregatorPair is deprecated. "
+                "Use LLMContextAggregatorPair instead.",
+                DeprecationWarning,
+                stacklevel=2,
+            )
+
     def user(self) -> GeminiLiveUserContextAggregator:
         """Get the user context aggregator.
 
@@ -609,7 +689,7 @@ class GeminiLiveLLMService(LLMService):
         self._run_llm_when_session_ready = False
 
         self._user_is_speaking = False
-        self._bot_is_speaking = False
+        self._bot_is_responding = False
         self._user_audio_buffer = bytearray()
         self._user_transcription_buffer = ""
         self._last_transcription_sent = ""
@@ -665,6 +745,9 @@ class GeminiLiveLLMService(LLMService):
         # Initialize the API client. Subclasses can override this if needed.
         self.create_client()
 
+        # Bookkeeping for tool calls
+        self._completed_tool_calls = set()
+
     def create_client(self):
         """Create the Gemini API client instance. Subclasses can override this."""
         self._client = Client(api_key=self._api_key, http_options=self._http_options)
@@ -787,9 +870,13 @@ class GeminiLiveLLMService(LLMService):
     #
 
     async def _handle_interruption(self):
-        await self._set_bot_is_speaking(False)
-        await self.push_frame(TTSStoppedFrame())
-        await self.push_frame(LLMFullResponseEndFrame())
+        if self._bot_is_responding:
+            await self._set_bot_is_responding(False)
+            if self._settings.get("modalities") == GeminiModalities.AUDIO:
+                await self.push_frame(TTSStoppedFrame())
+            # Do not send LLMFullResponseEndFrame here - an interruption
+            # already tells the assistant context aggregator that the response
+            # is over.
 
     async def _handle_user_started_speaking(self, frame):
         self._user_is_speaking = True
@@ -807,7 +894,6 @@ class GeminiLiveLLMService(LLMService):
 
     #
     # frame processing
-    #
     # StartFrame, StopFrame, CancelFrame implemented in base class
     #
 
@@ -820,7 +906,7 @@ class GeminiLiveLLMService(LLMService):
         """
         # Defer EndFrame handling until after the bot turn is finished
         if isinstance(frame, EndFrame):
-            if self._bot_is_speaking:
+            if self._bot_is_responding:
                 logger.debug("Deferring handling EndFrame until bot turn is finished")
                 self._end_frame_pending_bot_turn_finished = frame
                 return
@@ -829,22 +915,13 @@ class GeminiLiveLLMService(LLMService):
 
         if isinstance(frame, TranscriptionFrame):
             await self.push_frame(frame, direction)
-        elif isinstance(frame, OpenAILLMContextFrame):
-            context: GeminiLiveContext = GeminiLiveContext.upgrade(frame.context)
-            # For now, we'll only trigger inference here when either:
-            #   1. We have not seen a context frame before
-            #   2. The last message is a tool call result
-            if not self._context:
-                self._context = context
-                if frame.context.tools:
-                    self._tools = frame.context.tools
-                await self._create_initial_response()
-            elif context.messages and context.messages[-1].get("role") == "tool":
-                # Support just one tool call per context frame for now
-                tool_result_message = context.messages[-1]
-                await self._tool_result(tool_result_message)
-        elif isinstance(frame, LLMContextFrame):
-            raise NotImplementedError("Universal LLMContext is not yet supported for Gemini Live.")
+        elif isinstance(frame, (LLMContextFrame, OpenAILLMContextFrame)):
+            context = (
+                frame.context
+                if isinstance(frame, LLMContextFrame)
+                else LLMContext.from_openai_context(frame.context)
+            )
+            await self._handle_context(context)
         elif isinstance(frame, InputTextRawFrame):
             await self._send_user_text(frame.text)
             await self.push_frame(frame, direction)
@@ -883,13 +960,48 @@ class GeminiLiveLLMService(LLMService):
         else:
             await self.push_frame(frame, direction)
 
-    async def _set_bot_is_speaking(self, speaking: bool):
-        if self._bot_is_speaking == speaking:
+    async def _handle_context(self, context: LLMContext):
+        if not self._context:
+            # We got our initial context
+            self._context = context
+            if context.tools:
+                self._tools = context.tools
+            # Initialize our bookkeeping of already-completed tool calls in
+            # the context
+            await self._process_completed_function_calls(send_new_results=False)
+            await self._create_initial_response()
+        else:
+            # We got an updated context.
+            # This may contain a new user message or tool call result.
+            self._context = context
+            # Send results for newly-completed function calls, if any.
+            await self._process_completed_function_calls(send_new_results=True)
+
+    async def _process_completed_function_calls(self, send_new_results: bool):
+        # Check for set of completed function calls in the context
+        adapter: GeminiLLMAdapter = self.get_llm_adapter()
+        messages = adapter.get_llm_invocation_params(self._context).get("messages", [])
+        for message in messages:
+            if message.parts:
+                for part in message.parts:
+                    if part.function_response:
+                        tool_call_id = part.function_response.id
+                        tool_name = part.function_response.name
+                        if tool_call_id and tool_call_id not in self._completed_tool_calls:
+                            # Found a newly-completed function call - send the result to the service
+                            if send_new_results:
+                                await self._tool_result(
+                                    tool_call_id, tool_name, part.function_response.response
+                                )
+                            self._completed_tool_calls.add(tool_call_id)
+
+    async def _set_bot_is_responding(self, responding: bool):
+        if self._bot_is_responding == responding:
             return
 
-        self._bot_is_speaking = speaking
+        self._bot_is_responding = responding
 
-        if not self._bot_is_speaking and self._end_frame_pending_bot_turn_finished:
+        if not self._bot_is_responding and self._end_frame_pending_bot_turn_finished:
             await self.queue_frame(self._end_frame_pending_bot_turn_finished)
             self._end_frame_pending_bot_turn_finished = None
 
@@ -1116,6 +1228,7 @@ class GeminiLiveLLMService(LLMService):
             if self._session:
                 await self._session.close()
                 self._session = None
+            self._completed_tool_calls = set()
             self._disconnecting = False
         except Exception as e:
             logger.error(f"{self} error disconnecting: {e}")
@@ -1195,7 +1308,8 @@ class GeminiLiveLLMService(LLMService):
             self._run_llm_when_session_ready = True
             return
 
-        messages = self._context.get_messages_for_initializing_history()
+        adapter: GeminiLLMAdapter = self.get_llm_adapter()
+        messages = adapter.get_llm_invocation_params(self._context).get("messages", [])
         if not messages:
             return
 
@@ -1223,8 +1337,9 @@ class GeminiLiveLLMService(LLMService):
 
         # Create a throwaway context just for the purpose of getting messages
         # in the right format
-        context = GeminiLiveContext.upgrade(OpenAILLMContext(messages=messages_list))
-        messages = context.get_messages_for_initializing_history()
+        context = LLMContext(messages=messages_list)
+        adapter: GeminiLLMAdapter = self.get_llm_adapter()
+        messages = adapter.get_llm_invocation_params(context).get("messages", [])
 
         if not messages:
             return
@@ -1239,17 +1354,16 @@ class GeminiLiveLLMService(LLMService):
             await self._handle_send_error(e)
 
     @traced_gemini_live(operation="llm_tool_result")
-    async def _tool_result(self, tool_result_message):
+    async def _tool_result(
+        self, tool_call_id: str, tool_name: str, tool_result_message: Dict[str, Any]
+    ):
         """Send tool result back to the API."""
         if self._disconnecting or not self._session:
             return
 
         # For now we're shoving the name into the tool_call_id field, so this
         # will work until we revisit that.
-        id = tool_result_message.get("tool_call_id")
-        name = tool_result_message.get("tool_call_name")
-        result = json.loads(tool_result_message.get("content") or "")
-        response = FunctionResponse(name=name, id=id, response=result)
+        response = FunctionResponse(name=tool_name, id=tool_call_id, response=tool_result_message)
 
         try:
             await self._session.send_tool_response(function_responses=response)
@@ -1277,7 +1391,10 @@ class GeminiLiveLLMService(LLMService):
         # part.text is added when `modalities` is set to TEXT; otherwise, it's None
         text = part.text
         if text:
-            if not self._bot_text_buffer:
+            if not self._bot_is_responding:
+                # Update bot responding state and send service start frame
+                # (AUDIO modality case)
+                await self._set_bot_is_responding(True)
                 await self.push_frame(LLMFullResponseStartFrame())
 
             self._bot_text_buffer += text
@@ -1288,6 +1405,8 @@ class GeminiLiveLLMService(LLMService):
         if msg.server_content and msg.server_content.grounding_metadata:
             self._accumulated_grounding_metadata = msg.server_content.grounding_metadata
 
+        # If we have no audio, stop here.
+        # All logic below this point pertains to the AUDIO modality.
         inline_data = part.inline_data
         if not inline_data:
             return
@@ -1313,8 +1432,10 @@ class GeminiLiveLLMService(LLMService):
         if not audio:
             return
 
-        if not self._bot_is_speaking:
-            await self._set_bot_is_speaking(True)
+        # Update bot responding state and send service start frames
+        # (AUDIO modality case)
+        if not self._bot_is_responding:
+            await self._set_bot_is_responding(True)
             await self.push_frame(TTSStartedFrame())
             await self.push_frame(LLMFullResponseStartFrame())
 
@@ -1354,7 +1475,6 @@ class GeminiLiveLLMService(LLMService):
     @traced_gemini_live(operation="llm_response")
     async def _handle_msg_turn_complete(self, message: LiveServerMessage):
         """Handle the turn complete message."""
-        await self._set_bot_is_speaking(False)
         text = self._bot_text_buffer
 
         # Trace the complete LLM response (this will be handled by the decorator)
@@ -1373,13 +1493,15 @@ class GeminiLiveLLMService(LLMService):
         self._search_result_buffer = ""
         self._accumulated_grounding_metadata = None
 
-        # Only push the TTSStoppedFrame if the bot is outputting audio
-        # when text is found, modalities is set to TEXT and no audio
-        # is produced.
-        if not text:
-            await self.push_frame(TTSStoppedFrame())
-
-        await self.push_frame(LLMFullResponseEndFrame())
+        if self._bot_is_responding:
+            await self._set_bot_is_responding(False)
+            if not text:
+                # AUDIO modality case
+                await self.push_frame(TTSStoppedFrame())
+                await self.push_frame(LLMFullResponseEndFrame())
+            else:
+                # TEXT modality case
+                await self.push_frame(LLMFullResponseEndFrame())
 
     @traced_stt
     async def _handle_user_transcription(
@@ -1442,8 +1564,8 @@ class GeminiLiveLLMService(LLMService):
             return
 
         # This is the output transcription text when modalities is set to AUDIO.
-        # In this case, we push LLMTextFrame and TTSTextFrame to be handled by the
-        # downstream assistant context aggregator.
+        # In this case, we push TTSTextFrame to be handled by the downstream
+        # assistant context aggregator.
         text = message.server_content.output_transcription.text
 
         if not text:
@@ -1458,7 +1580,17 @@ class GeminiLiveLLMService(LLMService):
         # Collect text for tracing
         self._llm_output_buffer += text
 
-        await self.push_frame(LLMTextFrame(text=text))
+        # NOTE: Shoot. When using Vertex AI, output transcription messages
+        # arrive *before* the model_turn messages with audio, so we need to
+        # handle sending TTSStartedFrame and LLMFullResponseStartFrame here as
+        # well. These messages also contain much *more* text (it looks further
+        # ahead). That means that on an interruption our recorded context will
+        # contain some text that was actually never spoken.
+        if not self._bot_is_responding:
+            await self._set_bot_is_responding(True)
+            await self.push_frame(TTSStartedFrame())
+            await self.push_frame(LLMFullResponseStartFrame())
+
         await self.push_frame(TTSTextFrame(text=text))
 
     async def _handle_msg_grounding_metadata(self, message: LiveServerMessage):
@@ -1557,26 +1689,26 @@ class GeminiLiveLLMService(LLMService):
         *,
         user_params: LLMUserAggregatorParams = LLMUserAggregatorParams(),
         assistant_params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(),
-    ) -> GeminiLiveContextAggregatorPair:
+    ) -> LLMContextAggregatorPair:
         """Create an instance of GeminiLiveContextAggregatorPair from an OpenAILLMContext.
 
         Constructor keyword arguments for both the user and assistant aggregators can be provided.
 
+        NOTE: this method exists only for backward compatibility. New code
+        should instead do:
+            context = LLMContext(...)
+            context_aggregator = LLMContextAggregatorPair(context)
+
         Args:
             context: The LLM context to use.
             user_params: User aggregator parameters. Defaults to LLMUserAggregatorParams().
             assistant_params: Assistant aggregator parameters. Defaults to LLMAssistantAggregatorParams().
 
         Returns:
-            GeminiLiveContextAggregatorPair: A pair of context
-            aggregators, one for the user and one for the assistant,
-            encapsulated in an GeminiLiveContextAggregatorPair.
+            A pair of user and assistant context aggregators.
         """
-        context.set_llm_adapter(self.get_llm_adapter())
-
-        GeminiLiveContext.upgrade(context)
-        user = GeminiLiveUserContextAggregator(context, params=user_params)
-
+        context = LLMContext.from_openai_context(context)
         assistant_params.expect_stripped_words = False
-        assistant = GeminiLiveAssistantContextAggregator(context, params=assistant_params)
-        return GeminiLiveContextAggregatorPair(_user=user, _assistant=assistant)
+        return LLMContextAggregatorPair(
+            context, user_params=user_params, assistant_params=assistant_params
+        )
diff --git a/src/pipecat/services/hume/tts.py b/src/pipecat/services/hume/tts.py
index 2701c5d05..34947fb44 100644
--- a/src/pipecat/services/hume/tts.py
+++ b/src/pipecat/services/hume/tts.py
@@ -184,11 +184,15 @@ class HumeTTSService(TTSService):
             # Hume emits mono PCM at 48 kHz; downstream can resample if needed.
             # We buffer audio bytes before sending to prevent glitches.
             self._audio_bytes = b""
+
+            # Use version "2" by default if no description is provided
+            # Version "1" is needed when description is used
+            version = "1" if self._params.description is not None else "2"
             async for chunk in self._client.tts.synthesize_json_streaming(
                 utterances=[utterance],
                 format=pcm_fmt,
                 instant_mode=True,
-                version="2",
+                version=version,
             ):
                 audio_b64 = getattr(chunk, "audio", None)
                 if not audio_b64:
diff --git a/src/pipecat/services/openai/realtime/context.py b/src/pipecat/services/openai/realtime/context.py
index cb1c0a9f5..91c6e74d5 100644
--- a/src/pipecat/services/openai/realtime/context.py
+++ b/src/pipecat/services/openai/realtime/context.py
@@ -4,7 +4,85 @@
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
-"""OpenAI Realtime LLM context and aggregator implementations."""
+"""OpenAI Realtime LLM context and aggregator implementations.
+
+.. deprecated:: 0.0.92
+    OpenAI Realtime no longer uses types from this module under the hood.
+    It now uses `LLMContext` and `LLMContextAggregatorPair`.
+    Using the new patterns should allow you to not need types from this module.
+
+    BEFORE:
+    ```
+    # Setup
+    context = OpenAILLMContext(messages, tools)
+    context_aggregator = llm.create_context_aggregator(context)
+
+    # Context aggregator type
+    context_aggregator: OpenAIContextAggregatorPair
+
+    # Context frame type
+    frame: OpenAILLMContextFrame
+
+    # Context type
+    context: OpenAIRealtimeLLMContext
+    # or
+    context: OpenAILLMContext
+    ```
+
+    AFTER:
+    ```
+    # Setup
+    context = LLMContext(messages, tools)
+    context_aggregator = LLMContextAggregatorPair(context)
+
+    # Context aggregator type
+    context_aggregator: LLMContextAggregatorPair
+
+    # Context frame type
+    frame: LLMContextFrame
+
+    # Context type
+    context: LLMContext
+    ```
+"""
+
+import warnings
+
+with warnings.catch_warnings():
+    warnings.simplefilter("always")
+    warnings.warn(
+        "Types in pipecat.services.openai.realtime.llm (or "
+        "pipecat.services.openai_realtime.llm) are deprecated. \n"
+        "OpenAI Realtime no longer uses types from this module under the hood. \n"
+        "It now uses `LLMContext` and `LLMContextAggregatorPair`. \n"
+        "Using the new patterns should allow you to not need types from this module.\n\n"
+        "BEFORE:\n"
+        "```\n"
+        "# Setup\n"
+        "context = OpenAILLMContext(messages, tools)\n"
+        "context_aggregator = llm.create_context_aggregator(context)\n\n"
+        "# Context aggregator type\n"
+        "context_aggregator: OpenAIContextAggregatorPair\n\n"
+        "# Context frame type\n"
+        "frame: OpenAILLMContextFrame\n\n"
+        "# Context type\n"
+        "context: OpenAIRealtimeLLMContext\n"
+        "# or\n"
+        "context: OpenAILLMContext\n\n"
+        "```\n\n"
+        "AFTER:\n"
+        "```\n"
+        "# Setup\n"
+        "context = LLMContext(messages, tools)\n"
+        "context_aggregator = LLMContextAggregatorPair(context)\n\n"
+        "# Context aggregator type\n"
+        "context_aggregator: LLMContextAggregatorPair\n\n"
+        "# Context frame type\n"
+        "frame: LLMContextFrame\n\n"
+        "# Context type\n"
+        "context: LLMContext\n\n"
+        "```\n",
+    )
 
 import copy
 import json
diff --git a/src/pipecat/services/openai/realtime/frames.py b/src/pipecat/services/openai/realtime/frames.py
index 8617c6efd..39cfd9757 100644
--- a/src/pipecat/services/openai/realtime/frames.py
+++ b/src/pipecat/services/openai/realtime/frames.py
@@ -4,7 +4,28 @@
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
-"""Custom frame types for OpenAI Realtime API integration."""
+"""Custom frame types for OpenAI Realtime API integration.
+
+.. deprecated:: 0.0.92
+    OpenAI Realtime no longer uses types from this module under the hood.
+
+    It now works more like most LLM services in Pipecat, relying on updates to
+    its context, pushed by context aggregators, to update its internal state.
+
+    Listen for `LLMContextFrame`s for context updates.
+"""
+
+import warnings
+
+with warnings.catch_warnings():
+    warnings.simplefilter("always")
+    warnings.warn(
+        "Types in pipecat.services.openai.realtime.frames are deprecated. \n"
+        "OpenAI Realtime no longer uses types from this module under the hood. \n\n"
+        "It now works more like other LLM services in Pipecat, relying on updates to \n"
+        "its context, pushed by context aggregators, to update its internal state.\n\n"
+        "Listen for `LLMContextFrame`s for context updates.\n"
+    )
 
 from dataclasses import dataclass
 from typing import TYPE_CHECKING
diff --git a/src/pipecat/services/openai/realtime/llm.py b/src/pipecat/services/openai/realtime/llm.py
index 8b3d500eb..012604eb8 100644
--- a/src/pipecat/services/openai/realtime/llm.py
+++ b/src/pipecat/services/openai/realtime/llm.py
@@ -14,7 +14,9 @@ from typing import Optional
 
 from loguru import logger
 
-from pipecat.adapters.services.open_ai_realtime_adapter import OpenAIRealtimeLLMAdapter
+from pipecat.adapters.services.open_ai_realtime_adapter import (
+    OpenAIRealtimeLLMAdapter,
+)
 from pipecat.frames.frames import (
     BotStoppedSpeakingFrame,
     CancelFrame,
@@ -41,10 +43,12 @@ from pipecat.frames.frames import (
     UserStoppedSpeakingFrame,
 )
 from pipecat.metrics.metrics import LLMTokenUsage
+from pipecat.processors.aggregators.llm_context import LLMContext
 from pipecat.processors.aggregators.llm_response import (
     LLMAssistantAggregatorParams,
     LLMUserAggregatorParams,
 )
+from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
 from pipecat.processors.aggregators.openai_llm_context import (
     OpenAILLMContext,
     OpenAILLMContextFrame,
@@ -57,12 +61,6 @@ from pipecat.utils.time import time_now_iso8601
 from pipecat.utils.tracing.service_decorators import traced_openai_realtime, traced_stt
 
 from . import events
-from .context import (
-    OpenAIRealtimeAssistantContextAggregator,
-    OpenAIRealtimeLLMContext,
-    OpenAIRealtimeUserContextAggregator,
-)
-from .frames import RealtimeFunctionCallResultFrame, RealtimeMessagesUpdateFrame
 
 try:
     from websockets.asyncio.client import connect as websocket_connect
@@ -108,22 +106,39 @@ class OpenAIRealtimeLLMService(LLMService):
         base_url: str = "wss://api.openai.com/v1/realtime",
         session_properties: Optional[events.SessionProperties] = None,
         start_audio_paused: bool = False,
-        send_transcription_frames: bool = True,
+        send_transcription_frames: Optional[bool] = None,
         **kwargs,
     ):
         """Initialize the OpenAI Realtime LLM service.
 
         Args:
             api_key: OpenAI API key for authentication.
-            model: OpenAI model name. Defaults to "gpt-4o-realtime-preview-2025-06-03".
+            model: OpenAI model name. Defaults to "gpt-realtime".
             base_url: WebSocket base URL for the realtime API.
                 Defaults to "wss://api.openai.com/v1/realtime".
             session_properties: Configuration properties for the realtime session.
                 If None, uses default SessionProperties.
             start_audio_paused: Whether to start with audio input paused. Defaults to False.
-            send_transcription_frames: Whether to emit transcription frames. Defaults to True.
+            send_transcription_frames: Whether to emit transcription frames.
+
+                .. deprecated:: 0.0.92
+                    This parameter is deprecated and will be removed in a future version.
+                    Transcription frames are always sent.
+
             **kwargs: Additional arguments passed to parent LLMService.
         """
+        if send_transcription_frames is not None:
+            import warnings
+
+            with warnings.catch_warnings():
+                warnings.simplefilter("always")
+                warnings.warn(
+                    "`send_transcription_frames` is deprecated and will be removed in a future version. "
+                    "Transcription frames are always sent.",
+                    DeprecationWarning,
+                    stacklevel=2,
+                )
+
         full_url = f"{base_url}?model={model}"
         super().__init__(base_url=full_url, **kwargs)
 
@@ -135,10 +150,11 @@ class OpenAIRealtimeLLMService(LLMService):
             session_properties or events.SessionProperties()
         )
         self._audio_input_paused = start_audio_paused
-        self._send_transcription_frames = send_transcription_frames
         self._websocket = None
         self._receive_task = None
-        self._context = None
+        self._context: LLMContext = None
+
+        self._llm_needs_conversation_setup = True
 
         self._disconnecting = False
         self._api_session_ready = False
@@ -148,8 +164,8 @@ class OpenAIRealtimeLLMService(LLMService):
         self._current_audio_response = None
 
         self._messages_added_manually = {}
-        self._user_and_response_message_tuple = None
         self._pending_function_calls = {}  # Track function calls by call_id
+        self._completed_tool_calls = set()
 
         self._register_event_handler("on_conversation_item_created")
         self._register_event_handler("on_conversation_item_updated")
@@ -347,22 +363,13 @@ class OpenAIRealtimeLLMService(LLMService):
 
         if isinstance(frame, TranscriptionFrame):
             pass
-        elif isinstance(frame, OpenAILLMContextFrame):
-            context: OpenAIRealtimeLLMContext = OpenAIRealtimeLLMContext.upgrade_to_realtime(
+        elif isinstance(frame, (LLMContextFrame, OpenAILLMContextFrame)):
+            context = (
                 frame.context
+                if isinstance(frame, LLMContextFrame)
+                else LLMContext.from_openai_context(frame.context)
             )
-            if not self._context:
-                self._context = context
-            elif frame.context is not self._context:
-                # If the context has changed, reset the conversation
-                self._context = context
-                await self.reset_conversation()
-            # Run the LLM at next opportunity
-            await self._create_response()
-        elif isinstance(frame, LLMContextFrame):
-            raise NotImplementedError(
-                "Universal LLMContext is not yet supported for OpenAI Realtime."
-            )
+            await self._handle_context(context)
         elif isinstance(frame, InputAudioRawFrame):
             if not self._audio_input_paused:
                 await self._send_user_audio(frame)
@@ -376,29 +383,33 @@ class OpenAIRealtimeLLMService(LLMService):
             await self._handle_bot_stopped_speaking()
         elif isinstance(frame, LLMMessagesAppendFrame):
             await self._handle_messages_append(frame)
-        elif isinstance(frame, RealtimeMessagesUpdateFrame):
-            self._context = frame.context
         elif isinstance(frame, LLMUpdateSettingsFrame):
             self._session_properties = events.SessionProperties(**frame.settings)
             await self._update_settings()
         elif isinstance(frame, LLMSetToolsFrame):
             await self._update_settings()
-        elif isinstance(frame, RealtimeFunctionCallResultFrame):
-            await self._handle_function_call_result(frame.result_frame)
 
         await self.push_frame(frame, direction)
 
+    async def _handle_context(self, context: LLMContext):
+        if not self._context:
+            # We got our initial context
+            self._context = context
+            # Initialize our bookkeeping of already-completed tool calls in
+            # the context
+            await self._process_completed_function_calls(send_new_results=False)
+            # Run the LLM at next opportunity
+            await self._create_response()
+        else:
+            # We got an updated context.
+            # This may contain a new user message or tool call result.
+            self._context = context
+            # Send results for newly-completed function calls, if any.
+            await self._process_completed_function_calls(send_new_results=True)
+
     async def _handle_messages_append(self, frame):
         logger.error("!!! NEED TO IMPLEMENT MESSAGES APPEND")
 
-    async def _handle_function_call_result(self, frame):
-        item = events.ConversationItem(
-            type="function_call_output",
-            call_id=frame.tool_call_id,
-            output=json.dumps(frame.result),
-        )
-        await self.send_client_event(events.ConversationItemCreateEvent(item=item))
-
     #
     # websocket communication
     #
@@ -439,16 +450,21 @@ class OpenAIRealtimeLLMService(LLMService):
             if self._receive_task:
                 await self.cancel_task(self._receive_task, timeout=1.0)
                 self._receive_task = None
+            self._completed_tool_calls = set()
             self._disconnecting = False
         except Exception as e:
             logger.error(f"{self} error disconnecting: {e}")
 
     async def _ws_send(self, realtime_message):
         try:
-            if self._websocket:
+            if not self._disconnecting and self._websocket:
                 await self._websocket.send(json.dumps(realtime_message))
         except Exception as e:
-            if self._disconnecting:
+            if self._disconnecting or not self._websocket:
+                # We're in the process of disconnecting.
+                # (If not self._websocket, that could indicate that we
+                # somehow *started* the websocket send attempt while we still
+                # had a connection)
                 return
             logger.error(f"Error sending message to websocket: {e}")
             # In server-to-server contexts, a WebSocket error should be quite rare. Given how hard
@@ -459,13 +475,20 @@ class OpenAIRealtimeLLMService(LLMService):
 
     async def _update_settings(self):
         settings = self._session_properties
-        # tools given in the context override the tools in the session properties
-        if self._context and self._context.tools:
-            settings.tools = self._context.tools
-        # instructions in the context come from an initial "system" message in the
-        # messages list, and override instructions in the session properties
-        if self._context and self._context._session_instructions:
-            settings.instructions = self._context._session_instructions
+
+        if self._context:
+            adapter: OpenAIRealtimeLLMAdapter = self.get_llm_adapter()
+            llm_invocation_params = adapter.get_llm_invocation_params(self._context)
+
+            # tools given in the context override the tools in the session properties
+            if llm_invocation_params["tools"]:
+                settings.tools = llm_invocation_params["tools"]
+
+            # instructions in the context come from an initial "system" message in the
+            # messages list, and override instructions in the session properties
+            if llm_invocation_params["system_instruction"]:
+                settings.instructions = llm_invocation_params["system_instruction"]
+
         await self.send_client_event(events.SessionUpdateEvent(session=settings))
 
     #
@@ -571,12 +594,7 @@ class OpenAIRealtimeLLMService(LLMService):
             del self._messages_added_manually[evt.item.id]
             return
 
-        if evt.item.role == "user":
-            # We need to wait for completion of both user message and response message. Then we'll
-            # add both to the context. User message is complete when we have a "transcript" field
-            # that is not None. Response message is complete when we get a "response.done" event.
-            self._user_and_response_message_tuple = (evt.item, {"done": False, "output": []})
-        elif evt.item.role == "assistant":
+        if evt.item.role == "assistant":
             self._current_assistant_response = evt.item
             await self.push_frame(LLMFullResponseStartFrame())
 
@@ -587,11 +605,11 @@ class OpenAIRealtimeLLMService(LLMService):
         # For now, no additional logic needed beyond the event handler call
 
     async def _handle_evt_input_audio_transcription_delta(self, evt):
-        if self._send_transcription_frames:
-            await self.push_frame(
-                # no way to get a language code?
-                InterimTranscriptionFrame(evt.delta, "", time_now_iso8601(), result=evt)
-            )
+        await self.push_frame(
+            # no way to get a language code?
+            InterimTranscriptionFrame(evt.delta, "", time_now_iso8601(), result=evt),
+            direction=FrameDirection.UPSTREAM,
+        )
 
     @traced_stt
     async def _handle_user_transcription(
@@ -608,22 +626,12 @@ class OpenAIRealtimeLLMService(LLMService):
         """
         await self._call_event_handler("on_conversation_item_updated", evt.item_id, None)
 
-        if self._send_transcription_frames:
-            await self.push_frame(
-                # no way to get a language code?
-                TranscriptionFrame(evt.transcript, "", time_now_iso8601(), result=evt)
-            )
-            await self._handle_user_transcription(evt.transcript, True, Language.EN)
-        pair = self._user_and_response_message_tuple
-        if pair:
-            user, assistant = pair
-            user.content[0].transcript = evt.transcript
-            if assistant["done"]:
-                self._user_and_response_message_tuple = None
-                self._context.add_user_content_item_as_message(user)
-        else:
-            # User message without preceding conversation.item.created. Bug?
-            logger.warning(f"Transcript for unknown user message: {evt}")
+        await self.push_frame(
+            # no way to get a language code?
+            TranscriptionFrame(evt.transcript, "", time_now_iso8601(), result=evt),
+            FrameDirection.UPSTREAM,
+        )
+        await self._handle_user_transcription(evt.transcript, True, Language.EN)
 
     async def _handle_conversation_item_retrieved(self, evt: events.ConversationItemRetrieved):
         futures = self._retrieve_conversation_item_futures.pop(evt.item.id, None)
@@ -653,26 +661,17 @@ class OpenAIRealtimeLLMService(LLMService):
         # response content
         for item in evt.response.output:
             await self._call_event_handler("on_conversation_item_updated", item.id, item)
-        pair = self._user_and_response_message_tuple
-        if pair:
-            user, assistant = pair
-            assistant["done"] = True
-            assistant["output"] = evt.response.output
-            if user.content[0].transcript is not None:
-                self._user_and_response_message_tuple = None
-                self._context.add_user_content_item_as_message(user)
-        else:
-            # Response message without preceding user message (standalone response)
-            # Function calls in this response were already processed immediately when arguments were complete
-            logger.debug(f"Handling standalone response: {evt.response.id}")
 
     async def _handle_evt_text_delta(self, evt):
+        # We receive text deltas (as opposed to audio transcript deltas) when
+        # the output modality is "text"
         if evt.delta:
             await self.push_frame(LLMTextFrame(evt.delta))
 
     async def _handle_evt_audio_transcript_delta(self, evt):
+        # We receive audio transcript deltas (as opposed to text deltas) when
+        # the output modality is "audio" (the default)
         if evt.delta:
-            await self.push_frame(LLMTextFrame(evt.delta))
             await self.push_frame(TTSTextFrame(evt.delta))
 
     async def _handle_evt_function_call_arguments_done(self, evt):
@@ -760,9 +759,11 @@ class OpenAIRealtimeLLMService(LLMService):
         """
         logger.debug("Resetting conversation")
         await self._disconnect()
-        if self._context:
-            self._context.llm_needs_settings_update = True
-            self._context.llm_needs_initial_messages = True
+
+        # Prepare to setup server-side conversation from local context again
+        self._llm_needs_conversation_setup = True
+        await self._process_completed_function_calls(send_new_results=False)
+
         await self._connect()
 
     @traced_openai_realtime(operation="llm_request")
@@ -771,19 +772,29 @@ class OpenAIRealtimeLLMService(LLMService):
             self._run_llm_when_api_session_ready = True
             return
 
-        if self._context.llm_needs_initial_messages:
-            messages = self._context.get_messages_for_initializing_history()
+        adapter: OpenAIRealtimeLLMAdapter = self.get_llm_adapter()
+
+        # Configure the LLM for this session if needed
+        if self._llm_needs_conversation_setup:
+            logger.debug(
+                f"Setting up conversation on OpenAI Realtime LLM service with initial messages: {adapter.get_messages_for_logging(self._context)}"
+            )
+
+            # Send initial messages
+            llm_invocation_params = adapter.get_llm_invocation_params(self._context)
+            messages = llm_invocation_params["messages"]
             for item in messages:
                 evt = events.ConversationItemCreateEvent(item=item)
                 self._messages_added_manually[evt.item.id] = True
                 await self.send_client_event(evt)
-            self._context.llm_needs_initial_messages = False
 
-        if self._context.llm_needs_settings_update:
+            # Send new settings if needed
             await self._update_settings()
-            self._context.llm_needs_settings_update = False
 
-        logger.debug(f"Creating response: {self._context.get_messages_for_logging()}")
+            # We're done configuring the LLM for this session
+            self._llm_needs_conversation_setup = False
+
+        logger.debug(f"Creating response")
 
         await self.push_frame(LLMFullResponseStartFrame())
         await self.start_processing_metrics()
@@ -794,19 +805,50 @@ class OpenAIRealtimeLLMService(LLMService):
             )
         )
 
+    async def _process_completed_function_calls(self, send_new_results: bool):
+        # Check for set of completed function calls in the context
+        sent_new_result = False
+        for message in self._context.get_messages():
+            if message.get("role") and message.get("content") != "IN_PROGRESS":
+                tool_call_id = message.get("tool_call_id")
+                if tool_call_id and tool_call_id not in self._completed_tool_calls:
+                    # Found a newly-completed function call - send the result to the service
+                    if send_new_results:
+                        sent_new_result = True
+                        await self._send_tool_result(tool_call_id, message.get("content"))
+                    self._completed_tool_calls.add(tool_call_id)
+
+        # If we reported any new tool call results to the service, trigger
+        # another response
+        if sent_new_result:
+            await self._create_response()
+
     async def _send_user_audio(self, frame):
         payload = base64.b64encode(frame.audio).decode("utf-8")
         await self.send_client_event(events.InputAudioBufferAppendEvent(audio=payload))
 
+    async def _send_tool_result(self, tool_call_id: str, result: str):
+        item = events.ConversationItem(
+            type="function_call_output",
+            call_id=tool_call_id,
+            output=json.dumps(result),
+        )
+        await self.send_client_event(events.ConversationItemCreateEvent(item=item))
+
     def create_context_aggregator(
         self,
         context: OpenAILLMContext,
         *,
         user_params: LLMUserAggregatorParams = LLMUserAggregatorParams(),
         assistant_params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(),
-    ) -> OpenAIContextAggregatorPair:
+    ) -> LLMContextAggregatorPair:
         """Create an instance of OpenAIContextAggregatorPair from an OpenAILLMContext.
 
+        NOTE: this method exists only for backward compatibility. New code
+        should instead do:
+            context = LLMContext(...)
+            context_aggregator = LLMContextAggregatorPair(context)
+
         Constructor keyword arguments for both the user and assistant aggregators can be provided.
 
         Args:
@@ -819,11 +861,41 @@ class OpenAIRealtimeLLMService(LLMService):
             the user and one for the assistant, encapsulated in an
             OpenAIContextAggregatorPair.
         """
-        context.set_llm_adapter(self.get_llm_adapter())
-
-        OpenAIRealtimeLLMContext.upgrade_to_realtime(context)
-        user = OpenAIRealtimeUserContextAggregator(context, params=user_params)
+        # Log warning about transcription frame direction change in 0.0.92.
+        # We're putting this warning here rather than in the constructor so
+        # that it shows up for folks who haven't updated their code at all
+        # since 0.0.92, gives them a way to acknowledge and dismiss the
+        # warning, and encourages adoption of a new preferred pattern.
+        logger.warning(
+            "As of version 0.0.92, TranscriptionFrames and InterimTranscriptionFrames "
+            "now go upstream from OpenAIRealtimeLLMService, so if you're using "
+            "TranscriptProcessor, say, you'll want to adjust accordingly:\n\n"
+            "pipeline = Pipeline(\n"
+            "  [\n"
+            "    transport.input(),\n"
+            "    context_aggregator.user(),\n\n"
+            "    # BEFORE\n"
+            "    llm,\n"
+            "    transcript.user(),\n\n"
+            "    # AFTER\n"
+            "    transcript.user(),\n"
+            "    llm,\n\n"
+            "    transport.output(),\n"
+            "    transcript.assistant(),\n"
+            "    context_aggregator.assistant(),\n"
+            "  ]\n"
+            ")\n\n"
+            "Also, LLMTextFrames are no longer pushed from "
+            "OpenAIRealtimeLLMService when it's configured with "
+            "output_modalities=['audio']. Listen for TTSTextFrames instead.\n\n"
+            "Once you've made the appropriate changes (if needed), you can "
+            "dismiss this warning by updating to the new context-setup pattern:\n\n"
+            "  context = LLMContext(messages, tools)\n"
+            "  context_aggregator = LLMContextAggregatorPair(context)\n"
+        )
 
+        context = LLMContext.from_openai_context(context)
         assistant_params.expect_stripped_words = False
-        assistant = OpenAIRealtimeAssistantContextAggregator(context, params=assistant_params)
-        return OpenAIContextAggregatorPair(_user=user, _assistant=assistant)
+        return LLMContextAggregatorPair(
+            context, user_params=user_params, assistant_params=assistant_params
+        )
diff --git a/src/pipecat/services/openai_realtime/context.py b/src/pipecat/services/openai_realtime/context.py
index 58f1cfe75..79a01b980 100644
--- a/src/pipecat/services/openai_realtime/context.py
+++ b/src/pipecat/services/openai_realtime/context.py
@@ -4,18 +4,15 @@
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
-"""OpenAI Realtime LLM context and aggregator implementations."""
+"""OpenAI Realtime LLM context and aggregator implementations.
 
-import warnings
+.. deprecated:: 0.0.91
+    OpenAI Realtime no longer uses types from this module under the hood.
+    It now uses `LLMContext` and `LLMContextAggregatorPair`.
+    Using the new patterns should allow you to not need types from this module.
+
+    See deprecation warning in pipecat.services.openai.realtime.context for
+    more details.
+"""
 
 from pipecat.services.openai.realtime.context import *
-
-with warnings.catch_warnings():
-    warnings.simplefilter("always")
-    warnings.warn(
-        "Types in pipecat.services.openai_realtime.context are deprecated. "
-        "Please use the equivalent types from "
-        "pipecat.services.openai.realtime.context instead.",
-        DeprecationWarning,
-        stacklevel=2,
-    )
diff --git a/src/pipecat/services/simli/video.py b/src/pipecat/services/simli/video.py
index d48a744e0..383a8a3cb 100644
--- a/src/pipecat/services/simli/video.py
+++ b/src/pipecat/services/simli/video.py
@@ -7,9 +7,12 @@
 """Simli video service for real-time avatar generation."""
 
 import asyncio
+import warnings
+from typing import Optional
 
 import numpy as np
 from loguru import logger
+from pydantic import BaseModel
 
 from pipecat.frames.frames import (
     CancelFrame,
@@ -41,30 +44,103 @@ class SimliVideoService(FrameProcessor):
     audio resampling, video frame processing, and connection management.
     """
 
+    class InputParams(BaseModel):
+        """Input parameters for Simli video configuration.
+
+        Parameters:
+            max_session_length: Absolute maximum session duration in seconds.
+                Avatar will disconnect after this time even if it's speaking.
+            max_idle_time: Maximum duration in seconds the avatar is not speaking
+                before the avatar disconnects.
+        """
+
+        max_session_length: Optional[int] = None
+        max_idle_time: Optional[int] = None
+
     def __init__(
         self,
-        simli_config: SimliConfig,
+        *,
+        api_key: Optional[str] = None,
+        face_id: Optional[str] = None,
+        simli_config: Optional[SimliConfig] = None,
         use_turn_server: bool = False,
         latency_interval: int = 0,
         simli_url: str = "https://api.simli.ai",
         is_trinity_avatar: bool = False,
+        params: Optional[InputParams] = None,
+        **kwargs,
     ):
         """Initialize the Simli video service.
 
         Args:
+            api_key: Simli API key for authentication.
+            face_id: Simli Face ID. For Trinity avatars, specify "faceId/emotionId"
+                to use a different emotion than the default.
             simli_config: Configuration object for Simli client settings.
-            use_turn_server: Whether to use TURN server for connection. Defaults to False.
-            latency_interval: Latency interval setting for sending health checks to check the latency to Simli Servers. Defaults to 0.
-            simli_url: URL of the simli servers. Can be changed for custom deployments of enterprise users.
-            is_trinity_avatar: boolean to tell simli client that this is a Trinity avatar which reduces latency when using Trinity.
+                Use api_key and face_id instead.
 
+                .. deprecated:: 0.0.92
+                    The 'simli_config' parameter is deprecated and will be removed in a future version.
+                    Please use 'api_key' and 'face_id' parameters instead.
+
+            use_turn_server: Whether to use TURN server for connection. Defaults to False.
+            latency_interval: Latency interval setting for sending health checks to check
+                the latency to Simli Servers. Defaults to 0.
+            simli_url: URL of the simli servers. Can be changed for custom deployments
+                of enterprise users.
+            is_trinity_avatar: Boolean to tell simli client that this is a Trinity avatar
+                which reduces latency when using Trinity.
+            params: Additional input parameters for session configuration.
+            **kwargs: Additional arguments passed to the parent FrameProcessor.
         """
-        super().__init__()
+        super().__init__(**kwargs)
+
+        params = params or SimliVideoService.InputParams()
+
+        # Handle deprecated simli_config parameter
+        if simli_config is not None:
+            if api_key is not None or face_id is not None:
+                raise ValueError(
+                    "Cannot specify both simli_config and api_key/face_id. "
+                    "Please use api_key and face_id (simli_config is deprecated)."
+                )
+
+            warnings.warn(
+                "The 'simli_config' parameter is deprecated and will be removed in a future version. "
+                "Please use 'api_key' and 'face_id' parameters instead, with optional 'params' for "
+                "max_session_length and max_idle_time configuration.",
+                DeprecationWarning,
+                stacklevel=2,
+            )
+
+            # Use the provided simli_config
+            config = simli_config
+        else:
+            # Validate new parameters
+            if api_key is None:
+                raise ValueError("api_key is required")
+            if face_id is None:
+                raise ValueError("face_id is required")
+
+            # Build SimliConfig from new parameters
+            # Only pass optional parameters if explicitly provided to use SimliConfig defaults
+            config_kwargs = {
+                "apiKey": api_key,
+                "faceId": face_id,
+            }
+            if params.max_session_length is not None:
+                config_kwargs["maxSessionLength"] = params.max_session_length
+            if params.max_idle_time is not None:
+                config_kwargs["maxIdleTime"] = params.max_idle_time
+
+            config = SimliConfig(**config_kwargs)
+
         self._initialized = False
-        simli_config.maxIdleTime += 5
-        simli_config.maxSessionLength += 5
+        # Add buffer time to session limits
+        config.maxIdleTime += 5
+        config.maxSessionLength += 5
         self._simli_client = SimliClient(
-            simli_config,
+            config,
             use_turn_server,
             latency_interval,
             simliURL=simli_url,
diff --git a/src/pipecat/transports/daily/transport.py b/src/pipecat/transports/daily/transport.py
index 2e48c4d2e..18c36ee56 100644
--- a/src/pipecat/transports/daily/transport.py
+++ b/src/pipecat/transports/daily/transport.py
@@ -16,7 +16,7 @@ import time
 from concurrent.futures import CancelledError as FuturesCancelledError
 from concurrent.futures import ThreadPoolExecutor
 from dataclasses import dataclass
-from typing import Any, Awaitable, Callable, Dict, Mapping, Optional
+from typing import Any, Awaitable, Callable, Dict, Mapping, Optional, Tuple
 
 import aiohttp
 from loguru import logger
@@ -419,6 +419,11 @@ class DailyAudioTrack:
     track: CustomAudioTrack
 
 
+# This is just a type alias for the errors returned by daily-python. Right now
+# they are just a string.
+CallClientError = str
+
+
 class DailyTransportClient(EventHandler):
     """Core client for interacting with Daily's API.
 
@@ -553,14 +558,17 @@ class DailyTransportClient(EventHandler):
 
     async def send_message(
         self, frame: OutputTransportMessageFrame | OutputTransportMessageUrgentFrame
-    ):
+    ) -> Optional[CallClientError]:
         """Send an application message to participants.
 
         Args:
             frame: The message frame to send.
+
+        Returns:
+            error: An error description or None.
         """
         if not self._joined:
-            return
+            return "Unable to send messages before joining."
 
         participant_id = None
         if isinstance(
@@ -572,7 +580,7 @@ class DailyTransportClient(EventHandler):
         self._client.send_app_message(
             frame.message, participant_id, completion=completion_callback(future)
         )
-        await future
+        return await future
 
     async def read_next_audio_frame(self) -> Optional[InputAudioRawFrame]:
         """Reads the next 20ms audio frame from the virtual speaker."""
@@ -754,9 +762,6 @@ class DailyTransportClient(EventHandler):
 
             logger.info(f"Joined {self._room_url}")
 
-            if self._params.transcription_enabled:
-                await self.start_transcription(self._params.transcription_settings)
-
             await self._callbacks.on_joined(data)
 
             self._joined_event.set()
@@ -842,9 +847,6 @@ class DailyTransportClient(EventHandler):
         # Call callback before leaving.
         await self._callbacks.on_before_leave()
 
-        if self._params.transcription_enabled:
-            await self.stop_transcription()
-
         # Remove any custom tracks, if any.
         for track_name, _ in self._custom_audio_tracks.items():
             await self.remove_custom_audio_track(track_name)
@@ -873,7 +875,7 @@ class DailyTransportClient(EventHandler):
             self._client.release()
             self._client = None
 
-    def participants(self):
+    def participants(self) -> Mapping[str, Any]:
         """Get current participants in the room.
 
         Returns:
@@ -881,7 +883,7 @@ class DailyTransportClient(EventHandler):
         """
         return self._client.participants()
 
-    def participant_counts(self):
+    def participant_counts(self) -> Mapping[str, Any]:
         """Get participant count information.
 
         Returns:
@@ -889,165 +891,173 @@ class DailyTransportClient(EventHandler):
         """
         return self._client.participant_counts()
 
-    async def start_dialout(self, settings):
+    async def start_dialout(self, settings) -> Tuple[str, Optional[CallClientError]]:
         """Start a dial-out call to a phone number.
 
         Args:
             settings: Dial-out configuration settings.
-        """
-        logger.debug(f"Starting dialout: settings={settings}")
 
+        Returns:
+            session_id: Dail-out session ID.
+            error: An error description or None.
+        """
         future = self._get_event_loop().create_future()
         self._client.start_dialout(settings, completion=completion_callback(future))
-        error = await future
-        if error:
-            logger.error(f"Unable to start dialout: {error}")
+        return await future
 
-    async def stop_dialout(self, participant_id):
+    async def stop_dialout(self, participant_id) -> Optional[CallClientError]:
         """Stop a dial-out call for a specific participant.
 
         Args:
             participant_id: ID of the participant to stop dial-out for.
-        """
-        logger.debug(f"Stopping dialout: participant_id={participant_id}")
 
+        Returns:
+            error: An error description or None.
+        """
         future = self._get_event_loop().create_future()
         self._client.stop_dialout(participant_id, completion=completion_callback(future))
-        error = await future
-        if error:
-            logger.error(f"Unable to stop dialout: {error}")
+        return await future
 
-    async def send_dtmf(self, settings):
+    async def send_dtmf(self, settings) -> Optional[CallClientError]:
         """Send DTMF tones during a call.
 
         Args:
             settings: DTMF settings including tones and target session.
+
+        Returns:
+            error: An error description or None.
         """
         session_id = settings.get("sessionId") or self._dial_out_session_id
         if not session_id:
-            logger.error("Unable to send DTMF: 'sessionId' is not set")
-            return
+            return "Can't send DTMF if 'sessionId' is not set"
 
         # Update 'sessionId' field.
         settings["sessionId"] = session_id
 
         future = self._get_event_loop().create_future()
         self._client.send_dtmf(settings, completion=completion_callback(future))
-        await future
+        return await future
 
-    async def sip_call_transfer(self, settings):
+    async def sip_call_transfer(self, settings) -> Optional[CallClientError]:
         """Transfer a SIP call to another destination.
 
         Args:
             settings: SIP call transfer settings.
+
+        Returns:
+            error: An error description or None.
         """
         session_id = (
             settings.get("sessionId") or self._dial_out_session_id or self._dial_in_session_id
         )
         if not session_id:
-            logger.error("Unable to transfer SIP call: 'sessionId' is not set")
-            return
+            return "Can't transfer SIP call if 'sessionId' is not set"
 
         # Update 'sessionId' field.
         settings["sessionId"] = session_id
 
         future = self._get_event_loop().create_future()
         self._client.sip_call_transfer(settings, completion=completion_callback(future))
-        await future
+        return await future
 
-    async def sip_refer(self, settings):
+    async def sip_refer(self, settings) -> Optional[CallClientError]:
         """Send a SIP REFER request.
 
         Args:
             settings: SIP REFER settings.
+
+        Returns:
+            error: An error description or None.
         """
         future = self._get_event_loop().create_future()
         self._client.sip_refer(settings, completion=completion_callback(future))
-        await future
+        return await future
 
-    async def start_recording(self, streaming_settings, stream_id, force_new):
+    async def start_recording(
+        self, streaming_settings, stream_id, force_new
+    ) -> Tuple[str, Optional[CallClientError]]:
         """Start recording the call.
 
         Args:
             streaming_settings: Recording configuration settings.
             stream_id: Unique identifier for the recording stream.
             force_new: Whether to force a new recording session.
-        """
-        logger.debug(
-            f"Starting recording: stream_id={stream_id} force_new={force_new} settings={streaming_settings}"
-        )
 
+        Returns:
+            stream_id: Unique identifier for the recording stream.
+            error: An error description or None.
+        """
         future = self._get_event_loop().create_future()
         self._client.start_recording(
             streaming_settings, stream_id, force_new, completion=completion_callback(future)
         )
-        error = await future
-        if error:
-            logger.error(f"Unable to start recording: {error}")
+        return await future
 
-    async def stop_recording(self, stream_id):
+    async def stop_recording(self, stream_id) -> Optional[CallClientError]:
         """Stop recording the call.
 
         Args:
             stream_id: Unique identifier for the recording stream to stop.
-        """
-        logger.debug(f"Stopping recording: stream_id={stream_id}")
 
+        Returns:
+            error: An error description or None.
+        """
         future = self._get_event_loop().create_future()
         self._client.stop_recording(stream_id, completion=completion_callback(future))
-        error = await future
-        if error:
-            logger.error(f"Unable to stop recording: {error}")
+        return await future
 
-    async def start_transcription(self, settings):
+    async def start_transcription(self, settings) -> Optional[CallClientError]:
         """Start transcription for the call.
 
         Args:
             settings: Transcription configuration settings.
+
+        Returns:
+            error: An error description or None.
         """
         if not self._token:
-            logger.warning("Transcription can't be started without a room token")
-            return
-
-        logger.debug(f"Starting transcription: settings={settings}")
+            return "Transcription can't be started without a room token"
 
         future = self._get_event_loop().create_future()
         self._client.start_transcription(
             settings=self._params.transcription_settings.model_dump(exclude_none=True),
             completion=completion_callback(future),
         )
-        error = await future
-        if error:
-            logger.error(f"Unable to start transcription: {error}")
+        return await future
 
-    async def stop_transcription(self):
-        """Stop transcription for the call."""
+    async def stop_transcription(self) -> Optional[CallClientError]:
+        """Stop transcription for the call.
+
+        Returns:
+            error: An error description or None.
+        """
         if not self._token:
-            return
-
-        logger.debug(f"Stopping transcription")
+            return "Transcription can't be stopped without a room token"
 
         future = self._get_event_loop().create_future()
         self._client.stop_transcription(completion=completion_callback(future))
-        error = await future
-        if error:
-            logger.error(f"Unable to stop transcription: {error}")
+        return await future
 
-    async def send_prebuilt_chat_message(self, message: str, user_name: Optional[str] = None):
+    async def send_prebuilt_chat_message(
+        self, message: str, user_name: Optional[str] = None
+    ) -> Optional[CallClientError]:
         """Send a chat message to Daily's Prebuilt main room.
 
         Args:
             message: The chat message to send.
             user_name: Optional user name that will appear as sender of the message.
+
+        Returns:
+            error: An error description or None.
         """
         if not self._joined:
-            return
+            return "Can't send message if not joined"
 
         future = self._get_event_loop().create_future()
         self._client.send_prebuilt_chat_message(
             message, user_name=user_name, completion=completion_callback(future)
         )
-        await future
+        return await future
 
     async def capture_participant_transcription(self, participant_id: str):
         """Enable transcription capture for a specific participant.
@@ -1167,38 +1177,51 @@ class DailyTransportClient(EventHandler):
 
         return track
 
-    async def remove_custom_audio_track(self, track_name: str):
+    async def remove_custom_audio_track(self, track_name: str) -> Optional[CallClientError]:
         """Remove a custom audio track.
 
         Args:
             track_name: Name of the custom audio track to remove.
+
+        Returns:
+            error: An error description or None.
         """
         future = self._get_event_loop().create_future()
         self._client.remove_custom_audio_track(
             track_name=track_name,
             completion=completion_callback(future),
         )
-        await future
+        return await future
 
-    async def update_transcription(self, participants=None, instance_id=None):
+    async def update_transcription(
+        self, participants=None, instance_id=None
+    ) -> Optional[CallClientError]:
         """Update transcription settings for specific participants.
 
         Args:
             participants: List of participant IDs to enable transcription for.
             instance_id: Optional transcription instance ID.
+
+        Returns:
+            error: An error description or None.
         """
         future = self._get_event_loop().create_future()
         self._client.update_transcription(
             participants, instance_id, completion=completion_callback(future)
         )
-        await future
+        return await future
 
-    async def update_subscriptions(self, participant_settings=None, profile_settings=None):
+    async def update_subscriptions(
+        self, participant_settings=None, profile_settings=None
+    ) -> Optional[CallClientError]:
         """Update media subscription settings.
 
         Args:
             participant_settings: Per-participant subscription settings.
             profile_settings: Global subscription profile settings.
+
+        Returns:
+            error: An error description or None.
         """
         future = self._get_event_loop().create_future()
         self._client.update_subscriptions(
@@ -1206,32 +1229,42 @@ class DailyTransportClient(EventHandler):
             profile_settings=profile_settings,
             completion=completion_callback(future),
         )
-        await future
+        return await future
 
-    async def update_publishing(self, publishing_settings: Mapping[str, Any]):
+    async def update_publishing(
+        self, publishing_settings: Mapping[str, Any]
+    ) -> Optional[CallClientError]:
         """Update media publishing settings.
 
         Args:
             publishing_settings: Publishing configuration settings.
+
+        Returns:
+            error: An error description or None.
         """
         future = self._get_event_loop().create_future()
         self._client.update_publishing(
             publishing_settings=publishing_settings,
             completion=completion_callback(future),
         )
-        await future
+        return await future
 
-    async def update_remote_participants(self, remote_participants: Mapping[str, Any]):
+    async def update_remote_participants(
+        self, remote_participants: Mapping[str, Any]
+    ) -> Optional[CallClientError]:
         """Update settings for remote participants.
 
         Args:
             remote_participants: Remote participant configuration settings.
+
+        Returns:
+            error: An error description or None.
         """
         future = self._get_event_loop().create_future()
         self._client.update_remote_participants(
             remote_participants=remote_participants, completion=completion_callback(future)
         )
-        await future
+        return await future
 
     #
     #
@@ -1922,7 +1955,9 @@ class DailyOutputTransport(BaseOutputTransport):
         Args:
             frame: The transport message frame to send.
         """
-        await self._client.send_message(frame)
+        error = await self._client.send_message(frame)
+        if error:
+            logger.error(f"Unable to send message: {error}")
 
     async def register_video_destination(self, destination: str):
         """Register a video output destination.
@@ -2166,7 +2201,7 @@ class DailyTransport(BaseTransport):
         if self._output:
             await self._output.queue_frame(frame, FrameDirection.DOWNSTREAM)
 
-    def participants(self):
+    def participants(self) -> Mapping[str, Any]:
         """Get current participants in the room.
 
         Returns:
@@ -2174,7 +2209,7 @@ class DailyTransport(BaseTransport):
         """
         return self._client.participants()
 
-    def participant_counts(self):
+    def participant_counts(self) -> Mapping[str, Any]:
         """Get participant count information.
 
         Returns:
@@ -2182,76 +2217,155 @@ class DailyTransport(BaseTransport):
         """
         return self._client.participant_counts()
 
-    async def start_dialout(self, settings=None):
+    async def start_dialout(self, settings=None) -> Tuple[str, Optional[CallClientError]]:
         """Start a dial-out call to a phone number.
 
         Args:
             settings: Dial-out configuration settings.
-        """
-        await self._client.start_dialout(settings)
 
-    async def stop_dialout(self, participant_id):
+        Returns:
+            session_id: Dail-out session ID.
+            error: An error description or None.
+        """
+        logger.debug(f"Starting dialout: settings={settings}")
+
+        session_id, error = await self._client.start_dialout(settings)
+        if error:
+            logger.error(f"Unable to start dialout: {error}")
+        return session_id, error
+
+    async def stop_dialout(self, participant_id) -> Optional[CallClientError]:
         """Stop a dial-out call for a specific participant.
 
         Args:
             participant_id: ID of the participant to stop dial-out for.
-        """
-        await self._client.stop_dialout(participant_id)
 
-    async def sip_call_transfer(self, settings):
+        Returns:
+            error: An error description or None.
+        """
+        logger.debug(f"Stopping dialout: participant_id={participant_id}")
+
+        error = await self._client.stop_dialout(participant_id)
+        if error:
+            logger.error(f"Unable to stop dialout: {error}")
+        return error
+
+    async def sip_call_transfer(self, settings) -> Optional[CallClientError]:
         """Transfer a SIP call to another destination.
 
         Args:
             settings: SIP call transfer settings.
-        """
-        await self._client.sip_call_transfer(settings)
 
-    async def sip_refer(self, settings):
+        Returns:
+            error: An error description or None.
+        """
+        logger.debug(f"Staring SIP call transfer: settings={settings}")
+
+        error = await self._client.sip_call_transfer(settings)
+        if error:
+            logger.error(f"Unable to transfer SIP call: {error}")
+        return error
+
+    async def sip_refer(self, settings) -> Optional[CallClientError]:
         """Send a SIP REFER request.
 
         Args:
             settings: SIP REFER settings.
-        """
-        await self._client.sip_refer(settings)
 
-    async def start_recording(self, streaming_settings=None, stream_id=None, force_new=None):
+        Returns:
+            error: An error description or None.
+        """
+        logger.debug(f"Staring SIP REFER: settings={settings}")
+
+        error = await self._client.sip_refer(settings)
+        if error:
+            logger.error(f"Unable to perform SIP REFER: {error}")
+        return error
+
+    async def start_recording(
+        self, streaming_settings=None, stream_id=None, force_new=None
+    ) -> Tuple[str, Optional[CallClientError]]:
         """Start recording the call.
 
         Args:
             streaming_settings: Recording configuration settings.
             stream_id: Unique identifier for the recording stream.
             force_new: Whether to force a new recording session.
-        """
-        await self._client.start_recording(streaming_settings, stream_id, force_new)
 
-    async def stop_recording(self, stream_id=None):
+        Returns:
+            stream_id: Unique identifier for the recording stream.
+            error: An error description or None.
+        """
+        logger.debug(
+            f"Starting recording: stream_id={stream_id} force_new={force_new} settings={streaming_settings}"
+        )
+
+        r_id, error = await self._client.start_recording(streaming_settings, stream_id, force_new)
+        if error:
+            logger.error(f"Unable to start recording: {error}")
+        return r_id, error
+
+    async def stop_recording(self, stream_id=None) -> Optional[CallClientError]:
         """Stop recording the call.
 
         Args:
             stream_id: Unique identifier for the recording stream to stop.
-        """
-        await self._client.stop_recording(stream_id)
 
-    async def start_transcription(self, settings=None):
+        Returns:
+            error: An error description or None.
+        """
+        logger.debug(f"Stopping recording: stream_id={stream_id}")
+
+        error = await self._client.stop_recording(stream_id)
+        if error:
+            logger.error(f"Unable to stop recording: {error}")
+        return error
+
+    async def start_transcription(self, settings=None) -> Optional[CallClientError]:
         """Start transcription for the call.
 
         Args:
             settings: Transcription configuration settings.
+
+        Returns:
+            error: An error description or None.
         """
-        await self._client.start_transcription(settings)
+        logger.debug(f"Starting transcription: settings={settings}")
 
-    async def stop_transcription(self):
-        """Stop transcription for the call."""
-        await self._client.stop_transcription()
+        error = await self._client.start_transcription(settings)
+        if error:
+            logger.error(f"Unable to start transcription: {error}")
+        return error
 
-    async def send_prebuilt_chat_message(self, message: str, user_name: Optional[str] = None):
+    async def stop_transcription(self) -> Optional[CallClientError]:
+        """Stop transcription for the call.
+
+        Returns:
+            error: An error description or None.
+        """
+        logger.debug(f"Stopping transcription")
+
+        error = await self._client.stop_transcription()
+        if error:
+            logger.error(f"Unable to stop transcription: {error}")
+        return error
+
+    async def send_prebuilt_chat_message(
+        self, message: str, user_name: Optional[str] = None
+    ) -> Optional[CallClientError]:
         """Send a chat message to Daily's Prebuilt main room.
 
         Args:
             message: The chat message to send.
             user_name: Optional user name that will appear as sender of the message.
+
+        Returns:
+            error: An error description or None.
         """
-        await self._client.send_prebuilt_chat_message(message, user_name)
+        error = await self._client.send_prebuilt_chat_message(message, user_name)
+        if error:
+            logger.error(f"Unable to send prebuilt chat message: {error}")
+        return error
 
     async def capture_participant_transcription(self, participant_id: str):
         """Enable transcription capture for a specific participant.
@@ -2297,32 +2411,66 @@ class DailyTransport(BaseTransport):
                 participant_id, framerate, video_source, color_format
             )
 
-    async def update_publishing(self, publishing_settings: Mapping[str, Any]):
+    async def update_publishing(
+        self, publishing_settings: Mapping[str, Any]
+    ) -> Optional[CallClientError]:
         """Update media publishing settings.
 
         Args:
             publishing_settings: Publishing configuration settings.
-        """
-        await self._client.update_publishing(publishing_settings=publishing_settings)
 
-    async def update_subscriptions(self, participant_settings=None, profile_settings=None):
+        Returns:
+            error: An error description or None.
+        """
+        logger.debug(f"Updating publishing settings: settings={publishing_settings}")
+
+        error = await self._client.update_publishing(publishing_settings=publishing_settings)
+        if error:
+            logger.error(f"Unable to update publishing settings: {error}")
+        return error
+
+    async def update_subscriptions(
+        self, participant_settings=None, profile_settings=None
+    ) -> Optional[CallClientError]:
         """Update media subscription settings.
 
         Args:
             participant_settings: Per-participant subscription settings.
             profile_settings: Global subscription profile settings.
+
+        Returns:
+            error: An error description or None.
         """
-        await self._client.update_subscriptions(
-            participant_settings=participant_settings, profile_settings=profile_settings
+        logger.debug(
+            f"Updating subscriptions: participant_settings={participant_settings} profile_settings={profile_settings}"
         )
 
-    async def update_remote_participants(self, remote_participants: Mapping[str, Any]):
+        error = await self._client.update_subscriptions(
+            participant_settings=participant_settings, profile_settings=profile_settings
+        )
+        if error:
+            logger.error(f"Unable to update subscription settings: {error}")
+        return error
+
+    async def update_remote_participants(
+        self, remote_participants: Mapping[str, Any]
+    ) -> Optional[CallClientError]:
         """Update settings for remote participants.
 
         Args:
             remote_participants: Remote participant configuration settings.
+
+        Returns:
+            error: An error description or None.
         """
-        await self._client.update_remote_participants(remote_participants=remote_participants)
+        logger.debug(f"Updating remote participants: remote_participants={remote_participants}")
+
+        error = await self._client.update_remote_participants(
+            remote_participants=remote_participants
+        )
+        if error:
+            logger.error(f"Unable to update remote participants: {error}")
+        return error
 
     async def _on_active_speaker_changed(self, participant: Any):
         """Handle active speaker change events."""
@@ -2330,6 +2478,12 @@ class DailyTransport(BaseTransport):
 
     async def _on_joined(self, data):
         """Handle room joined events."""
+        if self._params.transcription_enabled:
+            # We report an error because we are starting transcription
+            # internally and if it fails we need to know.
+            error = await self.start_transcription(self._params.transcription_settings)
+            if error:
+                await self._on_error(f"Unable to start transcription: {error}")
         await self._call_event_handler("on_joined", data)
 
     async def _on_left(self):
@@ -2338,6 +2492,12 @@ class DailyTransport(BaseTransport):
 
     async def _on_before_leave(self):
         """Handle before leave room events."""
+        if self._params.transcription_enabled:
+            # We report an error because we are stopping transcription
+            # internally and if it fails we need to know.
+            error = await self.stop_transcription()
+            if error:
+                await self._on_error(f"Unable to stop transcription: {error}")
         await self._call_event_handler("on_before_leave")
 
     async def _on_error(self, error):
diff --git a/src/pipecat/utils/tracing/service_decorators.py b/src/pipecat/utils/tracing/service_decorators.py
index cf1ba912c..3935a4afc 100644
--- a/src/pipecat/utils/tracing/service_decorators.py
+++ b/src/pipecat/utils/tracing/service_decorators.py
@@ -905,7 +905,9 @@ def traced_openai_realtime(operation: str) -> Callable:
                             # Capture context messages being sent
                             if hasattr(self, "_context") and self._context:
                                 try:
-                                    messages = self._context.get_messages_for_logging()
+                                    messages = self.get_llm_adapter().get_messages_for_logging(
+                                        self._context
+                                    )
                                     if messages:
                                         operation_attrs["context_messages"] = json.dumps(messages)
                                 except Exception as e:
diff --git a/uv.lock b/uv.lock
index 8fae18c09..129491a61 100644
--- a/uv.lock
+++ b/uv.lock
@@ -1282,13 +1282,13 @@ wheels = [
 
 [[package]]
 name = "daily-python"
-version = "0.20.0"
+version = "0.21.0"
 source = { registry = "https://pypi.org/simple" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/9b/02/ce81ebf11a04cd133a5539e08f85060574711fff05a1d6ad29705f0755c1/daily_python-0.20.0-cp37-abi3-macosx_10_15_x86_64.whl", hash = "sha256:7da3f1df8cd9ef7f7fcc96ce688348dc903f62d82b6dd155a53bc64b7a74f3a7", size = 13259887, upload-time = "2025-10-16T22:14:12.262Z" },
-    { url = "https://files.pythonhosted.org/packages/4a/1e/51f06f3486c978e1184af2271e800ce6a6e8a8f95d61ee6624bae88ae9cd/daily_python-0.20.0-cp37-abi3-macosx_11_0_arm64.whl", hash = "sha256:d02fd7b8c8079ceaa550ef23db052cdf70a8ffaf8ab6a8bc1a1e97bf0b939464", size = 11642453, upload-time = "2025-10-16T22:14:14.477Z" },
-    { url = "https://files.pythonhosted.org/packages/71/c9/f767f0b479abd39330569ad61fb9db4661aae56cd74bb27c6f3483595463/daily_python-0.20.0-cp37-abi3-manylinux_2_28_aarch64.whl", hash = "sha256:a5c8718982c221dc18b41fb0692c9f8435f115f72e74994c94d3b9c6dad7c534", size = 13634216, upload-time = "2025-10-16T22:14:16.235Z" },
-    { url = "https://files.pythonhosted.org/packages/e8/10/5c6d7b000bee36c2a0587a092a34c7486d2de831fc8e44ed42b16a6bd99f/daily_python-0.20.0-cp37-abi3-manylinux_2_28_x86_64.whl", hash = "sha256:ca9132aef1bdb5be663d1894b440dab1f998ebb3f45dfc31d44effabded4bc08", size = 14282189, upload-time = "2025-10-16T22:14:18.229Z" },
+    { url = "https://files.pythonhosted.org/packages/ff/11/99590f8b7aad077f3f9b5b59d39b010aee0bd01b14dece8ae1e93d8080e7/daily_python-0.21.0-cp37-abi3-macosx_10_15_x86_64.whl", hash = "sha256:bdec96417825181559769bb2258ae688d1215949a1878336194e36fb452274a8", size = 13277066, upload-time = "2025-10-29T00:20:49.523Z" },
+    { url = "https://files.pythonhosted.org/packages/e5/db/8c57f1a1b713ba3393584ac2be32d8074d3022a2c2c17c28eb4cd2aa3629/daily_python-0.21.0-cp37-abi3-macosx_11_0_arm64.whl", hash = "sha256:18677fa1415a0dc48b891cdf2fb8fe9dabc70e1b019d5aaa3d0699ccc8d187c9", size = 11644908, upload-time = "2025-10-29T00:20:52.106Z" },
+    { url = "https://files.pythonhosted.org/packages/64/b6/b03f2f58a367d6ef4bb728715471542fdfa68afa8a177670139c3a2aadb7/daily_python-0.21.0-cp37-abi3-manylinux_2_28_aarch64.whl", hash = "sha256:97eb97352fe74227061b678e330b8befcfa4c694feb6eb2b09fe6eacec00ad6d", size = 13652356, upload-time = "2025-10-29T00:20:54.813Z" },
+    { url = "https://files.pythonhosted.org/packages/f6/76/bde65f6f8d4c1679dc6c185fa37dae9223f6ddb4b7ced728ef46504956f7/daily_python-0.21.0-cp37-abi3-manylinux_2_28_x86_64.whl", hash = "sha256:68c3e36f609fc2fce79e4d17ecf1021eadd836506db6c5125f95c682bcf3612a", size = 14304643, upload-time = "2025-10-29T00:20:57.194Z" },
 ]
 
 [[package]]
@@ -4637,7 +4637,7 @@ requires-dist = [
     { name = "azure-cognitiveservices-speech", marker = "extra == 'azure'", specifier = "~=1.42.0" },
     { name = "cartesia", marker = "extra == 'cartesia'", specifier = "~=2.0.3" },
     { name = "coremltools", marker = "extra == 'local-smart-turn'", specifier = ">=8.0" },
-    { name = "daily-python", marker = "extra == 'daily'", specifier = "~=0.20.0" },
+    { name = "daily-python", marker = "extra == 'daily'", specifier = "~=0.21.0" },
     { name = "deepgram-sdk", marker = "extra == 'deepgram'", specifier = "~=4.7.0" },
     { name = "docstring-parser", specifier = "~=0.16" },
     { name = "einops", marker = "extra == 'moondream'", specifier = "~=0.8.0" },