Added Sarvam TTS Websocket Implementation (#2356)

* Added Sarvam TTS Websocket Implementation

* Addressed some of the comments on PR

* added change voice logic

* added changes from main

* pushing text frames and added flush audio

* updated docs string for better docs

* Addressed comments and added some improvements

* pushed optional args down

* removed new line

* made aiohttp session mandatory in http service

* added push frame and removed unused function

* removed pong message

* added disconnecting logic

---------

Co-authored-by: vinayak-sarvam <vinayak@sarvam.ai>

examples/foundational/07z-interruptible-sarvam-http.py

@@ -0,0 +1,126 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+
+import os
+
+import aiohttp
+from dotenv import load_dotenv
+from loguru import logger
+
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.runner.types import RunnerArguments
+from pipecat.runner.utils import create_transport
+from pipecat.services.deepgram.stt import DeepgramSTTService
+from pipecat.services.openai.llm import OpenAILLMService
+from pipecat.services.sarvam.tts import SarvamHttpTTSService
+from pipecat.transcriptions.language import Language
+from pipecat.transports.base_transport import BaseTransport, TransportParams
+from pipecat.transports.network.fastapi_websocket import FastAPIWebsocketParams
+from pipecat.transports.services.daily import DailyParams
+
+load_dotenv(override=True)
+
+
+# We store functions so objects (e.g. SileroVADAnalyzer) don't get
+# instantiated. The function will be called when the desired transport gets
+# selected.
+transport_params = {
+    "daily": lambda: DailyParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+        vad_analyzer=SileroVADAnalyzer(),
+    ),
+    "twilio": lambda: FastAPIWebsocketParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+        vad_analyzer=SileroVADAnalyzer(),
+    ),
+    "webrtc": lambda: TransportParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+        vad_analyzer=SileroVADAnalyzer(),
+    ),
+}
+
+
+async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
+    logger.info(f"Starting bot")
+
+    # Create an HTTP session
+    async with aiohttp.ClientSession() as session:
+        stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
+
+        tts = SarvamHttpTTSService(
+            api_key=os.getenv("SARVAM_API_KEY"),
+            aiohttp_session=session,
+            params=SarvamHttpTTSService.InputParams(language=Language.EN),
+        )
+
+        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
+            ]
+        )
+
+        task = PipelineTask(
+            pipeline,
+            params=PipelineParams(
+                enable_metrics=True,
+                enable_usage_metrics=True,
+            ),
+            idle_timeout_secs=runner_args.pipeline_idle_timeout_secs,
+        )
+
+        @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()])
+
+        @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=runner_args.handle_sigint)
+
+        await runner.run(task)
+
+
+async def bot(runner_args: RunnerArguments):
+    """Main bot entry point compatible with Pipecat Cloud."""
+    transport = await create_transport(runner_args, transport_params)
+    await run_bot(transport, runner_args)
+
+
+if __name__ == "__main__":
+    from pipecat.runner.run import main
+
+    main()

examples/foundational/07z-interruptible-sarvam.py

@@ -5,6 +5,7 @@
 #
 
 
+import asyncio
 import os
 
 import aiohttp
@@ -12,6 +13,7 @@ from dotenv import load_dotenv
 from loguru import logger
 
 from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.frames.frames import TTSUpdateSettingsFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
@@ -21,7 +23,6 @@ from pipecat.runner.utils import create_transport
 from pipecat.services.deepgram.stt import DeepgramSTTService
 from pipecat.services.openai.llm import OpenAILLMService
 from pipecat.services.sarvam.tts import SarvamTTSService
-from pipecat.transcriptions.language import Language
 from pipecat.transports.base_transport import BaseTransport, TransportParams
 from pipecat.transports.network.fastapi_websocket import FastAPIWebsocketParams
 from pipecat.transports.services.daily import DailyParams
@@ -54,64 +55,64 @@ transport_params = {
 async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
     logger.info(f"Starting bot")
 
-    # Create an HTTP session
+    stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
-    async with aiohttp.ClientSession() as session:
-        stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
 
-        tts = SarvamTTSService(
+    tts = SarvamTTSService(
-            api_key=os.getenv("SARVAM_API_KEY"),
+        api_key=os.getenv("SARVAM_API_KEY"),
-            aiohttp_session=session,
+        model="bulbul:v2",
-            params=SarvamTTSService.InputParams(language=Language.EN),
+        voice_id="manisha",
-        )
+    )
+    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.",
+        },
+    ]
 
-        messages = [
+    context = OpenAILLMContext(messages)
-            {
+    context_aggregator = llm.create_context_aggregator(context)
-                "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.",
+    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
         ]
+    )
 
-        context = OpenAILLMContext(messages)
+    task = PipelineTask(
-        context_aggregator = llm.create_context_aggregator(context)
+        pipeline,
+        params=PipelineParams(
+            enable_metrics=True,
+            enable_usage_metrics=True,
+        ),
+    )
 
-        pipeline = Pipeline(
+    @transport.event_handler("on_client_connected")
-            [
+    async def on_client_connected(transport, client):
-                transport.input(),  # Transport user input
+        logger.info(f"Client connected")
-                stt,
+        # Kick off the conversation.
-                context_aggregator.user(),  # User responses
+        messages.append({"role": "system", "content": "Please introduce yourself to the user."})
-                llm,  # LLM
+        await task.queue_frames([context_aggregator.user().get_context_frame()])
-                tts,  # TTS
-                transport.output(),  # Transport bot output
-                context_aggregator.assistant(),  # Assistant spoken responses
-            ]
-        )
 
-        task = PipelineTask(
+        # Optionally, you can wait for 30 seconds and then change the voice.
-            pipeline,
+        # await asyncio.sleep(30)
-            params=PipelineParams(
+        # await task.queue_frame(TTSUpdateSettingsFrame(settings={"voice": "anushka"}))
-                enable_metrics=True,
-                enable_usage_metrics=True,
-            ),
-            idle_timeout_secs=runner_args.pipeline_idle_timeout_secs,
-        )
 
-        @transport.event_handler("on_client_connected")
+    @transport.event_handler("on_client_disconnected")
-        async def on_client_connected(transport, client):
+    async def on_client_disconnected(transport, client):
-            logger.info(f"Client connected")
+        logger.info(f"Client disconnected")
-            # Kick off the conversation.
+        await task.cancel()
-            messages.append({"role": "system", "content": "Please introduce yourself to the user."})
-            await task.queue_frames([context_aggregator.user().get_context_frame()])
 
-        @transport.event_handler("on_client_disconnected")
+    runner = PipelineRunner(handle_sigint=runner_args.handle_sigint)
-        async def on_client_disconnected(transport, client):
-            logger.info(f"Client disconnected")
-            await task.cancel()
 
-        runner = PipelineRunner(handle_sigint=runner_args.handle_sigint)
+    await runner.run(task)
-
-        await runner.run(task)
 
 
 async def bot(runner_args: RunnerArguments):

pyproject.toml

@@ -90,6 +90,7 @@ rime = [ "websockets>=13.1,<15.0" ]
 riva = [ "nvidia-riva-client~=2.21.1" ]
 runner = [ "python-dotenv>=1.0.0,<2.0.0", "uvicorn>=0.32.0,<1.0.0", "fastapi>=0.115.6,<0.117.0", "pipecat-ai-small-webrtc-prebuilt>=1.0.0"]
 sambanova = []
+sarvam = [ "websockets>=13.1,<15.0" ]
 sentry = [ "sentry-sdk~=2.23.1" ]
 local-smart-turn = [ "coremltools>=8.0", "transformers", "torch>=2.5.0,<3", "torchaudio>=2.5.0,<3" ]
 remote-smart-turn = []

src/pipecat/services/sarvam/tts.py

@@ -6,25 +6,42 @@
 
 """Sarvam AI text-to-speech service implementation."""
 
+import asyncio
 import base64
-from typing import AsyncGenerator, Optional
+import json
+import warnings
+from typing import Any, AsyncGenerator, Mapping, Optional
 
 import aiohttp
 from loguru import logger
 from pydantic import BaseModel, Field
 
 from pipecat.frames.frames import (
+    CancelFrame,
+    EndFrame,
     ErrorFrame,
     Frame,
+    LLMFullResponseEndFrame,
     StartFrame,
+    StartInterruptionFrame,
     TTSAudioRawFrame,
     TTSStartedFrame,
     TTSStoppedFrame,
 )
-from pipecat.services.tts_service import TTSService
+from pipecat.processors.frame_processor import FrameDirection
+from pipecat.services.tts_service import InterruptibleTTSService, TTSService
 from pipecat.transcriptions.language import Language
+from pipecat.utils.asyncio.watchdog_async_iterator import WatchdogAsyncIterator
 from pipecat.utils.tracing.service_decorators import traced_tts
 
+try:
+    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 Sarvam, you need to `pip install pipecat-ai[sarvam]`.")
+    raise Exception(f"Missing module: {e}")
+
 
 def language_to_sarvam_language(language: Language) -> Optional[str]:
     """Convert Pipecat Language enum to Sarvam AI language codes.
@@ -52,7 +69,7 @@ def language_to_sarvam_language(language: Language) -> Optional[str]:
     return LANGUAGE_MAP.get(language)
 
 
-class SarvamTTSService(TTSService):
+class SarvamHttpTTSService(TTSService):
     """Text-to-Speech service using Sarvam AI's API.
 
     Converts text to speech using Sarvam AI's TTS models with support for multiple
@@ -95,9 +112,9 @@ class SarvamTTSService(TTSService):
         self,
         *,
         api_key: str,
+        aiohttp_session: aiohttp.ClientSession,
         voice_id: str = "anushka",
         model: str = "bulbul:v2",
-        aiohttp_session: aiohttp.ClientSession,
         base_url: str = "https://api.sarvam.ai",
         sample_rate: Optional[int] = None,
         params: Optional[InputParams] = None,
@@ -107,9 +124,9 @@ class SarvamTTSService(TTSService):
 
         Args:
             api_key: Sarvam AI API subscription key.
+            aiohttp_session: Shared aiohttp session for making requests.
             voice_id: Speaker voice ID (e.g., "anushka", "meera"). Defaults to "anushka".
             model: TTS model to use ("bulbul:v1" or "bulbul:v2"). Defaults to "bulbul:v2".
-            aiohttp_session: Shared aiohttp session for making requests.
             base_url: Sarvam AI API base URL. Defaults to "https://api.sarvam.ai".
             sample_rate: Audio sample rate in Hz (8000, 16000, 22050, 24000). If None, uses default.
             params: Additional voice and preprocessing parameters. If None, uses defaults.
@@ -117,16 +134,16 @@ class SarvamTTSService(TTSService):
         """
         super().__init__(sample_rate=sample_rate, **kwargs)
 
-        params = params or SarvamTTSService.InputParams()
+        params = params or SarvamHttpTTSService.InputParams()
 
         self._api_key = api_key
         self._base_url = base_url
         self._session = aiohttp_session
 
         self._settings = {
-            "language": self.language_to_service_language(params.language)
+            "language": (
-            if params.language
+                self.language_to_service_language(params.language) if params.language else "en-IN"
-            else "en-IN",
+            ),
             "pitch": params.pitch,
             "pace": params.pace,
             "loudness": params.loudness,
@@ -186,7 +203,7 @@ class SarvamTTSService(TTSService):
                 "pitch": self._settings["pitch"],
                 "pace": self._settings["pace"],
                 "loudness": self._settings["loudness"],
-                "speech_sample_rate": self.sample_rate,
+                "sample_rate": self.sample_rate,
                 "enable_preprocessing": self._settings["enable_preprocessing"],
                 "model": self._model_name,
             }
@@ -240,3 +257,391 @@ class SarvamTTSService(TTSService):
         finally:
             await self.stop_ttfb_metrics()
             yield TTSStoppedFrame()
+
+
+class SarvamTTSService(InterruptibleTTSService):
+    """WebSocket-based text-to-speech service using Sarvam AI.
+
+    Provides streaming TTS with real-time audio generation for multiple Indian languages.
+    Supports voice control parameters like pitch, pace, and loudness adjustment.
+
+    Example::
+
+        tts = SarvamTTSService(
+            api_key="your-api-key",
+            voice_id="anushka",
+            model="bulbul:v2",
+            params=SarvamTTSService.InputParams(
+                language=Language.HI,
+                pitch=0.1,
+                pace=1.2
+            )
+        )
+    """
+
+    class InputParams(BaseModel):
+        """Configuration parameters for Sarvam TTS.
+
+        Parameters:
+            pitch: Voice pitch adjustment (-0.75 to 0.75). Defaults to 0.0.
+            pace: Speech pace multiplier (0.3 to 3.0). Defaults to 1.0.
+            loudness: Volume multiplier (0.1 to 3.0). Defaults to 1.0.
+            enable_preprocessing: Enable text preprocessing. Defaults to False.
+            min_buffer_size: Minimum number of characters to buffer before generating audio.
+                Lower values reduce latency but may affect quality. Defaults to 50.
+            max_chunk_length: Maximum number of characters processed in a single chunk.
+                Controls memory usage and processing efficiency. Defaults to 200.
+            output_audio_codec: Audio codec format. Defaults to "linear16".
+            output_audio_bitrate: Audio bitrate. Defaults to "128k".
+            language: Target language for synthesis. Supports Bengali (bn-IN), English (en-IN),
+                Gujarati (gu-IN), Hindi (hi-IN), Kannada (kn-IN), Malayalam (ml-IN),
+                Marathi (mr-IN), Odia (od-IN), Punjabi (pa-IN), Tamil (ta-IN),
+                Telugu (te-IN). Defaults to en-IN.
+
+                Available Speakers:
+            Female: anushka, manisha, vidya, arya
+            Male: abhilash, karun, hitesh
+        """
+
+        pitch: Optional[float] = Field(default=0.0, ge=-0.75, le=0.75)
+        pace: Optional[float] = Field(default=1.0, ge=0.3, le=3.0)
+        loudness: Optional[float] = Field(default=1.0, ge=0.1, le=3.0)
+        enable_preprocessing: Optional[bool] = False
+        min_buffer_size: Optional[int] = 50
+        max_chunk_length: Optional[int] = 200
+        output_audio_codec: Optional[str] = "linear16"
+        output_audio_bitrate: Optional[str] = "128k"
+        language: Optional[Language] = Language.EN
+
+    def __init__(
+        self,
+        *,
+        api_key: str,
+        model: str = "bulbul:v2",
+        voice_id: str = "anushka",
+        url: str = "wss://api.sarvam.ai/text-to-speech/ws",
+        aiohttp_session: Optional[aiohttp.ClientSession] = None,
+        aggregate_sentences: Optional[bool] = True,
+        sample_rate: Optional[int] = None,
+        params: Optional[InputParams] = None,
+        **kwargs,
+    ):
+        """Initialize the Sarvam TTS service with voice and transport configuration.
+
+        Args:
+            api_key: Sarvam API key for authenticating TTS requests.
+            model: Identifier of the Sarvam speech model (default "bulbul:v2").
+            voice_id: Voice identifier for synthesis (default "anushka").
+            url: WebSocket URL for connecting to the TTS backend (default production URL).
+            aiohttp_session: Optional shared aiohttp session. To maintain backward compatibility.
+                .. deprecated:: 0.0.81
+                    aiohttp_session is no longer used. This parameter will be removed in a future version.
+            aggregate_sentences: Whether to merge multiple sentences into one audio chunk (default True).
+            sample_rate: Desired sample rate for the output audio in Hz (overrides default if set).
+            params: Optional input parameters to override global configuration.
+            **kwargs: Optional keyword arguments forwarded to InterruptibleTTSService (such as
+                `push_stop_frames`, `sample_rate`, task manager parameters, event hooks, etc.)
+                to customize transport behavior or enable metrics support.
+
+        This method sets up the internal TTS configuration mapping, constructs the WebSocket
+        URL based on the chosen model, and initializes state flags before connecting.
+        """
+        # Initialize parent class first
+        super().__init__(
+            aggregate_sentences=aggregate_sentences,
+            push_text_frames=True,
+            pause_frame_processing=True,
+            push_stop_frames=True,
+            sample_rate=sample_rate,
+            **kwargs,
+        )
+        params = params or SarvamTTSService.InputParams()
+        if aiohttp_session is not None:
+            warnings.warn(
+                "The 'aiohttp_session' parameter is deprecated and will be removed in a future version. ",
+                DeprecationWarning,
+                stacklevel=2,
+            )
+        # WebSocket endpoint URL
+        self._websocket_url = f"{url}?model={model}"
+        self._api_key = api_key
+        self.set_model_name(model)
+        self.set_voice(voice_id)
+        # Configuration parameters
+        self._settings = {
+            "target_language_code": (
+                self.language_to_service_language(params.language) if params.language else "en-IN"
+            ),
+            "pitch": params.pitch,
+            "pace": params.pace,
+            "speaker": voice_id,
+            "loudness": params.loudness,
+            "speech_sample_rate": 0,
+            "enable_preprocessing": params.enable_preprocessing,
+            "min_buffer_size": params.min_buffer_size,
+            "max_chunk_length": params.max_chunk_length,
+            "output_audio_codec": params.output_audio_codec,
+            "output_audio_bitrate": params.output_audio_bitrate,
+        }
+        self._started = False
+
+        self._receive_task = None
+        self._keepalive_task = None
+        self._disconnecting = False
+
+    def can_generate_metrics(self) -> bool:
+        """Check if this service can generate processing metrics.
+
+        Returns:
+            True, as Sarvam service supports metrics generation.
+        """
+        return True
+
+    def language_to_service_language(self, language: Language) -> Optional[str]:
+        """Convert a Language enum to Sarvam AI language format.
+
+        Args:
+            language: The language to convert.
+
+        Returns:
+            The Sarvam AI-specific language code, or None if not supported.
+        """
+        return language_to_sarvam_language(language)
+
+    async def start(self, frame: StartFrame):
+        """Start the Sarvam TTS service.
+
+        Args:
+            frame: The start frame containing initialization parameters.
+        """
+        await super().start(frame)
+
+        self._settings["speech_sample_rate"] = self.sample_rate
+        await self._connect()
+
+    async def stop(self, frame: EndFrame):
+        """Stop the Sarvam TTS service.
+
+        Args:
+            frame: The end frame.
+        """
+        await super().stop(frame)
+        await self._disconnect()
+
+    async def cancel(self, frame: CancelFrame):
+        """Cancel the Sarvam TTS service.
+
+        Args:
+            frame: The cancel frame.
+        """
+        await super().cancel(frame)
+        await self._disconnect()
+
+    async def flush_audio(self):
+        """Flush any pending audio synthesis by sending stop command."""
+        if self._websocket:
+            msg = {"type": "flush"}
+            await self._websocket.send(json.dumps(msg))
+
+    async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM):
+        """Push a frame downstream with special handling for stop conditions.
+
+        Args:
+            frame: The frame to push.
+            direction: The direction to push the frame.
+        """
+        await super().push_frame(frame, direction)
+        if isinstance(frame, (TTSStoppedFrame, StartInterruptionFrame)):
+            self._started = False
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        """Process a frame and flush audio if it's the end of a full response."""
+        if isinstance(frame, LLMFullResponseEndFrame):
+            await self.flush_audio()
+        return await super().process_frame(frame, direction)
+
+    async def _update_settings(self, settings: Mapping[str, Any]):
+        """Update service settings and reconnect if voice changed."""
+        prev_voice = self._voice_id
+        await super()._update_settings(settings)
+        if not prev_voice == self._voice_id:
+            logger.info(f"Switching TTS voice to: [{self._voice_id}]")
+            await self._send_config()
+
+    async def _connect(self):
+        """Connect to Sarvam WebSocket and start background tasks."""
+        await self._connect_websocket()
+
+        if self._websocket and not self._receive_task:
+            self._receive_task = self.create_task(self._receive_task_handler(self._report_error))
+
+        if self._websocket and not self._keepalive_task:
+            self._keepalive_task = self.create_task(
+                self._keepalive_task_handler(),
+                watchdog_timeout_secs=25,
+            )
+
+    async def _disconnect(self):
+        """Disconnect from Sarvam WebSocket and clean up tasks."""
+        try:
+            # First, set a flag to prevent new operations
+            self._disconnecting = True
+
+            # Cancel background tasks BEFORE closing websocket
+            if self._receive_task:
+                await self.cancel_task(self._receive_task, timeout=2.0)
+                self._receive_task = None
+
+            if self._keepalive_task:
+                await self.cancel_task(self._keepalive_task, timeout=2.0)
+                self._keepalive_task = None
+
+            # Now close the websocket
+            await self._disconnect_websocket()
+
+        except Exception as e:
+            logger.error(f"Error during disconnect: {e}")
+        finally:
+            # Reset state only after everything is cleaned up
+            self._started = False
+            self._websocket = None
+            self._disconnecting = False
+
+    async def _connect_websocket(self):
+        """Establish WebSocket connection to Sarvam API."""
+        try:
+            if self._websocket and self._websocket.state is State.OPEN:
+                return
+
+            self._websocket = await websocket_connect(
+                self._websocket_url,
+                additional_headers={
+                    "api-subscription-key": self._api_key,
+                },
+            )
+            logger.debug("Connected to Sarvam TTS Websocket")
+            await self._send_config()
+
+        except Exception as e:
+            logger.error(f"{self} initialization error: {e}")
+            self._websocket = None
+            await self._call_event_handler("on_connection_error", f"{e}")
+
+    async def _send_config(self):
+        """Send initial configuration message."""
+        if not self._websocket:
+            raise Exception("WebSocket not connected")
+        self._settings["speaker"] = self._voice_id
+        logger.debug(f"Config being sent is {self._settings}")
+        config_message = {"type": "config", "data": self._settings}
+
+        try:
+            await self._websocket.send(json.dumps(config_message))
+            logger.debug("Configuration sent successfully")
+        except Exception as e:
+            logger.error(f"Failed to send config: {str(e)}")
+            await self.push_frame(ErrorFrame(f"Failed to send config: {str(e)}"))
+            raise
+
+    async def _disconnect_websocket(self):
+        """Close WebSocket connection and clean up state."""
+        try:
+            await self.stop_all_metrics()
+
+            if self._websocket:
+                logger.debug("Disconnecting from Sarvam")
+                await self._websocket.close()
+        except Exception as e:
+            logger.error(f"{self} error closing websocket: {e}")
+
+    def _get_websocket(self):
+        if self._websocket:
+            return self._websocket
+        raise Exception("Websocket not connected")
+
+    async def _receive_messages(self):
+        """Receive and process messages from Sarvam WebSocket."""
+        async for message in WatchdogAsyncIterator(
+            self._get_websocket(), manager=self.task_manager
+        ):
+            if isinstance(message, str):
+                msg = json.loads(message)
+                if msg.get("type") == "audio":
+                    # Check for interruption before processing audio
+                    await self.stop_ttfb_metrics()
+                    audio = base64.b64decode(msg["data"]["audio"])
+                    frame = TTSAudioRawFrame(audio, self.sample_rate, 1)
+                    await self.push_frame(frame)
+                elif msg.get("type") == "error":
+                    error_msg = msg["data"]["message"]
+                    logger.error(f"TTS Error: {error_msg}")
+
+                    # If it's a timeout error, the connection might need to be reset
+                    if "too long" in error_msg.lower() or "timeout" in error_msg.lower():
+                        logger.warning("Connection timeout detected, service may need restart")
+
+                    await self.push_frame(ErrorFrame(f"TTS Error: {error_msg}"))
+
+    async def _keepalive_task_handler(self):
+        """Handle keepalive messages to maintain WebSocket connection."""
+        KEEPALIVE_SLEEP = 20
+        while True:
+            self.reset_watchdog()
+            await asyncio.sleep(KEEPALIVE_SLEEP)
+            await self._send_keepalive()
+
+    async def _send_keepalive(self):
+        """Send keepalive message to maintain connection."""
+        if self._disconnecting:
+            return
+
+        if self._websocket and self._websocket.state == State.OPEN:
+            msg = {"type": "ping"}
+            await self._websocket.send(json.dumps(msg))
+
+    async def _send_text(self, text: str):
+        """Send text to Sarvam WebSocket for synthesis."""
+        if self._disconnecting:
+            logger.warning("Service is disconnecting, ignoring text send")
+            return
+
+        if self._websocket and self._websocket.state == State.OPEN:
+            msg = {"type": "text", "data": {"text": text}}
+            await self._websocket.send(json.dumps(msg))
+        else:
+            logger.warning("WebSocket not ready, cannot send text")
+
+    @traced_tts
+    async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
+        """Generate speech audio frames from input text using Sarvam TTS.
+
+        Sends text over WebSocket for synthesis and yields corresponding audio or status frames.
+
+        Args:
+            text: The text input to synthesize.
+
+        Yields:
+            Frame objects including TTSStartedFrame, TTSAudioRawFrame(s), or TTSStoppedFrame.
+        """
+        logger.debug(f"Generating TTS: [{text}]")
+
+        try:
+            if not self._websocket or self._websocket.state is State.CLOSED:
+                await self._connect()
+
+            try:
+                if not self._started:
+                    await self.start_ttfb_metrics()
+                    yield TTSStartedFrame()
+                    self._started = True
+                await self._send_text(text)
+                await self.start_tts_usage_metrics(text)
+            except Exception as e:
+                logger.error(f"{self} error sending message: {e}")
+                yield TTSStoppedFrame()
+                await self._disconnect()
+                await self._connect()
+                return
+            yield None
+        except Exception as e:
+            logger.error(f"{self} exception: {e}")

uv.lock

@@ -4337,6 +4337,9 @@ runner = [
     { name = "python-dotenv" },
     { name = "uvicorn" },
 ]
+sarvam = [
+    { name = "websockets" },
+]
 sentry = [
     { name = "sentry-sdk" },
 ]
@@ -4491,10 +4494,11 @@ requires-dist = [
     { name = "websockets", marker = "extra == 'openai'", specifier = ">=13.1,<15.0" },
     { name = "websockets", marker = "extra == 'playht'", specifier = ">=13.1,<15.0" },
     { name = "websockets", marker = "extra == 'rime'", specifier = ">=13.1,<15.0" },
+    { name = "websockets", marker = "extra == 'sarvam'", specifier = ">=13.1,<15.0" },
     { name = "websockets", marker = "extra == 'soniox'", specifier = ">=13.1,<15.0" },
     { name = "websockets", marker = "extra == 'websocket'", specifier = ">=13.1,<15.0" },
 ]
-provides-extras = ["anthropic", "assemblyai", "asyncai", "aws", "aws-nova-sonic", "azure", "cartesia", "cerebras", "deepseek", "daily", "deepgram", "elevenlabs", "fal", "fireworks", "fish", "gladia", "google", "grok", "groq", "gstreamer", "heygen", "inworld", "krisp", "koala", "langchain", "livekit", "lmnt", "local", "mcp", "mem0", "mistral", "mlx-whisper", "moondream", "nim", "neuphonic", "noisereduce", "openai", "openpipe", "openrouter", "perplexity", "playht", "qwen", "rime", "riva", "runner", "sambanova", "sentry", "local-smart-turn", "remote-smart-turn", "silero", "simli", "soniox", "soundfile", "speechmatics", "tavus", "together", "tracing", "ultravox", "webrtc", "websocket", "whisper"]
+provides-extras = ["anthropic", "assemblyai", "asyncai", "aws", "aws-nova-sonic", "azure", "cartesia", "cerebras", "deepseek", "daily", "deepgram", "elevenlabs", "fal", "fireworks", "fish", "gladia", "google", "grok", "groq", "gstreamer", "heygen", "inworld", "krisp", "koala", "langchain", "livekit", "lmnt", "local", "mcp", "mem0", "mistral", "mlx-whisper", "moondream", "nim", "neuphonic", "noisereduce", "openai", "openpipe", "openrouter", "perplexity", "playht", "qwen", "rime", "riva", "runner", "sambanova", "sarvam", "sentry", "local-smart-turn", "remote-smart-turn", "silero", "simli", "soniox", "soundfile", "speechmatics", "tavus", "together", "tracing", "ultravox", "webrtc", "websocket", "whisper"]
 
 [package.metadata.requires-dev]
 dev = [