diff --git a/src/pipecat/services/asyncai/tts.py b/src/pipecat/services/asyncai/tts.py
index e9170f8b6..69ed90ca1 100644
--- a/src/pipecat/services/asyncai/tts.py
+++ b/src/pipecat/services/asyncai/tts.py
@@ -9,7 +9,6 @@
 import asyncio
 import base64
 import json
-import uuid
 from typing import AsyncGenerator, Optional
 
 import aiohttp
@@ -148,7 +147,6 @@ class AsyncAITTSService(AudioContextTTSService):
 
         self._receive_task = None
         self._keepalive_task = None
-        self._context_id = None
 
     def can_generate_metrics(self) -> bool:
         """Check if this service can generate processing metrics.
@@ -255,7 +253,7 @@ class AsyncAITTSService(AudioContextTTSService):
             if self._websocket:
                 logger.debug("Disconnecting from Async")
                 # Close all contexts and the socket
-                if self._context_id:
+                if self.has_active_audio_context():
                     await self._websocket.send(json.dumps({"terminate": True}))
                 await self._websocket.close()
                 logger.debug("Disconnected from Async")
@@ -263,7 +261,7 @@ class AsyncAITTSService(AudioContextTTSService):
             await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e)
         finally:
             self._websocket = None
-            self._context_id = None
+            await self.remove_active_audio_context()
             await self._call_event_handler("on_disconnected")
 
     def _get_websocket(self):
@@ -271,26 +269,13 @@ class AsyncAITTSService(AudioContextTTSService):
             return self._websocket
         raise Exception("Websocket not connected")
 
-    def create_context_id(self) -> str:
-        """Generate a unique context ID for a TTS request in case we don't have one already in progress.
-
-        Returns:
-            A unique string identifier for the TTS context.
-        """
-        # If a context ID does not exist, create a new one.
-        # If an ID exists, continue using the current ID.
-        # When interruptions happen, user speech results in
-        # an interruption, which resets the context ID.
-        if not self._context_id:
-            return str(uuid.uuid4())
-        return self._context_id
-
     async def flush_audio(self):
         """Flush any pending audio."""
-        if not self._context_id or not self._websocket:
+        context_id = self.get_active_audio_context_id()
+        if not context_id or not self._websocket:
             return
         logger.trace(f"{self}: flushing audio")
-        msg = self._build_msg(text=" ", context_id=self._context_id, force=True)
+        msg = self._build_msg(text=" ", context_id=context_id, force=True)
         await self._websocket.send(msg)
 
     async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM):
@@ -318,11 +303,11 @@ class AsyncAITTSService(AudioContextTTSService):
 
             # Check if this message belongs to the current context.
             if not self.audio_context_available(received_ctx_id):
-                if self._context_id == received_ctx_id:
+                if self.get_active_audio_context_id() == received_ctx_id:
                     logger.debug(
-                        f"Received a delayed message, recreating the context: {self._context_id}"
+                        f"Received a delayed message, recreating the context: {received_ctx_id}"
                     )
-                    await self.create_audio_context(self._context_id)
+                    await self.create_audio_context(received_ctx_id)
                 else:
                     # This can happen if a message is received _after_ we have closed a context
                     # due to user interruption but _before_ the `isFinal` message for the context
@@ -343,10 +328,11 @@ class AsyncAITTSService(AudioContextTTSService):
             await asyncio.sleep(KEEPALIVE_SLEEP)
             try:
                 if self._websocket and self._websocket.state is State.OPEN:
-                    if self._context_id:
+                    context_id = self.get_active_audio_context_id()
+                    if context_id:
                         keepalive_message = {
                             "transcript": " ",
-                            "context_id": self._context_id,
+                            "context_id": context_id,
                         }
                         logger.trace("Sending keepalive message")
                     else:
@@ -362,19 +348,16 @@ class AsyncAITTSService(AudioContextTTSService):
 
     async def _handle_interruption(self, frame: InterruptionFrame, direction: FrameDirection):
         """Handle interruption by closing the current context."""
+        context_id = self.get_active_audio_context_id()
         await super()._handle_interruption(frame, direction)
-
         # Close the current context when interrupted without closing the websocket
-        if self._context_id and self._websocket:
+        if context_id and self._websocket:
             try:
                 await self._websocket.send(
-                    json.dumps(
-                        {"context_id": self._context_id, "close_context": True, "transcript": ""}
-                    )
+                    json.dumps({"context_id": context_id, "close_context": True, "transcript": ""})
                 )
             except Exception as e:
                 logger.error(f"Error closing context on interruption: {e}")
-            self._context_id = None
 
     @traced_tts
     async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
@@ -394,16 +377,13 @@ class AsyncAITTSService(AudioContextTTSService):
                 await self._connect()
 
             try:
-                if not self._context_id:
+                if not self.has_active_audio_context():
                     await self.start_ttfb_metrics()
                     yield TTSStartedFrame(context_id=context_id)
+                    if not self.audio_context_available(context_id):
+                        await self.create_audio_context(context_id)
 
-                    self._context_id = context_id
-
-                    if not self.audio_context_available(self._context_id):
-                        await self.create_audio_context(self._context_id)
-
-                msg = self._build_msg(text=text, force=True, context_id=self._context_id)
+                msg = self._build_msg(text=text, force=True, context_id=context_id)
                 await self._get_websocket().send(msg)
                 await self.start_tts_usage_metrics(text)
 
diff --git a/src/pipecat/services/cartesia/tts.py b/src/pipecat/services/cartesia/tts.py
index 1fa9a026a..e30acdcd1 100644
--- a/src/pipecat/services/cartesia/tts.py
+++ b/src/pipecat/services/cartesia/tts.py
@@ -8,7 +8,6 @@
 
 import base64
 import json
-import uuid
 import warnings
 from enum import Enum
 from typing import AsyncGenerator, List, Literal, Optional
@@ -307,7 +306,6 @@ class CartesiaTTSService(AudioContextWordTTSService):
         self.set_model_name(model)
         self.set_voice(voice_id)
 
-        self._context_id = None
         self._receive_task = None
 
     def can_generate_metrics(self) -> bool:
@@ -430,7 +428,7 @@ class CartesiaTTSService(AudioContextWordTTSService):
         msg = {
             "transcript": text,
             "continue": continue_transcript,
-            "context_id": self._context_id,
+            "context_id": self.get_active_audio_context_id(),
             "model_id": self.model_name,
             "voice": voice_config,
             "output_format": self._settings["output_format"],
@@ -523,7 +521,7 @@ class CartesiaTTSService(AudioContextWordTTSService):
         except Exception as e:
             await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e)
         finally:
-            self._context_id = None
+            await self.remove_active_audio_context()
             self._websocket = None
             await self._call_event_handler("on_disconnected")
 
@@ -533,35 +531,22 @@ class CartesiaTTSService(AudioContextWordTTSService):
         raise Exception("Websocket not connected")
 
     async def _handle_interruption(self, frame: InterruptionFrame, direction: FrameDirection):
+        context_id = self.get_active_audio_context_id()
         await super()._handle_interruption(frame, direction)
         await self.stop_all_metrics()
-        if self._context_id:
-            cancel_msg = json.dumps({"context_id": self._context_id, "cancel": True})
+        if context_id:
+            cancel_msg = json.dumps({"context_id": context_id, "cancel": True})
             await self._get_websocket().send(cancel_msg)
-            self._context_id = None
-
-    def create_context_id(self) -> str:
-        """Generate a unique context ID for a TTS request in case we don't have one already in progress.
-
-        Returns:
-            A unique string identifier for the TTS context.
-        """
-        # If a context ID does not exist, create a new one.
-        # If an ID exists, continue using the current ID.
-        # When interruptions happen, user speech results in
-        # an interruption, which resets the context ID.
-        if not self._context_id:
-            return str(uuid.uuid4())
-        return self._context_id
 
     async def flush_audio(self):
         """Flush any pending audio and finalize the current context."""
-        if not self._context_id or not self._websocket:
+        context_id = self.get_active_audio_context_id()
+        if not context_id or not self._websocket:
             return
         logger.trace(f"{self}: flushing audio")
         msg = self._build_msg(text="", continue_transcript=False)
         await self._websocket.send(msg)
-        self._context_id = None
+        self.reset_active_audio_context()
 
     async def _process_messages(self):
         async for message in self._get_websocket():
@@ -593,7 +578,7 @@ class CartesiaTTSService(AudioContextWordTTSService):
                 await self.push_frame(TTSStoppedFrame(context_id=ctx_id))
                 await self.stop_all_metrics()
                 await self.push_error(error_msg=f"Error: {msg}")
-                self._context_id = None
+                self.reset_active_audio_context()
             else:
                 await self.push_error(error_msg=f"Error, unknown message type: {msg}")
 
@@ -622,11 +607,10 @@ class CartesiaTTSService(AudioContextWordTTSService):
             if not self._websocket or self._websocket.state is State.CLOSED:
                 await self._connect()
 
-            if not self._context_id:
+            if not self.has_active_audio_context():
                 await self.start_ttfb_metrics()
                 yield TTSStartedFrame(context_id=context_id)
-                self._context_id = context_id
-                await self.create_audio_context(self._context_id)
+                await self.create_audio_context(context_id)
 
             msg = self._build_msg(text=text)
 
diff --git a/src/pipecat/services/elevenlabs/tts.py b/src/pipecat/services/elevenlabs/tts.py
index 7df891f0e..be57a1a3d 100644
--- a/src/pipecat/services/elevenlabs/tts.py
+++ b/src/pipecat/services/elevenlabs/tts.py
@@ -13,7 +13,6 @@ with support for streaming audio, word timestamps, and voice customization.
 import asyncio
 import base64
 import json
-import uuid
 from typing import Any, AsyncGenerator, Dict, List, Literal, Mapping, Optional, Tuple, Union
 
 import aiohttp
@@ -343,7 +342,6 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
         self._partial_word_start_time = 0.0
 
         # Context management for v1 multi API
-        self._context_id = None
         self._receive_task = None
         self._keepalive_task = None
 
@@ -411,18 +409,19 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
             )
             await self._disconnect()
             await self._connect()
-        elif voice_settings_changed and self._context_id:
+        elif voice_settings_changed and self.has_active_audio_context():
             # Voice settings can be updated by closing current context
             # so new one gets created with updated voice settings
             logger.debug(f"Voice settings changed, closing current context to apply changes")
+            context_id = self.get_active_audio_context_id()
             try:
                 if self._websocket:
                     await self._websocket.send(
-                        json.dumps({"context_id": self._context_id, "close_context": True})
+                        json.dumps({"context_id": context_id, "close_context": True})
                     )
             except Exception as e:
                 await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e)
-            self._context_id = None
+            self.reset_active_audio_context()
 
     async def start(self, frame: StartFrame):
         """Start the ElevenLabs TTS service.
@@ -454,10 +453,11 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
 
     async def flush_audio(self):
         """Flush any pending audio and finalize the current context."""
-        if not self._context_id or not self._websocket:
+        context_id = self.get_active_audio_context_id()
+        if not context_id or not self._websocket:
             return
         logger.trace(f"{self}: flushing audio")
-        msg = {"context_id": self._context_id, "flush": True}
+        msg = {"context_id": context_id, "flush": True}
         await self._websocket.send(json.dumps(msg))
 
     async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM):
@@ -470,7 +470,7 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
         await super().push_frame(frame, direction)
         if isinstance(frame, (TTSStoppedFrame, InterruptionFrame)):
             if isinstance(frame, TTSStoppedFrame):
-                await self.add_word_timestamps([("Reset", 0)], self._context_id)
+                await self.add_word_timestamps([("Reset", 0)], self.get_active_audio_context_id())
 
     async def _connect(self):
         await super()._connect()
@@ -545,14 +545,14 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
             if self._websocket:
                 logger.debug("Disconnecting from ElevenLabs")
                 # Close all contexts and the socket
-                if self._context_id:
+                if self.has_active_audio_context():
                     await self._websocket.send(json.dumps({"close_socket": True}))
                 await self._websocket.close()
                 logger.debug("Disconnected from ElevenLabs")
         except Exception as e:
             await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e)
         finally:
-            self._context_id = None
+            await self.remove_active_audio_context()
             self._websocket = None
             await self._call_event_handler("on_disconnected")
 
@@ -563,11 +563,12 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
 
     async def _handle_interruption(self, frame: InterruptionFrame, direction: FrameDirection):
         """Handle interruption by closing the current context."""
+        # Close the current context when interrupted without closing the websocket
+        context_id = self.get_active_audio_context_id()
         await super()._handle_interruption(frame, direction)
 
-        # Close the current context when interrupted without closing the websocket
-        if self._context_id and self._websocket:
-            logger.trace(f"Closing context {self._context_id} due to interruption")
+        if context_id and self._websocket:
+            logger.trace(f"Closing context {context_id} due to interruption")
             try:
                 # ElevenLabs requires that Pipecat manages the contexts and closes them
                 # when they're not longer in use. Since an InterruptionFrame is pushed
@@ -576,11 +577,10 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
                 # Note: We do not need to call remove_audio_context here, as the context is
                 # automatically reset when super ()._handle_interruption is called.
                 await self._websocket.send(
-                    json.dumps({"context_id": self._context_id, "close_context": True})
+                    json.dumps({"context_id": context_id, "close_context": True})
                 )
             except Exception as e:
                 await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e)
-            self._context_id = None
             self._partial_word = ""
             self._partial_word_start_time = 0.0
 
@@ -600,11 +600,11 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
 
             # Check if this message belongs to the current context.
             if not self.audio_context_available(received_ctx_id):
-                if self._context_id == received_ctx_id:
+                if self.get_active_audio_context_id() == received_ctx_id:
                     logger.debug(
-                        f"Received a delayed message, recreating the context: {self._context_id}"
+                        f"Received a delayed message, recreating the context: {received_ctx_id}"
                     )
-                    await self.create_audio_context(self._context_id)
+                    await self.create_audio_context(received_ctx_id)
                 else:
                     # This can happen if a message is received _after_ we have closed a context
                     # due to user interruption but _before_ the `isFinal` message for the context
@@ -657,13 +657,14 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
             await asyncio.sleep(KEEPALIVE_SLEEP)
             try:
                 if self._websocket and self._websocket.state is State.OPEN:
-                    if self._context_id:
+                    context_id = self.get_active_audio_context_id()
+                    if context_id:
                         # Send keepalive with context ID to keep the connection alive
                         keepalive_message = {
                             "text": "",
-                            "context_id": self._context_id,
+                            "context_id": context_id,
                         }
-                        logger.trace(f"Sending keepalive for context {self._context_id}")
+                        logger.trace(f"Sending keepalive for context {context_id}")
                     else:
                         # It's possible to have a user interruption which clears the context
                         # without generating a new TTS response. In this case, we'll just send
@@ -677,24 +678,11 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
 
     async def _send_text(self, text: str):
         """Send text to the WebSocket for synthesis."""
-        if self._websocket and self._context_id:
-            msg = {"text": text, "context_id": self._context_id}
+        context_id = self.get_active_audio_context_id()
+        if self._websocket and context_id:
+            msg = {"text": text, "context_id": context_id}
             await self._websocket.send(json.dumps(msg))
 
-    def create_context_id(self) -> str:
-        """Generate a unique context ID for a TTS request in case we don't have one already in progress.
-
-        Returns:
-            A unique string identifier for the TTS context.
-        """
-        # If a context ID does not exist, create a new one.
-        # If an ID exists, continue using the current ID.
-        # When interruptions happens, user speech results in
-        # an interruption, which resets the context ID.
-        if not self._context_id:
-            return str(uuid.uuid4())
-        return self._context_id
-
     @traced_tts
     async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
         """Generate speech from text using ElevenLabs' streaming WebSocket API.
@@ -713,19 +701,18 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
                 await self._connect()
 
             try:
-                if not self._context_id:
+                if not self.has_active_audio_context():
                     await self.start_ttfb_metrics()
                     yield TTSStartedFrame(context_id=context_id)
-                    self._context_id = context_id
                     self._cumulative_time = 0
                     self._partial_word = ""
                     self._partial_word_start_time = 0.0
 
-                    if not self.audio_context_available(self._context_id):
-                        await self.create_audio_context(self._context_id)
+                    if not self.audio_context_available(context_id):
+                        await self.create_audio_context(context_id)
 
                     # Initialize context with voice settings and pronunciation dictionaries
-                    msg = {"text": " ", "context_id": self._context_id}
+                    msg = {"text": " ", "context_id": context_id}
                     if self._voice_settings:
                         msg["voice_settings"] = self._voice_settings
                     if self._pronunciation_dictionary_locators:
@@ -734,7 +721,7 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
                             for locator in self._pronunciation_dictionary_locators
                         ]
                     await self._websocket.send(json.dumps(msg))
-                    logger.trace(f"Created new context {self._context_id}")
+                    logger.trace(f"Created new context {context_id}")
 
                 await self._send_text(text)
                 await self.start_tts_usage_metrics(text)
diff --git a/src/pipecat/services/gradium/tts.py b/src/pipecat/services/gradium/tts.py
index ef3cbbfde..98e08a9d3 100644
--- a/src/pipecat/services/gradium/tts.py
+++ b/src/pipecat/services/gradium/tts.py
@@ -6,7 +6,6 @@
 
 import base64
 import json
-import uuid
 from typing import Any, AsyncGenerator, Mapping, Optional
 
 from loguru import logger
@@ -97,7 +96,6 @@ class GradiumTTSService(AudioContextWordTTSService):
 
         # State tracking
         self._receive_task = None
-        self._context_id: Optional[str] = None
 
     def can_generate_metrics(self) -> bool:
         """Check if this service can generate processing metrics.
@@ -129,8 +127,9 @@ class GradiumTTSService(AudioContextWordTTSService):
     def _build_msg(self, text: str = "") -> dict:
         """Build JSON message for Gradium API."""
         msg = {"text": text, "type": "text"}
-        if self._context_id:
-            msg["client_req_id"] = self._context_id
+        context_id = self.get_active_audio_context_id()
+        if context_id:
+            msg["client_req_id"] = context_id
         return msg
 
     async def start(self, frame: StartFrame):
@@ -229,6 +228,7 @@ class GradiumTTSService(AudioContextWordTTSService):
         except Exception as e:
             await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e)
         finally:
+            await self.remove_active_audio_context()
             self._websocket = None
             await self._call_event_handler("on_disconnected")
 
@@ -240,12 +240,13 @@ class GradiumTTSService(AudioContextWordTTSService):
 
     async def flush_audio(self):
         """Flush any pending audio synthesis."""
-        if not self._context_id or not self._websocket:
+        context_id = self.get_active_audio_context_id()
+        if not context_id or not self._websocket:
             return
         try:
-            msg = {"type": "end_of_stream", "client_req_id": self._context_id}
+            msg = {"type": "end_of_stream", "client_req_id": context_id}
             await self._websocket.send(json.dumps(msg))
-            self._context_id = None
+            self.reset_active_audio_context()
         except ConnectionClosedOK:
             logger.debug(f"{self}: connection closed normally during flush")
         except Exception as e:
@@ -265,7 +266,6 @@ class GradiumTTSService(AudioContextWordTTSService):
         """
         await super()._handle_interruption(frame, direction)
         await self.stop_all_metrics()
-        self._context_id = None
 
     async def _receive_messages(self):
         """Process incoming websocket messages, demultiplexing by client_req_id."""
@@ -306,20 +306,6 @@ class GradiumTTSService(AudioContextWordTTSService):
                 await self.stop_all_metrics()
                 await self.push_error(error_msg=f"Error: {msg.get('message', msg)}")
 
-    def create_context_id(self) -> str:
-        """Generate a unique context ID for a TTS request in case we don't have one already in progress.
-
-        Returns:
-            A unique string identifier for the TTS context.
-        """
-        # If a context ID does not exist, create a new one.
-        # If an ID exists, continue using the current ID.
-        # When interruptions happens, user speech results in
-        # an interruption, which resets the context ID.
-        if not self._context_id:
-            return str(uuid.uuid4())
-        return self._context_id
-
     @traced_tts
     async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
         """Generate speech from text using Gradium's streaming API.
@@ -338,11 +324,10 @@ class GradiumTTSService(AudioContextWordTTSService):
                 await self._connect()
 
             try:
-                if not self._context_id:
+                if not self.has_active_audio_context():
                     await self.start_ttfb_metrics()
                     yield TTSStartedFrame(context_id=context_id)
-                    self._context_id = context_id
-                    await self.create_audio_context(self._context_id)
+                    await self.create_audio_context(context_id)
 
                 msg = self._build_msg(text=text)
                 await self._get_websocket().send(json.dumps(msg))
diff --git a/src/pipecat/services/inworld/tts.py b/src/pipecat/services/inworld/tts.py
index 9767c1b0a..adf72ce9f 100644
--- a/src/pipecat/services/inworld/tts.py
+++ b/src/pipecat/services/inworld/tts.py
@@ -517,7 +517,6 @@ class InworldTTSService(AudioContextWordTTSService):
 
         self._receive_task = None
         self._keepalive_task = None
-        self._context_id = None
 
         # Track cumulative time across generations for monotonic timestamps within a turn.
         # When auto_mode is enabled, the server controls generations and timestamps reset
@@ -573,9 +572,10 @@ class InworldTTSService(AudioContextWordTTSService):
         keeping the context open for subsequent text. The context is only
         closed on interruption, disconnect, or end of session.
         """
-        if self._context_id and self._websocket:
-            logger.trace(f"Flushing audio for context {self._context_id}")
-            await self._send_flush(self._context_id)
+        context_id = self.get_active_audio_context_id()
+        if context_id and self._websocket:
+            logger.trace(f"Flushing audio for context {context_id}")
+            await self._send_flush(context_id)
 
     async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM):
         """Push a frame and handle state changes.
@@ -640,7 +640,7 @@ class InworldTTSService(AudioContextWordTTSService):
             frame: The interruption frame.
             direction: The direction of the interruption.
         """
-        old_context_id = self._context_id
+        old_context_id = self.get_active_audio_context_id()
         logger.trace(f"{self}: Handling interruption, old context: {old_context_id}")
 
         await super()._handle_interruption(frame, direction)
@@ -652,7 +652,6 @@ class InworldTTSService(AudioContextWordTTSService):
             except Exception as e:
                 await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e)
 
-        self._context_id = None
         self._cumulative_time = 0.0
         self._generation_end_time = 0.0
         logger.trace(f"{self}: Interruption handled, context reset to None")
@@ -736,9 +735,10 @@ class InworldTTSService(AudioContextWordTTSService):
 
             if self._websocket:
                 logger.debug("Disconnecting from Inworld WebSocket TTS")
-                if self._context_id:
+                context_id = self.get_active_audio_context_id()
+                if context_id:
                     try:
-                        await self._send_close_context(self._context_id)
+                        await self._send_close_context(context_id)
                     except Exception:
                         pass
                 await self._websocket.close()
@@ -746,7 +746,7 @@ class InworldTTSService(AudioContextWordTTSService):
         except Exception as e:
             await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e)
         finally:
-            self._context_id = None
+            await self.remove_active_audio_context()
             self._websocket = None
             self._cumulative_time = 0.0
             self._generation_end_time = 0.0
@@ -772,7 +772,7 @@ class InworldTTSService(AudioContextWordTTSService):
             ]
             logger.debug(
                 f"{self}: Received message types={msg_types}, ctx_id={ctx_id}, "
-                f"current_ctx={self._context_id}, available={self.audio_context_available(ctx_id) if ctx_id else 'N/A'}"
+                f"current_ctx={self.get_active_audio_context_id()}, available={self.audio_context_available(ctx_id) if ctx_id else 'N/A'}"
             )
 
             # Check for errors
@@ -784,7 +784,9 @@ class InworldTTSService(AudioContextWordTTSService):
                 # Handle "Context not found" error (code 5)
                 # This can happen when a keepalive message is sent but no context is available.
                 if error_code == 5 and "not found" in error_msg.lower():
-                    logger.debug(f"{self}: Context {ctx_id or self._context_id} not found.")
+                    logger.debug(
+                        f"{self}: Context {ctx_id or self.get_active_audio_context_id()} not found."
+                    )
                     continue
 
                 # For other errors, push error frame
@@ -799,11 +801,9 @@ class InworldTTSService(AudioContextWordTTSService):
             # If the context isn't available but matches our current context ID,
             # recreate it (handles race conditions during interruption recovery).
             if ctx_id and not self.audio_context_available(ctx_id):
-                if self._context_id == ctx_id:
-                    logger.trace(
-                        f"{self}: Recreating audio context for current context: {self._context_id}"
-                    )
-                    await self.create_audio_context(self._context_id)
+                if self.get_active_audio_context_id() == ctx_id:
+                    logger.trace(f"{self}: Recreating audio context for current context: {ctx_id}")
+                    await self.create_audio_context(ctx_id)
                 else:
                     # This is a message from an old/closed context - skip it
                     logger.trace(f"{self}: Skipping message from unavailable context: {ctx_id}")
@@ -849,8 +849,8 @@ class InworldTTSService(AudioContextWordTTSService):
                 logger.trace(f"{self}: Context closed on server: {ctx_id}")
                 await self.stop_ttfb_metrics()
                 # Only reset if this is our current context
-                if ctx_id == self._context_id:
-                    self._context_id = None
+                if ctx_id == self.get_active_audio_context_id():
+                    self.reset_active_audio_context()
                 if ctx_id and self.audio_context_available(ctx_id):
                     await self.remove_audio_context(ctx_id)
                 await self.add_word_timestamps([("TTSStoppedFrame", 0), ("Reset", 0)], ctx_id)
@@ -862,12 +862,13 @@ class InworldTTSService(AudioContextWordTTSService):
             await asyncio.sleep(KEEPALIVE_SLEEP)
             try:
                 if self._websocket and self._websocket.state is State.OPEN:
-                    if self._context_id:
+                    context_id = self.get_active_audio_context_id()
+                    if context_id:
                         keepalive_message = {
                             "send_text": {"text": ""},
-                            "contextId": self._context_id,
+                            "contextId": context_id,
                         }
-                        logger.trace(f"Sending keepalive for context {self._context_id}")
+                        logger.trace(f"Sending keepalive for context {context_id}")
                     else:
                         keepalive_message = {"send_text": {"text": ""}}
                         logger.trace("Sending keepalive without context")
@@ -938,20 +939,6 @@ class InworldTTSService(AudioContextWordTTSService):
         msg = {"close_context": {}, "contextId": context_id}
         await self.send_with_retry(json.dumps(msg), self._report_error)
 
-    def create_context_id(self) -> str:
-        """Generate a unique context ID for a TTS request in case we don't have one already in progress.
-
-        Returns:
-            A unique string identifier for the TTS context.
-        """
-        # If a context ID does not exist, create a new one.
-        # If an ID exists, continue using the current ID.
-        # When interruptions happen, user speech results in
-        # an interruption, which resets the context ID.
-        if not self._context_id:
-            return str(uuid.uuid4())
-        return self._context_id
-
     @traced_tts
     async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
         """Generate TTS audio for the given text using the Inworld WebSocket TTS service.
@@ -970,19 +957,13 @@ class InworldTTSService(AudioContextWordTTSService):
                 await self._connect()
 
             try:
-                if not self._context_id:
+                if not self.has_active_audio_context():
                     await self.start_ttfb_metrics()
                     yield TTSStartedFrame(context_id=context_id)
-                    self._context_id = context_id
-                    logger.trace(f"{self}: Creating new context {self._context_id}")
-                    await self.create_audio_context(self._context_id)
-                    await self._send_context(self._context_id)
-                elif not self.audio_context_available(self._context_id):
-                    # Context exists on server but local tracking was removed
-                    logger.trace(f"{self}: Recreating local audio context {self._context_id}")
-                    await self.create_audio_context(self._context_id)
+                    await self.create_audio_context(context_id)
+                    await self._send_context(context_id)
 
-                await self._send_text(self._context_id, text)
+                await self._send_text(context_id, text)
                 await self.start_tts_usage_metrics(text)
 
             except Exception as e:
diff --git a/src/pipecat/services/resembleai/tts.py b/src/pipecat/services/resembleai/tts.py
index 964b9fa18..c51ccc07b 100644
--- a/src/pipecat/services/resembleai/tts.py
+++ b/src/pipecat/services/resembleai/tts.py
@@ -25,8 +25,6 @@ from pipecat.frames.frames import (
 )
 from pipecat.processors.frame_processor import FrameDirection
 from pipecat.services.tts_service import AudioContextWordTTSService
-from pipecat.transcriptions.language import Language
-from pipecat.utils.text.base_text_aggregator import BaseTextAggregator
 from pipecat.utils.tracing.service_decorators import traced_tts
 
 try:
@@ -70,6 +68,7 @@ class ResembleAITTSService(AudioContextWordTTSService):
         """
         super().__init__(
             sample_rate=sample_rate,
+            reuse_context_id_within_turn=False,
             **kwargs,
         )
 
diff --git a/src/pipecat/services/rime/tts.py b/src/pipecat/services/rime/tts.py
index cf3c6d5ca..c4b1c870a 100644
--- a/src/pipecat/services/rime/tts.py
+++ b/src/pipecat/services/rime/tts.py
@@ -12,7 +12,6 @@ using Rime's API for streaming and batch audio synthesis.
 
 import base64
 import json
-import uuid
 from typing import Any, AsyncGenerator, Mapping, Optional
 
 import aiohttp
@@ -167,7 +166,6 @@ class RimeTTSService(AudioContextWordTTSService):
         self._settings = self._build_settings()
 
         # State tracking
-        self._context_id = None  # Tracks current turn
         self._receive_task = None
         self._cumulative_time = 0  # Accumulates time across messages
         self._extra_msg_fields = {}  # Extra fields for next message
@@ -339,7 +337,7 @@ class RimeTTSService(AudioContextWordTTSService):
 
     def _build_msg(self, text: str = "") -> dict:
         """Build JSON message for Rime API."""
-        msg = {"text": text, "contextId": self._context_id}
+        msg = {"text": text, "contextId": self.get_active_audio_context_id()}
         if self._extra_msg_fields:
             msg |= self._extra_msg_fields
             self._extra_msg_fields = {}
@@ -427,7 +425,7 @@ class RimeTTSService(AudioContextWordTTSService):
         except Exception as e:
             await self.push_error(error_msg=f"Error disconnecting: {e}", exception=e)
         finally:
-            self._context_id = None
+            await self.remove_active_audio_context()
             self._websocket = None
             await self._call_event_handler("on_disconnected")
 
@@ -439,11 +437,11 @@ class RimeTTSService(AudioContextWordTTSService):
 
     async def _handle_interruption(self, frame: InterruptionFrame, direction: FrameDirection):
         """Handle interruption by clearing current context."""
+        context_id = self.get_active_audio_context_id()
         await super()._handle_interruption(frame, direction)
         await self.stop_all_metrics()
-        if self._context_id:
+        if context_id:
             await self._get_websocket().send(json.dumps(self._build_clear_msg()))
-            self._context_id = None
 
     def _calculate_word_times(self, words: list, starts: list, ends: list) -> list:
         """Calculate word timing pairs with proper spacing and punctuation.
@@ -474,28 +472,15 @@ class RimeTTSService(AudioContextWordTTSService):
 
         return word_pairs
 
-    def create_context_id(self) -> str:
-        """Generate a unique context ID for a TTS request in case we don't have one already in progress.
-
-        Returns:
-            A unique string identifier for the TTS context.
-        """
-        # If a context ID does not exist, create a new one.
-        # If an ID exists, continue using the current ID.
-        # When interruptions happen, user speech results in
-        # an interruption, which resets the context ID.
-        if not self._context_id:
-            return str(uuid.uuid4())
-        return self._context_id
-
     async def flush_audio(self):
         """Flush any pending audio synthesis."""
-        if not self._context_id or not self._websocket:
+        context_id = self.get_active_audio_context_id()
+        if not context_id or not self._websocket:
             return
 
         logger.trace(f"{self}: flushing audio")
         await self._get_websocket().send(json.dumps({"operation": "flush"}))
-        self._context_id = None
+        self.reset_active_audio_context()
 
     async def _receive_messages(self):
         """Process incoming websocket messages."""
@@ -537,7 +522,7 @@ class RimeTTSService(AudioContextWordTTSService):
                 await self.push_frame(TTSStoppedFrame())
                 await self.stop_all_metrics()
                 await self.push_error(error_msg=f"Error: {msg['message']}")
-                self._context_id = None
+                self.reset_active_audio_context()
 
     async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM):
         """Push frame and handle end-of-turn conditions.
@@ -568,12 +553,11 @@ class RimeTTSService(AudioContextWordTTSService):
                 await self._connect()
 
             try:
-                if not self._context_id:
+                if not self.has_active_audio_context():
                     await self.start_ttfb_metrics()
                     yield TTSStartedFrame(context_id=context_id)
                     self._cumulative_time = 0
-                    self._context_id = context_id
-                    await self.create_audio_context(self._context_id)
+                    await self.create_audio_context(context_id)
 
                 msg = self._build_msg(text=text)
                 await self._get_websocket().send(json.dumps(msg))
diff --git a/src/pipecat/services/tts_service.py b/src/pipecat/services/tts_service.py
index 02c799d0f..1948d3da2 100644
--- a/src/pipecat/services/tts_service.py
+++ b/src/pipecat/services/tts_service.py
@@ -1042,14 +1042,23 @@ class AudioContextTTSService(WebsocketTTSService):
     audio from context ID "A" will be played first.
     """
 
-    def __init__(self, *, reconnect_on_error: bool = True, **kwargs):
+    def __init__(
+        self,
+        *,
+        reuse_context_id_within_turn: bool = True,
+        reconnect_on_error: bool = True,
+        **kwargs,
+    ):
         """Initialize the Audio Context TTS service.
 
         Args:
+            reuse_context_id_within_turn: Whether the service should reuse context IDs within the same turn.
             reconnect_on_error: Whether to automatically reconnect on websocket errors.
             **kwargs: Additional arguments passed to the parent WebsocketTTSService.
         """
         super().__init__(reconnect_on_error=reconnect_on_error, **kwargs)
+        self._reuse_context_id_within_turn = reuse_context_id_within_turn
+        self._context_id = None
         self._contexts: Dict[str, asyncio.Queue] = {}
         self._audio_context_task = None
 
@@ -1059,6 +1068,10 @@ class AudioContextTTSService(WebsocketTTSService):
         Args:
             context_id: Unique identifier for the audio context.
         """
+        # Set the context ID if not already set
+        if not self._context_id:
+            self._context_id = context_id
+
         await self._contexts_queue.put(context_id)
         self._contexts[context_id] = asyncio.Queue()
         logger.trace(f"{self} created audio context {context_id}")
@@ -1091,6 +1104,32 @@ class AudioContextTTSService(WebsocketTTSService):
         else:
             logger.warning(f"{self} unable to remove context {context_id}")
 
+    def has_active_audio_context(self) -> bool:
+        """Check if there is an active audio context.
+
+        Returns:
+            True if an active audio context exists, False otherwise.
+        """
+        return self._context_id is not None and self.audio_context_available(self._context_id)
+
+    def get_active_audio_context_id(self) -> Optional[str]:
+        """Get the active audio context ID.
+
+        Returns:
+            The active context ID, or None if no context is active.
+        """
+        return self._context_id
+
+    async def remove_active_audio_context(self):
+        """Remove the active audio context."""
+        if self._context_id:
+            await self.remove_audio_context(self._context_id)
+            self.reset_active_audio_context()
+
+    def reset_active_audio_context(self):
+        """Reset the active audio context."""
+        self._context_id = None
+
     def audio_context_available(self, context_id: str) -> bool:
         """Check whether the given audio context is registered.
 
@@ -1102,6 +1141,20 @@ class AudioContextTTSService(WebsocketTTSService):
         """
         return context_id in self._contexts
 
+    def create_context_id(self) -> str:
+        """Generate or reuse a context ID based on concurrent TTS support.
+
+        If _reuse_context_id_within_turn is False and a context already exists,
+        the existing context ID is returned. Otherwise, a new unique context
+        ID is generated.
+
+        Returns:
+            A context ID string for the TTS request.
+        """
+        if self._reuse_context_id_within_turn and self._context_id:
+            return self._context_id
+        return super().create_context_id()
+
     async def start(self, frame: StartFrame):
         """Start the audio context TTS service.
 
@@ -1137,6 +1190,7 @@ class AudioContextTTSService(WebsocketTTSService):
     async def _handle_interruption(self, frame: InterruptionFrame, direction: FrameDirection):
         await super()._handle_interruption(frame, direction)
         await self._stop_audio_context_task()
+        self.reset_active_audio_context()
         self._create_audio_context_task()
 
     def _create_audio_context_task(self):
@@ -1155,6 +1209,7 @@ class AudioContextTTSService(WebsocketTTSService):
         running = True
         while running:
             context_id = await self._contexts_queue.get()
+            self._context_id = context_id
 
             if context_id:
                 # Process the audio context until the context doesn't have more
@@ -1163,11 +1218,15 @@ class AudioContextTTSService(WebsocketTTSService):
 
                 # We just finished processing the context, so we can safely remove it.
                 del self._contexts[context_id]
+                self.reset_active_audio_context()
 
                 # Append some silence between sentences.
                 silence = b"\x00" * self.sample_rate
                 frame = TTSAudioRawFrame(
-                    audio=silence, sample_rate=self.sample_rate, num_channels=1
+                    audio=silence,
+                    sample_rate=self.sample_rate,
+                    num_channels=1,
+                    context_id=context_id,
                 )
                 await self.push_frame(frame)
             else: