NeuphonicHttpTTSService: Refactor to use POST API

CHANGELOG.md

@@ -24,6 +24,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changed
 
+- 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.

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(
+        tts = NeuphonicHttpTTSService(
-        api_key=os.getenv("NEUPHONIC_API_KEY"),
+            api_key=os.getenv("NEUPHONIC_API_KEY"),
-        voice_id="fc854436-2dac-4d21-aa69-ae17b54e98eb",  # Emily
+            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 = [
+        messages = [
-        {
+            {
-            "role": "system",
+                "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.",
+                "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(
+        context = OpenAILLMContext(messages)
-        pipeline,
+        context_aggregator = llm.create_context_aggregator(context)
-        params=PipelineParams(
-            enable_metrics=True,
-            enable_usage_metrics=True,
-        ),
-    )
 
-    @transport.event_handler("on_client_connected")
+        pipeline = Pipeline(
-    async def on_client_connected(transport, client):
+            [
-        logger.info(f"Client connected")
+                transport.input(),  # Transport user input
-        # Kick off the conversation.
+                stt,
-        messages.append({"role": "system", "content": "Please introduce yourself to the user."})
+                context_aggregator.user(),  # User responses
-        await task.queue_frames([context_aggregator.user().get_context_frame()])
+                llm,  # LLM
+                tts,  # TTS
+                transport.output(),  # Transport bot output
+                context_aggregator.assistant(),  # Assistant spoken responses
+            ]
+        )
 
-    @transport.event_handler("on_client_disconnected")
+        task = PipelineTask(
-    async def on_client_disconnected(transport, client):
+            pipeline,
-        logger.info(f"Client disconnected")
+            params=PipelineParams(
-        await task.cancel()
+                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__":

pyproject.toml

@@ -73,7 +73,7 @@ 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,<15.0" ]
+neuphonic = [ "websockets>=13.1,<15.0" ]
 noisereduce = [ "noisereduce~=3.0.3" ]
 openai = [ "websockets>=13.1,<15.0" ]
 openpipe = [ "openpipe~=4.50.0" ]

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,7 +40,6 @@ from pipecat.utils.asyncio.watchdog_async_iterator import WatchdogAsyncIterator
 from pipecat.utils.tracing.service_decorators import traced_tts
 
 try:
-    from pyneuphonic import Neuphonic, TTSConfig
     from websockets.asyncio.client import connect as websocket_connect
     from websockets.protocol import State
 except ModuleNotFoundError as e:
@@ -407,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,
     ):
@@ -418,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".
@@ -429,13 +431,11 @@ class NeuphonicHttpTTSService(TTSService):
         params = params or NeuphonicHttpTTSService.InputParams()
 
         self._api_key = api_key
-        self._url = url
+        self._session = aiohttp_session
-        self._settings = {
+        self._base_url = url.rstrip("/")
-            "lang_code": self.language_to_service_language(params.language),
+        self._lang_code = self.language_to_service_language(params.language) or "en"
-            "speed": params.speed,
+        self._speed = params.speed
-            "encoding": encoding,
+        self._encoding = encoding
-            "sampling_rate": sample_rate,
-        }
         self.set_voice(voice_id)
 
     def can_generate_metrics(self) -> bool:
@@ -473,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.
@@ -485,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)
+            async with self._session.post(url, json=payload, headers=headers) as response:
-            yield TTSStartedFrame()
+                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:
+                await self.start_tts_usage_metrics(text)
-                if message.status_code != 200:
+                yield TTSStartedFrame()
-                    logger.error(f"{self} error: {message.errors}")
-                    yield ErrorFrame(error=f"Neuphonic API error: {message.errors}")
 
-                await self.stop_ttfb_metrics()
+                # Process SSE stream line by line
-                yield TTSAudioRawFrame(message.data.audio, self.sample_rate, 1)
+                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}")
+            logger.exception(f"Error in run_tts: {e}")
-            yield ErrorFrame(error=str(e))
+            yield ErrorFrame(error=f"Neuphonic TTS error: {str(e)}")
         finally:
+            await self.stop_ttfb_metrics()
             yield TTSStoppedFrame()