fix(speech): honor pcm response_format for Gemini TTS and reject unsupported containers

litellm/endpoints/speech/speech_to_completion_bridge/handler.py

@@ -115,9 +115,11 @@ class SpeechToCompletionBridgeHandler:
             **request_data,
         )
 
+        requested_response_format: Final = optional_params.get("response_format")
         if isinstance(result, ModelResponse):
             return self.transformation_handler.transform_response(
                 model_response=result,
+                response_format=requested_response_format if isinstance(requested_response_format, str) else None,
             )
         else:
             raise Exception(f"Unmapped response type. Got type: {type(result)}")

litellm/endpoints/speech/speech_to_completion_bridge/transformation.py

@@ -21,6 +21,8 @@ def _completion_response_cost(model_response: "ModelResponse") -> float | None:
 
 
 GEMINI_TTS_CHAT_AUDIO_FORMAT: Final = "pcm16"
+GEMINI_TTS_RAW_RESPONSE_FORMAT: Final = "pcm"
+GEMINI_TTS_SUPPORTED_RESPONSE_FORMATS: Final = frozenset({"wav", GEMINI_TTS_RAW_RESPONSE_FORMAT})
 
 
 class ChatAudioParam(TypedDict):
@@ -29,6 +31,26 @@ class ChatAudioParam(TypedDict):
 
 
 class SpeechToCompletionBridgeTransformationHandler:
+    def _validate_response_format(
+        self, model: str, custom_llm_provider: str, optional_params: Mapping[str, object]
+    ) -> None:
+        if not self._is_gemini_tts_model(model):
+            return
+        response_format: Final = optional_params.get("response_format")
+        if not isinstance(response_format, str) or response_format in GEMINI_TTS_SUPPORTED_RESPONSE_FORMATS:
+            return
+        from litellm.exceptions import BadRequestError
+
+        supported: Final = ", ".join(sorted(GEMINI_TTS_SUPPORTED_RESPONSE_FORMATS))
+        raise BadRequestError(
+            message=(
+                f"Gemini TTS only produces raw PCM16 audio, so response_format='{response_format}'"
+                f" is not supported. Supported response formats: {supported}."
+            ),
+            model=model,
+            llm_provider=custom_llm_provider,
+        )
+
     def _chat_completion_params(self, optional_params: Mapping[str, object]) -> Mapping[str, object]:
         return MappingProxyType(
             {
@@ -67,6 +89,7 @@ class SpeechToCompletionBridgeTransformationHandler:
         litellm_logging_obj: "LiteLLMLoggingObj",
         custom_llm_provider: str,
     ) -> dict:
+        self._validate_response_format(model, custom_llm_provider, optional_params)
         user_message: Final[ChatCompletionUserMessage] = {"role": "user", "content": input}
         return_kwargs: Final = {
             "model": model,
@@ -125,7 +148,14 @@ class SpeechToCompletionBridgeTransformationHandler:
         """Check if the model is a Gemini TTS model that returns PCM16 data."""
         return "gemini" in model.lower() and ("tts" in model.lower() or "preview-tts" in model.lower())
 
-    def transform_response(self, model_response: "ModelResponse") -> "HttpxBinaryResponseContent":
+    def _gemini_tts_response_body(self, decoded_audio: bytes, response_format: str | None) -> tuple[bytes, str]:
+        if response_format == GEMINI_TTS_RAW_RESPONSE_FORMAT:
+            return decoded_audio, "audio/pcm"
+        return self._convert_pcm16_to_wav(decoded_audio), "audio/wav"
+
+    def transform_response(
+        self, model_response: "ModelResponse", response_format: str | None
+    ) -> "HttpxBinaryResponseContent":
         import base64
 
         import httpx
@@ -136,23 +166,15 @@ class SpeechToCompletionBridgeTransformationHandler:
         audio_part: Final = cast(Choices, model_response.choices[0]).message.audio
         if audio_part is None:
             raise ValueError("No audio part found in the response")
-        audio_content: Final = audio_part.data
+        decoded_audio: Final = base64.b64decode(audio_part.data)
 
-        # Decode base64 to get binary content
-        binary_data = base64.b64decode(audio_content)
-
-        # Check if this is a Gemini TTS model that returns raw PCM16 data
         model: Final = getattr(model_response, "model", "")
-        headers: Final = {}
-        if self._is_gemini_tts_model(model):
-            # Convert PCM16 to WAV format for proper audio file playback
-            binary_data = self._convert_pcm16_to_wav(binary_data)
-            headers["Content-Type"] = "audio/wav"
-        else:
-            headers["Content-Type"] = "audio/mpeg"
-
-        # Create an httpx.Response object
-        response: Final = httpx.Response(status_code=200, content=binary_data, headers=headers)
+        content, content_type = (
+            self._gemini_tts_response_body(decoded_audio, response_format)
+            if self._is_gemini_tts_model(model)
+            else (decoded_audio, "audio/mpeg")
+        )
+        response: Final = httpx.Response(status_code=200, content=content, headers={"Content-Type": content_type})
         binary_response: Final = HttpxBinaryResponseContent(response)
         binary_response.set_response_cost(_completion_response_cost(model_response))
         return binary_response

tests/test_litellm/endpoints/speech/speech_to_completion_bridge/test_transformation.py

@@ -1,3 +1,4 @@
+import base64
 from typing import Final
 from unittest.mock import MagicMock
 
@@ -8,8 +9,17 @@ from litellm.constants import OPENAI_CHAT_COMPLETION_PARAMS
 from litellm.endpoints.speech.speech_to_completion_bridge.transformation import (
     SpeechToCompletionBridgeTransformationHandler,
 )
+from litellm.types.utils import ChatCompletionAudioResponse, Choices, Message, ModelResponse
 
 GEMINI_TTS_MODEL: Final = "gemini-3.1-flash-tts-preview"
+PCM_BYTES: Final = b"\x01\x02\x03\x04" * 6
+
+
+def _model_response(model: str, pcm: bytes) -> ModelResponse:
+    audio: Final = ChatCompletionAudioResponse(
+        data=base64.b64encode(pcm).decode(), expires_at=0, transcript="hello"
+    )
+    return ModelResponse(model=model, choices=[Choices(message=Message(content=None, audio=audio))])
 
 
 def _bridge_request(response_format: str | None) -> dict:
@@ -28,7 +38,7 @@ def _bridge_request(response_format: str | None) -> dict:
     )
 
 
-@pytest.mark.parametrize("response_format", ["wav", "mp3", "pcm", None])
+@pytest.mark.parametrize("response_format", ["wav", "pcm", None])
 def test_gemini_tts_request_keeps_speech_response_format_out_of_chat_params(response_format: str | None) -> None:
     request: Final = _bridge_request(response_format)
 
@@ -60,3 +70,48 @@ def test_non_gemini_request_forwards_speech_response_format_as_audio_format() ->
 
     assert "response_format" not in request
     assert request["audio"] == {"voice": "alloy", "format": "wav"}
+
+
+@pytest.mark.parametrize("response_format", ["mp3", "flac", "opus", "aac"])
+def test_gemini_tts_request_rejects_formats_gemini_cannot_produce(response_format: str) -> None:
+    with pytest.raises(litellm.BadRequestError) as excinfo:
+        _bridge_request(response_format)
+
+    assert excinfo.value.status_code == 400
+    assert response_format in str(excinfo.value)
+    assert "pcm" in str(excinfo.value)
+    assert "wav" in str(excinfo.value)
+
+
+def test_gemini_tts_pcm_response_returns_raw_pcm_bytes() -> None:
+    response: Final = SpeechToCompletionBridgeTransformationHandler().transform_response(
+        model_response=_model_response(GEMINI_TTS_MODEL, PCM_BYTES),
+        response_format="pcm",
+    )
+
+    assert response.response.content == PCM_BYTES
+    assert response.response.headers["content-type"] == "audio/pcm"
+
+
+@pytest.mark.parametrize("response_format", ["wav", None])
+def test_gemini_tts_wav_and_default_responses_wrap_pcm_in_wav(response_format: str | None) -> None:
+    response: Final = SpeechToCompletionBridgeTransformationHandler().transform_response(
+        model_response=_model_response(GEMINI_TTS_MODEL, PCM_BYTES),
+        response_format=response_format,
+    )
+
+    body: Final = response.response.content
+    assert body[:4] == b"RIFF"
+    assert body[8:12] == b"WAVE"
+    assert body[44:] == PCM_BYTES
+    assert response.response.headers["content-type"] == "audio/wav"
+
+
+def test_non_gemini_response_keeps_original_bytes_and_mpeg_content_type() -> None:
+    response: Final = SpeechToCompletionBridgeTransformationHandler().transform_response(
+        model_response=_model_response("gpt-4o-audio-preview", PCM_BYTES),
+        response_format="mp3",
+    )
+
+    assert response.response.content == PCM_BYTES
+    assert response.response.headers["content-type"] == "audio/mpeg"