From b8bbb4039e43296b005fad4ebf90f20eca00978f Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Thu, 29 Oct 2026 22:07:04 -0700 Subject: [PATCH] fix(proxy): encode mappings and models nested in Responses stream chunks as JSON objects --- .../common_utils/responses_stream_errors.py | 17 ++++++++- litellm/proxy/proxy_server.py | 4 +- .../proxy_server/test_streaming_helpers.py | 37 +++++++++++++++++++ 3 files changed, 55 insertions(+), 3 deletions(-) diff --git a/litellm/proxy/common_utils/responses_stream_errors.py b/litellm/proxy/common_utils/responses_stream_errors.py index 0024c210fab..ac4854f4014 100644 --- a/litellm/proxy/common_utils/responses_stream_errors.py +++ b/litellm/proxy/common_utils/responses_stream_errors.py @@ -6,6 +6,7 @@ from types import MappingProxyType from typing import Final from pydantic import BaseModel, ConfigDict, ValidationError, field_validator +from typing_extensions import TypeIs # noqa: TID251 # TypeIs reaches typing only on 3.13 from litellm._logging import redact_internal_details_from_client_message from litellm._uuid import uuid @@ -67,6 +68,20 @@ class _ErrorPayload(LiteLLMBaseModel): return {"message": self.error} if isinstance(self.error, str) else self.error +def _is_mapping( + value: object, +) -> TypeIs[Mapping[object, object]]: # guard-ok: isinstance decides it, keys and values stay object + return isinstance(value, Mapping) + + +def jsonable_stream_value(value: object) -> object: + if _is_mapping(value): + return {**value} + if isinstance(value, BaseModel): + return value.model_dump() + return str(value) + + def sse_error_object(data: str | bytes) -> Mapping[str, object] | None: try: return _ErrorPayload.model_validate_json(data).error_object() @@ -191,7 +206,7 @@ class ResponsesStreamErrorState: "sequence_number": self.sequence_number + 1, "error": error, }, - default=str, + default=jsonable_stream_value, ) return f"event: error\ndata: {payload}\n\n" diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index aef2bca785c..2ddb7a9241b 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -539,7 +539,7 @@ from litellm.proxy.common_utils.periodic_reload_schedule import ( ) from litellm.proxy.common_utils.proxy_state import ProxyState from litellm.proxy.common_utils.reset_budget_job import ResetBudgetJob -from litellm.proxy.common_utils.responses_stream_errors import ResponsesStreamErrorState +from litellm.proxy.common_utils.responses_stream_errors import ResponsesStreamErrorState, jsonable_stream_value from litellm.proxy.common_utils.scheduled_job_stagger import ( apply_scheduled_job_stagger, attach_job_timing_logger, @@ -10204,7 +10204,7 @@ def _format_streaming_sse_chunk(chunk: object) -> str | bytes: def _format_responses_sse_chunk(chunk: object) -> str | bytes: if isinstance(chunk, Mapping): - return f"data: {json.dumps({**chunk}, default=str)}\n\n" + return f"data: {json.dumps(chunk, default=jsonable_stream_value)}\n\n" return _format_streaming_sse_chunk(chunk=chunk) diff --git a/tests/unit/proxy/proxy_server/test_streaming_helpers.py b/tests/unit/proxy/proxy_server/test_streaming_helpers.py index b75a2494588..9899e7d2e64 100644 --- a/tests/unit/proxy/proxy_server/test_streaming_helpers.py +++ b/tests/unit/proxy/proxy_server/test_streaming_helpers.py @@ -40,6 +40,7 @@ from litellm.proxy.proxy_server import ( _apply_streaming_chunk_hooks, _fast_serialize_simple_model_response_stream, _format_fallback_metadata_sse_event, + _format_responses_sse_chunk, _format_streaming_sse_chunk, _get_client_requested_model_for_streaming, _get_streaming_fallback_metadata, @@ -182,6 +183,42 @@ async def test_responses_stream_sends_no_failure_event_after_an_upstream_error_e assert tuple(frame.split("\n", 1)[0] for frame in frames) == ("event: response.created", "event: error") assert json.loads(frames[-1].split("data: ", 1)[1])["message"] == "upstream failed" + +class _HookOutputItem(BaseModel): + id: str + type: str + + +@pytest.mark.parametrize( + "item", + [ + MappingProxyType({"id": "msg_nested", "type": "message"}), + _HookOutputItem(id="msg_nested", type="message"), + ], + ids=["mapping_proxy", "model"], +) +def test_responses_sse_chunk_encodes_values_nested_in_a_hook_chunk_as_json_objects( + item: Mapping[str, object] | BaseModel, +) -> None: + frame: Final = _format_responses_sse_chunk( + chunk=MappingProxyType({"type": "response.output_item.added", "output_index": 0, "item": item}) + ) + + assert isinstance(frame, str) + assert json.loads(frame.removeprefix("data: "))["item"] == {"id": "msg_nested", "type": "message"} + + +def test_responses_error_event_encodes_nested_error_fields_as_json_objects() -> None: + frame: Final = ResponsesStreamErrorState().format_error( + { + "message": "overloaded", + "provider_specific_fields": MappingProxyType({"upstream": MappingProxyType({"type": "overloaded_error"})}), + } + ) + + payload: Final = json.loads(frame.split("data: ", 1)[1]) + assert payload["error"]["provider_specific_fields"] == {"upstream": {"type": "overloaded_error"}} + @pytest.mark.asyncio @pytest.mark.parametrize( ("guardrail_error", "error_object"),