fix(streaming): surface in-body error payloads on OpenAI-compatible streams (#32237)

* fix(streaming): surface in-body error payloads on OpenAI-compatible streams

vLLM and sglang return HTTP 200 streams whose SSE body carries the error,
e.g. data: {"error": {"message": "...", "code": 400}}. The OpenAI-compatible
chunk parser had no detection for this shape: since #23931 the payload parsed
into an empty chunk (choices=[]) and the stream ended silently with 200,
losing the provider's error and never attempting configured fallbacks.

Detect the payload in OpenAIChatCompletionStreamingHandler.chunk_parser and
raise OpenAIError with the upstream message and status code. The existing
mid-stream gate then applies: 4xx surface directly to the client, 5xx wrap
into MidStreamFallbackError so the router can run configured fallbacks.

Fixes #25492

* fix(streaming): serialize messageless error payloads as JSON

Address review feedback: an error dict without a message field now
serializes via json.dumps instead of Python dict repr

litellm/llms/openai/chat/gpt_transformation.py

@@ -2,6 +2,7 @@
 Support for gpt model family
 """
 
+import json
 from typing import (
     TYPE_CHECKING,
     Any,
@@ -782,8 +783,30 @@ class OpenAIChatCompletionStreamingHandler(BaseModelResponseIterator):
                 delta["reasoning_content"] = delta.pop("reasoning")
         return choices
 
+    @staticmethod
+    def _extract_error_from_chunk(chunk: dict) -> Optional[tuple[str, int]]:
+        """OpenAI-compatible backends (vLLM, sglang) can return an HTTP 200
+        stream whose body carries an error payload, e.g.
+        ``data: {"error": {"message": "...", "code": 400}}``."""
+        error = chunk.get("error")
+        if not error:
+            return None
+        if not isinstance(error, dict):
+            return str(error), 500
+        message = error.get("message")
+        code = error.get("code")
+        status_code = code if isinstance(code, int) and 400 <= code < 600 else 500
+        return (message if isinstance(message, str) else json.dumps(error)), status_code
+
     def chunk_parser(self, chunk: dict) -> ModelResponseStream:
         try:
+            error_details = self._extract_error_from_chunk(chunk)
+            if error_details is not None:
+                error_message, error_status_code = error_details
+                raise OpenAIError(
+                    status_code=error_status_code,
+                    message=error_message,
+                )
             choices = chunk.get("choices", [])
             choices = self._map_reasoning_to_reasoning_content(choices)
 

tests/test_litellm/litellm_core_utils/test_streaming_handler.py

@@ -986,6 +986,86 @@ async def test_bedrock_validation_error_raises_directly(logging_obj: Logging):
     assert getattr(excinfo.value, "status_code", None) == 400
 
 
+def _hosted_vllm_stream_wrapper(logging_obj: Logging, error_payload: dict) -> CustomStreamWrapper:
+    """A CustomStreamWrapper over the real OpenAI-compatible line iterator,
+    fed an HTTP 200 SSE body that carries an in-body error payload the way
+    vLLM/sglang emit it."""
+    from litellm.llms.openai.chat.gpt_transformation import (
+        OpenAIChatCompletionStreamingHandler,
+    )
+
+    async def _stream():
+        yield f"data: {json.dumps(error_payload)}"
+        yield "data: [DONE]"
+
+    completion_stream = OpenAIChatCompletionStreamingHandler(
+        streaming_response=_stream(), sync_stream=False
+    )
+    return CustomStreamWrapper(
+        completion_stream=completion_stream,
+        model="qwen-vl",
+        logging_obj=logging_obj,
+        custom_llm_provider="hosted_vllm",
+    )
+
+
+@pytest.mark.asyncio
+async def test_in_body_stream_error_400_raises_bad_request(logging_obj: Logging):
+    """Regression for https://github.com/BerriAI/litellm/issues/25492: a 400
+    error returned inside a 200 SSE body must surface as BadRequestError with
+    the provider's message, not be parsed as an empty chunk that silently
+    ends the stream (and never as an internal MidStreamFallbackError)."""
+    from litellm.exceptions import MidStreamFallbackError
+
+    response = _hosted_vllm_stream_wrapper(
+        logging_obj,
+        {
+            "error": {
+                "object": "error",
+                "message": "The model is not multimodal. Please remove image inputs.",
+                "type": "BadRequestError",
+                "param": None,
+                "code": 400,
+            }
+        },
+    )
+
+    with pytest.raises(litellm.BadRequestError) as excinfo:
+        await response.__anext__()
+
+    assert not isinstance(excinfo.value, MidStreamFallbackError)
+    assert excinfo.value.status_code == 400
+    assert "not multimodal" in str(excinfo.value)
+
+
+@pytest.mark.asyncio
+async def test_in_body_stream_error_500_wraps_for_midstream_fallback(
+    logging_obj: Logging,
+):
+    """An in-body 5xx error wraps into MidStreamFallbackError so the Router's
+    FallbackStreamWrapper can switch to a configured fallback deployment."""
+    from litellm.exceptions import MidStreamFallbackError
+
+    response = _hosted_vllm_stream_wrapper(
+        logging_obj,
+        {
+            "error": {
+                "object": "error",
+                "message": "internal engine crash",
+                "type": "InternalServerError",
+                "param": None,
+                "code": 500,
+            }
+        },
+    )
+
+    with pytest.raises(MidStreamFallbackError) as excinfo:
+        await response.__anext__()
+
+    assert excinfo.value.is_pre_first_chunk is True
+    assert "internal engine crash" in str(excinfo.value)
+
+
 @pytest.mark.asyncio
 async def test_async_streaming_read_timeout_triggers_midstream_fallback(
     logging_obj: Logging,

tests/test_litellm/llms/openai/chat/test_openai_gpt_transformation.py

@@ -187,6 +187,84 @@ class TestOpenAIChatCompletionStreamingHandler:
         assert result.usage.completion_tokens == 350
         assert result.usage.total_tokens == 14147
 
+    def test_chunk_parser_raises_on_in_body_error_payload(self):
+        """vLLM/sglang return HTTP 200 streams whose body carries the error,
+        e.g. data: {"error": {..., "code": 400}}. chunk_parser must surface it
+        as a provider error instead of parsing an empty chunk that silently
+        ends the stream (https://github.com/BerriAI/litellm/issues/25492)."""
+        from litellm.llms.openai.common_utils import OpenAIError
+
+        handler = OpenAIChatCompletionStreamingHandler(
+            streaming_response=None, sync_stream=True
+        )
+
+        error_chunk = {
+            "error": {
+                "object": "error",
+                "message": "The model is not multimodal. Please remove image inputs.",
+                "type": "BadRequestError",
+                "param": None,
+                "code": 400,
+            }
+        }
+
+        with pytest.raises(OpenAIError) as excinfo:
+            handler.chunk_parser(error_chunk)
+
+        assert excinfo.value.status_code == 400
+        assert "not multimodal" in excinfo.value.message
+
+    def test_chunk_parser_error_payload_without_usable_code_maps_to_500(self):
+        """OpenAI-style error payloads may carry a string code (e.g.
+        "invalid_api_key") or none at all; those must map to 500, not crash."""
+        from litellm.llms.openai.common_utils import OpenAIError
+
+        handler = OpenAIChatCompletionStreamingHandler(
+            streaming_response=None, sync_stream=True
+        )
+
+        with pytest.raises(OpenAIError) as excinfo:
+            handler.chunk_parser(
+                {"error": {"message": "engine crashed", "code": "server_error"}}
+            )
+        assert excinfo.value.status_code == 500
+        assert "engine crashed" in excinfo.value.message
+
+        with pytest.raises(OpenAIError) as excinfo:
+            handler.chunk_parser({"error": "plain string error"})
+        assert excinfo.value.status_code == 500
+        assert "plain string error" in excinfo.value.message
+
+        with pytest.raises(OpenAIError) as excinfo:
+            handler.chunk_parser({"error": {"type": "overloaded", "code": 503}})
+        assert excinfo.value.status_code == 503
+        assert excinfo.value.message == '{"type": "overloaded", "code": 503}'
+
+    def test_chunk_parser_tolerates_null_error_field(self):
+        """A chunk that carries "error": null alongside real data must parse
+        normally, not raise."""
+        handler = OpenAIChatCompletionStreamingHandler(
+            streaming_response=None, sync_stream=True
+        )
+
+        chunk = {
+            "id": "gen-123",
+            "created": 1234567890,
+            "model": "openai/gpt-4o-mini",
+            "object": "chat.completion.chunk",
+            "error": None,
+            "choices": [
+                {
+                    "index": 0,
+                    "delta": {"role": "assistant", "content": "Hello"},
+                    "finish_reason": None,
+                }
+            ],
+        }
+
+        result = handler.chunk_parser(chunk)
+        assert result.choices[0].delta.content == "Hello"
+
     def test_chunk_parser_without_usage(self):
         """Test that chunk_parser works normally for chunks without usage."""
         handler = OpenAIChatCompletionStreamingHandler(