added cambai tts integration

examples/foundational/07za-interruptible-camb.py

@@ -0,0 +1,207 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""Camb.ai MARS-8 TTS example with interruption handling.
+
+This example demonstrates:
+- Basic TTS synthesis with Camb.ai MARS-8
+- Voice selection
+- Speed control
+- Handling interruptions
+
+Requirements:
+- CAMB_API_KEY environment variable
+- OPENAI_API_KEY environment variable (for LLM)
+- DEEPGRAM_API_KEY environment variable (for STT)
+
+Usage:
+    export CAMB_API_KEY=your_camb_api_key
+    export OPENAI_API_KEY=your_openai_api_key
+    export DEEPGRAM_API_KEY=your_deepgram_api_key
+    python 07za-interruptible-camb.py --transport daily
+
+For more information:
+- Camb.ai API docs: https://camb.mintlify.app/
+- Pipecat docs: https://docs.pipecat.ai/
+"""
+
+import os
+
+import aiohttp
+from dotenv import load_dotenv
+from loguru import logger
+
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.audio.vad.vad_analyzer import VADParams
+from pipecat.frames.frames import LLMRunFrame
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.processors.aggregators.llm_response_universal import (
+    LLMContextAggregatorPair,
+)
+from pipecat.runner.types import RunnerArguments
+from pipecat.runner.utils import create_transport
+from pipecat.services.camb.tts import CambTTSService
+from pipecat.services.deepgram.stt import DeepgramSTTService
+from pipecat.services.openai.llm import OpenAILLMService
+from pipecat.transports.base_transport import BaseTransport, TransportParams
+from pipecat.transports.daily.transport import DailyParams
+from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
+
+load_dotenv(override=True)
+
+
+# Transport configuration for different platforms
+transport_params = {
+    "daily": lambda: DailyParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+        vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)),
+    ),
+    "twilio": lambda: FastAPIWebsocketParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+        vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)),
+    ),
+    "webrtc": lambda: TransportParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+        vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)),
+    ),
+}
+
+
+async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
+    """Run the bot with Camb.ai TTS.
+
+    Args:
+        transport: The transport to use for audio I/O.
+        runner_args: Runner arguments from the CLI.
+    """
+    logger.info("Starting Camb.ai TTS bot")
+
+    # Create an HTTP session for the TTS service
+    async with aiohttp.ClientSession() as session:
+        # Initialize Deepgram STT for speech recognition
+        stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
+
+        # Initialize Camb.ai TTS with MARS-8-flash model (fastest)
+        tts = CambTTSService(
+            api_key=os.getenv("CAMB_API_KEY"),
+            aiohttp_session=session,
+            voice_id=2681,  # Attic voice (default)
+            model="mars-8-flash",  # Fast inference model
+            params=CambTTSService.InputParams(
+                speed=1.0,  # Normal speed (0.5-2.0 range)
+            ),
+        )
+
+        # Initialize OpenAI LLM
+        llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
+
+        # System prompt for the assistant
+        messages = [
+            {
+                "role": "system",
+                "content": """You are a helpful voice assistant powered by Camb.ai's MARS-8
+text-to-speech technology. Your goal is to have natural conversations and demonstrate
+high-quality speech synthesis. Keep your responses concise and conversational since
+they will be spoken aloud. Avoid special characters, emojis, or bullet points that
+can't easily be spoken.""",
+            },
+        ]
+
+        # Set up context management
+        context = LLMContext(messages)
+        context_aggregator = LLMContextAggregatorPair(context)
+
+        # Build the pipeline
+        pipeline = Pipeline(
+            [
+                transport.input(),  # Transport user input
+                stt,  # Speech-to-text
+                context_aggregator.user(),  # User context aggregation
+                llm,  # Language model
+                tts,  # Camb.ai TTS
+                transport.output(),  # Transport bot output
+                context_aggregator.assistant(),  # Assistant context aggregation
+            ]
+        )
+
+        # Create the pipeline task
+        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("Client connected")
+            # Start the conversation with a greeting
+            messages.append(
+                {
+                    "role": "system",
+                    "content": "Please introduce yourself briefly and ask how you can help.",
+                }
+            )
+            await task.queue_frames([LLMRunFrame()])
+
+        @transport.event_handler("on_client_disconnected")
+        async def on_client_disconnected(transport, client):
+            logger.info("Client disconnected")
+            await task.cancel()
+
+        # Run the pipeline
+        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.
+
+    Args:
+        runner_args: Arguments passed from the runner.
+    """
+    transport = await create_transport(runner_args, transport_params)
+    await run_bot(transport, runner_args)
+
+
+async def list_available_voices():
+    """Helper function to list available Camb.ai voices.
+
+    Run this to see what voices are available for your API key.
+    """
+    async with aiohttp.ClientSession() as session:
+        voices = await CambTTSService.list_voices(
+            api_key=os.getenv("CAMB_API_KEY"),
+            aiohttp_session=session,
+        )
+        print("\nAvailable Camb.ai voices:")
+        print("-" * 50)
+        for voice in voices:
+            print(f"  ID: {voice['id']}, Name: {voice['name']}, Gender: {voice['gender']}")
+        print("-" * 50)
+        print(f"Total: {len(voices)} voices\n")
+
+
+if __name__ == "__main__":
+    import sys
+
+    # If --list-voices flag is passed, list voices and exit
+    if "--list-voices" in sys.argv:
+        import asyncio
+
+        asyncio.run(list_available_voices())
+    else:
+        from pipecat.runner.run import main
+
+        main()

examples/foundational/test_camb_quick.py

@@ -0,0 +1,181 @@
+#!/usr/bin/env python3
+"""Quick test script to verify Camb.ai TTS integration works.
+
+Usage:
+    export CAMB_API_KEY=your_api_key
+    python test_camb_quick.py
+"""
+
+import asyncio
+import os
+import sys
+
+# Add the src directory to the path so we can import the module
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "src"))
+
+import aiohttp
+from dotenv import load_dotenv
+
+load_dotenv()
+
+
+async def test_list_voices():
+    """Test listing available voices."""
+    from pipecat.services.camb.tts import CambTTSService
+
+    api_key = os.getenv("CAMB_API_KEY")
+    if not api_key:
+        print("ERROR: CAMB_API_KEY environment variable not set!")
+        return False
+
+    print("\n1. Testing list_voices()...")
+    async with aiohttp.ClientSession() as session:
+        try:
+            voices = await CambTTSService.list_voices(
+                api_key=api_key,
+                aiohttp_session=session,
+            )
+            print(f"   SUCCESS: Found {len(voices)} voices")
+            if voices:
+                print(f"   First voice: ID={voices[0]['id']}, Name={voices[0]['name']}")
+            return True
+        except Exception as e:
+            print(f"   FAILED: {e}")
+            import traceback
+            traceback.print_exc()
+            return False
+
+
+async def test_tts_synthesis():
+    """Test basic TTS synthesis."""
+    from pipecat.services.camb.tts import CambTTSService
+    from pipecat.frames.frames import TTSAudioRawFrame, TTSStartedFrame, TTSStoppedFrame, ErrorFrame
+
+    api_key = os.getenv("CAMB_API_KEY")
+    if not api_key:
+        print("ERROR: CAMB_API_KEY environment variable not set!")
+        return False
+
+    print("\n2. Testing TTS synthesis...")
+    async with aiohttp.ClientSession() as session:
+        tts = CambTTSService(
+            api_key=api_key,
+            aiohttp_session=session,
+            voice_id=2681,  # Attic voice
+            model="mars-8-flash",
+        )
+
+        # Manually set sample rate (normally done by StartFrame)
+        tts._sample_rate = 24000
+
+        text = "Hello! This is a test of the Camb.ai text to speech integration."
+        print(f"   Synthesizing: '{text}'")
+
+        audio_bytes = 0
+        frames_received = []
+
+        try:
+            async for frame in tts.run_tts(text):
+                frames_received.append(type(frame).__name__)
+                if isinstance(frame, TTSAudioRawFrame):
+                    audio_bytes += len(frame.audio)
+                elif isinstance(frame, ErrorFrame):
+                    print(f"   FAILED: {frame.error}")
+                    return False
+
+            print(f"   Frames received: {frames_received}")
+            print(f"   Audio bytes received: {audio_bytes}")
+
+            if audio_bytes > 0:
+                print("   SUCCESS: TTS synthesis works!")
+
+                # Optionally save and play audio
+                save_audio = input("\n   Save audio to test_output.wav? (y/n): ").strip().lower()
+                if save_audio == 'y':
+                    await save_audio_to_file(tts, text)
+                    # Try to play the audio
+                    play_audio = input("   Play the audio? (y/n): ").strip().lower()
+                    if play_audio == 'y':
+                        play_wav_file("test_output.wav")
+
+                return True
+            else:
+                print("   FAILED: No audio received")
+                return False
+
+        except Exception as e:
+            print(f"   FAILED: {e}")
+            import traceback
+            traceback.print_exc()
+            return False
+
+
+async def save_audio_to_file(tts, text):
+    """Save synthesized audio to a WAV file."""
+    import wave
+    from pipecat.frames.frames import TTSAudioRawFrame
+
+    audio_data = bytearray()
+    async for frame in tts.run_tts(text):
+        if isinstance(frame, TTSAudioRawFrame):
+            audio_data.extend(frame.audio)
+
+    if audio_data:
+        with wave.open("test_output.wav", "wb") as wav_file:
+            wav_file.setnchannels(1)  # Mono
+            wav_file.setsampwidth(2)  # 16-bit
+            wav_file.setframerate(24000)  # 24kHz
+            wav_file.writeframes(bytes(audio_data))
+        print("   Saved to test_output.wav")
+
+
+def play_wav_file(filepath):
+    """Play a WAV file using the system's default player."""
+    import subprocess
+    import platform
+
+    system = platform.system()
+    try:
+        if system == "Darwin":  # macOS
+            subprocess.run(["afplay", filepath], check=True)
+        elif system == "Linux":
+            subprocess.run(["aplay", filepath], check=True)
+        elif system == "Windows":
+            subprocess.run(["powershell", "-c", f"(New-Object Media.SoundPlayer '{filepath}').PlaySync()"], check=True)
+        else:
+            print(f"   Unsupported platform: {system}. Please play {filepath} manually.")
+    except Exception as e:
+        print(f"   Could not play audio: {e}")
+
+
+async def main():
+    print("=" * 50)
+    print("Camb.ai TTS Integration Test")
+    print("=" * 50)
+
+    results = []
+
+    # Test 1: List voices
+    results.append(await test_list_voices())
+
+    # Test 2: TTS synthesis
+    results.append(await test_tts_synthesis())
+
+    # Summary
+    print("\n" + "=" * 50)
+    print("Summary:")
+    print(f"  List voices: {'PASS' if results[0] else 'FAIL'}")
+    print(f"  TTS synthesis: {'PASS' if results[1] else 'FAIL'}")
+    print("=" * 50)
+
+    if all(results):
+        print("\nAll tests passed!")
+        return 0
+    else:
+        print("\nSome tests failed!")
+        return 1
+
+
+if __name__ == "__main__":
+    exit_code = asyncio.run(main())
+    sys.exit(exit_code)

pyproject.toml

@@ -53,6 +53,7 @@ aws = [ "aioboto3~=15.5.0", "pipecat-ai[websockets-base]" ]
 aws-nova-sonic = [ "aws_sdk_bedrock_runtime~=0.2.0; python_version>='3.12'" ]
 azure = [ "azure-cognitiveservices-speech~=1.44.0"]
 cartesia = [ "cartesia~=2.0.3", "pipecat-ai[websockets-base]" ]
+camb = [ "pipecat-ai[websockets-base]" ]
 cerebras = []
 daily = [ "daily-python~=0.23.0" ]
 deepgram = [ "deepgram-sdk~=4.7.0", "pipecat-ai[websockets-base]" ]

src/pipecat/services/camb/__init__.py

@@ -0,0 +1,8 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+
+from .tts import *

src/pipecat/services/camb/tts.py

@@ -0,0 +1,467 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""Camb.ai MARS-8 text-to-speech service implementation.
+
+This module provides TTS functionality using Camb.ai's MARS-8 model family,
+offering high-quality text-to-speech synthesis with HTTP streaming support.
+
+Features:
+    - MARS-8 models: mars-8, mars-8-flash, mars-8-instruct
+    - 140+ languages supported
+    - Real-time HTTP streaming
+    - 24kHz audio output
+    - Voice customization (speed, instructions)
+"""
+
+import asyncio
+from typing import Any, AsyncGenerator, Dict, List, Mapping, Optional
+
+import aiohttp
+from loguru import logger
+from pydantic import BaseModel, Field
+
+from pipecat.frames.frames import (
+    ErrorFrame,
+    Frame,
+    StartFrame,
+    TTSAudioRawFrame,
+    TTSStartedFrame,
+    TTSStoppedFrame,
+)
+from pipecat.services.tts_service import TTSService
+from pipecat.transcriptions.language import Language, resolve_language
+from pipecat.utils.tracing.service_decorators import traced_tts
+
+
+# Default configuration
+DEFAULT_VOICE_ID = 2681  # Attic voice (publicly available)
+DEFAULT_LANGUAGE = "en-us"
+DEFAULT_MODEL = "mars-8-flash"  # Faster inference
+DEFAULT_BASE_URL = "https://client.camb.ai/apis"
+DEFAULT_SAMPLE_RATE = 24000  # 24kHz
+DEFAULT_TIMEOUT = 60.0  # Seconds (minimum recommended by Camb.ai)
+MIN_TEXT_LENGTH = 3
+MAX_TEXT_LENGTH = 3000
+
+
+def language_to_camb_language(language: Language) -> Optional[str]:
+    """Convert a Pipecat Language enum to Camb.ai language code.
+
+    Args:
+        language: The Language enum value to convert.
+
+    Returns:
+        The corresponding Camb.ai language code (BCP-47 format), or None if not supported.
+    """
+    LANGUAGE_MAP = {
+        Language.EN: "en-us",
+        Language.EN_US: "en-us",
+        Language.EN_GB: "en-gb",
+        Language.EN_AU: "en-au",
+        Language.ES: "es-es",
+        Language.ES_ES: "es-es",
+        Language.ES_MX: "es-mx",
+        Language.FR: "fr-fr",
+        Language.FR_FR: "fr-fr",
+        Language.FR_CA: "fr-ca",
+        Language.DE: "de-de",
+        Language.DE_DE: "de-de",
+        Language.IT: "it-it",
+        Language.PT: "pt-pt",
+        Language.PT_BR: "pt-br",
+        Language.PT_PT: "pt-pt",
+        Language.NL: "nl-nl",
+        Language.PL: "pl-pl",
+        Language.RU: "ru-ru",
+        Language.JA: "ja-jp",
+        Language.KO: "ko-kr",
+        Language.ZH: "zh-cn",
+        Language.ZH_CN: "zh-cn",
+        Language.ZH_TW: "zh-tw",
+        Language.AR: "ar-sa",
+        Language.HI: "hi-in",
+        Language.TR: "tr-tr",
+        Language.VI: "vi-vn",
+        Language.TH: "th-th",
+        Language.ID: "id-id",
+        Language.MS: "ms-my",
+        Language.SV: "sv-se",
+        Language.DA: "da-dk",
+        Language.NO: "no-no",
+        Language.FI: "fi-fi",
+        Language.CS: "cs-cz",
+        Language.EL: "el-gr",
+        Language.HE: "he-il",
+        Language.HU: "hu-hu",
+        Language.RO: "ro-ro",
+        Language.SK: "sk-sk",
+        Language.UK: "uk-ua",
+        Language.BG: "bg-bg",
+        Language.HR: "hr-hr",
+        Language.SR: "sr-rs",
+        Language.SL: "sl-si",
+        Language.CA: "ca-es",
+        Language.EU: "eu-es",
+        Language.GL: "gl-es",
+        Language.AF: "af-za",
+        Language.SW: "sw-ke",
+        Language.TA: "ta-in",
+        Language.TE: "te-in",
+        Language.BN: "bn-in",
+        Language.MR: "mr-in",
+        Language.GU: "gu-in",
+        Language.KN: "kn-in",
+        Language.ML: "ml-in",
+        Language.PA: "pa-in",
+        Language.UR: "ur-pk",
+        Language.FA: "fa-ir",
+        Language.TL: "tl-ph",
+    }
+
+    return resolve_language(language, LANGUAGE_MAP, use_base_code=True)
+
+
+class CambTTSService(TTSService):
+    """Camb.ai MARS-8 HTTP-based text-to-speech service.
+
+    Converts text to speech using Camb.ai's MARS-8 TTS models with support for
+    multiple languages. Provides control over voice characteristics like speed
+    and custom instructions (for mars-8-instruct model).
+
+    Example::
+
+        tts = CambTTSService(
+            api_key="your-api-key",
+            voice_id=2681,
+            model="mars-8-flash",
+            aiohttp_session=session,
+            params=CambTTSService.InputParams(
+                language=Language.EN,
+                speed=1.0
+            )
+        )
+
+        # For mars-8-instruct with custom instructions:
+        tts_instruct = CambTTSService(
+            api_key="your-api-key",
+            voice_id=2681,
+            model="mars-8-instruct",
+            aiohttp_session=session,
+            params=CambTTSService.InputParams(
+                language=Language.EN,
+                user_instructions="Speak with excitement and energy"
+            )
+        )
+    """
+
+    class InputParams(BaseModel):
+        """Input parameters for Camb.ai TTS configuration.
+
+        Parameters:
+            language: Language for synthesis (BCP-47 format). Defaults to English.
+            speed: Speech speed multiplier (0.5 to 2.0). Defaults to 1.0.
+            user_instructions: Custom instructions for mars-8-instruct model only.
+                Ignored for other models. Max 1000 characters.
+        """
+
+        language: Optional[Language] = Language.EN
+        speed: Optional[float] = Field(default=1.0, ge=0.5, le=2.0)
+        user_instructions: Optional[str] = Field(
+            default=None,
+            max_length=1000,
+            description="Custom instructions for mars-8-instruct model only. "
+            "Use to control tone, style, or pronunciation. Max 1000 characters.",
+        )
+
+    def __init__(
+        self,
+        *,
+        api_key: str,
+        aiohttp_session: aiohttp.ClientSession,
+        voice_id: int = DEFAULT_VOICE_ID,
+        model: str = DEFAULT_MODEL,
+        base_url: str = DEFAULT_BASE_URL,
+        sample_rate: Optional[int] = None,
+        params: Optional[InputParams] = None,
+        **kwargs,
+    ):
+        """Initialize the Camb.ai TTS service.
+
+        Args:
+            api_key: Camb.ai API key for authentication.
+            aiohttp_session: Shared aiohttp session for making HTTP requests.
+            voice_id: Voice ID to use (e.g., 2681 for Attic). Defaults to 2681.
+            model: TTS model to use. Options: "mars-8", "mars-8-flash", "mars-8-instruct".
+                Defaults to "mars-8-flash" (fastest).
+            base_url: Camb.ai API base URL. Defaults to production URL.
+            sample_rate: Audio sample rate in Hz. If None, uses Camb.ai default (24000).
+            params: Additional voice parameters. If None, uses defaults.
+            **kwargs: Additional arguments passed to parent TTSService.
+        """
+        super().__init__(sample_rate=sample_rate, **kwargs)
+
+        params = params or CambTTSService.InputParams()
+
+        self._api_key = api_key
+        self._session = aiohttp_session
+
+        # Remove trailing slash from base URL
+        if base_url.endswith("/"):
+            logger.warning("Base URL ends with a slash, removing it.")
+            base_url = base_url[:-1]
+
+        self._base_url = base_url
+
+        # Build settings
+        self._settings = {
+            "language": (
+                self.language_to_service_language(params.language)
+                if params.language
+                else DEFAULT_LANGUAGE
+            ),
+            "speed": params.speed or 1.0,
+            "user_instructions": params.user_instructions,
+        }
+
+        self.set_model_name(model)
+        self.set_voice(str(voice_id))
+        self._voice_id_int = voice_id
+
+    def can_generate_metrics(self) -> bool:
+        """Check if this service can generate processing metrics.
+
+        Returns:
+            True, as Camb.ai service supports metrics generation.
+        """
+        return True
+
+    def language_to_service_language(self, language: Language) -> Optional[str]:
+        """Convert a Language enum to Camb.ai language format.
+
+        Args:
+            language: The language to convert.
+
+        Returns:
+            The Camb.ai-specific language code, or None if not supported.
+        """
+        return language_to_camb_language(language)
+
+    async def start(self, frame: StartFrame):
+        """Start the Camb.ai TTS service.
+
+        Args:
+            frame: The start frame containing initialization parameters.
+        """
+        await super().start(frame)
+        # Use Camb.ai's native sample rate if not specified
+        if not self._init_sample_rate:
+            self._sample_rate = DEFAULT_SAMPLE_RATE
+        self._settings["sample_rate"] = self._sample_rate
+
+    async def _update_settings(self, settings: Mapping[str, Any]):
+        """Update service settings dynamically.
+
+        Args:
+            settings: Dictionary of settings to update.
+        """
+        await super()._update_settings(settings)
+
+        for key, value in settings.items():
+            if key in self._settings:
+                if key == "language" and isinstance(value, Language):
+                    self._settings[key] = language_to_camb_language(value)
+                else:
+                    self._settings[key] = value
+                logger.debug(f"Updated Camb.ai TTS setting {key} to: {value}")
+            elif key == "voice_id":
+                self._voice_id_int = int(value)
+                self.set_voice(str(value))
+
+    @traced_tts
+    async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
+        """Generate speech from text using Camb.ai's TTS API.
+
+        Args:
+            text: The text to synthesize into speech (3-3000 characters).
+
+        Yields:
+            Frame: Audio frames containing the synthesized speech.
+        """
+        logger.debug(f"{self}: Generating TTS [{text}]")
+
+        # Validate text length
+        if len(text) < MIN_TEXT_LENGTH:
+            logger.warning(f"Text too short for Camb.ai TTS (min {MIN_TEXT_LENGTH} chars): {text}")
+            yield TTSStoppedFrame()
+            return
+
+        if len(text) > MAX_TEXT_LENGTH:
+            logger.warning(
+                f"Text too long for Camb.ai TTS (max {MAX_TEXT_LENGTH} chars), truncating"
+            )
+            text = text[:MAX_TEXT_LENGTH]
+
+        # Build request payload
+        payload = {
+            "text": text,
+            "voice_id": self._voice_id_int,
+            "language": self._settings["language"],
+            "speech_model": self._model_name,
+            "output_configuration": {"format": "pcm_s16le"},
+            "voice_settings": {"speed": self._settings["speed"]},
+        }
+
+        # Add user instructions if using mars-8-instruct model
+        if self._model_name == "mars-8-instruct" and self._settings.get("user_instructions"):
+            payload["user_instructions"] = self._settings["user_instructions"]
+
+        headers = {
+            "x-api-key": self._api_key,
+            "Accept": "application/json",
+            "Content-Type": "application/json",
+        }
+
+        try:
+            await self.start_ttfb_metrics()
+
+            async with self._session.post(
+                f"{self._base_url}/tts-stream",
+                json=payload,
+                headers=headers,
+                timeout=aiohttp.ClientTimeout(total=DEFAULT_TIMEOUT),
+            ) as response:
+                if response.status != 200:
+                    error_text = await response.text()
+                    error_msg = self._format_error_message(response.status, error_text)
+                    logger.error(f"{self}: {error_msg}")
+                    yield ErrorFrame(error=error_msg)
+                    return
+
+                await self.start_tts_usage_metrics(text)
+                yield TTSStartedFrame()
+
+                CHUNK_SIZE = self.chunk_size
+
+                async for frame in self._stream_audio_frames_from_iterator(
+                    response.content.iter_chunked(CHUNK_SIZE), strip_wav_header=False
+                ):
+                    await self.stop_ttfb_metrics()
+                    yield frame
+
+        except aiohttp.ClientError as e:
+            error_msg = f"Network error communicating with Camb.ai: {e}"
+            logger.error(f"{self}: {error_msg}")
+            yield ErrorFrame(error=error_msg)
+        except asyncio.TimeoutError:
+            error_msg = f"Timeout waiting for Camb.ai TTS response (>{DEFAULT_TIMEOUT}s)"
+            logger.error(f"{self}: {error_msg}")
+            yield ErrorFrame(error=error_msg)
+        except Exception as e:
+            error_msg = f"Unexpected error in Camb.ai TTS: {e}"
+            logger.error(f"{self}: {error_msg}")
+            yield ErrorFrame(error=error_msg)
+        finally:
+            logger.debug(f"{self}: Finished TTS [{text}]")
+            await self.stop_ttfb_metrics()
+            yield TTSStoppedFrame()
+
+    def _format_error_message(self, status: int, error_text: str) -> str:
+        """Format error message based on HTTP status code.
+
+        Args:
+            status: HTTP status code.
+            error_text: Error response body.
+
+        Returns:
+            Formatted, user-friendly error message.
+        """
+        if status == 401:
+            return (
+                "Invalid Camb.ai API key. "
+                "Set CAMB_API_KEY environment variable with your API key from https://camb.ai"
+            )
+        elif status == 403:
+            return (
+                f"Voice ID {self._voice_id_int} is not accessible with your API key. "
+                "Use list_voices() to see available voices."
+            )
+        elif status == 404:
+            return (
+                f"Invalid voice ID: {self._voice_id_int}. "
+                "Use list_voices() to see available voices."
+            )
+        elif status == 429:
+            return "Camb.ai rate limit exceeded. Please wait before making more requests."
+        elif status >= 500:
+            return f"Camb.ai server error (status {status}): {error_text}"
+        else:
+            return f"Camb.ai API error (status {status}): {error_text}"
+
+    @staticmethod
+    async def list_voices(
+        api_key: str,
+        aiohttp_session: aiohttp.ClientSession,
+        base_url: str = DEFAULT_BASE_URL,
+    ) -> List[Dict[str, Any]]:
+        """Fetch available voices from Camb.ai API.
+
+        Args:
+            api_key: Camb.ai API key for authentication.
+            aiohttp_session: aiohttp ClientSession for making HTTP requests.
+            base_url: Camb.ai API base URL.
+
+        Returns:
+            List of voice dictionaries with id, name, gender, and language fields.
+
+        Raises:
+            Exception: If the API request fails.
+
+        Example::
+
+            async with aiohttp.ClientSession() as session:
+                voices = await CambTTSService.list_voices(
+                    api_key="your-api-key",
+                    aiohttp_session=session,
+                )
+                for voice in voices:
+                    print(f"{voice['id']}: {voice['name']}")
+        """
+        if base_url.endswith("/"):
+            base_url = base_url[:-1]
+
+        headers = {
+            "x-api-key": api_key,
+            "Accept": "application/json",
+        }
+
+        gender_map = {
+            0: "Not Specified",
+            1: "Male",
+            2: "Female",
+            9: "Not Applicable",
+        }
+
+        async with aiohttp_session.get(
+            f"{base_url}/list-voices",
+            headers=headers,
+            timeout=aiohttp.ClientTimeout(total=30.0),
+        ) as response:
+            if response.status != 200:
+                error_text = await response.text()
+                raise Exception(f"Failed to list voices (status {response.status}): {error_text}")
+
+            data = await response.json()
+            return [
+                {
+                    "id": v["id"],
+                    "name": v.get("voice_name", "Unknown"),
+                    "gender": gender_map.get(v.get("gender"), "Unknown"),
+                    "age": v.get("age"),
+                    "language": v.get("language"),
+                }
+                for v in data
+            ]

tests/test_camb_tts.py

@@ -0,0 +1,431 @@
+#
+# Copyright (c) 2024-2025 Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""Tests for CambTTSService.
+
+These tests use mock servers to simulate the Camb.ai API responses.
+"""
+
+import asyncio
+
+import aiohttp
+import pytest
+from aiohttp import web
+
+from pipecat.frames.frames import (
+    AggregatedTextFrame,
+    ErrorFrame,
+    TTSAudioRawFrame,
+    TTSSpeakFrame,
+    TTSStartedFrame,
+    TTSStoppedFrame,
+    TTSTextFrame,
+)
+from pipecat.services.camb.tts import CambTTSService, language_to_camb_language
+from pipecat.tests.utils import run_test
+from pipecat.transcriptions.language import Language
+
+
+@pytest.mark.asyncio
+async def test_run_camb_tts_success(aiohttp_client):
+    """Test successful TTS generation with chunked PCM audio.
+
+    Verifies the frame sequence: TTSStartedFrame -> TTSAudioRawFrame* -> TTSStoppedFrame
+    """
+
+    async def handler(request):
+        # Verify request headers
+        assert request.headers.get("x-api-key") == "test-api-key"
+        assert request.headers.get("Content-Type") == "application/json"
+
+        # Parse and verify request body
+        body = await request.json()
+        assert "text" in body
+        assert body["voice_id"] == 2681
+        assert body["language"] == "en-us"
+        assert body["speech_model"] == "mars-8-flash"
+        assert body["output_configuration"]["format"] == "pcm_s16le"
+
+        # Prepare a StreamResponse with chunked PCM data
+        resp = web.StreamResponse(
+            status=200,
+            reason="OK",
+            headers={"Content-Type": "audio/raw"},
+        )
+        await resp.prepare(request)
+
+        # Write out chunked PCM byte data (16-bit samples)
+        # Use smaller chunks for more predictable frame count
+        data = b"\x00\x01" * 4800  # Small chunk of audio
+        await resp.write(data)
+        await resp.write_eof()
+
+        return resp
+
+    # Create an aiohttp test server
+    app = web.Application()
+    app.router.add_post("/tts-stream", handler)
+    client = await aiohttp_client(app)
+
+    base_url = str(client.make_url("")).rstrip("/")
+
+    async with aiohttp.ClientSession() as session:
+        tts_service = CambTTSService(
+            api_key="test-api-key",
+            aiohttp_session=session,
+            base_url=base_url,
+            voice_id=2681,
+            model="mars-8-flash",
+        )
+
+        # Manually set sample rate (normally done by StartFrame)
+        tts_service._sample_rate = 24000
+
+        # Test run_tts directly to avoid frame count variability
+        text = "Hello world, this is a test."
+        frames = []
+        async for frame in tts_service.run_tts(text):
+            frames.append(frame)
+
+        # Verify we got the expected frame types
+        frame_types = [type(f).__name__ for f in frames]
+        assert "TTSStartedFrame" in frame_types, "Should have TTSStartedFrame"
+        assert "TTSAudioRawFrame" in frame_types, "Should have TTSAudioRawFrame"
+        assert "TTSStoppedFrame" in frame_types, "Should have TTSStoppedFrame"
+
+        audio_frames = [f for f in frames if isinstance(f, TTSAudioRawFrame)]
+        assert len(audio_frames) > 0, "Should have at least one audio frame"
+
+        # Verify sample rate matches Camb.ai's output (24kHz)
+        for a_frame in audio_frames:
+            assert a_frame.sample_rate == 24000, "Sample rate should be 24000 Hz"
+            assert a_frame.num_channels == 1, "Should be mono audio"
+
+
+@pytest.mark.asyncio
+async def test_run_camb_tts_error_401(aiohttp_client):
+    """Test handling of invalid API key (401 Unauthorized)."""
+
+    async def handler(request):
+        return web.Response(
+            status=401,
+            text="Unauthorized: Invalid API key",
+        )
+
+    app = web.Application()
+    app.router.add_post("/tts-stream", handler)
+    client = await aiohttp_client(app)
+    base_url = str(client.make_url("")).rstrip("/")
+
+    async with aiohttp.ClientSession() as session:
+        tts_service = CambTTSService(
+            api_key="invalid-key",
+            aiohttp_session=session,
+            base_url=base_url,
+        )
+
+        frames_to_send = [
+            TTSSpeakFrame(text="This should fail."),
+        ]
+
+        expected_down_frames = [AggregatedTextFrame, TTSStoppedFrame, TTSTextFrame]
+        expected_up_frames = [ErrorFrame]
+
+        frames_received = await run_test(
+            tts_service,
+            frames_to_send=frames_to_send,
+            expected_down_frames=expected_down_frames,
+            expected_up_frames=expected_up_frames,
+        )
+        up_frames = frames_received[1]
+
+        assert isinstance(up_frames[0], ErrorFrame), "Must receive an ErrorFrame for 401"
+        assert "Invalid Camb.ai API key" in up_frames[0].error, (
+            "ErrorFrame should mention invalid API key"
+        )
+
+
+@pytest.mark.asyncio
+async def test_run_camb_tts_error_404(aiohttp_client):
+    """Test handling of invalid voice ID (404 Not Found)."""
+
+    async def handler(request):
+        return web.Response(
+            status=404,
+            text="Voice not found",
+        )
+
+    app = web.Application()
+    app.router.add_post("/tts-stream", handler)
+    client = await aiohttp_client(app)
+    base_url = str(client.make_url("")).rstrip("/")
+
+    async with aiohttp.ClientSession() as session:
+        tts_service = CambTTSService(
+            api_key="test-api-key",
+            aiohttp_session=session,
+            base_url=base_url,
+            voice_id=99999,  # Invalid voice ID
+        )
+
+        frames_to_send = [
+            TTSSpeakFrame(text="This should fail."),
+        ]
+
+        expected_down_frames = [AggregatedTextFrame, TTSStoppedFrame, TTSTextFrame]
+        expected_up_frames = [ErrorFrame]
+
+        frames_received = await run_test(
+            tts_service,
+            frames_to_send=frames_to_send,
+            expected_down_frames=expected_down_frames,
+            expected_up_frames=expected_up_frames,
+        )
+        up_frames = frames_received[1]
+
+        assert isinstance(up_frames[0], ErrorFrame), "Must receive an ErrorFrame for 404"
+        assert "Invalid voice ID" in up_frames[0].error, (
+            "ErrorFrame should mention invalid voice ID"
+        )
+
+
+@pytest.mark.asyncio
+async def test_run_camb_tts_error_429(aiohttp_client):
+    """Test handling of rate limit (429 Too Many Requests)."""
+
+    async def handler(request):
+        return web.Response(
+            status=429,
+            text="Rate limit exceeded",
+        )
+
+    app = web.Application()
+    app.router.add_post("/tts-stream", handler)
+    client = await aiohttp_client(app)
+    base_url = str(client.make_url("")).rstrip("/")
+
+    async with aiohttp.ClientSession() as session:
+        tts_service = CambTTSService(
+            api_key="test-api-key",
+            aiohttp_session=session,
+            base_url=base_url,
+        )
+
+        frames_to_send = [
+            TTSSpeakFrame(text="This should fail due to rate limit."),
+        ]
+
+        expected_down_frames = [AggregatedTextFrame, TTSStoppedFrame, TTSTextFrame]
+        expected_up_frames = [ErrorFrame]
+
+        frames_received = await run_test(
+            tts_service,
+            frames_to_send=frames_to_send,
+            expected_down_frames=expected_down_frames,
+            expected_up_frames=expected_up_frames,
+        )
+        up_frames = frames_received[1]
+
+        assert isinstance(up_frames[0], ErrorFrame), "Must receive an ErrorFrame for 429"
+        assert "rate limit" in up_frames[0].error.lower(), (
+            "ErrorFrame should mention rate limit"
+        )
+
+
+@pytest.mark.asyncio
+async def test_list_voices(aiohttp_client):
+    """Test voice listing endpoint."""
+
+    async def handler(request):
+        # Verify API key header
+        assert request.headers.get("x-api-key") == "test-api-key"
+
+        # Return mock voice data (matching actual API response structure)
+        voices = [
+            {
+                "id": 2681,
+                "voice_name": "Attic",
+                "gender": 1,
+                "age": 25,
+                "language": None,
+                "transcript": None,
+                "description": None,
+                "is_published": None,
+            },
+            {
+                "id": 2682,
+                "voice_name": "Cellar",
+                "gender": 2,
+                "age": 30,
+                "language": 1,
+                "transcript": None,
+                "description": None,
+                "is_published": False,
+            },
+        ]
+        return web.json_response(voices)
+
+    app = web.Application()
+    app.router.add_get("/list-voices", handler)
+    client = await aiohttp_client(app)
+    base_url = str(client.make_url("")).rstrip("/")
+
+    async with aiohttp.ClientSession() as session:
+        voices = await CambTTSService.list_voices(
+            api_key="test-api-key",
+            aiohttp_session=session,
+            base_url=base_url,
+        )
+
+        # Should return all voices
+        assert len(voices) == 2, "Should return all voices"
+
+        # Verify voice data structure
+        attic_voice = next(v for v in voices if v["id"] == 2681)
+        assert attic_voice["name"] == "Attic"
+        assert attic_voice["gender"] == "Male"
+        assert attic_voice["age"] == 25
+
+
+@pytest.mark.asyncio
+async def test_text_length_validation_too_short(aiohttp_client):
+    """Test that text shorter than 3 characters is handled gracefully."""
+
+    async def handler(request):
+        # This should not be called for short text
+        pytest.fail("Handler should not be called for text < 3 chars")
+
+    app = web.Application()
+    app.router.add_post("/tts-stream", handler)
+    client = await aiohttp_client(app)
+    base_url = str(client.make_url("")).rstrip("/")
+
+    async with aiohttp.ClientSession() as session:
+        tts_service = CambTTSService(
+            api_key="test-api-key",
+            aiohttp_session=session,
+            base_url=base_url,
+        )
+
+        frames_to_send = [
+            TTSSpeakFrame(text="Hi"),  # Only 2 characters
+        ]
+
+        # For short text, we expect TTSStoppedFrame but no audio
+        expected_down_frames = [AggregatedTextFrame, TTSStoppedFrame, TTSTextFrame]
+
+        frames_received = await run_test(
+            tts_service,
+            frames_to_send=frames_to_send,
+            expected_down_frames=expected_down_frames,
+        )
+        down_frames = frames_received[0]
+
+        # Verify no audio frames were generated
+        audio_frames = [f for f in down_frames if isinstance(f, TTSAudioRawFrame)]
+        assert len(audio_frames) == 0, "Should not generate audio for text < 3 chars"
+
+
+@pytest.mark.asyncio
+async def test_input_params():
+    """Test InputParams model validation and defaults."""
+
+    # Test defaults
+    params = CambTTSService.InputParams()
+    assert params.language == Language.EN
+    assert params.speed == 1.0
+    assert params.user_instructions is None
+
+    # Test custom values
+    params = CambTTSService.InputParams(
+        language=Language.ES,
+        speed=1.5,
+        user_instructions="Speak slowly and clearly",
+    )
+    assert params.language == Language.ES
+    assert params.speed == 1.5
+    assert params.user_instructions == "Speak slowly and clearly"
+
+
+@pytest.mark.asyncio
+async def test_language_mapping():
+    """Test language enum to Camb.ai language code conversion."""
+
+    # Test common languages
+    assert language_to_camb_language(Language.EN) == "en-us"
+    assert language_to_camb_language(Language.EN_US) == "en-us"
+    assert language_to_camb_language(Language.EN_GB) == "en-gb"
+    assert language_to_camb_language(Language.ES) == "es-es"
+    assert language_to_camb_language(Language.FR) == "fr-fr"
+    assert language_to_camb_language(Language.DE) == "de-de"
+    assert language_to_camb_language(Language.JA) == "ja-jp"
+    assert language_to_camb_language(Language.ZH) == "zh-cn"
+
+
+@pytest.mark.asyncio
+async def test_mars8_instruct_model(aiohttp_client):
+    """Test that user_instructions are included for mars-8-instruct model."""
+
+    received_payload = {}
+
+    async def handler(request):
+        nonlocal received_payload
+        received_payload = await request.json()
+
+        # Return minimal successful response
+        resp = web.StreamResponse(status=200, headers={"Content-Type": "audio/raw"})
+        await resp.prepare(request)
+        await resp.write(b"\x00" * 1000)
+        await resp.write_eof()
+        return resp
+
+    app = web.Application()
+    app.router.add_post("/tts-stream", handler)
+    client = await aiohttp_client(app)
+    base_url = str(client.make_url("")).rstrip("/")
+
+    async with aiohttp.ClientSession() as session:
+        tts_service = CambTTSService(
+            api_key="test-api-key",
+            aiohttp_session=session,
+            base_url=base_url,
+            model="mars-8-instruct",
+            params=CambTTSService.InputParams(user_instructions="Speak with excitement"),
+        )
+
+        frames_to_send = [
+            TTSSpeakFrame(text="This is exciting news!"),
+        ]
+
+        await run_test(
+            tts_service,
+            frames_to_send=frames_to_send,
+            expected_down_frames=[
+                AggregatedTextFrame,
+                TTSStartedFrame,
+                TTSAudioRawFrame,
+                TTSStoppedFrame,
+                TTSTextFrame,
+            ],
+        )
+
+        # Verify user_instructions was included in the request
+        assert received_payload.get("speech_model") == "mars-8-instruct"
+        assert received_payload.get("user_instructions") == "Speak with excitement"
+
+
+@pytest.mark.asyncio
+async def test_base_url_trailing_slash():
+    """Test that trailing slash in base URL is handled correctly."""
+    async with aiohttp.ClientSession() as session:
+        tts = CambTTSService(
+            api_key="test-key",
+            aiohttp_session=session,
+            base_url="https://api.example.com/",  # With trailing slash
+        )
+
+        # Should have removed the trailing slash
+        assert tts._base_url == "https://api.example.com"

uv.lock

@@ -4259,6 +4259,9 @@ aws-nova-sonic = [
 azure = [
     { name = "azure-cognitiveservices-speech" },
 ]
+camb = [
+    { name = "websockets" },
+]
 cartesia = [
     { name = "cartesia" },
     { name = "websockets" },
@@ -4525,6 +4528,7 @@ requires-dist = [
     { name = "pipecat-ai", extras = ["websockets-base"], marker = "extra == 'assemblyai'" },
     { name = "pipecat-ai", extras = ["websockets-base"], marker = "extra == 'asyncai'" },
     { name = "pipecat-ai", extras = ["websockets-base"], marker = "extra == 'aws'" },
+    { name = "pipecat-ai", extras = ["websockets-base"], marker = "extra == 'camb'" },
     { name = "pipecat-ai", extras = ["websockets-base"], marker = "extra == 'cartesia'" },
     { name = "pipecat-ai", extras = ["websockets-base"], marker = "extra == 'deepgram'" },
     { name = "pipecat-ai", extras = ["websockets-base"], marker = "extra == 'elevenlabs'" },
@@ -4573,7 +4577,7 @@ requires-dist = [
     { name = "wait-for2", marker = "python_full_version < '3.12'", specifier = ">=0.4.1" },
     { name = "websockets", marker = "extra == 'websockets-base'", specifier = ">=13.1,<16.0" },
 ]
-provides-extras = ["aic", "anthropic", "assemblyai", "asyncai", "aws", "aws-nova-sonic", "azure", "cartesia", "cerebras", "daily", "deepgram", "deepseek", "elevenlabs", "fal", "fireworks", "fish", "gladia", "google", "gradium", "grok", "groq", "gstreamer", "heygen", "hume", "inworld", "koala", "krisp", "langchain", "livekit", "lmnt", "local", "local-smart-turn", "local-smart-turn-v3", "mcp", "mem0", "mistral", "mlx-whisper", "moondream", "neuphonic", "noisereduce", "nvidia", "openai", "rnnoise", "openpipe", "openrouter", "perplexity", "playht", "qwen", "remote-smart-turn", "rime", "riva", "runner", "sagemaker", "sambanova", "sarvam", "sentry", "silero", "simli", "soniox", "soundfile", "speechmatics", "strands", "tavus", "together", "tracing", "ultravox", "webrtc", "websocket", "websockets-base", "whisper"]
+provides-extras = ["aic", "anthropic", "assemblyai", "asyncai", "aws", "aws-nova-sonic", "azure", "cartesia", "camb", "cerebras", "daily", "deepgram", "deepseek", "elevenlabs", "fal", "fireworks", "fish", "gladia", "google", "gradium", "grok", "groq", "gstreamer", "heygen", "hume", "inworld", "koala", "krisp", "langchain", "livekit", "lmnt", "local", "local-smart-turn", "local-smart-turn-v3", "mcp", "mem0", "mistral", "mlx-whisper", "moondream", "neuphonic", "noisereduce", "nvidia", "openai", "rnnoise", "openpipe", "openrouter", "perplexity", "playht", "qwen", "remote-smart-turn", "rime", "riva", "runner", "sagemaker", "sambanova", "sarvam", "sentry", "silero", "simli", "soniox", "soundfile", "speechmatics", "strands", "tavus", "together", "tracing", "ultravox", "webrtc", "websocket", "websockets-base", "whisper"]
 
 [package.metadata.requires-dev]
 dev = [