sketch for runner as a module

Foundational requirements.txt: add silero, websocket optional dep, re…

CHANGELOG.md

@@ -19,8 +19,40 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Added `LocalSmartTurnAnalyzerV2`, which supports local on-device inference
   with the new `smart-turn-v2` turn detection model.
 
+- Added `set_log_level` to `DailyTransport`, allowing setting the logging level
+  for Daily's internal logging system.
+
 ### Changed
 
+- Play delayed messages from `ElevenLabsTTSService` if they still belong to the
+  current context.
+
+- Dependency compatibility improvements: Relaxed version constraints for core
+  dependencies to support broader version ranges while maintaining stability:
+
+  - `aiohttp`, `Markdown`, `nltk`, `numpy`, `Pillow`, `pydantic`, `openai`,
+    `numba`: Now support up to the next major version (e.g. `numpy>=1.26.4,<3`)
+  - `pyht`: Relaxed to `>=0.1.6` to resolve `grpcio` conflicts with
+    `nvidia-riva-client`
+  - `fastapi`: Updated to support versions `>=0.115.6,<0.117.0`
+  - `torch`/`torchaudio`: Changed from exact pinning (`==2.5.0`) to compatible
+    range (`~=2.5.0`)
+  - `aws_sdk_bedrock_runtime`: Added Python 3.12+ constraint via environment
+    marker
+  - `numba`: Reduced minimum version to `0.60.0` for better compatibility
+
+- Changed `NeuphonicHttpTTSService` to use a POST based request instead of the
+  `pyneuphonic` package. This removes a package requirement, allowing Neuphonic
+  to work with more services.
+
+- Updated the `deepgram` optional dependency to 4.7.0, which downgrades the
+  `tasks cancelled error` to a debug log. This removes the log from appearing
+  in Pipecat logs upon leaving.
+
+- Upgraded the `websockets` implementation to the new asyncio implementation.
+  Along with this change, we're updating support for versions >=13.1.0 and
+  <15.0.0. All services have been update to use the asyncio implementation.
+
 - Updated `MiniMaxHttpTTSService` with a `base_url` arg where you can specify
   the Global endpoint (default) or Mainland China.
 
@@ -57,6 +89,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   it was only meant to do that when there were no "regular" (non-function-call)
   messages in the context, to ensure that inference would run properly.
 
+- Fixed an issue in `LiveKitTransport` where the `on_audio_track_subscribed` was never emitted.
+
 ## [0.0.76] - 2025-07-11
 
 ### Added

dev-requirements.txt

@@ -11,3 +11,10 @@ ruff~=0.12.1
 setuptools~=78.1.1
 setuptools_scm~=8.3.1
 python-dotenv~=1.1.1
+
+# For running examples
+uvicorn
+python-dotenv
+fastapi
+aiohttp
+aiortc

examples/deployment/pipecat-cloud-daily-pstn-server/nextjs-webhook-server/package-lock.json

@@ -8,7 +8,7 @@
       "name": "my-daily-app",
       "version": "0.1.0",
       "dependencies": {
-        "axios": "^1.10.0",
+        "axios": "^1.11.0",
         "next": "^14.0.0",
         "pino": "^8.15.0",
         "react": "^18.2.0",
@@ -1165,13 +1165,13 @@
       }
     },
     "node_modules/axios": {
-      "version": "1.10.0",
-      "resolved": "https://registry.npmjs.org/axios/-/axios-1.10.0.tgz",
-      "integrity": "sha512-/1xYAC4MP/HEG+3duIhFr4ZQXR4sQXOIe+o6sdqzeykGLx6Upp/1p8MHqhINOvGeP7xyNHe7tsiJByc4SSVUxw==",
+      "version": "1.11.0",
+      "resolved": "https://registry.npmjs.org/axios/-/axios-1.11.0.tgz",
+      "integrity": "sha512-1Lx3WLFQWm3ooKDYZD1eXmoGO9fxYQjrycfHFC8P0sCfQVXyROp0p9PFWBehewBOdCwHc+f/b8I0fMto5eSfwA==",
       "license": "MIT",
       "dependencies": {
         "follow-redirects": "^1.15.6",
-        "form-data": "^4.0.0",
+        "form-data": "^4.0.4",
         "proxy-from-env": "^1.1.0"
       }
     },

examples/deployment/pipecat-cloud-daily-pstn-server/nextjs-webhook-server/package.json

@@ -9,7 +9,7 @@
     "lint": "next lint"
   },
   "dependencies": {
-    "axios": "^1.10.0",
+    "axios": "^1.11.0",
     "next": "^14.0.0",
     "pino": "^8.15.0",
     "react": "^18.2.0",

examples/foundational/04a-transports-daily.py

@@ -20,7 +20,7 @@ from pipecat.pipeline.task import PipelineParams, PipelineTask
 from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
 from pipecat.services.cartesia.tts import CartesiaTTSService
 from pipecat.services.openai.llm import OpenAILLMService
-from pipecat.transports.services.daily import DailyParams, DailyTransport
+from pipecat.transports.services.daily import DailyLogLevel, DailyParams, DailyTransport
 
 load_dotenv(override=True)
 
@@ -43,6 +43,7 @@ async def main():
                 vad_analyzer=SileroVADAnalyzer(),
             ),
         )
+        transport.set_log_level(DailyLogLevel.Info)
 
         tts = CartesiaTTSService(
             api_key=os.getenv("CARTESIA_API_KEY"),

examples/foundational/07v-interruptible-neuphonic-http.py

@@ -7,6 +7,7 @@
 import argparse
 import os
 
+import aiohttp
 from dotenv import load_dotenv
 from loguru import logger
 
@@ -50,60 +51,63 @@ transport_params = {
 async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_sigint: bool):
     logger.info(f"Starting bot")
 
-    stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
+    # Create an HTTP session
+    async with aiohttp.ClientSession() as session:
+        stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
 
-    tts = NeuphonicHttpTTSService(
-        api_key=os.getenv("NEUPHONIC_API_KEY"),
-        voice_id="fc854436-2dac-4d21-aa69-ae17b54e98eb",  # Emily
-    )
+        tts = NeuphonicHttpTTSService(
+            api_key=os.getenv("NEUPHONIC_API_KEY"),
+            voice_id="fc854436-2dac-4d21-aa69-ae17b54e98eb",  # Emily
+            aiohttp_session=session,
+        )
 
-    llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
+        llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
 
-    messages = [
-        {
-            "role": "system",
-            "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.",
-        },
-    ]
-
-    context = OpenAILLMContext(messages)
-    context_aggregator = llm.create_context_aggregator(context)
-
-    pipeline = Pipeline(
-        [
-            transport.input(),  # Transport user input
-            stt,
-            context_aggregator.user(),  # User responses
-            llm,  # LLM
-            tts,  # TTS
-            transport.output(),  # Transport bot output
-            context_aggregator.assistant(),  # Assistant spoken responses
+        messages = [
+            {
+                "role": "system",
+                "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.",
+            },
         ]
-    )
 
-    task = PipelineTask(
-        pipeline,
-        params=PipelineParams(
-            enable_metrics=True,
-            enable_usage_metrics=True,
-        ),
-    )
+        context = OpenAILLMContext(messages)
+        context_aggregator = llm.create_context_aggregator(context)
 
-    @transport.event_handler("on_client_connected")
-    async def on_client_connected(transport, client):
-        logger.info(f"Client connected")
-        # Kick off the conversation.
-        messages.append({"role": "system", "content": "Please introduce yourself to the user."})
-        await task.queue_frames([context_aggregator.user().get_context_frame()])
+        pipeline = Pipeline(
+            [
+                transport.input(),  # Transport user input
+                stt,
+                context_aggregator.user(),  # User responses
+                llm,  # LLM
+                tts,  # TTS
+                transport.output(),  # Transport bot output
+                context_aggregator.assistant(),  # Assistant spoken responses
+            ]
+        )
 
-    @transport.event_handler("on_client_disconnected")
-    async def on_client_disconnected(transport, client):
-        logger.info(f"Client disconnected")
-        await task.cancel()
+        task = PipelineTask(
+            pipeline,
+            params=PipelineParams(
+                enable_metrics=True,
+                enable_usage_metrics=True,
+            ),
+        )
 
-    runner = PipelineRunner(handle_sigint=handle_sigint)
+        @transport.event_handler("on_client_connected")
+        async def on_client_connected(transport, client):
+            logger.info(f"Client connected")
+            # Kick off the conversation.
+            messages.append({"role": "system", "content": "Please introduce yourself to the user."})
+            await task.queue_frames([context_aggregator.user().get_context_frame()])
 
-    await runner.run(task)
+        @transport.event_handler("on_client_disconnected")
+        async def on_client_disconnected(transport, client):
+            logger.info(f"Client disconnected")
+            await task.cancel()
+
+        runner = PipelineRunner(handle_sigint=handle_sigint)
+
+        await runner.run(task)
 
 
 if __name__ == "__main__":

examples/foundational/requirements.txt

@@ -1,5 +0,0 @@
-fastapi
-uvicorn
-python-dotenv
-pipecat-ai[webrtc,daily,deepgram,cartesia]
-pipecat-ai-small-webrtc-prebuilt

examples/word-wrangler-gemini-live/README.md

@@ -295,6 +295,22 @@ This project uses TypeScript, React, and Next.js, making it a perfect fit for [V
 
 Again, we'll use Pipecat Cloud. Follow the steps from above. The only difference will be the secrets required; in addition to a GOOGLE_API_KEY, you'll need `GOOGLE_APPLICATION_CREDENTIALS` in the format of a .json file with your [Google Cloud service account](https://console.cloud.google.com/iam-admin/serviceaccounts) information.
 
+You'll need to modify the Dockerfile so that the credentials.json and word_list.py are accessible. This Dockerfile will work:
+
+```Dockerfile
+FROM dailyco/pipecat-base:latest
+
+COPY ./requirements.txt requirements.txt
+
+RUN pip install --no-cache-dir --upgrade -r requirements.txt
+
+COPY ./word_list.py word_list.py
+COPY ./credentials.json credentials.json
+COPY ./bot_phone_twilio.py bot.py
+```
+
+Note: Your `credentials.json` file should have your Google service account credentials.
+
 #### Buy and Configure a Twilio Number
 
 Check out the [Twilio Websocket Telephony guide](https://docs.pipecat.daily.co/pipecat-in-production/telephony/twilio-mediastreams) for a step-by-step walkthrough on how to purchase a phone number, configure your TwiML, and make or receive calls.

pyproject.toml

@@ -20,22 +20,22 @@ classifiers = [
     "Topic :: Scientific/Engineering :: Artificial Intelligence"
 ]
 dependencies = [
-    "aiohttp~=3.11.12",
+    "aiohttp>=3.11.12,<4",
     "audioop-lts~=0.2.1; python_version>='3.13'",
     "docstring_parser~=0.16",
     "loguru~=0.7.3",
-    "Markdown~=3.7",
-    "nltk>=3.9.1",
-    "numpy>=1.26.4",
-    "Pillow~=11.1.0",
+    "Markdown>=3.7,<4",
+    "nltk>=3.9.1,<4",
+    "numpy>=1.26.4,<3",
+    "Pillow>=11.1.0,<12",
     "protobuf~=5.29.3",
-    "pydantic~=2.10.6",
+    "pydantic>=2.10.6,<3",
     "pyloudnorm~=0.1.1",
     "resampy~=0.4.3",
     "soxr~=0.5.0",
-    "openai~=1.74.0",
+    "openai>=1.74.0,<2",
     # Explicit dependency pins for Python 3.11+ compatibility
-    "numba>=0.61.0",
+    "numba>=0.60.0,<1",
 ]
 
 [project.urls]
@@ -44,21 +44,21 @@ Website = "https://pipecat.ai"
 
 [project.optional-dependencies]
 anthropic = [ "anthropic~=0.49.0" ]
-assemblyai = [ "websockets~=13.1" ]
-aws = [ "aioboto3~=15.0.0", "websockets~=13.1" ]
-aws-nova-sonic = [ "aws_sdk_bedrock_runtime~=0.0.2" ]
+assemblyai = [ "websockets>=13.1,<15.0" ]
+aws = [ "aioboto3~=15.0.0", "websockets>=13.1,<15.0" ]
+aws-nova-sonic = [ "aws_sdk_bedrock_runtime~=0.0.2; python_version>='3.12'" ]
 azure = [ "azure-cognitiveservices-speech~=1.42.0"]
-cartesia = [ "cartesia~=2.0.3", "websockets~=13.1" ]
+cartesia = [ "cartesia~=2.0.3", "websockets>=13.1,<15.0" ]
 cerebras = []
 deepseek = []
 daily = [ "daily-python~=0.19.4" ]
-deepgram = [ "deepgram-sdk~=4.1.0" ]
-elevenlabs = [ "websockets~=13.1" ]
+deepgram = [ "deepgram-sdk~=4.7.0" ]
+elevenlabs = [ "websockets>=13.1,<15.0" ]
 fal = [ "fal-client~=0.5.9" ]
 fireworks = []
-fish = [ "ormsgpack~=1.7.0", "websockets~=13.1" ]
-gladia = [ "websockets~=13.1" ]
-google = [ "google-cloud-speech~=2.32.0", "google-cloud-texttospeech~=2.26.0", "google-genai~=1.24.0", "websockets~=13.1" ]
+fish = [ "ormsgpack~=1.7.0", "websockets>=13.1,<15.0" ]
+gladia = [ "websockets>=13.1,<15.0" ]
+google = [ "google-cloud-speech~=2.32.0", "google-cloud-texttospeech~=2.26.0", "google-genai~=1.24.0", "websockets>=13.1,<15.0" ]
 grok = []
 groq = [ "groq~=0.23.0" ]
 gstreamer = [ "pygobject~=3.50.0" ]
@@ -66,30 +66,30 @@ krisp = [ "pipecat-ai-krisp~=0.4.0" ]
 koala = [ "pvkoala~=2.0.3" ]
 langchain = [ "langchain~=0.3.20", "langchain-community~=0.3.20", "langchain-openai~=0.3.9" ]
 livekit = [ "livekit~=0.22.0", "livekit-api~=0.8.2", "tenacity>=8.2.3,<10.0.0" ]
-lmnt = [ "websockets~=13.1" ]
+lmnt = [ "websockets>=13.1,<15.0" ]
 local = [ "pyaudio~=0.2.14" ]
 mcp = [ "mcp[cli]~=1.9.4" ]
 mem0 = [ "mem0ai~=0.1.94" ]
 mlx-whisper = [ "mlx-whisper~=0.4.2" ]
 moondream = [ "einops~=0.8.0", "timm~=1.0.13", "transformers>=4.48.0" ]
 nim = []
-neuphonic = [ "pyneuphonic~=1.5.13", "websockets~=13.1" ]
+neuphonic = [ "websockets>=13.1,<15.0" ]
 noisereduce = [ "noisereduce~=3.0.3" ]
-openai = [ "websockets~=13.1" ]
+openai = [ "websockets>=13.1,<15.0" ]
 openpipe = [ "openpipe~=4.50.0" ]
 openrouter = []
 perplexity = []
-playht = [ "pyht~=0.1.12", "websockets~=13.1" ]
+playht = [ "pyht>=0.1.6", "websockets>=13.1,<15.0" ]
 qwen = []
-rime = [ "websockets~=13.1" ]
+rime = [ "websockets>=13.1,<15.0" ]
 riva = [ "nvidia-riva-client~=2.21.1" ]
 sambanova = []
 sentry = [ "sentry-sdk~=2.23.1" ]
-local-smart-turn = [ "coremltools>=8.0", "transformers", "torch==2.5.0", "torchaudio==2.5.0" ]
+local-smart-turn = [ "coremltools>=8.0", "transformers", "torch~=2.5.0", "torchaudio~=2.5.0" ]
 remote-smart-turn = []
 silero = [ "onnxruntime~=1.20.1" ]
 simli = [ "simli-ai~=0.1.10"]
-soniox = [ "websockets~=13.1" ]
+soniox = [ "websockets>=13.1,<15.0" ]
 soundfile = [ "soundfile~=0.13.0" ]
 speechmatics = [ "speechmatics-rt>=0.3.1" ]
 tavus=[]
@@ -97,7 +97,7 @@ together = []
 tracing = [ "opentelemetry-sdk>=1.33.0", "opentelemetry-api>=1.33.0", "opentelemetry-instrumentation>=0.54b0" ]
 ultravox = [ "transformers>=4.48.0", "vllm~=0.7.3" ]
 webrtc = [ "aiortc~=1.11.0", "opencv-python~=4.11.0.86" ]
-websocket = [ "websockets~=13.1", "fastapi~=0.115.6" ]
+websocket = [ "websockets>=13.1,<15.0", "fastapi>=0.115.6,<0.117.0" ]
 whisper = [ "faster-whisper~=1.1.1" ]
 
 [tool.setuptools.packages.find]
@@ -152,3 +152,6 @@ convention = "google"
 command_line = "--module pytest"
 source = ["src"]
 omit = ["*/tests/*"]
+
+[project.scripts]
+pipecat = "pipecat.__main__:main"

src/pipecat/__main__.py

@@ -0,0 +1,101 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import argparse
+import importlib.util
+import sys
+from pathlib import Path
+from typing import Any, Callable
+
+from loguru import logger
+
+
+def load_bot_module(file_path: str, function_name: str = "run_example"):
+    """Load a bot module from a Python file and return the specified function.
+
+    Args:
+        file_path: Path to the Python file containing the bot
+        function_name: Name of the function to load (default: run_example)
+
+    Returns:
+        The callable function from the module
+
+    Raises:
+        SystemExit: If the file doesn't exist, isn't a Python file, or the function isn't found
+    """
+    logger.info(f"Loading bot module from: {file_path}")
+    logger.info(f"Looking for function: {function_name}")
+
+    file_path_obj = Path(file_path)
+    if not file_path_obj.exists():
+        print(f"Error: File '{file_path}' not found", file=sys.stderr)
+        sys.exit(1)
+
+    if not file_path_obj.suffix == ".py":
+        print(f"Error: File '{file_path}' is not a Python file", file=sys.stderr)
+        sys.exit(1)
+
+    # Import the module
+    try:
+        logger.info(f"Importing module from: {file_path}")
+        spec = importlib.util.spec_from_file_location("bot_module", file_path_obj)
+        if spec is None or spec.loader is None:
+            print(f"Error: Could not load module from '{file_path}'", file=sys.stderr)
+            sys.exit(1)
+
+        module = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(module)
+        logger.info(f"Successfully imported module: {module.__name__}")
+    except Exception as e:
+        print(f"Error importing module from '{file_path}': {e}", file=sys.stderr)
+        sys.exit(1)
+
+    # Find the function to run
+    if not hasattr(module, function_name):
+        print(f"Error: Function '{function_name}' not found in '{file_path}'", file=sys.stderr)
+        print(
+            f"Available functions: {[name for name in dir(module) if not name.startswith('_')]}", file=sys.stderr)
+        sys.exit(1)
+
+    run_example = getattr(module, function_name)
+    if not callable(run_example):
+        print(f"Error: '{function_name}' is not a callable function", file=sys.stderr)
+        sys.exit(1)
+
+    logger.info(f"Successfully loaded function: {function_name}")
+    return run_example
+
+
+def main():
+    """Main entry point for the pipecat command line tool.
+
+    This function is called by the entry point script and handles argument parsing
+    and module loading before calling the actual main execution logic.
+    """
+    # Set up argument parser for our specific arguments
+    parser = argparse.ArgumentParser(description="Run a Pipecat bot from a Python file")
+    parser.add_argument("file", help="Python file containing the bot to run")
+    parser.add_argument("--function", "-f", default="run_example",
+                        help="Function name to run (default: run_example)")
+
+    # Parse our arguments first
+    args, remaining_args = parser.parse_known_args()
+
+    # Load the bot module and get the function
+    run_example = load_bot_module(args.file, args.function)
+
+    # Set sys.argv to the remaining arguments for the run_main function
+    sys.argv = [sys.argv[0]] + remaining_args
+
+    # Import run_main only when we need it
+    from pipecat.examples.run import main as run_main
+
+    # Call the main function from pipecat.examples.run
+    run_main(run_example)
+
+
+if __name__ == "__main__":
+    main()

src/pipecat/adapters/base_llm_adapter.py

@@ -11,45 +11,21 @@ adapters that handle tool format conversion and standardization.
 """
 
 from abc import ABC, abstractmethod
-from typing import Any, Generic, List, TypedDict, TypeVar, Union, cast
+from typing import Any, List, Union, cast
 
 from loguru import logger
 
 from pipecat.adapters.schemas.tools_schema import ToolsSchema
-from pipecat.processors.aggregators.llm_context import LLMContext
-
-# Should be a TypedDict
-TLLMInvocationParams = TypeVar("TLLMInvocationParams", bound=dict[str, Any])
 
 
-# TODO: fix everywhere we subclass BaseLLMAdapter...
-class BaseLLMAdapter(ABC, Generic[TLLMInvocationParams]):
+class BaseLLMAdapter(ABC):
     """Abstract base class for LLM provider adapters.
 
-    Provides a standard interface for converting to provider-specific formats.
-
-    Handles:
-    - Extracting provider-specific parameters for LLM invocation from a
-      universal LLM context
-    - Converting standardized tools schema to provider-specific tool formats.
-    - Extracting messages from the LLM context for the purposes of logging
-      about the specific provider.
-
-    Subclasses must implement provider-specific conversion logic.
+    Provides a standard interface for converting between Pipecat's standardized
+    tool schemas and provider-specific tool formats. Subclasses must implement
+    provider-specific conversion logic.
     """
 
-    @abstractmethod
-    def get_llm_invocation_params(self, context: LLMContext) -> TLLMInvocationParams:
-        """Get provider-specific LLM invocation parameters from a universal LLM context.
-
-        Args:
-            context: The LLM context containing messages, tools, etc.
-
-        Returns:
-            Provider-specific parameters for invoking the LLM.
-        """
-        pass
-
     @abstractmethod
     def to_provider_tools_format(self, tools_schema: ToolsSchema) -> List[Any]:
         """Convert tools schema to the provider's specific format.
@@ -62,20 +38,6 @@ class BaseLLMAdapter(ABC, Generic[TLLMInvocationParams]):
         """
         pass
 
-    @abstractmethod
-    def get_messages_for_logging(self, context: LLMContext) -> List[dict[str, Any]]:
-        """Get messages from the LLM context in a format ready for logging about this provider.
-
-        Args:
-            context: The LLM context containing messages.
-
-        Returns:
-            List of messages in a format ready for logging about this
-            provider.
-        """
-        pass
-
-    # TODO: should this also be able to return NotGiven?
     def from_standard_tools(self, tools: Any) -> List[Any]:
         """Convert tools from standard format to provider format.
 
@@ -92,38 +54,4 @@ class BaseLLMAdapter(ABC, Generic[TLLMInvocationParams]):
         # Fallback to return the same tools in case they are not in a standard format
         return tools
 
-    def create_wav_header(self, sample_rate, num_channels, bits_per_sample, data_size):
-        """Create a WAV file header for audio data.
-
-        Args:
-            sample_rate: Audio sample rate in Hz.
-            num_channels: Number of audio channels.
-            bits_per_sample: Bits per audio sample.
-            data_size: Size of audio data in bytes.
-
-        Returns:
-            WAV header as a bytearray.
-        """
-        # RIFF chunk descriptor
-        header = bytearray()
-        header.extend(b"RIFF")  # ChunkID
-        header.extend((data_size + 36).to_bytes(4, "little"))  # ChunkSize: total size - 8
-        header.extend(b"WAVE")  # Format
-        # "fmt " sub-chunk
-        header.extend(b"fmt ")  # Subchunk1ID
-        header.extend((16).to_bytes(4, "little"))  # Subchunk1Size (16 for PCM)
-        header.extend((1).to_bytes(2, "little"))  # AudioFormat (1 for PCM)
-        header.extend(num_channels.to_bytes(2, "little"))  # NumChannels
-        header.extend(sample_rate.to_bytes(4, "little"))  # SampleRate
-        # Calculate byte rate and block align
-        byte_rate = sample_rate * num_channels * (bits_per_sample // 8)
-        block_align = num_channels * (bits_per_sample // 8)
-        header.extend(byte_rate.to_bytes(4, "little"))  # ByteRate
-        header.extend(block_align.to_bytes(2, "little"))  # BlockAlign
-        header.extend(bits_per_sample.to_bytes(2, "little"))  # BitsPerSample
-        # "data" sub-chunk
-        header.extend(b"data")  # Subchunk2ID
-        header.extend(data_size.to_bytes(4, "little"))  # Subchunk2Size
-        return header
-
     # TODO: we can move the logic to also handle the Messages here

src/pipecat/adapters/services/gemini_adapter.py

@@ -6,68 +6,21 @@
 
 """Gemini LLM adapter for Pipecat."""
 
-import base64
-import json
-from dataclasses import dataclass
-from typing import Any, Dict, List, Optional, TypedDict, Union
-
-from loguru import logger
+from typing import Any, Dict, List, Union
 
 from pipecat.adapters.base_llm_adapter import BaseLLMAdapter
 from pipecat.adapters.schemas.tools_schema import AdapterType, ToolsSchema
-from pipecat.processors.aggregators.llm_context import LLMContext, LLMContextMessage
-
-try:
-    from google.genai.types import (
-        Blob,
-        Content,
-        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]`.")
-    raise Exception(f"Missing module: {e}")
 
 
-class GeminiLLMInvocationParams(TypedDict):
-    """Context-based parameters for invoking Gemini LLM."""
+class GeminiLLMAdapter(BaseLLMAdapter):
+    """LLM adapter for Google's Gemini service.
 
-    system_instruction: Optional[str]
-    messages: List[Content]
-    tools: List[Any]
-
-
-class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
-    """Gemini-specific adapter for Pipecat.
-
-    Handles:
-    - Extracting parameters for Gemini's API from a universal
-      LLM context
-    - Converting Pipecat's standardized tools schema to Gemini's function-calling format.
-    - Extracting and sanitizing messages from the LLM context for logging with Gemini.
+    Provides tool schema conversion functionality to transform standard tool
+    definitions into Gemini's specific function-calling format for use with
+    Gemini LLM models.
     """
 
-    def get_llm_invocation_params(self, context: LLMContext) -> GeminiLLMInvocationParams:
-        """Get Gemini-specific LLM invocation parameters from a universal LLM context.
-
-        Args:
-            context: The LLM context containing messages, tools, etc.
-
-        Returns:
-            Dictionary of parameters for Gemini's API.
-        """
-        # TODO: remove when done testing
-        print(f"[pk] {self}: Getting LLM invocation params...")
-        messages = self._from_standard_messages(context.messages)
-        return {
-            "system_instruction": messages.system_instruction,
-            "messages": messages.messages,
-            "tools": self.from_standard_tools(context.tools),
-        }
-
-    def to_provider_tools_format(self, tools_schema: ToolsSchema) -> List[Any]:
+    def to_provider_tools_format(self, tools_schema: ToolsSchema) -> List[Dict[str, Any]]:
         """Convert tool schemas to Gemini's function-calling format.
 
         Args:
@@ -86,217 +39,3 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
             custom_gemini_tools = tools_schema.custom_tools.get(AdapterType.GEMINI, [])
 
         return formatted_standard_tools + custom_gemini_tools
-
-    def get_messages_for_logging(self, context: LLMContext) -> List[dict[str, Any]]:
-        """Get messages from the LLM context in a format ready for logging about Gemini.
-
-        Removes or truncates sensitive data like image content for safe logging.
-
-        Args:
-            context: The LLM context containing messages.
-
-        Returns:
-            List of messages in a format ready for logging about Gemini.
-        """
-        # Get messages in Gemini's format
-        messages = self._from_standard_messages(context.messages).messages
-
-        # Sanitize messages for logging
-        messages_for_logging = []
-        for message in messages:
-            obj = message.to_json_dict()
-            try:
-                if "parts" in obj:
-                    for part in obj["parts"]:
-                        if "inline_data" in part:
-                            part["inline_data"]["data"] = "..."
-            except Exception as e:
-                logger.debug(f"Error: {e}")
-            messages_for_logging.append(obj)
-        return messages_for_logging
-
-    @dataclass
-    class ConvertedMessages:
-        """Container for converted messages.
-
-        Holds the converted messages in a format suitable for Gemini's API.
-        """
-
-        messages: List[Content]
-        system_instruction: Optional[str] = None
-
-    def _from_standard_messages(
-        self, standard_messages: List[LLMContextMessage]
-    ) -> ConvertedMessages:
-        """Restructures messages to ensure proper Google format and message ordering.
-
-        This method handles conversion of OpenAI-formatted messages to Google format,
-        with special handling for function calls, function responses, and system messages.
-        System messages are added back to the context as user messages when needed.
-
-        The final message order is preserved as:
-        1. Function calls (from model)
-        2. Function responses (from user)
-        3. Text messages (converted from system messages)
-
-        Note:
-            System messages are only added back when there are no regular text
-            messages in the context, ensuring proper conversation continuity
-            after function calls.
-        """
-        system_instruction = None
-        messages = []
-
-        # Process each message, preserving Google-formatted messages and converting others
-        for message in standard_messages:
-            if isinstance(message, Content):
-                # Keep existing Google-formatted messages (e.g., function calls/responses)
-                # TODO: this branch is probably not needed anymore, since LLMContext contains a universal format
-                messages.append(message)
-                continue
-
-            # Convert standard format to Google format
-            converted = self._from_standard_message(message)
-            if isinstance(converted, Content):
-                # Regular (non-system) message
-                messages.append(converted)
-            else:
-                # System instruction
-                system_instruction = converted
-
-        # Check if we only have function-related messages (no regular text)
-        has_regular_messages = any(
-            len(msg.parts) == 1
-            and getattr(msg.parts[0], "text", None)
-            and not getattr(msg.parts[0], "function_call", None)
-            and not getattr(msg.parts[0], "function_response", None)
-            for msg in messages
-        )
-
-        # Add system instruction back as a user message if we only have function messages
-        if system_instruction and not has_regular_messages:
-            messages.append(Content(role="user", parts=[Part(text=system_instruction)]))
-
-        # Remove any empty messages
-        messages = [m for m in messages if m.parts]
-
-        return self.ConvertedMessages(messages=messages, system_instruction=system_instruction)
-
-    def _from_standard_message(self, message: LLMContextMessage) -> Content | str:
-        """Convert standard format message to Google Content object.
-
-        Handles conversion of text, images, and function calls to Google's
-        format.
-        System instructions are returned as a plain string.
-
-        Args:
-            message: Message in standard format.
-
-        Returns:
-            Content object with role and parts, or a plain string for system
-            messages.
-
-        Examples:
-            Standard text message::
-
-                {
-                    "role": "user",
-                    "content": "Hello there"
-                }
-
-            Converts to Google Content with::
-
-                Content(
-                    role="user",
-                    parts=[Part(text="Hello there")]
-                )
-
-            Standard function call message::
-
-                {
-                    "role": "assistant",
-                    "tool_calls": [
-                        {
-                            "function": {
-                                "name": "search",
-                                "arguments": '{"query": "test"}'
-                            }
-                        }
-                    ]
-                }
-
-            Converts to Google Content with::
-
-                Content(
-                    role="model",
-                    parts=[Part(function_call=FunctionCall(name="search", args={"query": "test"}))]
-                )
-        """
-        role = message["role"]
-        content = message.get("content", [])
-        if role == "system":
-            # System instructions are returned as plain text
-            # TODO: here we've always assumed that system instructions are plain text...is that a safe assumption?
-            return content
-        elif role == "assistant":
-            role = "model"
-
-        parts = []
-        if message.get("tool_calls"):
-            for tc in message["tool_calls"]:
-                parts.append(
-                    Part(
-                        function_call=FunctionCall(
-                            name=tc["function"]["name"],
-                            args=json.loads(tc["function"]["arguments"]),
-                        )
-                    )
-                )
-        elif role == "tool":
-            role = "model"
-            parts.append(
-                Part(
-                    function_response=FunctionResponse(
-                        name="tool_call_result",  # seems to work to hard-code the same name every time
-                        response=json.loads(message["content"]),
-                    )
-                )
-            )
-        elif isinstance(content, str):
-            parts.append(Part(text=content))
-        elif isinstance(content, list):
-            for c in content:
-                if c["type"] == "text":
-                    parts.append(Part(text=c["text"]))
-                elif c["type"] == "image_url":
-                    parts.append(
-                        Part(
-                            inline_data=Blob(
-                                mime_type="image/jpeg",
-                                data=base64.b64decode(c["image_url"]["url"].split(",")[1]),
-                            )
-                        )
-                    )
-                elif c["type"] == "input_audio":
-                    input_audio = c["input_audio"]
-                    parts.append(
-                        Part(
-                            inline_data=Blob(
-                                mime_type="audio/wav",
-                                data=(
-                                    bytes(
-                                        self.create_wav_header(
-                                            input_audio["sample_rate"],
-                                            input_audio["num_channels"],
-                                            16,
-                                            len(input_audio["data"]),
-                                        )
-                                        + input_audio["data"]
-                                    )
-                                ),
-                            )
-                        )
-                    )
-
-        message = Content(role=role, parts=parts)
-        return message

src/pipecat/adapters/services/open_ai_adapter.py

@@ -6,62 +6,22 @@
 
 """OpenAI LLM adapter for Pipecat."""
 
-import copy
-import json
-from typing import Any, List, TypedDict
+from typing import List
 
-from openai._types import NOT_GIVEN as OPEN_AI_NOT_GIVEN
-from openai._types import NotGiven as OpenAINotGiven
-from openai.types.chat import (
-    ChatCompletionMessageParam,
-    ChatCompletionToolChoiceOptionParam,
-    ChatCompletionToolParam,
-)
+from openai.types.chat import ChatCompletionToolParam
 
 from pipecat.adapters.base_llm_adapter import BaseLLMAdapter
 from pipecat.adapters.schemas.tools_schema import ToolsSchema
-from pipecat.processors.aggregators.llm_context import (
-    LLMContext,
-    LLMContextMessage,
-    LLMContextToolChoice,
-    NotGiven,
-)
 
 
-class OpenAILLMInvocationParams(TypedDict):
-    """Context-based parameters for invoking OpenAI ChatCompletion API."""
+class OpenAILLMAdapter(BaseLLMAdapter):
+    """Adapter for converting tool schemas to OpenAI's format.
 
-    messages: List[ChatCompletionMessageParam]
-    tools: List[ChatCompletionToolParam] | OpenAINotGiven
-    tool_choice: ChatCompletionToolChoiceOptionParam | OpenAINotGiven
-
-
-class OpenAILLMAdapter(BaseLLMAdapter[OpenAILLMInvocationParams]):
-    """OpenAI-specific adapter for Pipecat.
-
-    Handles:
-    - Extracting parameters for OpenAI's ChatCompletion API from a universal
-      LLM context
-    - Converting Pipecat's standardized tools schema to OpenAI's function-calling format.
-    - Extracting and sanitizing messages from the LLM context for logging with OpenAI.
+    Provides conversion utilities for transforming Pipecat's standard tool
+    schemas into the format expected by OpenAI's ChatCompletion API for
+    function calling capabilities.
     """
 
-    def get_llm_invocation_params(self, context: LLMContext) -> OpenAILLMInvocationParams:
-        """Get OpenAI-specific LLM invocation parameters from a universal LLM context.
-
-        Args:
-            context: The LLM context containing messages, tools, etc.
-
-        Returns:
-            Dictionary of parameters for OpenAI's ChatCompletion API.
-        """
-        return {
-            "messages": self._from_standard_messages(context.messages),
-            # TODO: doesn't seem right that we may or may not need to convert tools here; they should already be guaranteed to exist in a universal format in the LLMContext, right?
-            "tools": self.from_standard_tools(context.tools),
-            "tool_choice": context.tool_choice,
-        }
-
     def to_provider_tools_format(self, tools_schema: ToolsSchema) -> List[ChatCompletionToolParam]:
         """Convert function schemas to OpenAI's function-calling format.
 
@@ -77,40 +37,3 @@ class OpenAILLMAdapter(BaseLLMAdapter[OpenAILLMInvocationParams]):
             ChatCompletionToolParam(type="function", function=func.to_default_dict())
             for func in functions_schema
         ]
-
-    def get_messages_for_logging(self, context: LLMContext) -> List[dict[str, Any]]:
-        """Get messages from the LLM context in a format ready for logging about OpenAI.
-
-        Removes or truncates sensitive data like image content for safe logging.
-
-        Args:
-            context: The LLM context containing messages.
-
-        Returns:
-            List of messages in a format ready for logging about OpenAI.
-        """
-        msgs = []
-        for message in context.messages:
-            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 "mime_type" in msg and msg["mime_type"].startswith("image/"):
-                msg["data"] = "..."
-            msgs.append(msg)
-        return json.dumps(msgs, ensure_ascii=False)
-
-    def _from_standard_messages(
-        self, messages: List[LLMContextMessage]
-    ) -> List[ChatCompletionMessageParam]:
-        # Just a pass-through: messages is already the right type
-        return messages
-
-    def _from_standard_tool_choice(
-        self, tool_choice: LLMContextToolChoice | NotGiven
-    ) -> ChatCompletionToolChoiceOptionParam | OpenAINotGiven:
-        # Just a pass-through: tool_choice is already the right type
-        return tool_choice

src/pipecat/frames/frames.py

@@ -378,7 +378,7 @@ class TranslationFrame(TextFrame):
 
 
 @dataclass
-class LLMContextAssistantTimestampFrame(DataFrame):
+class OpenAILLMContextAssistantTimestampFrame(DataFrame):
     """Timestamp information for assistant messages in LLM context.
 
     Parameters:

src/pipecat/processors/aggregators/llm_context.py

@@ -1,211 +0,0 @@
-#
-# Copyright (c) 2025, Daily
-#
-# SPDX-License-Identifier: BSD 2-Clause License
-#
-
-"""Universal LLM context management for LLM services in Pipecat.
-
-Context contents are represented in a generic format (extended from OpenAI)
-that supports a union of known Pipecat LLM service functionality.
-
-Whenever an LLM service needs to access context, it does a just-in-time
-translation from this universal context into whatever format it needs, using a
-service-specific adapter.
-"""
-
-import base64
-import io
-from dataclasses import dataclass
-from typing import Any, List, Optional
-
-from openai._types import NOT_GIVEN as OPEN_AI_NOT_GIVEN
-from openai._types import NotGiven as OpenAINotGiven
-from openai.types.chat import (
-    ChatCompletionMessageParam,
-    ChatCompletionToolChoiceOptionParam,
-    ChatCompletionToolParam,
-)
-from PIL import Image
-
-from pipecat.adapters.schemas.tools_schema import ToolsSchema
-from pipecat.frames.frames import AudioRawFrame, Frame
-
-# "Re-export" types from OpenAI that we're using as universal context types.
-# NOTE: this is just for convenience, for now. As soon as the universal types
-# diverge from OpenAI's, we should ditch this. In fact, audio frames already
-# diverge from OpenAI's standard format...we really ought to do this.
-LLMContextMessage = ChatCompletionMessageParam
-LLMContextTool = ChatCompletionToolParam
-LLMContextToolChoice = ChatCompletionToolChoiceOptionParam
-NOT_GIVEN = OPEN_AI_NOT_GIVEN
-NotGiven = OpenAINotGiven
-
-
-class LLMContext:
-    """Manages conversation context for LLM interactions.
-
-    Handles message history, tool definitions, tool choices, and multimedia
-    content for LLM conversations. Provides methods for message manipulation,
-    and content formatting.
-    """
-
-    def __init__(
-        self,
-        messages: Optional[List[LLMContextMessage]] = None,
-        tools: List[LLMContextTool] | NotGiven | ToolsSchema = NOT_GIVEN,
-        tool_choice: LLMContextToolChoice | NotGiven = NOT_GIVEN,
-    ):
-        """Initialize the LLM context.
-
-        Args:
-            messages: Initial list of conversation messages.
-            tools: Available tools for the LLM to use.
-            tool_choice: Tool selection strategy for the LLM.
-        """
-        self._messages: List[LLMContextMessage] = messages if messages else []
-        self._tools: List[LLMContextTool] | NotGiven | ToolsSchema = tools
-        self._tool_choice: LLMContextToolChoice | NotGiven = tool_choice
-
-    @property
-    def messages(self) -> List[LLMContextMessage]:
-        """Get the current messages list.
-
-        Returns:
-            List of conversation messages.
-        """
-        return self._messages
-
-    @property
-    def tools(self) -> List[LLMContextTool] | NotGiven | List[Any]:
-        """Get the tools list.
-
-        Returns:
-            Tools list.
-        """
-        return self._tools
-
-    @property
-    def tool_choice(self) -> LLMContextToolChoice | NotGiven:
-        """Get the current tool choice setting.
-
-        Returns:
-            The tool choice configuration.
-        """
-        return self._tool_choice
-
-    def add_message(self, message: LLMContextMessage):
-        """Add a single message to the context.
-
-        Args:
-            message: The message to add to the conversation history.
-        """
-        self._messages.append(message)
-
-    def add_messages(self, messages: List[LLMContextMessage]):
-        """Add multiple messages to the context.
-
-        Args:
-            messages: List of messages to add to the conversation history.
-        """
-        self._messages.extend(messages)
-
-    def set_messages(self, messages: List[LLMContextMessage]):
-        """Replace all messages in the context.
-
-        Args:
-            messages: New list of messages to replace the current history.
-        """
-        self._messages[:] = messages
-
-    def set_tools(self, tools: List[LLMContextTool] | NotGiven | ToolsSchema = NOT_GIVEN):
-        """Set the available tools for the LLM.
-
-        Args:
-            tools: List of tools available to the LLM, a ToolsSchema, or NOT_GIVEN to disable tools.
-        """
-        # TODO: convert empty ToolsSchema to NOT_GIVEN if needed?
-        # TODO: maybe someday also convert provider-specific tools to ToolsSchema so it's always in a provider-neutral format here? See open_ai_adapter.py for related comment. Pipecat Flows is currently converting provider-specific tools to ToolsSchema...
-        if isinstance(tools, list) and len(tools) == 0:
-            tools = NOT_GIVEN
-        self._tools = tools
-
-    def set_tool_choice(self, tool_choice: LLMContextToolChoice | NotGiven):
-        """Set the tool choice configuration.
-
-        Args:
-            tool_choice: Tool selection strategy for the LLM.
-        """
-        self._tool_choice = tool_choice
-
-    def add_image_frame_message(
-        self, *, format: str, size: tuple[int, int], image: bytes, text: str = None
-    ):
-        """Add a message containing an image frame.
-
-        Args:
-            format: Image format (e.g., 'RGB', 'RGBA').
-            size: Image dimensions as (width, height) tuple.
-            image: Raw image bytes.
-            text: Optional text to include with the image.
-        """
-        buffer = io.BytesIO()
-        Image.frombytes(format, size, image).save(buffer, format="JPEG")
-        # TODO: we might not want the universal format to be base64 encoded, since encoding is not needed by all LLM services; today, te Gemini adapter has to decode from base64, which is less than ideal.
-        encoded_image = base64.b64encode(buffer.getvalue()).decode("utf-8")
-
-        content = []
-        if text:
-            content.append({"type": "text", "text": text})
-        content.append(
-            {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{encoded_image}"}},
-        )
-        self.add_message({"role": "user", "content": content})
-
-    # NOTE: today we've only built support for audio frames with the Google
-    # LLM, so this "universal" representation skews towards that.
-    # When we add support for other LLMs, we may need to adjust this.
-    def add_audio_frames_message(
-        self, *, audio_frames: list[AudioRawFrame], text: str = "Audio follows"
-    ):
-        """Add a message containing audio frames.
-
-        Args:
-            audio_frames: List of audio frame objects to include.
-            text: Optional text to include with the audio.
-        """
-        if not audio_frames:
-            return
-
-        sample_rate = audio_frames[0].sample_rate
-        num_channels = audio_frames[0].num_channels
-
-        content = []
-        content.append({"type": "text", "text": text})
-        data = b"".join(frame.audio for frame in audio_frames)
-        # TODO: filter this out in OpenAI adapter, since it doesn't support audio frames
-        content.append(
-            {
-                "type": "input_audio",
-                "input_audio": {
-                    "data": data,
-                    "sample_rate": sample_rate,
-                    "num_channels": num_channels,
-                },
-            }
-        )
-        self.add_message({"role": "user", "content": content})
-
-
-@dataclass
-class LLMContextFrame(Frame):
-    """Frame containing LLM context.
-
-    Used as a signal to LLM services to ingest the provided context and
-    generate a response based on it.
-
-    Parameters:
-        context: The LLM context containing messages, tools, and configuration.
-    """
-
-    context: LLMContext

src/pipecat/processors/aggregators/llm_response.py

@@ -36,7 +36,6 @@ from pipecat.frames.frames import (
     FunctionCallsStartedFrame,
     InputAudioRawFrame,
     InterimTranscriptionFrame,
-    LLMContextAssistantTimestampFrame,
     LLMFullResponseEndFrame,
     LLMFullResponseStartFrame,
     LLMMessagesAppendFrame,
@@ -45,6 +44,7 @@ from pipecat.frames.frames import (
     LLMSetToolChoiceFrame,
     LLMSetToolsFrame,
     LLMTextFrame,
+    OpenAILLMContextAssistantTimestampFrame,
     SpeechControlParamsFrame,
     StartFrame,
     StartInterruptionFrame,
@@ -864,7 +864,7 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator):
         await self.push_context_frame()
 
         # Push timestamp frame with current time
-        timestamp_frame = LLMContextAssistantTimestampFrame(timestamp=time_now_iso8601())
+        timestamp_frame = OpenAILLMContextAssistantTimestampFrame(timestamp=time_now_iso8601())
         await self.push_frame(timestamp_frame)
 
     async def _handle_llm_messages_append(self, frame: LLMMessagesAppendFrame):

src/pipecat/processors/aggregators/llm_response_universal.py

@@ -1,874 +0,0 @@
-#
-# Copyright (c) 2024–2025, Daily
-#
-# SPDX-License-Identifier: BSD 2-Clause License
-#
-
-"""LLM response aggregators for handling conversation context and message aggregation.
-
-This module provides aggregators that process and accumulate LLM responses, user inputs,
-and conversation context. These aggregators handle the flow between speech-to-text,
-LLM processing, and text-to-speech components in conversational AI pipelines.
-"""
-
-import asyncio
-import json
-from dataclasses import dataclass
-from typing import Any, Dict, List, Literal, Optional, Set
-
-from loguru import logger
-
-from pipecat.audio.interruptions.base_interruption_strategy import BaseInterruptionStrategy
-from pipecat.audio.turn.smart_turn.base_smart_turn import SmartTurnParams
-from pipecat.audio.vad.vad_analyzer import VADParams
-from pipecat.frames.frames import (
-    BotInterruptionFrame,
-    BotStartedSpeakingFrame,
-    BotStoppedSpeakingFrame,
-    CancelFrame,
-    EmulateUserStartedSpeakingFrame,
-    EmulateUserStoppedSpeakingFrame,
-    EndFrame,
-    Frame,
-    FunctionCallCancelFrame,
-    FunctionCallInProgressFrame,
-    FunctionCallResultFrame,
-    FunctionCallsStartedFrame,
-    InputAudioRawFrame,
-    InterimTranscriptionFrame,
-    LLMContextAssistantTimestampFrame,
-    LLMFullResponseEndFrame,
-    LLMFullResponseStartFrame,
-    LLMMessagesAppendFrame,
-    LLMMessagesUpdateFrame,
-    LLMSetToolChoiceFrame,
-    LLMSetToolsFrame,
-    SpeechControlParamsFrame,
-    StartFrame,
-    StartInterruptionFrame,
-    TextFrame,
-    TranscriptionFrame,
-    UserImageRawFrame,
-    UserStartedSpeakingFrame,
-    UserStoppedSpeakingFrame,
-)
-from pipecat.processors.aggregators.llm_context import LLMContext, LLMContextFrame
-from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
-from pipecat.utils.time import time_now_iso8601
-
-
-@dataclass
-class LLMUserContextAggregatorParams:
-    """Parameters for configuring LLM user context aggregation behavior.
-
-    Parameters:
-        aggregation_timeout: Maximum time in seconds to wait for additional
-            transcription content before pushing aggregated result. This
-            timeout is used only when the transcription is slow to arrive.
-        turn_emulated_vad_timeout: Maximum time in seconds to wait for emulated
-            VAD when using turn-based analysis. Applied when transcription is
-            received but VAD didn't detect speech (e.g., whispered utterances).
-    """
-
-    aggregation_timeout: float = 0.5
-    turn_emulated_vad_timeout: float = 0.8
-
-
-@dataclass
-class LLMAssistantContextAggregatorParams:
-    """Parameters for configuring LLM assistant context aggregation behavior.
-
-    Parameters:
-        expect_stripped_words: Whether to expect and handle stripped words
-            in text frames by adding spaces between tokens.
-    """
-
-    expect_stripped_words: bool = True
-
-
-class LLMContextAggregator(FrameProcessor):
-    """Base LLM aggregator that uses an LLMContext for conversation storage.
-
-    This aggregator maintains conversation state using an LLMContext and
-    pushes LLMContextFrame objects as aggregation frames. It provides
-    common functionality for context-based conversation management.
-    """
-
-    def __init__(self, *, context: LLMContext, role: str, **kwargs):
-        """Initialize the context response aggregator.
-
-        Args:
-            context: The LLM context to use for conversation storage.
-            role: The role this aggregator represents (e.g. "user", "assistant").
-            **kwargs: Additional arguments passed to parent class.
-        """
-        super().__init__(**kwargs)
-        self._context = context
-        self._role = role
-
-        self._aggregation: str = ""
-
-    @property
-    def messages(self) -> List[dict]:
-        """Get messages from the LLM context.
-
-        Returns:
-            List of message dictionaries from the context.
-        """
-        return self._context.messages
-
-    @property
-    def role(self) -> str:
-        """Get the role for this aggregator.
-
-        Returns:
-            The role string for this aggregator.
-        """
-        return self._role
-
-    @property
-    def context(self):
-        """Get the LLM context.
-
-        Returns:
-            The LLMContext instance used by this aggregator.
-        """
-        return self._context
-
-    def get_context_frame(self) -> LLMContextFrame:
-        """Create a context frame with the current context.
-
-        Returns:
-            LLMContextFrame containing the current context.
-        """
-        return LLMContextFrame(context=self._context)
-
-    async def push_context_frame(self, direction: FrameDirection = FrameDirection.DOWNSTREAM):
-        """Push a context frame in the specified direction.
-
-        Args:
-            direction: The direction to push the frame (upstream or downstream).
-        """
-        frame = self.get_context_frame()
-        await self.push_frame(frame, direction)
-
-    def add_messages(self, messages):
-        """Add messages to the context.
-
-        Args:
-            messages: Messages to add to the conversation context.
-        """
-        self._context.add_messages(messages)
-
-    def set_messages(self, messages):
-        """Set the context messages.
-
-        Args:
-            messages: Messages to replace the current context messages.
-        """
-        self._context.set_messages(messages)
-
-    def set_tools(self, tools: List):
-        """Set tools in the context.
-
-        Args:
-            tools: List of tool definitions to set in the context.
-        """
-        self._context.set_tools(tools)
-
-    def set_tool_choice(self, tool_choice: Literal["none", "auto", "required"] | dict):
-        """Set tool choice in the context.
-
-        Args:
-            tool_choice: Tool choice configuration for the context.
-        """
-        self._context.set_tool_choice(tool_choice)
-
-    async def reset(self):
-        """Reset the aggregation state."""
-        self._aggregation = ""
-
-
-# NOTE: the "universal" suffix is just meant to distinguish this aggregator
-# from the old LLMUserContextAggregator while we gradually migrate service to
-# use the new universal LLMContext and associated patterns. The suffix will go
-# away once the migration is complete and the other LLMUserContextAggregator is
-# deprecated.
-class LLMUserContextAggregator_Universal(LLMContextAggregator):
-    """User LLM aggregator that processes speech-to-text transcriptions.
-
-    This aggregator handles the complex logic of aggregating user speech transcriptions
-    from STT services. It manages multiple scenarios including:
-
-    - Transcriptions received between VAD events
-    - Transcriptions received outside VAD events
-    - Interim vs final transcriptions
-    - User interruptions during bot speech
-    - Emulated VAD for whispered or short utterances
-
-    The aggregator uses timeouts to handle cases where transcriptions arrive
-    after VAD events or when no VAD is available.
-    """
-
-    def __init__(
-        self,
-        context: LLMContext,
-        *,
-        params: Optional[LLMUserContextAggregatorParams] = None,
-        **kwargs,
-    ):
-        """Initialize the user context aggregator.
-
-        Args:
-            context: The LLM context for conversation storage.
-            params: Configuration parameters for aggregation behavior.
-            **kwargs: Additional arguments. Supports deprecated 'aggregation_timeout'.
-        """
-        super().__init__(context=context, role="user", **kwargs)
-        self._params = params or LLMUserContextAggregatorParams()
-        self._vad_params: Optional[VADParams] = None
-        self._turn_params: Optional[SmartTurnParams] = None
-
-        if "aggregation_timeout" in kwargs:
-            import warnings
-
-            with warnings.catch_warnings():
-                warnings.simplefilter("always")
-                warnings.warn(
-                    "Parameter 'aggregation_timeout' is deprecated, use 'params' instead.",
-                    DeprecationWarning,
-                )
-
-            self._params.aggregation_timeout = kwargs["aggregation_timeout"]
-
-        self._user_speaking = False
-        self._bot_speaking = False
-        self._was_bot_speaking = False
-        self._emulating_vad = False
-        self._seen_interim_results = False
-        self._waiting_for_aggregation = False
-
-        self._aggregation_event = asyncio.Event()
-        self._aggregation_task = None
-
-    async def reset(self):
-        """Reset the aggregation state and interruption strategies."""
-        await super().reset()
-        self._was_bot_speaking = False
-        self._seen_interim_results = False
-        self._waiting_for_aggregation = False
-        [await s.reset() for s in self._interruption_strategies]
-
-    async def process_frame(self, frame: Frame, direction: FrameDirection):
-        """Process frames for user speech aggregation and context management.
-
-        Args:
-            frame: The frame to process.
-            direction: The direction of frame flow in the pipeline.
-        """
-        await super().process_frame(frame, direction)
-
-        if isinstance(frame, StartFrame):
-            # Push StartFrame before start(), because we want StartFrame to be
-            # processed by every processor before any other frame is processed.
-            await self.push_frame(frame, direction)
-            await self._start(frame)
-        elif isinstance(frame, EndFrame):
-            # Push EndFrame before stop(), because stop() waits on the task to
-            # finish and the task finishes when EndFrame is processed.
-            await self.push_frame(frame, direction)
-            await self._stop(frame)
-        elif isinstance(frame, CancelFrame):
-            await self._cancel(frame)
-            await self.push_frame(frame, direction)
-        elif isinstance(frame, InputAudioRawFrame):
-            await self._handle_input_audio(frame)
-            await self.push_frame(frame, direction)
-        elif isinstance(frame, UserStartedSpeakingFrame):
-            await self._handle_user_started_speaking(frame)
-            await self.push_frame(frame, direction)
-        elif isinstance(frame, UserStoppedSpeakingFrame):
-            await self._handle_user_stopped_speaking(frame)
-            await self.push_frame(frame, direction)
-        elif isinstance(frame, BotStartedSpeakingFrame):
-            await self._handle_bot_started_speaking(frame)
-            await self.push_frame(frame, direction)
-        elif isinstance(frame, BotStoppedSpeakingFrame):
-            await self._handle_bot_stopped_speaking(frame)
-            await self.push_frame(frame, direction)
-        elif isinstance(frame, TranscriptionFrame):
-            await self._handle_transcription(frame)
-        elif isinstance(frame, InterimTranscriptionFrame):
-            await self._handle_interim_transcription(frame)
-        elif isinstance(frame, LLMMessagesAppendFrame):
-            await self._handle_llm_messages_append(frame)
-        elif isinstance(frame, LLMMessagesUpdateFrame):
-            await self._handle_llm_messages_update(frame)
-        elif isinstance(frame, LLMSetToolsFrame):
-            self.set_tools(frame.tools)
-        elif isinstance(frame, LLMSetToolChoiceFrame):
-            self.set_tool_choice(frame.tool_choice)
-        elif isinstance(frame, SpeechControlParamsFrame):
-            self._vad_params = frame.vad_params
-            self._turn_params = frame.turn_params
-            await self.push_frame(frame, direction)
-        else:
-            await self.push_frame(frame, direction)
-
-    async def _process_aggregation(self):
-        """Process the current aggregation and push it downstream."""
-        aggregation = self._aggregation
-        await self.reset()
-        self._context.add_message({"role": self.role, "content": aggregation})
-        frame = LLMContextFrame(self._context)
-        await self.push_frame(frame)
-
-    async def _push_aggregation(self):
-        """Push the current aggregation based on interruption strategies and conditions."""
-        if len(self._aggregation) > 0:
-            if self.interruption_strategies and self._bot_speaking:
-                should_interrupt = await self._should_interrupt_based_on_strategies()
-
-                if should_interrupt:
-                    logger.debug(
-                        "Interruption conditions met - pushing BotInterruptionFrame and aggregation"
-                    )
-                    await self.push_frame(BotInterruptionFrame(), FrameDirection.UPSTREAM)
-                    await self._process_aggregation()
-                else:
-                    logger.debug("Interruption conditions not met - not pushing aggregation")
-                    # Don't process aggregation, just reset it
-                    await self.reset()
-            else:
-                # No interruption config - normal behavior (always push aggregation)
-                await self._process_aggregation()
-        # Handles the case where both the user and the bot are not speaking,
-        # and the bot was previously speaking before the user interruption.
-        # Normally, when the user stops speaking, new text is expected,
-        # which triggers the bot to respond. However, if no new text
-        # is received, this safeguard ensures
-        # the bot doesn't hang indefinitely while waiting to speak again.
-        elif not self._seen_interim_results and self._was_bot_speaking and not self._bot_speaking:
-            logger.warning("User stopped speaking but no new aggregation received.")
-            # Resetting it so we don't trigger this twice
-            self._was_bot_speaking = False
-            # TODO: we are not enabling this for now, due to some STT services which can take as long as 2 seconds two return a transcription
-            # So we need more tests and probably make this feature configurable, disabled it by default.
-            # We are just pushing the same previous context to be processed again in this case
-            # await self.push_frame(LLMContextFrame(self._context))
-
-    async def _should_interrupt_based_on_strategies(self) -> bool:
-        """Check if interruption should occur based on configured strategies.
-
-        Returns:
-            True if any interruption strategy indicates interruption should occur.
-        """
-
-        async def should_interrupt(strategy: BaseInterruptionStrategy):
-            await strategy.append_text(self._aggregation)
-            return await strategy.should_interrupt()
-
-        return any([await should_interrupt(s) for s in self._interruption_strategies])
-
-    async def _start(self, frame: StartFrame):
-        self._create_aggregation_task()
-
-    async def _stop(self, frame: EndFrame):
-        await self._cancel_aggregation_task()
-
-    async def _cancel(self, frame: CancelFrame):
-        await self._cancel_aggregation_task()
-
-    async def _handle_llm_messages_append(self, frame: LLMMessagesAppendFrame):
-        self.add_messages(frame.messages)
-        if frame.run_llm:
-            await self.push_context_frame()
-
-    async def _handle_llm_messages_update(self, frame: LLMMessagesUpdateFrame):
-        self.set_messages(frame.messages)
-        if frame.run_llm:
-            await self.push_context_frame()
-
-    async def _handle_input_audio(self, frame: InputAudioRawFrame):
-        for s in self.interruption_strategies:
-            await s.append_audio(frame.audio, frame.sample_rate)
-
-    async def _handle_user_started_speaking(self, frame: UserStartedSpeakingFrame):
-        self._user_speaking = True
-        self._waiting_for_aggregation = True
-        self._was_bot_speaking = self._bot_speaking
-
-        # If we get a non-emulated UserStartedSpeakingFrame but we are in the
-        # middle of emulating VAD, let's stop emulating VAD (i.e. don't send the
-        # EmulateUserStoppedSpeakingFrame).
-        if not frame.emulated and self._emulating_vad:
-            self._emulating_vad = False
-
-    async def _handle_user_stopped_speaking(self, _: UserStoppedSpeakingFrame):
-        self._user_speaking = False
-        # We just stopped speaking. Let's see if there's some aggregation to
-        # push. If the last thing we saw is an interim transcription, let's wait
-        # pushing the aggregation as we will probably get a final transcription.
-        if len(self._aggregation) > 0:
-            if not self._seen_interim_results:
-                await self._push_aggregation()
-        # Handles the case where both the user and the bot are not speaking,
-        # and the bot was previously speaking before the user interruption.
-        # So in this case we are resetting the aggregation timer
-        elif not self._seen_interim_results and self._was_bot_speaking and not self._bot_speaking:
-            # Reset aggregation timer.
-            self._aggregation_event.set()
-
-    async def _handle_bot_started_speaking(self, _: BotStartedSpeakingFrame):
-        self._bot_speaking = True
-
-    async def _handle_bot_stopped_speaking(self, _: BotStoppedSpeakingFrame):
-        self._bot_speaking = False
-
-    async def _handle_transcription(self, frame: TranscriptionFrame):
-        text = frame.text
-
-        # Make sure we really have some text.
-        if not text.strip():
-            return
-
-        self._aggregation += f" {text}" if self._aggregation else text
-        # We just got a final result, so let's reset interim results.
-        self._seen_interim_results = False
-        # Reset aggregation timer.
-        self._aggregation_event.set()
-
-    async def _handle_interim_transcription(self, _: InterimTranscriptionFrame):
-        self._seen_interim_results = True
-
-    def _create_aggregation_task(self):
-        if not self._aggregation_task:
-            self._aggregation_task = self.create_task(self._aggregation_task_handler())
-
-    async def _cancel_aggregation_task(self):
-        if self._aggregation_task:
-            await self.cancel_task(self._aggregation_task)
-            self._aggregation_task = None
-
-    async def _aggregation_task_handler(self):
-        while True:
-            try:
-                # The _aggregation_task_handler handles two distinct timeout scenarios:
-                #
-                # 1. When emulating_vad=True: Wait for emulated VAD timeout before
-                #    pushing aggregation (simulating VAD behavior when no actual VAD
-                #    detection occurred).
-                #
-                # 2. When emulating_vad=False: Use aggregation_timeout as a buffer
-                #    to wait for potential late-arriving transcription frames after
-                #    a real VAD event.
-                #
-                # For emulated VAD scenarios, the timeout strategy depends on whether
-                # a turn analyzer is configured:
-                #
-                # - WITH turn analyzer: Use turn_emulated_vad_timeout parameter because
-                #   the VAD's stop_secs is set very low (e.g. 0.2s) for rapid speech
-                #   chunking to feed the turn analyzer. This low value is too fast
-                #   for emulated VAD scenarios where we need to allow users time to
-                #   finish speaking (e.g. 0.8s).
-                #
-                # - WITHOUT turn analyzer: Use VAD's stop_secs directly to maintain
-                #   consistent user experience between real VAD detection and
-                #   emulated VAD scenarios.
-                if not self._emulating_vad:
-                    timeout = self._params.aggregation_timeout
-                elif self._turn_params:
-                    timeout = self._params.turn_emulated_vad_timeout
-                else:
-                    # Use VAD stop_secs when no turn analyzer is present, fallback if no VAD params
-                    timeout = (
-                        self._vad_params.stop_secs
-                        if self._vad_params
-                        else self._params.turn_emulated_vad_timeout
-                    )
-                await asyncio.wait_for(self._aggregation_event.wait(), timeout)
-                await self._maybe_emulate_user_speaking()
-            except asyncio.TimeoutError:
-                if not self._user_speaking:
-                    await self._push_aggregation()
-
-                # If we are emulating VAD we still need to send the user stopped
-                # speaking frame.
-                if self._emulating_vad:
-                    await self.push_frame(
-                        EmulateUserStoppedSpeakingFrame(), FrameDirection.UPSTREAM
-                    )
-                    self._emulating_vad = False
-            finally:
-                self.reset_watchdog()
-                self._aggregation_event.clear()
-
-    async def _maybe_emulate_user_speaking(self):
-        """Maybe emulate user speaking based on transcription.
-
-        Emulate user speaking if we got a transcription but it was not
-        detected by VAD. Only do that if the bot is not speaking.
-        """
-        # Check if we received a transcription but VAD was not able to detect
-        # voice (e.g. when you whisper a short utterance). In that case, we need
-        # to emulate VAD (i.e. user start/stopped speaking), but we do it only
-        # if the bot is not speaking. If the bot is speaking and we really have
-        # a short utterance we don't really want to interrupt the bot.
-        if (
-            not self._user_speaking
-            and not self._waiting_for_aggregation
-            and len(self._aggregation) > 0
-        ):
-            if self._bot_speaking:
-                # If we reached this case and the bot is speaking, let's ignore
-                # what the user said.
-                logger.debug("Ignoring user speaking emulation, bot is speaking.")
-                await self.reset()
-            else:
-                # The bot is not speaking so, let's trigger user speaking
-                # emulation.
-                await self.push_frame(EmulateUserStartedSpeakingFrame(), FrameDirection.UPSTREAM)
-                self._emulating_vad = True
-
-
-# NOTE: the "universal" suffix is just meant to distinguish this aggregator
-# from the old LLMAssistantContextAggregator while we gradually migrate service
-# to use the new universal LLMContext and associated patterns. The suffix will
-# go away once the migration is complete and the other
-# LLMAssistantContextAggregator is deprecated.
-class LLMAssistantContextAggregator_Universal(LLMContextAggregator):
-    """Assistant LLM aggregator that processes bot responses and function calls.
-
-    This aggregator handles the complex logic of processing assistant responses including:
-
-    - Text frame aggregation between response start/end markers
-    - Function call lifecycle management
-    - Context updates with timestamps
-    - Tool execution and result handling
-    - Interruption handling during responses
-
-    The aggregator manages function calls in progress and coordinates between
-    text generation and tool execution phases of LLM responses.
-    """
-
-    def __init__(
-        self,
-        context: LLMContext,
-        *,
-        params: Optional[LLMAssistantContextAggregatorParams] = None,
-        **kwargs,
-    ):
-        """Initialize the assistant context aggregator.
-
-        Args:
-            context: The OpenAI LLM context for conversation storage.
-            params: Configuration parameters for aggregation behavior.
-            **kwargs: Additional arguments. Supports deprecated 'expect_stripped_words'.
-        """
-        super().__init__(context=context, role="assistant", **kwargs)
-        self._params = params or LLMAssistantContextAggregatorParams()
-
-        if "expect_stripped_words" in kwargs:
-            import warnings
-
-            with warnings.catch_warnings():
-                warnings.simplefilter("always")
-                warnings.warn(
-                    "Parameter 'expect_stripped_words' is deprecated, use 'params' instead.",
-                    DeprecationWarning,
-                )
-
-            self._params.expect_stripped_words = kwargs["expect_stripped_words"]
-
-        self._started = 0
-        self._function_calls_in_progress: Dict[str, Optional[FunctionCallInProgressFrame]] = {}
-        self._context_updated_tasks: Set[asyncio.Task] = set()
-
-    @property
-    def has_function_calls_in_progress(self) -> bool:
-        """Check if there are any function calls currently in progress.
-
-        Returns:
-            True if function calls are in progress, False otherwise.
-        """
-        return bool(self._function_calls_in_progress)
-
-    async def process_frame(self, frame: Frame, direction: FrameDirection):
-        """Process frames for assistant response aggregation and function call management.
-
-        Args:
-            frame: The frame to process.
-            direction: The direction of frame flow in the pipeline.
-        """
-        await super().process_frame(frame, direction)
-
-        if isinstance(frame, StartInterruptionFrame):
-            await self._handle_interruptions(frame)
-            await self.push_frame(frame, direction)
-        elif isinstance(frame, LLMFullResponseStartFrame):
-            await self._handle_llm_start(frame)
-        elif isinstance(frame, LLMFullResponseEndFrame):
-            await self._handle_llm_end(frame)
-        elif isinstance(frame, TextFrame):
-            await self._handle_text(frame)
-        elif isinstance(frame, LLMMessagesAppendFrame):
-            await self._handle_llm_messages_append(frame)
-        elif isinstance(frame, LLMMessagesUpdateFrame):
-            await self._handle_llm_messages_update(frame)
-        elif isinstance(frame, LLMSetToolsFrame):
-            self.set_tools(frame.tools)
-        elif isinstance(frame, LLMSetToolChoiceFrame):
-            self.set_tool_choice(frame.tool_choice)
-        elif isinstance(frame, FunctionCallsStartedFrame):
-            await self._handle_function_calls_started(frame)
-        elif isinstance(frame, FunctionCallInProgressFrame):
-            await self._handle_function_call_in_progress(frame)
-        elif isinstance(frame, FunctionCallResultFrame):
-            await self._handle_function_call_result(frame)
-        elif isinstance(frame, FunctionCallCancelFrame):
-            await self._handle_function_call_cancel(frame)
-        elif isinstance(frame, UserImageRawFrame) and frame.request and frame.request.tool_call_id:
-            await self._handle_user_image_frame(frame)
-        elif isinstance(frame, BotStoppedSpeakingFrame):
-            await self._push_aggregation()
-            await self.push_frame(frame, direction)
-        else:
-            await self.push_frame(frame, direction)
-
-    async def _push_aggregation(self):
-        """Push the current assistant aggregation with timestamp."""
-        if not self._aggregation:
-            return
-
-        aggregation = self._aggregation.strip()
-        await self.reset()
-
-        if aggregation:
-            self._context.add_message({"role": "assistant", "content": aggregation})
-
-        # Push context frame
-        await self.push_context_frame()
-
-        # Push timestamp frame with current time
-        timestamp_frame = LLMContextAssistantTimestampFrame(timestamp=time_now_iso8601())
-        await self.push_frame(timestamp_frame)
-
-    async def _handle_llm_messages_append(self, frame: LLMMessagesAppendFrame):
-        self.add_messages(frame.messages)
-        if frame.run_llm:
-            await self.push_context_frame(FrameDirection.UPSTREAM)
-
-    async def _handle_llm_messages_update(self, frame: LLMMessagesUpdateFrame):
-        self.set_messages(frame.messages)
-        if frame.run_llm:
-            await self.push_context_frame(FrameDirection.UPSTREAM)
-
-    async def _handle_interruptions(self, frame: StartInterruptionFrame):
-        await self._push_aggregation()
-        self._started = 0
-        await self.reset()
-
-    async def _handle_function_calls_started(self, frame: FunctionCallsStartedFrame):
-        function_names = [f"{f.function_name}:{f.tool_call_id}" for f in frame.function_calls]
-        logger.debug(f"{self} FunctionCallsStartedFrame: {function_names}")
-        for function_call in frame.function_calls:
-            self._function_calls_in_progress[function_call.tool_call_id] = None
-
-    async def _handle_function_call_in_progress(self, frame: FunctionCallInProgressFrame):
-        logger.debug(
-            f"{self} FunctionCallInProgressFrame: [{frame.function_name}:{frame.tool_call_id}]"
-        )
-
-        # Update context with the in-progress function call
-        self._context.add_message(
-            {
-                "role": "assistant",
-                "tool_calls": [
-                    {
-                        "id": frame.tool_call_id,
-                        "function": {
-                            "name": frame.function_name,
-                            "arguments": json.dumps(frame.arguments),
-                        },
-                        "type": "function",
-                    }
-                ],
-            }
-        )
-        self._context.add_message(
-            {
-                "role": "tool",
-                "content": "IN_PROGRESS",
-                "tool_call_id": frame.tool_call_id,
-            }
-        )
-
-        self._function_calls_in_progress[frame.tool_call_id] = frame
-
-    async def _handle_function_call_result(self, frame: FunctionCallResultFrame):
-        logger.debug(
-            f"{self} FunctionCallResultFrame: [{frame.function_name}:{frame.tool_call_id}]"
-        )
-        if frame.tool_call_id not in self._function_calls_in_progress:
-            logger.warning(
-                f"FunctionCallResultFrame tool_call_id [{frame.tool_call_id}] is not running"
-            )
-            return
-
-        del self._function_calls_in_progress[frame.tool_call_id]
-
-        properties = frame.properties
-
-        # Update context with the function call result
-        if frame.result:
-            result = json.dumps(frame.result)
-            self._update_function_call_result(frame.function_name, frame.tool_call_id, result)
-        else:
-            self._update_function_call_result(frame.function_name, frame.tool_call_id, "COMPLETED")
-
-        run_llm = False
-
-        # Run inference if the function call result requires it.
-        if frame.result:
-            if properties and properties.run_llm is not None:
-                # If the tool call result has a run_llm property, use it.
-                run_llm = properties.run_llm
-            elif frame.run_llm is not None:
-                # If the frame is indicating we should run the LLM, do it.
-                run_llm = frame.run_llm
-            else:
-                # If this is the last function call in progress, run the LLM.
-                run_llm = not bool(self._function_calls_in_progress)
-
-        if run_llm:
-            await self.push_context_frame(FrameDirection.UPSTREAM)
-
-        # Call the `on_context_updated` callback once the function call result
-        # is added to the context. Also, run this in a separate task to make
-        # sure we don't block the pipeline.
-        if properties and properties.on_context_updated:
-            task_name = f"{frame.function_name}:{frame.tool_call_id}:on_context_updated"
-            task = self.create_task(properties.on_context_updated(), task_name)
-            self._context_updated_tasks.add(task)
-            task.add_done_callback(self._context_updated_task_finished)
-
-    async def _handle_function_call_cancel(self, frame: FunctionCallCancelFrame):
-        logger.debug(
-            f"{self} FunctionCallCancelFrame: [{frame.function_name}:{frame.tool_call_id}]"
-        )
-        if frame.tool_call_id not in self._function_calls_in_progress:
-            return
-
-        if self._function_calls_in_progress[frame.tool_call_id].cancel_on_interruption:
-            # Update context with the function call cancellation
-            self._update_function_call_result(frame.function_name, frame.tool_call_id, "CANCELLED")
-            del self._function_calls_in_progress[frame.tool_call_id]
-
-    def _update_function_call_result(self, function_name: str, tool_call_id: str, result: Any):
-        for message in self._context.messages:
-            if (
-                message["role"] == "tool"
-                and message["tool_call_id"]
-                and message["tool_call_id"] == tool_call_id
-            ):
-                message["content"] = result
-
-    async def _handle_user_image_frame(self, frame: UserImageRawFrame):
-        logger.debug(
-            f"{self} UserImageRawFrame: [{frame.request.function_name}:{frame.request.tool_call_id}]"
-        )
-
-        if frame.request.tool_call_id not in self._function_calls_in_progress:
-            logger.warning(
-                f"UserImageRawFrame tool_call_id [{frame.request.tool_call_id}] is not running"
-            )
-            return
-
-        del self._function_calls_in_progress[frame.request.tool_call_id]
-
-        # Update context with the image frame
-        await self._update_function_call_result(
-            frame.request.function_name, frame.request.tool_call_id, "COMPLETED"
-        )
-        self._context.add_image_frame_message(
-            format=frame.format,
-            size=frame.size,
-            image=frame.image,
-            text=frame.request.context,
-        )
-
-        await self._push_aggregation()
-        await self.push_context_frame(FrameDirection.UPSTREAM)
-
-    async def _handle_llm_start(self, _: LLMFullResponseStartFrame):
-        self._started += 1
-
-    async def _handle_llm_end(self, _: LLMFullResponseEndFrame):
-        self._started -= 1
-        await self._push_aggregation()
-
-    async def _handle_text(self, frame: TextFrame):
-        if not self._started:
-            return
-
-        if self._params.expect_stripped_words:
-            self._aggregation += f" {frame.text}" if self._aggregation else frame.text
-        else:
-            self._aggregation += frame.text
-
-    def _context_updated_task_finished(self, task: asyncio.Task):
-        self._context_updated_tasks.discard(task)
-        # The task is finished so this should exit immediately. We need to do
-        # this because otherwise the task manager would report a dangling task
-        # if we don't remove it.
-        asyncio.run_coroutine_threadsafe(self.wait_for_task(task), self.get_event_loop())
-
-
-@dataclass
-class LLMContextAggregatorPair:
-    """Pair of LLM context aggregators for user and assistant messages.
-
-    Parameters:
-        _user: User context aggregator for processing user messages.
-        _assistant: Assistant context aggregator for processing assistant messages.
-    """
-
-    _user: LLMUserContextAggregator_Universal
-    _assistant: LLMAssistantContextAggregator_Universal
-
-    @staticmethod
-    def create(
-        context: LLMContext,
-        *,
-        user_params: LLMUserContextAggregatorParams = LLMUserContextAggregatorParams(),
-        assistant_params: LLMAssistantContextAggregatorParams = LLMAssistantContextAggregatorParams(),
-    ) -> "LLMContextAggregatorPair":
-        """Factory method to create an LLMContextAggregatorPair.
-
-        Args:
-            context: The context managed by the aggregators.
-            user_params: Parameters for the user context aggregator.
-            assistant_params: Parameters for the assistant context aggregator.
-
-        Returns:
-            LLMContextAggregatorPair: A new instance with configured aggregators.
-        """
-        user = LLMUserContextAggregator_Universal(context, params=user_params)
-        assistant = LLMAssistantContextAggregator_Universal(context, params=assistant_params)
-        return LLMContextAggregatorPair(_user=user, _assistant=assistant)
-
-    def user(self) -> LLMUserContextAggregator_Universal:
-        """Get the user context aggregator.
-
-        Returns:
-            The user context aggregator instance.
-        """
-        return self._user
-
-    def assistant(self) -> LLMAssistantContextAggregator_Universal:
-        """Get the assistant context aggregator.
-
-        Returns:
-            The assistant context aggregator instance.
-        """
-        return self._assistant

src/pipecat/processors/frameworks/rtvi.py

@@ -72,6 +72,7 @@ from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 from pipecat.services.llm_service import (
     FunctionCallParams,  # TODO(aleix): we shouldn't import `services` from `processors`
 )
+from pipecat.services.openai.llm import OpenAIContextAggregatorPair
 from pipecat.transports.base_input import BaseInputTransport
 from pipecat.transports.base_output import BaseOutputTransport
 from pipecat.transports.base_transport import BaseTransport

src/pipecat/services/assemblyai/stt.py

@@ -44,6 +44,7 @@ from .models import (
 
 try:
     import websockets
+    from websockets.asyncio.client import connect as websocket_connect
 except ModuleNotFoundError as e:
     logger.error(f"Exception: {e}")
     logger.error('In order to use AssemblyAI, you need to `pip install "pipecat-ai[assemblyai]"`.')
@@ -190,9 +191,9 @@ class AssemblyAISTTService(STTService):
                 "Authorization": self._api_key,
                 "User-Agent": f"AssemblyAI/1.0 (integration=Pipecat/{pipecat_version})",
             }
-            self._websocket = await websockets.connect(
+            self._websocket = await websocket_connect(
                 ws_url,
-                extra_headers=headers,
+                additional_headers=headers,
             )
             self._connected = True
             self._receive_task = self.create_task(self._receive_task_handler())

src/pipecat/services/aws/stt.py

@@ -36,6 +36,8 @@ from pipecat.utils.tracing.service_decorators import traced_stt
 
 try:
     import websockets
+    from websockets.asyncio.client import connect as websocket_connect
+    from websockets.protocol import State
 except ModuleNotFoundError as e:
     logger.error(f"Exception: {e}")
     logger.error("In order to use AWS services, you need to `pip install pipecat-ai[aws]`.")
@@ -133,7 +135,7 @@ class AWSTranscribeSTTService(STTService):
         while retry_count < max_retries:
             try:
                 await self._connect()
-                if self._ws_client and self._ws_client.open:
+                if self._ws_client and self._ws_client.state is State.OPEN:
                     logger.info("Successfully established WebSocket connection")
                     return
                 logger.warning("WebSocket connection not established after connect")
@@ -174,7 +176,7 @@ class AWSTranscribeSTTService(STTService):
         """
         try:
             # Ensure WebSocket is connected
-            if not self._ws_client or not self._ws_client.open:
+            if not self._ws_client or self._ws_client.state is State.CLOSED:
                 logger.debug("WebSocket not connected, attempting to reconnect...")
                 try:
                     await self._connect()
@@ -208,7 +210,7 @@ class AWSTranscribeSTTService(STTService):
 
     async def _connect(self):
         """Connect to AWS Transcribe with connection state management."""
-        if self._ws_client and self._ws_client.open and self._receive_task:
+        if self._ws_client and self._ws_client.state is State.OPEN and self._receive_task:
             logger.debug(f"{self} Already connected")
             return
 
@@ -238,7 +240,7 @@ class AWSTranscribeSTTService(STTService):
                 )
 
                 # Add required headers
-                extra_headers = {
+                additional_headers = {
                     "Origin": "https://localhost",
                     "Sec-WebSocket-Key": websocket_key,
                     "Sec-WebSocket-Version": "13",
@@ -268,9 +270,9 @@ class AWSTranscribeSTTService(STTService):
                 logger.debug(f"{self} Connecting to WebSocket with URL: {presigned_url[:100]}...")
 
                 # Connect with the required headers and settings
-                self._ws_client = await websockets.connect(
+                self._ws_client = await websocket_connect(
                     presigned_url,
-                    extra_headers=extra_headers,
+                    additional_headers=additional_headers,
                     subprotocols=["mqtt"],
                     ping_interval=None,
                     ping_timeout=None,
@@ -299,7 +301,7 @@ class AWSTranscribeSTTService(STTService):
             self._receive_task = None
 
         try:
-            if self._ws_client and self._ws_client.open:
+            if self._ws_client and self._ws_client.state is State.OPEN:
                 # Send end-stream message
                 end_stream = {"message-type": "event", "event": "end"}
                 await self._ws_client.send(json.dumps(end_stream))
@@ -341,7 +343,7 @@ class AWSTranscribeSTTService(STTService):
     async def _receive_loop(self):
         """Background task to receive and process messages from AWS Transcribe."""
         while True:
-            if not self._ws_client or not self._ws_client.open:
+            if not self._ws_client or self._ws_client.state is State.CLOSED:
                 logger.warning(f"{self} WebSocket closed in receive loop")
                 break
 

src/pipecat/services/cartesia/stt.py

@@ -15,7 +15,6 @@ import json
 import urllib.parse
 from typing import AsyncGenerator, Optional
 
-import websockets
 from loguru import logger
 
 from pipecat.frames.frames import (
@@ -34,6 +33,15 @@ from pipecat.transcriptions.language import Language
 from pipecat.utils.time import time_now_iso8601
 from pipecat.utils.tracing.service_decorators import traced_stt
 
+try:
+    import websockets
+    from websockets.asyncio.client import connect as websocket_connect
+    from websockets.protocol import State
+except ModuleNotFoundError as e:
+    logger.error(f"Exception: {e}")
+    logger.error("In order to use Cartesia, you need to `pip install pipecat-ai[cartesia]`.")
+    raise Exception(f"Missing module: {e}")
+
 
 class CartesiaLiveOptions:
     """Configuration options for Cartesia Live STT service.
@@ -216,7 +224,7 @@ class CartesiaSTTService(STTService):
             None - transcription results are handled via WebSocket responses.
         """
         # If the connection is closed, due to timeout, we need to reconnect when the user starts speaking again
-        if not self._connection or self._connection.closed:
+        if not self._connection or self._connection.state is State.CLOSED:
             await self._connect()
 
         await self._connection.send(audio)
@@ -229,7 +237,7 @@ class CartesiaSTTService(STTService):
         headers = {"Cartesia-Version": "2025-04-16", "X-API-Key": self._api_key}
 
         try:
-            self._connection = await websockets.connect(ws_url, extra_headers=headers)
+            self._connection = await websocket_connect(ws_url, additional_headers=headers)
             # Setup the receiver task to handle the incoming messages from the Cartesia server
             if self._receiver_task is None or self._receiver_task.done():
                 self._receiver_task = asyncio.create_task(self._receive_messages())
@@ -240,7 +248,7 @@ class CartesiaSTTService(STTService):
     async def _receive_messages(self):
         try:
             while True:
-                if not self._connection or self._connection.closed:
+                if not self._connection or self._connection.state is State.CLOSED:
                     break
 
                 message = await self._connection.recv()
@@ -320,7 +328,7 @@ class CartesiaSTTService(STTService):
                 logger.exception(f"Unexpected exception while cancelling task: {e}")
             self._receiver_task = None
 
-        if self._connection and self._connection.open:
+        if self._connection and self._connection.state is State.OPEN:
             logger.debug("Disconnecting from Cartesia")
 
             await self._connection.close()
@@ -344,5 +352,5 @@ class CartesiaSTTService(STTService):
             await self.start_metrics()
         elif isinstance(frame, UserStoppedSpeakingFrame):
             # Send finalize command to flush the transcription session
-            if self._connection and self._connection.open:
+            if self._connection and self._connection.state is State.OPEN:
                 await self._connection.send("finalize")

src/pipecat/services/cartesia/tts.py

@@ -36,8 +36,9 @@ from pipecat.utils.tracing.service_decorators import traced_tts
 
 # See .env.example for Cartesia configuration needed
 try:
-    import websockets
     from cartesia import AsyncCartesia
+    from websockets.asyncio.client import connect as websocket_connect
+    from websockets.protocol import State
 except ModuleNotFoundError as e:
     logger.error(f"Exception: {e}")
     logger.error("In order to use Cartesia, you need to `pip install pipecat-ai[cartesia]`.")
@@ -288,10 +289,10 @@ class CartesiaTTSService(AudioContextWordTTSService):
 
     async def _connect_websocket(self):
         try:
-            if self._websocket and self._websocket.open:
+            if self._websocket and self._websocket.state is State.OPEN:
                 return
             logger.debug("Connecting to Cartesia")
-            self._websocket = await websockets.connect(
+            self._websocket = await websocket_connect(
                 f"{self._url}?api_key={self._api_key}&cartesia_version={self._cartesia_version}"
             )
         except Exception as e:
@@ -380,7 +381,7 @@ class CartesiaTTSService(AudioContextWordTTSService):
         logger.debug(f"{self}: Generating TTS [{text}]")
 
         try:
-            if not self._websocket or self._websocket.closed:
+            if not self._websocket or self._websocket.state is State.CLOSED:
                 await self._connect()
 
             if not self._context_id:

src/pipecat/services/elevenlabs/tts.py

@@ -44,6 +44,8 @@ from pipecat.utils.tracing.service_decorators import traced_tts
 # See .env.example for ElevenLabs configuration needed
 try:
     import websockets
+    from websockets.asyncio.client import connect as websocket_connect
+    from websockets.protocol import State
 except ModuleNotFoundError as e:
     logger.error(f"Exception: {e}")
     logger.error("In order to use ElevenLabs, you need to `pip install pipecat-ai[elevenlabs]`.")
@@ -447,7 +449,7 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
 
     async def _connect_websocket(self):
         try:
-            if self._websocket and self._websocket.open:
+            if self._websocket and self._websocket.state is State.OPEN:
                 return
 
             logger.debug("Connecting to ElevenLabs")
@@ -474,8 +476,8 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
                 )
 
             # Set max websocket message size to 16MB for large audio responses
-            self._websocket = await websockets.connect(
-                url, max_size=16 * 1024 * 1024, extra_headers={"xi-api-key": self._api_key}
+            self._websocket = await websocket_connect(
+                url, max_size=16 * 1024 * 1024, additional_headers={"xi-api-key": self._api_key}
             )
 
         except Exception as e:
@@ -546,8 +548,14 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
             # Check if this message belongs to the current context.
             # This should never happen, so warn about it.
             if not self.audio_context_available(received_ctx_id):
-                logger.warning(f"Ignoring message from unavailable context: {received_ctx_id}")
-                continue
+                if self._context_id == received_ctx_id:
+                    logger.debug(
+                        f"Received a delayed message, recreating the context: {self._context_id}"
+                    )
+                    await self.create_audio_context(self._context_id)
+                else:
+                    logger.warning(f"Ignoring message from unavailable context: {received_ctx_id}")
+                    continue
 
             if msg.get("audio"):
                 await self.stop_ttfb_metrics()
@@ -587,7 +595,7 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
             self.reset_watchdog()
             await asyncio.sleep(KEEPALIVE_SLEEP)
             try:
-                if self._websocket and self._websocket.open:
+                if self._websocket and self._websocket.state is State.OPEN:
                     if self._context_id:
                         # Send keepalive with context ID to keep the connection alive
                         keepalive_message = {
@@ -625,7 +633,7 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
         logger.debug(f"{self}: Generating TTS [{text}]")
 
         try:
-            if not self._websocket or self._websocket.closed:
+            if not self._websocket or self._websocket.state is State.CLOSED:
                 await self._connect()
 
             try:

src/pipecat/services/fish/tts.py

@@ -34,7 +34,8 @@ from pipecat.utils.tracing.service_decorators import traced_tts
 
 try:
     import ormsgpack
-    import websockets
+    from websockets.asyncio.client import connect as websocket_connect
+    from websockets.protocol import State
 except ModuleNotFoundError as e:
     logger.error(f"Exception: {e}")
     logger.error("In order to use Fish Audio, you need to `pip install pipecat-ai[fish]`.")
@@ -210,13 +211,13 @@ class FishAudioTTSService(InterruptibleTTSService):
 
     async def _connect_websocket(self):
         try:
-            if self._websocket and self._websocket.open:
+            if self._websocket and self._websocket.state is State.OPEN:
                 return
 
             logger.debug("Connecting to Fish Audio")
             headers = {"Authorization": f"Bearer {self._api_key}"}
             headers["model"] = self.model_name
-            self._websocket = await websockets.connect(self._base_url, extra_headers=headers)
+            self._websocket = await websocket_connect(self._base_url, additional_headers=headers)
 
             # Send initial start message with ormsgpack
             start_message = {"event": "start", "request": {"text": "", **self._settings}}
@@ -246,7 +247,7 @@ class FishAudioTTSService(InterruptibleTTSService):
     async def flush_audio(self):
         """Flush any buffered audio by sending a flush event to Fish Audio."""
         logger.trace(f"{self}: Flushing audio buffers")
-        if not self._websocket or self._websocket.closed:
+        if not self._websocket or self._websocket.state is State.CLOSED:
             return
         flush_message = {"event": "flush"}
         await self._get_websocket().send(ormsgpack.packb(flush_message))
@@ -292,7 +293,7 @@ class FishAudioTTSService(InterruptibleTTSService):
         """
         logger.debug(f"{self}: Generating Fish TTS: [{text}]")
         try:
-            if not self._websocket or self._websocket.closed:
+            if not self._websocket or self._websocket.state is State.CLOSED:
                 await self._connect()
 
             if not self._request_id:

src/pipecat/services/gemini_multimodal_live/gemini.py

@@ -75,7 +75,7 @@ from . import events
 from .file_api import GeminiFileAPI
 
 try:
-    import websockets
+    from websockets.asyncio.client import connect as websocket_connect
 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]`.")
@@ -791,7 +791,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
         try:
             logger.info(f"Connecting to wss://{self._base_url}")
             uri = f"wss://{self._base_url}?key={self._api_key}"
-            self._websocket = await websockets.connect(uri=uri)
+            self._websocket = await websocket_connect(uri=uri)
             self._receive_task = self.create_task(self._receive_task_handler())
 
             # Create the basic configuration

src/pipecat/services/gladia/stt.py

@@ -37,6 +37,8 @@ from pipecat.utils.tracing.service_decorators import traced_stt
 
 try:
     import websockets
+    from websockets.asyncio.client import connect as websocket_connect
+    from websockets.protocol import State
 except ModuleNotFoundError as e:
     logger.error(f"Exception: {e}")
     logger.error("In order to use Gladia, you need to `pip install pipecat-ai[gladia]`.")
@@ -402,7 +404,7 @@ class GladiaSTTService(STTService):
                 logger.warning(f"Audio buffer exceeded max size, trimmed {trim_size} bytes")
 
         # Send audio if connected
-        if self._connection_active and self._websocket and not self._websocket.closed:
+        if self._connection_active and self._websocket and self._websocket.state is State.OPEN:
             try:
                 await self._send_audio(audio)
             except websockets.exceptions.ConnectionClosed as e:
@@ -423,7 +425,7 @@ class GladiaSTTService(STTService):
                     self._reconnection_attempts = 0
 
                 # Connect with automatic reconnection
-                async with websockets.connect(self._session_url) as websocket:
+                async with websocket_connect(self._session_url) as websocket:
                     try:
                         self._websocket = websocket
                         self._connection_active = True
@@ -507,7 +509,7 @@ class GladiaSTTService(STTService):
 
     async def _send_audio(self, audio: bytes):
         """Send audio chunk with proper message format."""
-        if self._websocket and not self._websocket.closed:
+        if self._websocket and self._websocket.state is State.OPEN:
             data = base64.b64encode(audio).decode("utf-8")
             message = {"type": "audio_chunk", "data": {"chunk": data}}
             await self._websocket.send(json.dumps(message))
@@ -520,7 +522,7 @@ class GladiaSTTService(STTService):
                 await self._send_audio(bytes(self._audio_buffer))
 
     async def _send_stop_recording(self):
-        if self._websocket and not self._websocket.closed:
+        if self._websocket and self._websocket.state is State.OPEN:
             await self._websocket.send(json.dumps({"type": "stop_recording"}))
 
     async def _keepalive_task_handler(self):
@@ -531,7 +533,7 @@ class GladiaSTTService(STTService):
                 self.reset_watchdog()
                 # Send keepalive (Gladia times out after 30 seconds)
                 await asyncio.sleep(KEEPALIVE_SLEEP)
-                if self._websocket and not self._websocket.closed:
+                if self._websocket and self._websocket.state is State.OPEN:
                     # Send an empty audio chunk as keepalive
                     empty_audio = b""
                     await self._send_audio(empty_audio)

src/pipecat/services/google/llm.py

@@ -10,29 +10,49 @@ This module provides Google Gemini integration for the Pipecat framework,
 including LLM services, context management, and message aggregation.
 """
 
+import base64
+import io
+import json
 import os
 import uuid
+from dataclasses import dataclass
 from typing import Any, Dict, List, Optional
 
 from loguru import logger
 from PIL import Image
 from pydantic import BaseModel, Field
 
-from pipecat.adapters.services.gemini_adapter import GeminiLLMAdapter, GeminiLLMInvocationParams
+from pipecat.adapters.services.gemini_adapter import GeminiLLMAdapter
 from pipecat.frames.frames import (
+    AudioRawFrame,
     Frame,
+    FunctionCallCancelFrame,
+    FunctionCallInProgressFrame,
+    FunctionCallResultFrame,
     LLMFullResponseEndFrame,
     LLMFullResponseStartFrame,
     LLMMessagesFrame,
     LLMTextFrame,
     LLMUpdateSettingsFrame,
+    UserImageRawFrame,
     VisionImageRawFrame,
 )
 from pipecat.metrics.metrics import LLMTokenUsage
-from pipecat.processors.aggregators.llm_context import LLMContext, LLMContextFrame
+from pipecat.processors.aggregators.llm_response import (
+    LLMAssistantAggregatorParams,
+    LLMUserAggregatorParams,
+)
+from pipecat.processors.aggregators.openai_llm_context import (
+    OpenAILLMContext,
+    OpenAILLMContextFrame,
+)
 from pipecat.processors.frame_processor import FrameDirection
 from pipecat.services.google.frames import LLMSearchResponseFrame
 from pipecat.services.llm_service import FunctionCallFromLLM, LLMService
+from pipecat.services.openai.llm import (
+    OpenAIAssistantContextAggregator,
+    OpenAIUserContextAggregator,
+)
 from pipecat.utils.asyncio.watchdog_async_iterator import WatchdogAsyncIterator
 from pipecat.utils.tracing.service_decorators import traced_llm
 
@@ -43,8 +63,13 @@ try:
     from google import genai
     from google.api_core.exceptions import DeadlineExceeded
     from google.genai.types import (
+        Blob,
+        Content,
+        FunctionCall,
+        FunctionResponse,
         GenerateContentConfig,
         HttpOptions,
+        Part,
     )
 except ModuleNotFoundError as e:
     logger.error(f"Exception: {e}")
@@ -52,12 +77,577 @@ except ModuleNotFoundError as e:
     raise Exception(f"Missing module: {e}")
 
 
+class GoogleUserContextAggregator(OpenAIUserContextAggregator):
+    """Google-specific user context aggregator.
+
+    Extends OpenAI user context aggregator to handle Google AI's specific
+    Content and Part message format for user messages.
+    """
+
+    async def push_aggregation(self):
+        """Push aggregated user text as a Google Content message."""
+        if len(self._aggregation) > 0:
+            self._context.add_message(Content(role="user", parts=[Part(text=self._aggregation)]))
+
+            # Reset the aggregation. Reset it before pushing it down, otherwise
+            # if the tasks gets cancelled we won't be able to clear things up.
+            self._aggregation = ""
+
+            # Push context frame
+            frame = OpenAILLMContextFrame(self._context)
+            await self.push_frame(frame)
+
+            # Reset our accumulator state.
+            await self.reset()
+
+
+class GoogleAssistantContextAggregator(OpenAIAssistantContextAggregator):
+    """Google-specific assistant context aggregator.
+
+    Extends OpenAI assistant context aggregator to handle Google AI's specific
+    Content and Part message format for assistant responses and function calls.
+    """
+
+    async def handle_aggregation(self, aggregation: str):
+        """Handle aggregated assistant text response.
+
+        Args:
+            aggregation: The aggregated text response from the assistant.
+        """
+        self._context.add_message(Content(role="model", parts=[Part(text=aggregation)]))
+
+    async def handle_function_call_in_progress(self, frame: FunctionCallInProgressFrame):
+        """Handle function call in progress frame.
+
+        Args:
+            frame: Frame containing function call details.
+        """
+        self._context.add_message(
+            Content(
+                role="model",
+                parts=[
+                    Part(
+                        function_call=FunctionCall(
+                            id=frame.tool_call_id, name=frame.function_name, args=frame.arguments
+                        )
+                    )
+                ],
+            )
+        )
+        self._context.add_message(
+            Content(
+                role="user",
+                parts=[
+                    Part(
+                        function_response=FunctionResponse(
+                            id=frame.tool_call_id,
+                            name=frame.function_name,
+                            response={"response": "IN_PROGRESS"},
+                        )
+                    )
+                ],
+            )
+        )
+
+    async def handle_function_call_result(self, frame: FunctionCallResultFrame):
+        """Handle function call result frame.
+
+        Args:
+            frame: Frame containing function call result.
+        """
+        if frame.result:
+            await self._update_function_call_result(
+                frame.function_name, frame.tool_call_id, frame.result
+            )
+        else:
+            await self._update_function_call_result(
+                frame.function_name, frame.tool_call_id, "COMPLETED"
+            )
+
+    async def handle_function_call_cancel(self, frame: FunctionCallCancelFrame):
+        """Handle function call cancellation frame.
+
+        Args:
+            frame: Frame containing function call cancellation details.
+        """
+        await self._update_function_call_result(
+            frame.function_name, frame.tool_call_id, "CANCELLED"
+        )
+
+    async def _update_function_call_result(
+        self, function_name: str, tool_call_id: str, result: Any
+    ):
+        for message in self._context.messages:
+            if message.role == "user":
+                for part in message.parts:
+                    if part.function_response and part.function_response.id == tool_call_id:
+                        part.function_response.response = {"value": json.dumps(result)}
+
+    async def handle_user_image_frame(self, frame: UserImageRawFrame):
+        """Handle user image frame.
+
+        Args:
+            frame: Frame containing user image data and request context.
+        """
+        await self._update_function_call_result(
+            frame.request.function_name, frame.request.tool_call_id, "COMPLETED"
+        )
+        self._context.add_image_frame_message(
+            format=frame.format,
+            size=frame.size,
+            image=frame.image,
+            text=frame.request.context,
+        )
+
+
+@dataclass
+class GoogleContextAggregatorPair:
+    """Pair of Google context aggregators for user and assistant messages.
+
+    Parameters:
+        _user: User context aggregator for handling user messages.
+        _assistant: Assistant context aggregator for handling assistant responses.
+    """
+
+    _user: GoogleUserContextAggregator
+    _assistant: GoogleAssistantContextAggregator
+
+    def user(self) -> GoogleUserContextAggregator:
+        """Get the user context aggregator.
+
+        Returns:
+            The user context aggregator instance.
+        """
+        return self._user
+
+    def assistant(self) -> GoogleAssistantContextAggregator:
+        """Get the assistant context aggregator.
+
+        Returns:
+            The assistant context aggregator instance.
+        """
+        return self._assistant
+
+
+class GoogleLLMContext(OpenAILLMContext):
+    """Google AI LLM context that extends OpenAI context for Google-specific formatting.
+
+    This class handles conversion between OpenAI-style messages and Google AI's
+    Content/Part format, including system messages, function calls, and media.
+    """
+
+    def __init__(
+        self,
+        messages: Optional[List[dict]] = None,
+        tools: Optional[List[dict]] = None,
+        tool_choice: Optional[dict] = None,
+    ):
+        """Initialize GoogleLLMContext.
+
+        Args:
+            messages: Initial messages in OpenAI format.
+            tools: Available tools/functions for the model.
+            tool_choice: Tool choice configuration.
+        """
+        super().__init__(messages=messages, tools=tools, tool_choice=tool_choice)
+        self.system_message = None
+
+    @staticmethod
+    def upgrade_to_google(obj: OpenAILLMContext) -> "GoogleLLMContext":
+        """Upgrade an OpenAI context to a Google context.
+
+        Args:
+            obj: OpenAI LLM context to upgrade.
+
+        Returns:
+            GoogleLLMContext instance with converted messages.
+        """
+        if isinstance(obj, OpenAILLMContext) and not isinstance(obj, GoogleLLMContext):
+            logger.debug(f"Upgrading to Google: {obj}")
+            obj.__class__ = GoogleLLMContext
+            obj._restructure_from_openai_messages()
+        return obj
+
+    def set_messages(self, messages: List):
+        """Set messages and restructure them for Google format.
+
+        Args:
+            messages: List of messages to set.
+        """
+        self._messages[:] = messages
+        self._restructure_from_openai_messages()
+
+    def add_messages(self, messages: List):
+        """Add messages to the context, converting to Google format as needed.
+
+        Args:
+            messages: List of messages to add (can be mixed formats).
+        """
+        # Convert each message individually
+        converted_messages = []
+        for msg in messages:
+            if isinstance(msg, Content):
+                # Already in Gemini format
+                converted_messages.append(msg)
+            else:
+                # Convert from standard format to Gemini format
+                converted = self.from_standard_message(msg)
+                if converted is not None:
+                    converted_messages.append(converted)
+
+        # Add the converted messages to our existing messages
+        self._messages.extend(converted_messages)
+
+    def get_messages_for_logging(self):
+        """Get messages formatted for logging with sensitive data redacted.
+
+        Returns:
+            List of message dictionaries with inline data redacted.
+        """
+        msgs = []
+        for message in self.messages:
+            obj = message.to_json_dict()
+            try:
+                if "parts" in obj:
+                    for part in obj["parts"]:
+                        if "inline_data" in part:
+                            part["inline_data"]["data"] = "..."
+            except Exception as e:
+                logger.debug(f"Error: {e}")
+            msgs.append(obj)
+        return msgs
+
+    def add_image_frame_message(
+        self, *, format: str, size: tuple[int, int], image: bytes, text: str = None
+    ):
+        """Add an image message to the context.
+
+        Args:
+            format: Image format (e.g., 'RGB', 'RGBA').
+            size: Image dimensions as (width, height).
+            image: Raw image bytes.
+            text: Optional text to accompany the image.
+        """
+        buffer = io.BytesIO()
+        Image.frombytes(format, size, image).save(buffer, format="JPEG")
+
+        parts = []
+        if text:
+            parts.append(Part(text=text))
+        parts.append(Part(inline_data=Blob(mime_type="image/jpeg", data=buffer.getvalue())))
+
+        self.add_message(Content(role="user", parts=parts))
+
+    def add_audio_frames_message(
+        self, *, audio_frames: list[AudioRawFrame], text: str = "Audio follows"
+    ):
+        """Add audio frames as a message to the context.
+
+        Args:
+            audio_frames: List of audio frames to add.
+            text: Text description of the audio content.
+        """
+        if not audio_frames:
+            return
+
+        sample_rate = audio_frames[0].sample_rate
+        num_channels = audio_frames[0].num_channels
+
+        parts = []
+        data = b"".join(frame.audio for frame in audio_frames)
+        # NOTE(aleix): According to the docs only text or inline_data should be needed.
+        # (see https://cloud.google.com/vertex-ai/generative-ai/docs/model-reference/inference)
+        parts.append(Part(text=text))
+        parts.append(
+            Part(
+                inline_data=Blob(
+                    mime_type="audio/wav",
+                    data=(
+                        bytes(
+                            self.create_wav_header(sample_rate, num_channels, 16, len(data)) + data
+                        )
+                    ),
+                )
+            ),
+        )
+        self.add_message(Content(role="user", parts=parts))
+        # message = {"mime_type": "audio/mp3", "data": bytes(data + create_wav_header(sample_rate, num_channels, 16, len(data)))}
+        # self.add_message(message)
+
+    def from_standard_message(self, message):
+        """Convert standard format message to Google Content object.
+
+        Handles conversion of text, images, and function calls to Google's format.
+        System messages are stored separately and return None.
+
+        Args:
+            message: Message in standard format.
+
+        Returns:
+            Content object with role and parts, or None for system messages.
+
+        Examples:
+            Standard text message::
+
+                {
+                    "role": "user",
+                    "content": "Hello there"
+                }
+
+            Converts to Google Content with::
+
+                Content(
+                    role="user",
+                    parts=[Part(text="Hello there")]
+                )
+
+            Standard function call message::
+
+                {
+                    "role": "assistant",
+                    "tool_calls": [
+                        {
+                            "function": {
+                                "name": "search",
+                                "arguments": '{"query": "test"}'
+                            }
+                        }
+                    ]
+                }
+
+            Converts to Google Content with::
+
+                Content(
+                    role="model",
+                    parts=[Part(function_call=FunctionCall(name="search", args={"query": "test"}))]
+                )
+
+            System message returns None and stores content in self.system_message.
+        """
+        role = message["role"]
+        content = message.get("content", [])
+        if role == "system":
+            self.system_message = content
+            return None
+        elif role == "assistant":
+            role = "model"
+
+        parts = []
+        if message.get("tool_calls"):
+            for tc in message["tool_calls"]:
+                parts.append(
+                    Part(
+                        function_call=FunctionCall(
+                            name=tc["function"]["name"],
+                            args=json.loads(tc["function"]["arguments"]),
+                        )
+                    )
+                )
+        elif role == "tool":
+            role = "model"
+            parts.append(
+                Part(
+                    function_response=FunctionResponse(
+                        name="tool_call_result",  # seems to work to hard-code the same name every time
+                        response=json.loads(message["content"]),
+                    )
+                )
+            )
+        elif isinstance(content, str):
+            parts.append(Part(text=content))
+        elif isinstance(content, list):
+            for c in content:
+                if c["type"] == "text":
+                    parts.append(Part(text=c["text"]))
+                elif c["type"] == "image_url":
+                    parts.append(
+                        Part(
+                            inline_data=Blob(
+                                mime_type="image/jpeg",
+                                data=base64.b64decode(c["image_url"]["url"].split(",")[1]),
+                            )
+                        )
+                    )
+
+        message = Content(role=role, parts=parts)
+        return message
+
+    def to_standard_messages(self, obj) -> list:
+        """Convert Google Content object to standard structured format.
+
+        Handles text, images, and function calls from Google's Content/Part objects.
+
+        Args:
+            obj: Google Content object with role and parts.
+
+        Returns:
+            List containing a single message in standard format.
+
+        Examples:
+            Google Content with text::
+
+                Content(
+                    role="user",
+                    parts=[Part(text="Hello")]
+                )
+
+            Converts to::
+
+                [
+                    {
+                        "role": "user",
+                        "content": [{"type": "text", "text": "Hello"}]
+                    }
+                ]
+
+            Google Content with function call::
+
+                Content(
+                    role="model",
+                    parts=[Part(function_call=FunctionCall(name="search", args={"q": "test"}))]
+                )
+
+            Converts to::
+
+                [
+                    {
+                        "role": "assistant",
+                        "tool_calls": [
+                            {
+                                "id": "search",
+                                "type": "function",
+                                "function": {
+                                    "name": "search",
+                                    "arguments": '{"q": "test"}'
+                                }
+                            }
+                        ]
+                    }
+                ]
+
+            Google Content with image::
+
+                Content(
+                    role="user",
+                    parts=[Part(inline_data=Blob(mime_type="image/jpeg", data=bytes_data))]
+                )
+
+            Converts to::
+
+                [
+                    {
+                        "role": "user",
+                        "content": [
+                            {
+                                "type": "image_url",
+                                "image_url": {"url": "data:image/jpeg;base64,<encoded_data>"}
+                            }
+                        ]
+                    }
+                ]
+        """
+        msg = {"role": obj.role, "content": []}
+        if msg["role"] == "model":
+            msg["role"] = "assistant"
+
+        for part in obj.parts:
+            if part.text:
+                msg["content"].append({"type": "text", "text": part.text})
+            elif part.inline_data:
+                encoded = base64.b64encode(part.inline_data.data).decode("utf-8")
+                msg["content"].append(
+                    {
+                        "type": "image_url",
+                        "image_url": {"url": f"data:{part.inline_data.mime_type};base64,{encoded}"},
+                    }
+                )
+            elif part.function_call:
+                args = part.function_call.args if hasattr(part.function_call, "args") else {}
+                msg["tool_calls"] = [
+                    {
+                        "id": part.function_call.name,
+                        "type": "function",
+                        "function": {
+                            "name": part.function_call.name,
+                            "arguments": json.dumps(args),
+                        },
+                    }
+                ]
+
+            elif part.function_response:
+                msg["role"] = "tool"
+                resp = (
+                    part.function_response.response
+                    if hasattr(part.function_response, "response")
+                    else {}
+                )
+                msg["tool_call_id"] = part.function_response.name
+                msg["content"] = json.dumps(resp)
+
+        # there might be no content parts for tool_calls messages
+        if not msg["content"]:
+            del msg["content"]
+        return [msg]
+
+    def _restructure_from_openai_messages(self):
+        """Restructures messages to ensure proper Google format and message ordering.
+
+        This method handles conversion of OpenAI-formatted messages to Google format,
+        with special handling for function calls, function responses, and system messages.
+        System messages are added back to the context as user messages when needed.
+
+        The final message order is preserved as:
+        1. Function calls (from model)
+        2. Function responses (from user)
+        3. Text messages (converted from system messages)
+
+        Note:
+            System messages are only added back when there are no regular text
+            messages in the context, ensuring proper conversation continuity
+            after function calls.
+        """
+        self.system_message = None
+        converted_messages = []
+
+        # Process each message, preserving Google-formatted messages and converting others
+        for message in self._messages:
+            if isinstance(message, Content):
+                # Keep existing Google-formatted messages (e.g., function calls/responses)
+                converted_messages.append(message)
+                continue
+
+            # Convert OpenAI format to Google format, system messages return None
+            converted = self.from_standard_message(message)
+            if converted is not None:
+                converted_messages.append(converted)
+
+        # Update message list
+        self._messages[:] = converted_messages
+
+        # Check if we only have function-related messages (no regular text)
+        has_regular_messages = any(
+            len(msg.parts) == 1
+            and getattr(msg.parts[0], "text", None)
+            and not getattr(msg.parts[0], "function_call", None)
+            and not getattr(msg.parts[0], "function_response", None)
+            for msg in self._messages
+        )
+
+        # Add system message back as a user message if we only have function messages
+        if self.system_message and not has_regular_messages:
+            self._messages.append(Content(role="user", parts=[Part(text=self.system_message)]))
+
+        # Remove any empty messages
+        self._messages = [m for m in self._messages if m.parts]
+
+
 class GoogleLLMService(LLMService):
     """Google AI (Gemini) LLM service implementation.
 
     This class implements inference with Google's AI models, translating internally
-    from the universal LLMContext to the message format expected by the Google
-    AI model.
+    from OpenAILLMContext to the messages format expected by the Google AI model.
+    We use OpenAILLMContext as a lingua franca for all LLM services to enable
+    easy switching between different LLMs.
     """
 
     # Overriding the default adapter to use the Gemini one.
@@ -160,7 +750,7 @@ class GoogleLLMService(LLMService):
             logger.exception(f"Failed to unset thinking budget: {e}")
 
     @traced_llm
-    async def _process_context(self, context: LLMContext):
+    async def _process_context(self, context: OpenAILLMContext):
         await self.push_frame(LLMFullResponseStartFrame())
 
         prompt_tokens = 0
@@ -173,31 +763,19 @@ class GoogleLLMService(LLMService):
         search_result = ""
 
         try:
-            adapter = self.get_llm_adapter()
-            llm_invocation_params: GeminiLLMInvocationParams = adapter.get_llm_invocation_params(
-                context
-            )
-
             logger.debug(
-                # TODO: figure out a nice way to also log system instruction
-                # f"{self}: Generating chat [{self._system_instruction}] | [{adapter.get_messages_for_logging(context)}]"
-                f"{self}: Generating chat [{adapter.get_messages_for_logging(context)}]"
+                # f"{self}: Generating chat [{self._system_instruction}] | [{context.get_messages_for_logging()}]"
+                f"{self}: Generating chat [{context.get_messages_for_logging()}]"
             )
 
-            messages = llm_invocation_params["messages"]
-            if (
-                llm_invocation_params.get("system_instruction")
-                and self._system_instruction != llm_invocation_params["system_instruction"]
-            ):
-                logger.debug(
-                    f"System instruction changed: {llm_invocation_params['system_instruction']}"
-                )
-                self._system_instruction = llm_invocation_params["system_instruction"]
+            messages = context.messages
+            if context.system_message and self._system_instruction != context.system_message:
+                logger.debug(f"System instruction changed: {context.system_message}")
+                self._system_instruction = context.system_message
 
-            # TODO: test what happens when there are no tools
             tools = []
-            if llm_invocation_params.get("tools"):
-                tools = llm_invocation_params["tools"]
+            if context.tools:
+                tools = context.tools
             elif self._tools:
                 tools = self._tools
             tool_config = None
@@ -344,12 +922,12 @@ class GoogleLLMService(LLMService):
 
         context = None
 
-        if isinstance(frame, LLMContextFrame):
-            context = frame.context
+        if isinstance(frame, OpenAILLMContextFrame):
+            context = GoogleLLMContext.upgrade_to_google(frame.context)
         elif isinstance(frame, LLMMessagesFrame):
-            context = LLMContext(messages=frame.messages)
+            context = GoogleLLMContext(frame.messages)
         elif isinstance(frame, VisionImageRawFrame):
-            context = LLMContext()
+            context = GoogleLLMContext()
             context.add_image_frame_message(
                 format=frame.format, size=frame.size, image=frame.image, text=frame.text
             )
@@ -360,3 +938,34 @@ class GoogleLLMService(LLMService):
 
         if context:
             await self._process_context(context)
+
+    def create_context_aggregator(
+        self,
+        context: OpenAILLMContext,
+        *,
+        user_params: LLMUserAggregatorParams = LLMUserAggregatorParams(),
+        assistant_params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(),
+    ) -> GoogleContextAggregatorPair:
+        """Create Google-specific context aggregators.
+
+        Creates a pair of context aggregators optimized for Google's message format,
+        including support for function calls, tool usage, and image handling.
+
+        Args:
+            context: The LLM context to create aggregators for.
+            user_params: Parameters for user message aggregation.
+            assistant_params: Parameters for assistant message aggregation.
+
+        Returns:
+            GoogleContextAggregatorPair: A pair of context aggregators, one for
+            the user and one for the assistant, encapsulated in an
+            GoogleContextAggregatorPair.
+
+        """
+        context.set_llm_adapter(self.get_llm_adapter())
+
+        if isinstance(context, OpenAILLMContext):
+            context = GoogleLLMContext.upgrade_to_google(context)
+        user = GoogleUserContextAggregator(context, params=user_params)
+        assistant = GoogleAssistantContextAggregator(context, params=assistant_params)
+        return GoogleContextAggregatorPair(_user=user, _assistant=assistant)

src/pipecat/services/llm_service.py

@@ -41,7 +41,6 @@ from pipecat.frames.frames import (
     StartInterruptionFrame,
     UserImageRequestFrame,
 )
-from pipecat.processors.aggregators.llm_context import LLMContext
 from pipecat.processors.aggregators.llm_response import (
     LLMAssistantAggregatorParams,
     LLMUserAggregatorParams,
@@ -90,8 +89,7 @@ class FunctionCallParams:
     tool_call_id: str
     arguments: Mapping[str, Any]
     llm: "LLMService"
-    # TODO: after migration of all services to universal LLMContext, OpenAILLMContext can be removed
-    context: LLMContext | OpenAILLMContext
+    context: OpenAILLMContext
     result_callback: FunctionCallResultCallback
 
 
@@ -420,10 +418,7 @@ class LLMService(AIService):
             else:
                 await self._sequential_runner_queue.put(runner_item)
 
-    # TODO: after migration of all services to universal LLMContext, OpenAILLMContext can be removed
-    async def _call_start_function(
-        self, context: LLMContext | OpenAILLMContext, function_name: str
-    ):
+    async def _call_start_function(self, context: OpenAILLMContext, function_name: str):
         if function_name in self._start_callbacks.keys():
             await self._start_callbacks[function_name](function_name, self, context)
         elif None in self._start_callbacks.keys():

src/pipecat/services/lmnt/tts.py

@@ -29,7 +29,8 @@ from pipecat.utils.tracing.service_decorators import traced_tts
 
 # See .env.example for LMNT configuration needed
 try:
-    import websockets
+    from websockets.asyncio.client import connect as websocket_connect
+    from websockets.protocol import State
 except ModuleNotFoundError as e:
     logger.error(f"Exception: {e}")
     logger.error("In order to use LMNT, you need to `pip install pipecat-ai[lmnt]`.")
@@ -200,7 +201,7 @@ class LmntTTSService(InterruptibleTTSService):
     async def _connect_websocket(self):
         """Connect to LMNT websocket."""
         try:
-            if self._websocket and self._websocket.open:
+            if self._websocket and self._websocket.state is State.OPEN:
                 return
 
             logger.debug("Connecting to LMNT")
@@ -216,7 +217,7 @@ class LmntTTSService(InterruptibleTTSService):
             }
 
             # Connect to LMNT's websocket directly
-            self._websocket = await websockets.connect("wss://api.lmnt.com/v1/ai/speech/stream")
+            self._websocket = await websocket_connect("wss://api.lmnt.com/v1/ai/speech/stream")
 
             # Send initialization message
             await self._websocket.send(json.dumps(init_msg))
@@ -251,7 +252,7 @@ class LmntTTSService(InterruptibleTTSService):
 
     async def flush_audio(self):
         """Flush any pending audio synthesis."""
-        if not self._websocket or self._websocket.closed:
+        if not self._websocket or self._websocket.state is State.CLOSED:
             return
         await self._get_websocket().send(json.dumps({"flush": True}))
 
@@ -292,7 +293,7 @@ class LmntTTSService(InterruptibleTTSService):
         logger.debug(f"{self}: Generating TTS [{text}]")
 
         try:
-            if not self._websocket or self._websocket.closed:
+            if not self._websocket or self._websocket.state is State.CLOSED:
                 await self._connect()
 
             try:

src/pipecat/services/neuphonic/tts.py

@@ -15,6 +15,7 @@ import base64
 import json
 from typing import Any, AsyncGenerator, Mapping, Optional
 
+import aiohttp
 from loguru import logger
 from pydantic import BaseModel
 
@@ -39,8 +40,8 @@ from pipecat.utils.asyncio.watchdog_async_iterator import WatchdogAsyncIterator
 from pipecat.utils.tracing.service_decorators import traced_tts
 
 try:
-    import websockets
-    from pyneuphonic import Neuphonic, TTSConfig
+    from websockets.asyncio.client import connect as websocket_connect
+    from websockets.protocol import State
 except ModuleNotFoundError as e:
     logger.error(f"Exception: {e}")
     logger.error("In order to use Neuphonic, you need to `pip install pipecat-ai[neuphonic]`.")
@@ -271,7 +272,7 @@ class NeuphonicTTSService(InterruptibleTTSService):
     async def _connect_websocket(self):
         """Establish WebSocket connection to Neuphonic API."""
         try:
-            if self._websocket and self._websocket.open:
+            if self._websocket and self._websocket.state is State.OPEN:
                 return
 
             logger.debug("Connecting to Neuphonic")
@@ -292,7 +293,7 @@ class NeuphonicTTSService(InterruptibleTTSService):
 
             headers = {"x-api-key": self._api_key}
 
-            self._websocket = await websockets.connect(url, extra_headers=headers)
+            self._websocket = await websocket_connect(url, additional_headers=headers)
         except Exception as e:
             logger.error(f"{self} initialization error: {e}")
             self._websocket = None
@@ -359,7 +360,7 @@ class NeuphonicTTSService(InterruptibleTTSService):
         logger.debug(f"Generating TTS: [{text}]")
 
         try:
-            if not self._websocket or self._websocket.closed:
+            if not self._websocket or self._websocket.state is State.CLOSED:
                 await self._connect()
 
             try:
@@ -406,9 +407,10 @@ class NeuphonicHttpTTSService(TTSService):
         *,
         api_key: str,
         voice_id: Optional[str] = None,
+        aiohttp_session: aiohttp.ClientSession,
         url: str = "https://api.neuphonic.com",
         sample_rate: Optional[int] = 22050,
-        encoding: str = "pcm_linear",
+        encoding: Optional[str] = "pcm_linear",
         params: Optional[InputParams] = None,
         **kwargs,
     ):
@@ -417,6 +419,7 @@ class NeuphonicHttpTTSService(TTSService):
         Args:
             api_key: Neuphonic API key for authentication.
             voice_id: ID of the voice to use for synthesis.
+            aiohttp_session: Shared aiohttp session for HTTP requests.
             url: Base URL for the Neuphonic HTTP API.
             sample_rate: Audio sample rate in Hz. Defaults to 22050.
             encoding: Audio encoding format. Defaults to "pcm_linear".
@@ -428,13 +431,11 @@ class NeuphonicHttpTTSService(TTSService):
         params = params or NeuphonicHttpTTSService.InputParams()
 
         self._api_key = api_key
-        self._url = url
-        self._settings = {
-            "lang_code": self.language_to_service_language(params.language),
-            "speed": params.speed,
-            "encoding": encoding,
-            "sampling_rate": sample_rate,
-        }
+        self._session = aiohttp_session
+        self._base_url = url.rstrip("/")
+        self._lang_code = self.language_to_service_language(params.language) or "en"
+        self._speed = params.speed
+        self._encoding = encoding
         self.set_voice(voice_id)
 
     def can_generate_metrics(self) -> bool:
@@ -472,6 +473,40 @@ class NeuphonicHttpTTSService(TTSService):
         """
         pass
 
+    def _parse_sse_message(self, message: str) -> dict | None:
+        """Parse a Server-Sent Event message.
+
+        Args:
+            message: The SSE message to parse.
+
+        Returns:
+            Parsed message dictionary or None if not a data message.
+        """
+        message = message.strip()
+
+        if not message or "data" not in message:
+            return None
+
+        try:
+            # Split on ": " and take the part after "data: "
+            _, data_content = message.split(": ", 1)
+
+            if not data_content or data_content == "[DONE]":
+                return None
+
+            message_dict = json.loads(data_content)
+
+            # Check for errors in the response
+            if message_dict.get("errors") is not None:
+                raise Exception(
+                    f"Neuphonic API error {message_dict.get('status_code', 'unknown')}: {message_dict['errors']}"
+                )
+
+            return message_dict
+        except (ValueError, json.JSONDecodeError) as e:
+            logger.warning(f"Failed to parse SSE message: {e}")
+            return None
+
     @traced_tts
     async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
         """Generate speech from text using Neuphonic streaming API.
@@ -484,26 +519,71 @@ class NeuphonicHttpTTSService(TTSService):
         """
         logger.debug(f"Generating TTS: [{text}]")
 
-        client = Neuphonic(api_key=self._api_key, base_url=self._url.replace("https://", ""))
+        url = f"{self._base_url}/sse/speak/{self._lang_code}"
 
-        sse = client.tts.AsyncSSEClient()
+        headers = {
+            "X-API-KEY": self._api_key,
+            "Content-Type": "application/json",
+        }
+
+        payload = {
+            "text": text,
+            "lang_code": self._lang_code,
+            "encoding": self._encoding,
+            "sampling_rate": self.sample_rate,
+            "speed": self._speed,
+        }
+
+        if self._voice_id:
+            payload["voice_id"] = self._voice_id
 
         try:
             await self.start_ttfb_metrics()
-            response = sse.send(text, TTSConfig(**self._settings, voice_id=self._voice_id))
 
-            await self.start_tts_usage_metrics(text)
-            yield TTSStartedFrame()
+            async with self._session.post(url, json=payload, headers=headers) as response:
+                if response.status != 200:
+                    error_text = await response.text()
+                    error_message = f"Neuphonic API error: HTTP {response.status} - {error_text}"
+                    logger.error(error_message)
+                    yield ErrorFrame(error=error_message)
+                    return
 
-            async for message in response:
-                if message.status_code != 200:
-                    logger.error(f"{self} error: {message.errors}")
-                    yield ErrorFrame(error=f"Neuphonic API error: {message.errors}")
+                await self.start_tts_usage_metrics(text)
+                yield TTSStartedFrame()
 
-                await self.stop_ttfb_metrics()
-                yield TTSAudioRawFrame(message.data.audio, self.sample_rate, 1)
+                # Process SSE stream line by line
+                async for line in response.content:
+                    if not line:
+                        continue
+
+                    message = line.decode("utf-8", errors="ignore")
+                    if not message.strip():
+                        continue
+
+                    try:
+                        parsed_message = self._parse_sse_message(message)
+
+                        if (
+                            parsed_message is not None
+                            and parsed_message.get("data", {}).get("audio") is not None
+                        ):
+                            audio_b64 = parsed_message["data"]["audio"]
+                            audio_bytes = base64.b64decode(audio_b64)
+
+                            await self.stop_ttfb_metrics()
+                            yield TTSAudioRawFrame(audio_bytes, self.sample_rate, 1)
+
+                    except Exception as e:
+                        logger.error(f"Error processing SSE message: {e}")
+                        # Don't yield error frame for individual message failures
+                        continue
+
+        except asyncio.CancelledError:
+            logger.debug("TTS generation cancelled")
+            raise
         except Exception as e:
-            logger.error(f"Error in run_tts: {e}")
-            yield ErrorFrame(error=str(e))
+            logger.exception(f"Error in run_tts: {e}")
+            yield ErrorFrame(error=f"Neuphonic TTS error: {str(e)}")
         finally:
+            await self.stop_ttfb_metrics()
             yield TTSStoppedFrame()

src/pipecat/services/openai/base_llm.py

@@ -4,7 +4,7 @@
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
-"""Base LLM service implementation for services that use the AsyncOpenAI client."""
+"""Base OpenAI LLM service implementation."""
 
 import base64
 import json
@@ -21,7 +21,6 @@ from openai import (
 from openai.types.chat import ChatCompletionChunk, ChatCompletionMessageParam
 from pydantic import BaseModel, Field
 
-from pipecat.adapters.services.open_ai_adapter import OpenAILLMInvocationParams
 from pipecat.frames.frames import (
     Frame,
     LLMFullResponseEndFrame,
@@ -32,9 +31,9 @@ from pipecat.frames.frames import (
     VisionImageRawFrame,
 )
 from pipecat.metrics.metrics import LLMTokenUsage
-from pipecat.processors.aggregators.llm_context import (
-    LLMContext,
-    LLMContextFrame,
+from pipecat.processors.aggregators.openai_llm_context import (
+    OpenAILLMContext,
+    OpenAILLMContextFrame,
 )
 from pipecat.processors.frame_processor import FrameDirection
 from pipecat.services.llm_service import FunctionCallFromLLM, LLMService
@@ -45,8 +44,8 @@ from pipecat.utils.tracing.service_decorators import traced_llm
 class BaseOpenAILLMService(LLMService):
     """Base class for all services that use the AsyncOpenAI client.
 
-    This service consumes LLMContextFrame frames, which contain a reference to
-    an LLMContext object. The context defines what is sent to the LLM for
+    This service consumes OpenAILLMContextFrame frames, which contain a reference
+    to an OpenAILLMContext object. The context defines what is sent to the LLM for
     completion, including user, assistant, and system messages, as well as tool
     choices and function call configurations.
     """
@@ -174,13 +173,13 @@ class BaseOpenAILLMService(LLMService):
         return True
 
     async def get_chat_completions(
-        self, params_from_context: OpenAILLMInvocationParams
+        self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam]
     ) -> AsyncStream[ChatCompletionChunk]:
         """Get streaming chat completions from OpenAI API.
 
         Args:
-            params_from_context: Parameters, derived from the LLM context, to
-                use for the chat completion. Contains messages, tools, and tool choice.
+            context: The LLM context containing tools and configuration.
+            messages: List of chat completion messages to send.
 
         Returns:
             Async stream of chat completion chunks.
@@ -188,6 +187,9 @@ class BaseOpenAILLMService(LLMService):
         params = {
             "model": self.model_name,
             "stream": True,
+            "messages": messages,
+            "tools": context.tools,
+            "tool_choice": context.tool_choice,
             "stream_options": {"include_usage": True},
             "frequency_penalty": self._settings["frequency_penalty"],
             "presence_penalty": self._settings["presence_penalty"],
@@ -198,28 +200,39 @@ class BaseOpenAILLMService(LLMService):
             "max_completion_tokens": self._settings["max_completion_tokens"],
         }
 
-        # Messages, tools, tool_choice
-        params.update(params_from_context)
-
         params.update(self._settings["extra"])
 
         chunks = await self._client.chat.completions.create(**params)
         return chunks
 
     async def _stream_chat_completions(
-        self, context: LLMContext
+        self, context: OpenAILLMContext
     ) -> AsyncStream[ChatCompletionChunk]:
-        adapter = self.get_llm_adapter()
-        logger.debug(f"{self}: Generating chat [{adapter.get_messages_for_logging(context)}]")
+        logger.debug(f"{self}: Generating chat [{context.get_messages_for_logging()}]")
 
-        params: OpenAILLMInvocationParams = adapter.get_llm_invocation_params(context)
+        messages: List[ChatCompletionMessageParam] = context.get_messages()
 
-        chunks = await self.get_chat_completions(params)
+        # base64 encode any images
+        for message in messages:
+            if message.get("mime_type") == "image/jpeg":
+                encoded_image = base64.b64encode(message["data"].getvalue()).decode("utf-8")
+                text = message["content"]
+                message["content"] = [
+                    {"type": "text", "text": text},
+                    {
+                        "type": "image_url",
+                        "image_url": {"url": f"data:image/jpeg;base64,{encoded_image}"},
+                    },
+                ]
+                del message["data"]
+                del message["mime_type"]
+
+        chunks = await self.get_chat_completions(context, messages)
 
         return chunks
 
     @traced_llm
-    async def _process_context(self, context: LLMContext):
+    async def _process_context(self, context: OpenAILLMContext):
         functions_list = []
         arguments_list = []
         tool_id_list = []
@@ -318,7 +331,7 @@ class BaseOpenAILLMService(LLMService):
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         """Process frames for LLM completion requests.
 
-        Handles LLMContextFrame, LLMMessagesFrame, VisionImageRawFrame,
+        Handles OpenAILLMContextFrame, LLMMessagesFrame, VisionImageRawFrame,
         and LLMUpdateSettingsFrame to trigger LLM completions and manage settings.
 
         Args:
@@ -328,12 +341,12 @@ class BaseOpenAILLMService(LLMService):
         await super().process_frame(frame, direction)
 
         context = None
-        if isinstance(frame, LLMContextFrame):
-            context: LLMContext = frame.context
+        if isinstance(frame, OpenAILLMContextFrame):
+            context: OpenAILLMContext = frame.context
         elif isinstance(frame, LLMMessagesFrame):
-            context = LLMContext(messages=frame.messages)
+            context = OpenAILLMContext.from_messages(frame.messages)
         elif isinstance(frame, VisionImageRawFrame):
-            context = LLMContext()
+            context = OpenAILLMContext()
             context.add_image_frame_message(
                 format=frame.format, size=frame.size, image=frame.image, text=frame.text
             )

src/pipecat/services/openai/llm.py

@@ -16,16 +16,45 @@ from pipecat.frames.frames import (
     FunctionCallResultFrame,
     UserImageRawFrame,
 )
-from pipecat.processors.aggregators.llm_context import LLMContext
-from pipecat.processors.aggregators.llm_response_universal import (
-    LLMAssistantContextAggregator_Universal,
-    LLMAssistantContextAggregatorParams,
-    LLMUserContextAggregator_Universal,
-    LLMUserContextAggregatorParams,
+from pipecat.processors.aggregators.llm_response import (
+    LLMAssistantAggregatorParams,
+    LLMAssistantContextAggregator,
+    LLMUserAggregatorParams,
+    LLMUserContextAggregator,
 )
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
 from pipecat.services.openai.base_llm import BaseOpenAILLMService
 
 
+@dataclass
+class OpenAIContextAggregatorPair:
+    """Pair of OpenAI context aggregators for user and assistant messages.
+
+    Parameters:
+        _user: User context aggregator for processing user messages.
+        _assistant: Assistant context aggregator for processing assistant messages.
+    """
+
+    _user: "OpenAIUserContextAggregator"
+    _assistant: "OpenAIAssistantContextAggregator"
+
+    def user(self) -> "OpenAIUserContextAggregator":
+        """Get the user context aggregator.
+
+        Returns:
+            The user context aggregator instance.
+        """
+        return self._user
+
+    def assistant(self) -> "OpenAIAssistantContextAggregator":
+        """Get the assistant context aggregator.
+
+        Returns:
+            The assistant context aggregator instance.
+        """
+        return self._assistant
+
+
 class OpenAILLMService(BaseOpenAILLMService):
     """OpenAI LLM service implementation.
 
@@ -49,3 +78,141 @@ class OpenAILLMService(BaseOpenAILLMService):
             **kwargs: Additional arguments passed to the parent BaseOpenAILLMService.
         """
         super().__init__(model=model, params=params, **kwargs)
+
+    def create_context_aggregator(
+        self,
+        context: OpenAILLMContext,
+        *,
+        user_params: LLMUserAggregatorParams = LLMUserAggregatorParams(),
+        assistant_params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(),
+    ) -> OpenAIContextAggregatorPair:
+        """Create OpenAI-specific context aggregators.
+
+        Creates a pair of context aggregators optimized for OpenAI's message format,
+        including support for function calls, tool usage, and image handling.
+
+        Args:
+            context: The LLM context to create aggregators for.
+            user_params: Parameters for user message aggregation.
+            assistant_params: Parameters for assistant message aggregation.
+
+        Returns:
+            OpenAIContextAggregatorPair: A pair of context aggregators, one for
+            the user and one for the assistant, encapsulated in an
+            OpenAIContextAggregatorPair.
+
+        """
+        context.set_llm_adapter(self.get_llm_adapter())
+        user = OpenAIUserContextAggregator(context, params=user_params)
+        assistant = OpenAIAssistantContextAggregator(context, params=assistant_params)
+        return OpenAIContextAggregatorPair(_user=user, _assistant=assistant)
+
+
+class OpenAIUserContextAggregator(LLMUserContextAggregator):
+    """OpenAI-specific user context aggregator.
+
+    Handles aggregation of user messages for OpenAI LLM services.
+    Inherits all functionality from the base LLMUserContextAggregator.
+    """
+
+    pass
+
+
+class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator):
+    """OpenAI-specific assistant context aggregator.
+
+    Handles aggregation of assistant messages for OpenAI LLM services,
+    with specialized support for OpenAI's function calling format,
+    tool usage tracking, and image message handling.
+    """
+
+    async def handle_function_call_in_progress(self, frame: FunctionCallInProgressFrame):
+        """Handle a function call in progress.
+
+        Adds the function call to the context with an IN_PROGRESS status
+        to track ongoing function execution.
+
+        Args:
+            frame: Frame containing function call progress information.
+        """
+        self._context.add_message(
+            {
+                "role": "assistant",
+                "tool_calls": [
+                    {
+                        "id": frame.tool_call_id,
+                        "function": {
+                            "name": frame.function_name,
+                            "arguments": json.dumps(frame.arguments),
+                        },
+                        "type": "function",
+                    }
+                ],
+            }
+        )
+        self._context.add_message(
+            {
+                "role": "tool",
+                "content": "IN_PROGRESS",
+                "tool_call_id": frame.tool_call_id,
+            }
+        )
+
+    async def handle_function_call_result(self, frame: FunctionCallResultFrame):
+        """Handle the result of a function call.
+
+        Updates the context with the function call result, replacing any
+        previous IN_PROGRESS status.
+
+        Args:
+            frame: Frame containing the function call result.
+        """
+        if frame.result:
+            result = json.dumps(frame.result)
+            await self._update_function_call_result(frame.function_name, frame.tool_call_id, result)
+        else:
+            await self._update_function_call_result(
+                frame.function_name, frame.tool_call_id, "COMPLETED"
+            )
+
+    async def handle_function_call_cancel(self, frame: FunctionCallCancelFrame):
+        """Handle a cancelled function call.
+
+        Updates the context to mark the function call as cancelled.
+
+        Args:
+            frame: Frame containing the function call cancellation information.
+        """
+        await self._update_function_call_result(
+            frame.function_name, frame.tool_call_id, "CANCELLED"
+        )
+
+    async def _update_function_call_result(
+        self, function_name: str, tool_call_id: str, result: Any
+    ):
+        for message in self._context.messages:
+            if (
+                message["role"] == "tool"
+                and message["tool_call_id"]
+                and message["tool_call_id"] == tool_call_id
+            ):
+                message["content"] = result
+
+    async def handle_user_image_frame(self, frame: UserImageRawFrame):
+        """Handle a user image frame from a function call request.
+
+        Marks the associated function call as completed and adds the image
+        to the context for processing.
+
+        Args:
+            frame: Frame containing the user image and request context.
+        """
+        await self._update_function_call_result(
+            frame.request.function_name, frame.request.tool_call_id, "COMPLETED"
+        )
+        self._context.add_image_frame_message(
+            format=frame.format,
+            size=frame.size,
+            image=frame.image,
+            text=frame.request.context,
+        )

src/pipecat/services/openai_realtime_beta/azure.py

@@ -11,7 +11,7 @@ from loguru import logger
 from .openai import OpenAIRealtimeBetaLLMService
 
 try:
-    import websockets
+    from websockets.asyncio.client import connect as websocket_connect
 except ModuleNotFoundError as e:
     logger.error(f"Exception: {e}")
     logger.error(
@@ -55,9 +55,9 @@ class AzureRealtimeBetaLLMService(OpenAIRealtimeBetaLLMService):
                 return
 
             logger.info(f"Connecting to {self.base_url}, api key: {self.api_key}")
-            self._websocket = await websockets.connect(
+            self._websocket = await websocket_connect(
                 uri=self.base_url,
-                extra_headers={
+                additional_headers={
                     "api-key": self.api_key,
                 },
             )

src/pipecat/services/openai_realtime_beta/openai.py

@@ -66,7 +66,7 @@ from .context import (
 from .frames import RealtimeFunctionCallResultFrame, RealtimeMessagesUpdateFrame
 
 try:
-    import websockets
+    from websockets.asyncio.client import connect as websocket_connect
 except ModuleNotFoundError as e:
     logger.error(f"Exception: {e}")
     logger.error("In order to use OpenAI, you need to `pip install pipecat-ai[openai]`.")
@@ -387,9 +387,9 @@ class OpenAIRealtimeBetaLLMService(LLMService):
                 # Here we assume that if we have a websocket, we are connected. We
                 # handle disconnections in the send/recv code paths.
                 return
-            self._websocket = await websockets.connect(
+            self._websocket = await websocket_connect(
                 uri=self.base_url,
-                extra_headers={
+                additional_headers={
                     "Authorization": f"Bearer {self.api_key}",
                     "OpenAI-Beta": "realtime=v1",
                 },

src/pipecat/services/playht/tts.py

@@ -17,7 +17,6 @@ import uuid
 from typing import AsyncGenerator, Optional
 
 import aiohttp
-import websockets
 from loguru import logger
 from pydantic import BaseModel
 
@@ -41,6 +40,8 @@ try:
     from pyht.async_client import AsyncClient
     from pyht.client import Format, TTSOptions
     from pyht.client import Language as PlayHTLanguage
+    from websockets.asyncio.client import connect as websocket_connect
+    from websockets.protocol import State
 except ModuleNotFoundError as e:
     logger.error(f"Exception: {e}")
     logger.error("In order to use PlayHT, you need to `pip install pipecat-ai[playht]`.")
@@ -244,7 +245,7 @@ class PlayHTTTSService(InterruptibleTTSService):
     async def _connect_websocket(self):
         """Connect to PlayHT websocket."""
         try:
-            if self._websocket and self._websocket.open:
+            if self._websocket and self._websocket.state is State.OPEN:
                 return
 
             logger.debug("Connecting to PlayHT")
@@ -255,7 +256,7 @@ class PlayHTTTSService(InterruptibleTTSService):
             if not isinstance(self._websocket_url, str):
                 raise ValueError("WebSocket URL is not a string")
 
-            self._websocket = await websockets.connect(self._websocket_url)
+            self._websocket = await websocket_connect(self._websocket_url)
         except ValueError as e:
             logger.error(f"{self} initialization error: {e}")
             self._websocket = None
@@ -362,7 +363,7 @@ class PlayHTTTSService(InterruptibleTTSService):
 
         try:
             # Reconnect if the websocket is closed
-            if not self._websocket or self._websocket.closed:
+            if not self._websocket or self._websocket.state is State.CLOSED:
                 await self._connect()
 
             if not self._request_id:

src/pipecat/services/rime/tts.py

@@ -39,7 +39,8 @@ from pipecat.utils.text.skip_tags_aggregator import SkipTagsAggregator
 from pipecat.utils.tracing.service_decorators import traced_tts
 
 try:
-    import websockets
+    from websockets.asyncio.client import connect as websocket_connect
+    from websockets.protocol import State
 except ModuleNotFoundError as e:
     logger.error(f"Exception: {e}")
     logger.error("In order to use Rime, you need to `pip install pipecat-ai[rime]`.")
@@ -238,13 +239,13 @@ class RimeTTSService(AudioContextWordTTSService):
     async def _connect_websocket(self):
         """Connect to Rime websocket API with configured settings."""
         try:
-            if self._websocket and self._websocket.open:
+            if self._websocket and self._websocket.state is State.OPEN:
                 return
 
             params = "&".join(f"{k}={v}" for k, v in self._settings.items())
             url = f"{self._url}?{params}"
             headers = {"Authorization": f"Bearer {self._api_key}"}
-            self._websocket = await websockets.connect(url, extra_headers=headers)
+            self._websocket = await websocket_connect(url, additional_headers=headers)
         except Exception as e:
             logger.error(f"{self} initialization error: {e}")
             self._websocket = None
@@ -380,7 +381,7 @@ class RimeTTSService(AudioContextWordTTSService):
         """
         logger.debug(f"{self}: Generating TTS [{text}]")
         try:
-            if not self._websocket or self._websocket.closed:
+            if not self._websocket or self._websocket.state is State.CLOSED:
                 await self._connect()
 
             try:

src/pipecat/services/soniox/stt.py

@@ -32,6 +32,8 @@ from pipecat.utils.tracing.service_decorators import traced_stt
 
 try:
     import websockets
+    from websockets.asyncio.client import connect as websocket_connect
+    from websockets.protocol import State
 except ModuleNotFoundError as e:
     logger.error(f"Exception: {e}")
     logger.error("In order to use Soniox, you need to `pip install pipecat-ai[soniox]`.")
@@ -162,7 +164,7 @@ class SonioxSTTService(STTService):
         if self._websocket:
             return
 
-        self._websocket = await websockets.connect(self._url)
+        self._websocket = await websocket_connect(self._url)
 
         if not self._websocket:
             logger.error(f"Unable to connect to Soniox API at {self._url}")
@@ -244,7 +246,7 @@ class SonioxSTTService(STTService):
             Frame: None (transcription results come via WebSocket callbacks).
         """
         await self.start_processing_metrics()
-        if self._websocket and not self._websocket.closed:
+        if self._websocket and self._websocket.state is State.OPEN:
             await self._websocket.send(audio)
         await self.stop_processing_metrics()
 
@@ -268,13 +270,13 @@ class SonioxSTTService(STTService):
 
         if isinstance(frame, UserStoppedSpeakingFrame) and self._vad_force_turn_endpoint:
             # Send finalize message to Soniox so we get the final tokens asap.
-            if self._websocket and not self._websocket.closed:
+            if self._websocket and self._websocket.state is State.OPEN:
                 await self._websocket.send(FINALIZE_MESSAGE)
                 logger.debug(f"Triggered finalize event on: {frame.name=}, {direction=}")
 
     async def _send_stop_recording(self):
         """Send stop recording message to Soniox."""
-        if self._websocket and not self._websocket.closed:
+        if self._websocket and self._websocket.state is State.OPEN:
             # Send stop recording message
             await self._websocket.send("")
 
@@ -283,7 +285,7 @@ class SonioxSTTService(STTService):
         try:
             while True:
                 logger.debug("Sending keepalive message")
-                if self._websocket and not self._websocket.closed:
+                if self._websocket and self._websocket.state is State.OPEN:
                     await self._websocket.send(KEEPALIVE_MESSAGE)
                 else:
                     logger.debug("WebSocket connection closed.")

src/pipecat/services/websocket_service.py

@@ -43,7 +43,7 @@ class WebsocketService(ABC):
             True if connection is verified working, False otherwise.
         """
         try:
-            if not self._websocket or self._websocket.closed:
+            if not self._websocket or self._websocket.state is State.CLOSED:
                 return False
             await self._websocket.ping()
             return True
@@ -82,7 +82,7 @@ class WebsocketService(ABC):
             try:
                 await self._receive_messages()
                 retry_count = 0  # Reset counter on successful message receive
-                if self._websocket and self._websocket.state == State.CLOSED:
+                if self._websocket and self._websocket.state is State.CLOSED:
                     raise websockets.ConnectionClosedOK(
                         self._websocket.close_rcvd,
                         self._websocket.close_sent,

src/pipecat/transports/network/websocket_client.py

@@ -20,6 +20,7 @@ from typing import Awaitable, Callable, Optional
 import websockets
 from loguru import logger
 from pydantic.main import BaseModel
+from websockets.asyncio.client import connect as websocket_connect
 
 from pipecat.frames.frames import (
     CancelFrame,
@@ -129,7 +130,7 @@ class WebsocketClientSession:
             return
 
         try:
-            self._websocket = await websockets.connect(uri=self._uri, open_timeout=10)
+            self._websocket = await websocket_connect(uri=self._uri, open_timeout=10)
             self._client_task = self.task_manager.create_task(
                 self._client_task_handler(),
                 f"{self._transport_name}::WebsocketClientSession::_client_task_handler",

src/pipecat/transports/network/websocket_server.py

@@ -39,6 +39,8 @@ from pipecat.transports.base_transport import BaseTransport, TransportParams
 
 try:
     import websockets
+    from websockets.asyncio.server import serve as websocket_serve
+    from websockets.protocol import State
 except ModuleNotFoundError as e:
     logger.error(f"Exception: {e}")
     logger.error("In order to use websockets, you need to `pip install pipecat-ai[websocket]`.")
@@ -177,11 +179,11 @@ class WebsocketServerInputTransport(BaseInputTransport):
     async def _server_task_handler(self):
         """Handle WebSocket server startup and client connections."""
         logger.info(f"Starting websocket server on {self._host}:{self._port}")
-        async with websockets.serve(self._client_handler, self._host, self._port) as server:
+        async with websocket_serve(self._client_handler, self._host, self._port) as server:
             await self._callbacks.on_websocket_ready()
             await self._stop_server_event.wait()
 
-    async def _client_handler(self, websocket: websockets.WebSocketServerProtocol, path):
+    async def _client_handler(self, websocket: websockets.WebSocketServerProtocol):
         """Handle individual client connections and message processing."""
         logger.info(f"New client connection from {websocket.remote_address}")
         if self._websocket:
@@ -231,7 +233,7 @@ class WebsocketServerInputTransport(BaseInputTransport):
         """Monitor WebSocket connection for session timeout."""
         try:
             await asyncio.sleep(session_timeout)
-            if not websocket.closed:
+            if websocket.state is not State.CLOSED:
                 await self._callbacks.on_session_timeout(websocket)
         except asyncio.CancelledError:
             logger.info(f"Monitoring task cancelled for: {websocket.remote_address}")

src/pipecat/transports/services/daily.py

@@ -62,6 +62,9 @@ try:
         VirtualCameraDevice,
         VirtualSpeakerDevice,
     )
+    from daily import (
+        LogLevel as DailyLogLevel,
+    )
 except ModuleNotFoundError as e:
     logger.error(f"Exception: {e}")
     logger.error(
@@ -1924,6 +1927,18 @@ class DailyTransport(BaseTransport):
         """
         return self._client.participant_id
 
+    def set_log_level(self, level: DailyLogLevel):
+        """Set the logging level for Daily's internal logging system.
+
+        Args:
+            level: The log level to set. Should be a member of the DailyLogLevel enum,
+                  such as DailyLogLevel.Info, DailyLogLevel.Debug, etc.
+
+        Example:
+            transport.set_log_level(DailyLogLevel.Info)
+        """
+        Daily.set_log_level(level)
+
     async def send_image(self, frame: OutputImageRawFrame | SpriteFrame):
         """Send an image frame to the Daily call.
 

src/pipecat/transports/services/livekit.py

@@ -439,6 +439,7 @@ class LiveKitTransportClient:
                 self._process_audio_stream(audio_stream, participant.sid),
                 f"{self}::_process_audio_stream",
             )
+            await self._callbacks.on_audio_track_subscribed(participant.sid)
 
     async def _async_on_track_unsubscribed(
         self,

tests/test_context_aggregators.py

@@ -17,9 +17,9 @@ from pipecat.frames.frames import (
     FunctionCallResultFrame,
     FunctionCallResultProperties,
     InterimTranscriptionFrame,
-    LLMContextAssistantTimestampFrame,
     LLMFullResponseEndFrame,
     LLMFullResponseStartFrame,
+    OpenAILLMContextAssistantTimestampFrame,
     SpeechControlParamsFrame,
     StartInterruptionFrame,
     TextFrame,
@@ -738,7 +738,7 @@ class TestAnthropicAssistantContextAggregator(
 ):
     CONTEXT_CLASS = AnthropicLLMContext
     AGGREGATOR_CLASS = AnthropicAssistantContextAggregator
-    EXPECTED_CONTEXT_FRAMES = [OpenAILLMContextFrame, LLMContextAssistantTimestampFrame]
+    EXPECTED_CONTEXT_FRAMES = [OpenAILLMContextFrame, OpenAILLMContextAssistantTimestampFrame]
 
     def check_message_multi_content(
         self, context: OpenAILLMContext, content_index: int, index: int, content: str
@@ -773,7 +773,7 @@ class TestAWSBedrockAssistantContextAggregator(
 ):
     CONTEXT_CLASS = AWSBedrockLLMContext
     AGGREGATOR_CLASS = AWSBedrockAssistantContextAggregator
-    EXPECTED_CONTEXT_FRAMES = [OpenAILLMContextFrame, LLMContextAssistantTimestampFrame]
+    EXPECTED_CONTEXT_FRAMES = [OpenAILLMContextFrame, OpenAILLMContextAssistantTimestampFrame]
 
     def check_message_multi_content(
         self, context: OpenAILLMContext, content_index: int, index: int, content: str
@@ -814,7 +814,7 @@ class TestGoogleAssistantContextAggregator(
 ):
     CONTEXT_CLASS = GoogleLLMContext
     AGGREGATOR_CLASS = GoogleAssistantContextAggregator
-    EXPECTED_CONTEXT_FRAMES = [OpenAILLMContextFrame, LLMContextAssistantTimestampFrame]
+    EXPECTED_CONTEXT_FRAMES = [OpenAILLMContextFrame, OpenAILLMContextAssistantTimestampFrame]
 
     def check_message_content(self, context: OpenAILLMContext, index: int, content: str):
         obj = context.messages[index].to_json_dict()
@@ -848,4 +848,4 @@ class TestOpenAIAssistantContextAggregator(
 ):
     CONTEXT_CLASS = OpenAILLMContext
     AGGREGATOR_CLASS = OpenAIAssistantContextAggregator
-    EXPECTED_CONTEXT_FRAMES = [OpenAILLMContextFrame, LLMContextAssistantTimestampFrame]
+    EXPECTED_CONTEXT_FRAMES = [OpenAILLMContextFrame, OpenAILLMContextAssistantTimestampFrame]