feat: add CAMB AI text-to-speech provider integration

litellm/llms/camb_ai/common_utils.py

@@ -0,0 +1,5 @@
+from litellm.llms.base_llm.chat.transformation import BaseLLMException
+
+
+class CambAIException(BaseLLMException):
+    pass

litellm/llms/camb_ai/text_to_speech/transformation.py

@@ -0,0 +1,220 @@
+"""
+CAMB AI Text-to-Speech transformation
+
+Maps OpenAI TTS spec to CAMB AI TTS streaming API
+"""
+
+from typing import TYPE_CHECKING, Any, Dict, Optional, Tuple, Union
+
+import httpx
+from httpx import Headers
+
+import litellm
+from litellm.llms.base_llm.chat.transformation import BaseLLMException
+from litellm.llms.base_llm.text_to_speech.transformation import (
+    BaseTextToSpeechConfig,
+    TextToSpeechRequestData,
+)
+from litellm.secret_managers.main import get_secret_str
+
+from ..common_utils import CambAIException
+
+if TYPE_CHECKING:
+    from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
+    from litellm.types.llms.openai import HttpxBinaryResponseContent
+else:
+    LiteLLMLoggingObj = Any
+    HttpxBinaryResponseContent = Any
+
+
+class CambAITextToSpeechConfig(BaseTextToSpeechConfig):
+    """
+    Configuration for CAMB AI Text-to-Speech
+
+    Reference: https://docs.camb.ai
+    """
+
+    TTS_BASE_URL = "https://client.camb.ai/apis"
+    TTS_ENDPOINT_PATH = "/tts-stream"
+
+    # Response format mappings from OpenAI to CAMB AI output_configuration
+    FORMAT_MAPPINGS = {
+        "mp3": "mp3",
+        "wav": "wav",
+        "pcm": "pcm",
+        "flac": "flac",
+    }
+
+    def get_supported_openai_params(self, model: str) -> list:
+        return ["voice", "response_format", "language"]
+
+    def map_openai_params(
+        self,
+        model: str,
+        optional_params: Dict,
+        voice: Optional[Union[str, Dict]] = None,
+        drop_params: bool = False,
+        kwargs: Optional[Dict[str, Any]] = None,
+    ) -> Tuple[Optional[str], Dict]:
+        mapped_params: Dict[str, Any] = {}
+        params = dict(optional_params) if optional_params else {}
+
+        # Extract voice_id — CAMB AI uses integer voice IDs
+        voice_id: Optional[str] = None
+        if isinstance(voice, str) and voice.strip():
+            voice_id = voice.strip()
+        elif isinstance(voice, dict):
+            for key in ("voice_id", "id", "name"):
+                candidate = voice.get(key)
+                if isinstance(candidate, (str, int)) and str(candidate).strip():
+                    voice_id = str(candidate).strip()
+                    break
+
+        if voice_id is not None:
+            try:
+                mapped_params["voice_id"] = int(voice_id)
+            except (TypeError, ValueError):
+                mapped_params["voice_id"] = voice_id
+
+        # Response format
+        response_format = params.pop("response_format", None)
+        if isinstance(response_format, str):
+            mapped_format = self.FORMAT_MAPPINGS.get(response_format, response_format)
+            mapped_params["output_configuration"] = {"format": mapped_format}
+
+        # Speed — drop silently (CAMB AI doesn't support it directly)
+        params.pop("speed", None)
+
+        # Instructions — OpenAI-specific, omit
+        params.pop("instructions", None)
+
+        # Pass through remaining params
+        for key, value in params.items():
+            if value is None:
+                continue
+            mapped_params[key] = value
+
+        return voice_id, mapped_params
+
+    def validate_environment(
+        self,
+        headers: dict,
+        model: str,
+        api_key: Optional[str] = None,
+        api_base: Optional[str] = None,
+    ) -> dict:
+        api_key = (
+            api_key
+            or litellm.api_key
+            or get_secret_str("CAMB_API_KEY")
+        )
+
+        if api_key is None:
+            raise ValueError(
+                "CAMB AI API key is required. Set CAMB_API_KEY environment variable."
+            )
+
+        headers.update(
+            {
+                "x-api-key": api_key,
+                "Content-Type": "application/json",
+            }
+        )
+
+        return headers
+
+    def get_error_class(
+        self, error_message: str, status_code: int, headers: Union[dict, Headers]
+    ) -> BaseLLMException:
+        return CambAIException(
+            message=error_message, status_code=status_code, headers=headers
+        )
+
+    def transform_text_to_speech_request(
+        self,
+        model: str,
+        input: str,
+        voice: Optional[str],
+        optional_params: Dict,
+        litellm_params: Dict,
+        headers: dict,
+    ) -> TextToSpeechRequestData:
+        params = dict(optional_params) if optional_params else {}
+        extra_body = params.pop("extra_body", None)
+
+        # Get language from kwargs/litellm_params, default to English (US)
+        language = litellm_params.get("language", params.pop("language", "en-us"))
+
+        request_body: Dict[str, Any] = {
+            "text": input,
+            "language": language,
+            "speech_model": model,
+        }
+
+        # Apply response_format → output_configuration mapping
+        response_format = params.pop("response_format", None)
+        if isinstance(response_format, str):
+            mapped_format = self.FORMAT_MAPPINGS.get(response_format, response_format)
+            request_body["output_configuration"] = {"format": mapped_format}
+
+        # Drop speed silently (CAMB AI doesn't support it)
+        params.pop("speed", None)
+
+        # Add voice_id if present
+        voice_id = params.pop("voice_id", None)
+        if voice_id is not None:
+            request_body["voice_id"] = voice_id
+        elif voice is not None:
+            # Support dict-style voice extraction
+            if isinstance(voice, dict):
+                for key in ("voice_id", "id", "name"):
+                    candidate = voice.get(key)
+                    if isinstance(candidate, (str, int)) and str(candidate).strip():
+                        voice = str(candidate).strip()
+                        break
+            try:
+                request_body["voice_id"] = int(voice)
+            except (TypeError, ValueError):
+                request_body["voice_id"] = voice
+
+        # Add remaining params
+        for key, value in params.items():
+            if value is None:
+                continue
+            request_body[key] = value
+
+        if isinstance(extra_body, dict):
+            for key, value in extra_body.items():
+                if value is None:
+                    continue
+                request_body[key] = value
+
+        return TextToSpeechRequestData(
+            dict_body=request_body,
+            headers={"Content-Type": "application/json"},
+        )
+
+    def transform_text_to_speech_response(
+        self,
+        model: str,
+        raw_response: httpx.Response,
+        logging_obj: LiteLLMLoggingObj,
+    ) -> "HttpxBinaryResponseContent":
+        from litellm.types.llms.openai import HttpxBinaryResponseContent
+
+        return HttpxBinaryResponseContent(raw_response)
+
+    def get_complete_url(
+        self,
+        model: str,
+        api_base: Optional[str],
+        litellm_params: dict,
+    ) -> str:
+        base_url = (
+            api_base
+            or get_secret_str("CAMB_API_BASE")
+            or self.TTS_BASE_URL
+        )
+        base_url = base_url.rstrip("/")
+
+        return f"{base_url}{self.TTS_ENDPOINT_PATH}"

litellm/main.py

@@ -6834,6 +6834,44 @@ def speech(  # noqa: PLR0915
             client=client,
             _is_async=aspeech or False,
         )
+    elif custom_llm_provider == "camb_ai":
+        from litellm.llms.camb_ai.text_to_speech.transformation import (
+            CambAITextToSpeechConfig,
+        )
+
+        if text_to_speech_provider_config is None:
+            text_to_speech_provider_config = CambAITextToSpeechConfig()
+
+        camb_ai_config = cast(
+            CambAITextToSpeechConfig, text_to_speech_provider_config
+        )
+
+        voice_id = voice if isinstance(voice, str) else None
+
+        if api_base is not None:
+            litellm_params_dict["api_base"] = api_base
+        if api_key is not None:
+            litellm_params_dict["api_key"] = api_key
+
+        # Pass language through litellm_params if provided in kwargs
+        language = kwargs.pop("language", None)
+        if language is not None:
+            litellm_params_dict["language"] = language
+
+        response = base_llm_http_handler.text_to_speech_handler(
+            model=model,
+            input=input,
+            voice=voice_id,
+            text_to_speech_provider_config=camb_ai_config,
+            text_to_speech_optional_params=optional_params,
+            custom_llm_provider=custom_llm_provider,
+            litellm_params=litellm_params_dict,
+            logging_obj=logging_obj,
+            timeout=timeout,
+            extra_headers=extra_headers,
+            client=client,
+            _is_async=aspeech or False,
+        )
     elif custom_llm_provider == "vertex_ai" or custom_llm_provider == "vertex_ai_beta":
         from litellm.llms.vertex_ai.text_to_speech.transformation import (
             VertexAITextToSpeechConfig,

litellm/model_prices_and_context_window_backup.json

@@ -12098,6 +12098,33 @@
             "/v1/audio/speech"
         ]
     },
+    "camb_ai/mars-pro": {
+        "input_cost_per_character": 0.0,
+        "litellm_provider": "camb_ai",
+        "mode": "audio_speech",
+        "source": "https://camb.ai",
+        "supported_endpoints": [
+            "/v1/audio/speech"
+        ]
+    },
+    "camb_ai/mars-flash": {
+        "input_cost_per_character": 0.0,
+        "litellm_provider": "camb_ai",
+        "mode": "audio_speech",
+        "source": "https://camb.ai",
+        "supported_endpoints": [
+            "/v1/audio/speech"
+        ]
+    },
+    "camb_ai/mars-instruct": {
+        "input_cost_per_character": 0.0,
+        "litellm_provider": "camb_ai",
+        "mode": "audio_speech",
+        "source": "https://camb.ai",
+        "supported_endpoints": [
+            "/v1/audio/speech"
+        ]
+    },
     "embed-english-light-v2.0": {
         "input_cost_per_token": 1e-07,
         "litellm_provider": "cohere",

litellm/types/utils.py

@@ -3202,6 +3202,7 @@ class LlmProviders(str, Enum):
     CHUTES = "chutes"
     XIAOMI_MIMO = "xiaomi_mimo"
     LITELLM_AGENT = "litellm_agent"
+    CAMB_AI = "camb_ai"
     CURSOR = "cursor"
 
 

litellm/utils.py

@@ -8935,6 +8935,12 @@ class ProviderConfigManager:
             )
 
             return AWSPollyTextToSpeechConfig()
+        elif litellm.LlmProviders.CAMB_AI == provider:
+            from litellm.llms.camb_ai.text_to_speech.transformation import (
+                CambAITextToSpeechConfig,
+            )
+
+            return CambAITextToSpeechConfig()
         return None
 
     @staticmethod

tests/audio_tests/test_audio_speech.py

@@ -684,3 +684,73 @@ async def test_aws_polly_tts_real_api():
     assert speech_file_path.stat().st_size > 0
 
     print(f"AWS Polly TTS audio saved to: {speech_file_path}")
+
+
+@pytest.mark.asyncio
+async def test_camb_ai_tts_request_body():
+    """
+    Test CAMB AI TTS request body is formatted correctly.
+    Verifies the full litellm.aspeech() -> HTTP handler flow.
+    """
+    import json
+    from unittest.mock import MagicMock, patch
+    import httpx
+
+    mock_response_content = b"fake_audio_data"
+    mock_httpx_response = MagicMock(spec=httpx.Response)
+    mock_httpx_response.content = mock_response_content
+    mock_httpx_response.status_code = 200
+    mock_httpx_response.headers = httpx.Headers({"content-type": "audio/mpeg"})
+
+    with patch("litellm.llms.custom_httpx.http_handler.AsyncHTTPHandler.post") as mock_post:
+        mock_post.return_value = mock_httpx_response
+
+        response = await litellm.aspeech(
+            model="camb_ai/mars-flash",
+            voice="123",
+            input="Hello world",
+            api_key="test-key",
+            language="en-us",
+        )
+
+        assert mock_post.called
+
+        call_args = mock_post.call_args
+        request_body = call_args.kwargs.get("json")
+
+        assert request_body["text"] == "Hello world"
+        assert request_body["speech_model"] == "mars-flash"
+        assert request_body["voice_id"] == 123
+        assert request_body["language"] == "en-us"
+
+
+@pytest.mark.asyncio
+async def test_camb_ai_tts_request_url_and_headers():
+    """
+    Test CAMB AI TTS sends to the correct URL with proper auth headers.
+    """
+    import json
+    from unittest.mock import MagicMock, patch
+    import httpx
+
+    mock_httpx_response = MagicMock(spec=httpx.Response)
+    mock_httpx_response.content = b"fake_audio_data"
+    mock_httpx_response.status_code = 200
+    mock_httpx_response.headers = httpx.Headers({"content-type": "audio/mpeg"})
+
+    with patch("litellm.llms.custom_httpx.http_handler.AsyncHTTPHandler.post") as mock_post:
+        mock_post.return_value = mock_httpx_response
+
+        await litellm.aspeech(
+            model="camb_ai/mars-flash",
+            voice="456",
+            input="Test URL and headers",
+            api_key="my-secret-key",
+        )
+
+        call_args = mock_post.call_args
+        url = call_args.kwargs.get("url", call_args.args[0] if call_args.args else None)
+        headers = call_args.kwargs.get("headers", {})
+
+        assert "client.camb.ai" in str(url)
+        assert headers.get("x-api-key") == "my-secret-key"

tests/test_litellm/test_camb_ai.py

@@ -0,0 +1,176 @@
+"""
+Tests for CAMB AI TTS integration
+"""
+
+import httpx
+import pytest
+from unittest.mock import MagicMock, patch
+
+from litellm.llms.camb_ai.text_to_speech.transformation import (
+    CambAITextToSpeechConfig,
+)
+from litellm.types.utils import LlmProviders
+
+
+class TestCambAITextToSpeechConfig:
+    def setup_method(self):
+        self.config = CambAITextToSpeechConfig()
+
+    def test_provider_enum_exists(self):
+        assert LlmProviders.CAMB_AI == "camb_ai"
+
+    def test_get_supported_openai_params(self):
+        params = self.config.get_supported_openai_params(model="mars-flash")
+        assert "voice" in params
+        assert "response_format" in params
+        assert "language" in params
+        assert "speed" not in params
+
+    def test_map_openai_params_voice_string(self):
+        voice, params = self.config.map_openai_params(
+            model="mars-flash",
+            optional_params={},
+            voice="123",
+        )
+        assert voice == "123"
+        assert params["voice_id"] == 123
+
+    def test_map_openai_params_voice_dict(self):
+        voice, params = self.config.map_openai_params(
+            model="mars-flash",
+            optional_params={},
+            voice={"voice_id": "456"},
+        )
+        assert voice == "456"
+        assert params["voice_id"] == 456
+
+    def test_map_openai_params_response_format(self):
+        voice, params = self.config.map_openai_params(
+            model="mars-flash",
+            optional_params={"response_format": "mp3"},
+            voice="123",
+        )
+        assert params["output_configuration"] == {"format": "mp3"}
+
+    def test_map_openai_params_speed_dropped(self):
+        voice, params = self.config.map_openai_params(
+            model="mars-flash",
+            optional_params={"speed": 1.5},
+            voice="123",
+        )
+        assert "speed" not in params
+
+    def test_get_complete_url_default(self):
+        url = self.config.get_complete_url(
+            model="mars-flash",
+            api_base=None,
+            litellm_params={},
+        )
+        assert url == "https://client.camb.ai/apis/tts-stream"
+
+    def test_get_complete_url_custom_base(self):
+        url = self.config.get_complete_url(
+            model="mars-flash",
+            api_base="https://custom.camb.ai/v2",
+            litellm_params={},
+        )
+        assert url == "https://custom.camb.ai/v2/tts-stream"
+
+    def test_validate_environment_with_key(self):
+        headers = self.config.validate_environment(
+            headers={},
+            model="mars-flash",
+            api_key="test-key-123",
+        )
+        assert headers["x-api-key"] == "test-key-123"
+        assert headers["Content-Type"] == "application/json"
+
+    @patch.dict("os.environ", {}, clear=False)
+    def test_validate_environment_missing_key(self, monkeypatch):
+        monkeypatch.delenv("CAMB_API_KEY", raising=False)
+        monkeypatch.delenv("CAMB_AI_API_KEY", raising=False)
+        import litellm
+        original_api_key = litellm.api_key
+        litellm.api_key = None
+        try:
+            with pytest.raises(ValueError, match="CAMB AI API key is required"):
+                self.config.validate_environment(
+                    headers={},
+                    model="mars-flash",
+                    api_key=None,
+                )
+        finally:
+            litellm.api_key = original_api_key
+
+    def test_transform_text_to_speech_request(self):
+        result = self.config.transform_text_to_speech_request(
+            model="mars-flash",
+            input="Hello world",
+            voice="123",
+            optional_params={"voice_id": 123},
+            litellm_params={"language": "en-us"},
+            headers={},
+        )
+        body = result["dict_body"]
+        assert body["text"] == "Hello world"
+        assert body["language"] == "en-us"
+        assert body["speech_model"] == "mars-flash"
+        assert body["voice_id"] == 123
+
+    def test_transform_text_to_speech_request_default_language(self):
+        result = self.config.transform_text_to_speech_request(
+            model="mars-pro",
+            input="Test",
+            voice="456",
+            optional_params={},
+            litellm_params={},
+            headers={},
+        )
+        body = result["dict_body"]
+        assert body["language"] == "en-us"
+
+    def test_transform_text_to_speech_request_response_format(self):
+        result = self.config.transform_text_to_speech_request(
+            model="mars-flash",
+            input="Test",
+            voice="123",
+            optional_params={"response_format": "wav"},
+            litellm_params={},
+            headers={},
+        )
+        body = result["dict_body"]
+        assert body["output_configuration"] == {"format": "wav"}
+
+    def test_transform_text_to_speech_request_speed_dropped(self):
+        result = self.config.transform_text_to_speech_request(
+            model="mars-flash",
+            input="Test",
+            voice="123",
+            optional_params={"speed": 1.5},
+            litellm_params={},
+            headers={},
+        )
+        body = result["dict_body"]
+        assert "speed" not in body
+
+    def test_transform_text_to_speech_response(self):
+        mock_response = MagicMock(spec=httpx.Response)
+        mock_logging = MagicMock()
+        result = self.config.transform_text_to_speech_response(
+            model="mars-flash",
+            raw_response=mock_response,
+            logging_obj=mock_logging,
+        )
+        assert result is not None
+
+    def test_get_error_class(self):
+        from litellm.llms.camb_ai.common_utils import CambAIException
+
+        error = self.config.get_error_class(
+            error_message="test error",
+            status_code=400,
+            headers={},
+        )
+        assert isinstance(error, CambAIException)
+        assert error.status_code == 400
+        assert error.message == "test error"