fix(proxy): handle non-standard SSE frames in Anthropic passthrough logging

litellm/proxy/pass_through_endpoints/llm_provider_handlers/anthropic_passthrough_logging_handler.py

@@ -468,6 +468,13 @@ def _build_complete_streaming_response_legacy(
             # Process each individual event
             for event_str in individual_events:
                 try:
+                    # Skip OpenAI-style [DONE] sentinels some Anthropic-compatible
+                    # providers emit. Match the whole SSE line so a valid chunk whose
+                    # text payload happens to contain "[DONE]" is not dropped.
+                    if any(
+                        line.strip() == "data: [DONE]" for line in event_str.split("\n")
+                    ):
+                        continue
                     transformed_openai_chunk = anthropic_model_response_iterator.convert_str_chunk_to_generic_chunk(
                         chunk=event_str
                     )
@@ -476,6 +483,14 @@ def _build_complete_streaming_response_legacy(
 
                 except (StopIteration, StopAsyncIteration):
                     break
+                except json.JSONDecodeError:
+                    # Some upstreams emit non-JSON SSE lines; skip them so the
+                    # logging pipeline is not broken by a single bad frame.
+                    verbose_proxy_logger.debug(
+                        "Skipping non-JSON SSE event: %s",
+                        event_str[:200],
+                    )
+                    continue
 
         complete_streaming_response = litellm.stream_chunk_builder(
             chunks=all_openai_chunks,

tests/test_litellm/proxy/pass_through_endpoints/llm_provider_handlers/test_anthropic_passthrough_logging_handler.py

@@ -724,6 +724,72 @@ def test_store_batch_managed_object_success(
             )
 
 
+class TestBuildCompleteStreamingResponseRobustness:
+    """_build_complete_streaming_response must tolerate non-standard SSE frames."""
+
+    def _build(self, chunks: List[str]):
+        return AnthropicPassthroughLoggingHandler._build_complete_streaming_response(
+            all_chunks=chunks,
+            litellm_logging_obj=MagicMock(),
+            model="claude-3-sonnet-20240229",
+        )
+
+    def test_done_frame_is_skipped(self):
+        """A bare 'data: [DONE]' control frame must not break reconstruction."""
+        chunks = [
+            'event: message_start\ndata: {"type":"message_start","message":{"id":"msg_1","type":"message","role":"assistant","content":[],"model":"claude-3-sonnet-20240229","stop_reason":null,"stop_sequence":null,"usage":{"input_tokens":10,"output_tokens":1}}}',
+            'event: content_block_start\ndata: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}',
+            'event: content_block_delta\ndata: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Hi"}}',
+            'event: content_block_stop\ndata: {"type":"content_block_stop","index":0}',
+            'event: message_delta\ndata: {"type":"message_delta","delta":{"stop_reason":"end_turn","stop_sequence":null},"usage":{"output_tokens":2}}',
+            'event: message_stop\ndata: {"type":"message_stop"}',
+            "data: [DONE]",
+        ]
+        result = self._build(chunks)
+        assert result is not None
+        assert result.choices[0].message.content == "Hi"
+
+    def test_non_json_sse_line_is_skipped(self):
+        """Non-JSON SSE lines (comments, keep-alive pings) must be skipped."""
+        chunks = [
+            ": ping",
+            'event: message_start\ndata: {"type":"message_start","message":{"id":"msg_1","type":"message","role":"assistant","content":[],"model":"claude-3-sonnet-20240229","stop_reason":null,"stop_sequence":null,"usage":{"input_tokens":10,"output_tokens":1}}}',
+            "this is not json at all",
+        ]
+        # Must not raise; a malformed stream simply yields no usable response.
+        result = self._build(chunks)
+        assert result is None or hasattr(result, "choices")
+
+    def test_mixed_valid_and_invalid_frames(self):
+        """Valid events are still collected when interleaved with invalid ones."""
+        chunks = [
+            'event: message_start\ndata: {"type":"message_start","message":{"id":"msg_1","type":"message","role":"assistant","content":[],"model":"claude-3-sonnet-20240229","stop_reason":null,"stop_sequence":null,"usage":{"input_tokens":10,"output_tokens":1}}}',
+            "data: [DONE]",
+            ": keep-alive",
+            "not-json",
+            'event: content_block_start\ndata: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}',
+            'event: content_block_delta\ndata: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Hello"}}',
+            'event: content_block_stop\ndata: {"type":"content_block_stop","index":0}',
+            'event: message_delta\ndata: {"type":"message_delta","delta":{"stop_reason":"end_turn","stop_sequence":null},"usage":{"output_tokens":2}}',
+            'event: message_stop\ndata: {"type":"message_stop"}',
+        ]
+        result = self._build(chunks)
+        assert result is not None
+        assert result.choices[0].message.content == "Hello"
+
+    def test_done_in_text_payload_is_not_dropped(self):
+        """A valid event whose text content contains '[DONE]' must NOT be skipped."""
+        chunks = [
+            'event: message_start\ndata: {"type":"message_start","message":{"id":"msg_1","type":"message","role":"assistant","content":[],"model":"claude-3-sonnet-20240229","stop_reason":null,"stop_sequence":null,"usage":{"input_tokens":10,"output_tokens":1}}}',
+            'event: content_block_start\ndata: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}',
+            'event: content_block_delta\ndata: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"The stream ends with [DONE]"}}',
+            'event: content_block_stop\ndata: {"type":"content_block_stop","index":0}',
+            'event: message_delta\ndata: {"type":"message_delta","delta":{"stop_reason":"end_turn","stop_sequence":null},"usage":{"output_tokens":8}}',
+            'event: message_stop\ndata: {"type":"message_stop"}',
+        ]
+        result = self._build(chunks)
+        assert result is not None
+        assert result.choices[0].message.content == "The stream ends with [DONE]"
 class TestPureTextFastPathParity:
     """
     The pure-text fast path in _build_complete_streaming_response must produce