diff --git a/litellm/llms/camb_ai/__init__.py b/litellm/llms/camb_ai/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/litellm/llms/camb_ai/common_utils.py b/litellm/llms/camb_ai/common_utils.py new file mode 100644 index 00000000000..e3a43273fe6 --- /dev/null +++ b/litellm/llms/camb_ai/common_utils.py @@ -0,0 +1,5 @@ +from litellm.llms.base_llm.chat.transformation import BaseLLMException + + +class CambAIException(BaseLLMException): + pass diff --git a/litellm/llms/camb_ai/text_to_speech/__init__.py b/litellm/llms/camb_ai/text_to_speech/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/litellm/llms/camb_ai/text_to_speech/transformation.py b/litellm/llms/camb_ai/text_to_speech/transformation.py new file mode 100644 index 00000000000..f63c7fae084 --- /dev/null +++ b/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}" diff --git a/litellm/main.py b/litellm/main.py index 794d30ed384..af991890a00 100644 --- a/litellm/main.py +++ b/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, diff --git a/litellm/model_prices_and_context_window_backup.json b/litellm/model_prices_and_context_window_backup.json index 1e0a7899422..ee02e6df290 100644 --- a/litellm/model_prices_and_context_window_backup.json +++ b/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", diff --git a/litellm/types/utils.py b/litellm/types/utils.py index 3c818387744..900dfa4e65a 100644 --- a/litellm/types/utils.py +++ b/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" diff --git a/litellm/utils.py b/litellm/utils.py index fca5914dbab..142744dfe6d 100644 --- a/litellm/utils.py +++ b/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 diff --git a/tests/audio_tests/test_audio_speech.py b/tests/audio_tests/test_audio_speech.py index 67e0dbffa61..1e981075f85 100644 --- a/tests/audio_tests/test_audio_speech.py +++ b/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" diff --git a/tests/test_litellm/test_camb_ai.py b/tests/test_litellm/test_camb_ai.py new file mode 100644 index 00000000000..c73b9c5b2d5 --- /dev/null +++ b/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"