From 131339d8e59dafd8be074318e852ddc46efee384 Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Wed, 5 Aug 2026 16:15:04 -0700 Subject: [PATCH 1/3] fix(proxy): send keepalive pings on anthropic messages SSE streams during upstream silence --- litellm/__init__.py | 1 + litellm/proxy/common_request_processing.py | 6 +- litellm/proxy/common_utils/sse_keepalive.py | 43 ++++++ .../proxy/common_utils/test_sse_keepalive.py | 122 ++++++++++++++++++ 4 files changed, 171 insertions(+), 1 deletion(-) create mode 100644 litellm/proxy/common_utils/sse_keepalive.py create mode 100644 tests/test_litellm/proxy/common_utils/test_sse_keepalive.py diff --git a/litellm/__init__.py b/litellm/__init__.py index 319da4e25eb..717ff0c404b 100644 --- a/litellm/__init__.py +++ b/litellm/__init__.py @@ -244,6 +244,7 @@ use_chat_completions_url_for_anthropic_messages: bool = bool( # Or via `litellm_settings.strip_anthropic_total_tokens: true` in # config.yaml. strip_anthropic_total_tokens: bool = False +anthropic_sse_ping_interval_seconds: float = 15.0 route_all_chat_openai_to_responses: bool = ( os.getenv("LITELLM_ROUTE_ALL_CHAT_OPENAI_TO_RESPONSES", "false").lower() == "true" ) # When True, routes all OpenAI /chat/completions requests through the Responses API bridge diff --git a/litellm/proxy/common_request_processing.py b/litellm/proxy/common_request_processing.py index 0b9a2d5e4c0..6b4d3143992 100644 --- a/litellm/proxy/common_request_processing.py +++ b/litellm/proxy/common_request_processing.py @@ -46,6 +46,7 @@ from litellm.proxy.common_utils.callback_utils import ( get_logging_caching_headers, get_remaining_tokens_and_requests_from_request_data, ) +from litellm.proxy.common_utils.sse_keepalive import wrap_sse_stream_with_keepalive_pings from litellm.proxy.dd_span_tagger import DDSpanTagger from litellm.proxy.route_llm_request import route_request from litellm.proxy.utils import ProxyLogging, _check_and_merge_model_level_guardrails @@ -1950,7 +1951,10 @@ class ProxyBaseLLMRequestProcessing: request=request, ) return await create_response( - generator=selected_data_generator, + generator=wrap_sse_stream_with_keepalive_pings( + stream=selected_data_generator, + ping_interval_seconds=litellm.anthropic_sse_ping_interval_seconds, + ), media_type="text/event-stream", headers=custom_headers, request=request, diff --git a/litellm/proxy/common_utils/sse_keepalive.py b/litellm/proxy/common_utils/sse_keepalive.py new file mode 100644 index 00000000000..3ea17b7b1e8 --- /dev/null +++ b/litellm/proxy/common_utils/sse_keepalive.py @@ -0,0 +1,43 @@ +import asyncio +import contextlib +from collections.abc import AsyncGenerator +from typing import Final + +import anyio + +ANTHROPIC_PING_SSE_CHUNK: Final = 'event: ping\ndata: {"type": "ping"}\n\n' + + +def wrap_sse_stream_with_keepalive_pings( + stream: AsyncGenerator[str, None], + ping_interval_seconds: float, +) -> AsyncGenerator[str, None]: + if ping_interval_seconds <= 0: + return stream + return _keepalive_ping_stream(stream=stream, ping_interval_seconds=ping_interval_seconds) + + +async def _keepalive_ping_stream( + stream: AsyncGenerator[str, None], + ping_interval_seconds: float, +) -> AsyncGenerator[str, None]: + pending = asyncio.ensure_future( + stream.__anext__() + ) # rebind-ok: re-armed with the next __anext__ after each delivered chunk + try: + while True: + await asyncio.wait({pending}, timeout=ping_interval_seconds) + if not pending.done(): + yield ANTHROPIC_PING_SSE_CHUNK + continue + try: + yield pending.result() + except StopAsyncIteration: + return + pending = asyncio.ensure_future(stream.__anext__()) + finally: + pending.cancel() + with anyio.CancelScope(shield=True): + with contextlib.suppress(BaseException): + await pending + await stream.aclose() diff --git a/tests/test_litellm/proxy/common_utils/test_sse_keepalive.py b/tests/test_litellm/proxy/common_utils/test_sse_keepalive.py new file mode 100644 index 00000000000..bc6f4060163 --- /dev/null +++ b/tests/test_litellm/proxy/common_utils/test_sse_keepalive.py @@ -0,0 +1,122 @@ +import asyncio +from collections.abc import AsyncGenerator +from typing import Final + +import pytest +from fastapi.responses import StreamingResponse + +from litellm.proxy.common_request_processing import create_response +from litellm.proxy.common_utils.sse_keepalive import ( + ANTHROPIC_PING_SSE_CHUNK, + wrap_sse_stream_with_keepalive_pings, +) + +MESSAGE_START_CHUNK: Final = 'data: {"type": "message_start"}\n\n' +TEXT_DELTA_CHUNK: Final = 'data: {"type": "content_block_delta"}\n\n' + + +@pytest.mark.asyncio +async def test_pings_fill_mid_stream_silence_and_preserve_chunk_order(): + async def gappy_stream() -> AsyncGenerator[str, None]: + yield MESSAGE_START_CHUNK + await asyncio.sleep(0.3) + yield TEXT_DELTA_CHUNK + + wrapped: Final = wrap_sse_stream_with_keepalive_pings(stream=gappy_stream(), ping_interval_seconds=0.05) + collected: Final = [chunk async for chunk in wrapped] + + assert collected[0] == MESSAGE_START_CHUNK + assert collected[-1] == TEXT_DELTA_CHUNK + assert ANTHROPIC_PING_SSE_CHUNK in collected[1:-1] + assert [chunk for chunk in collected if chunk != ANTHROPIC_PING_SSE_CHUNK] == [ + MESSAGE_START_CHUNK, + TEXT_DELTA_CHUNK, + ] + + +@pytest.mark.asyncio +async def test_ping_emitted_while_waiting_for_first_chunk(): + async def slow_start_stream() -> AsyncGenerator[str, None]: + await asyncio.sleep(0.2) + yield MESSAGE_START_CHUNK + + wrapped: Final = wrap_sse_stream_with_keepalive_pings(stream=slow_start_stream(), ping_interval_seconds=0.05) + collected: Final = [chunk async for chunk in wrapped] + + assert collected[0] == ANTHROPIC_PING_SSE_CHUNK + assert collected[-1] == MESSAGE_START_CHUNK + + +@pytest.mark.asyncio +async def test_no_pings_when_chunks_arrive_faster_than_interval(): + async def fast_stream() -> AsyncGenerator[str, None]: + yield MESSAGE_START_CHUNK + yield TEXT_DELTA_CHUNK + yield TEXT_DELTA_CHUNK + + wrapped: Final = wrap_sse_stream_with_keepalive_pings(stream=fast_stream(), ping_interval_seconds=1.0) + collected: Final = [chunk async for chunk in wrapped] + + assert collected == [MESSAGE_START_CHUNK, TEXT_DELTA_CHUNK, TEXT_DELTA_CHUNK] + + +@pytest.mark.asyncio +async def test_upstream_exception_propagates(): + async def failing_stream() -> AsyncGenerator[str, None]: + yield MESSAGE_START_CHUNK + raise ValueError("upstream broke") + + wrapped: Final = wrap_sse_stream_with_keepalive_pings(stream=failing_stream(), ping_interval_seconds=5.0) + + assert await wrapped.__anext__() == MESSAGE_START_CHUNK + with pytest.raises(ValueError, match="upstream broke"): + await wrapped.__anext__() + + +@pytest.mark.asyncio +async def test_aclose_mid_silence_cancels_upstream_and_runs_its_cleanup(): + upstream_cleaned_up: Final = asyncio.Event() + + async def hung_stream() -> AsyncGenerator[str, None]: + try: + yield MESSAGE_START_CHUNK + await asyncio.Event().wait() + yield TEXT_DELTA_CHUNK + finally: + upstream_cleaned_up.set() + + wrapped: Final = wrap_sse_stream_with_keepalive_pings(stream=hung_stream(), ping_interval_seconds=0.05) + + assert await wrapped.__anext__() == MESSAGE_START_CHUNK + assert await wrapped.__anext__() == ANTHROPIC_PING_SSE_CHUNK + await wrapped.aclose() + + assert upstream_cleaned_up.is_set() + + +@pytest.mark.asyncio +async def test_non_positive_interval_returns_stream_unwrapped(): + async def any_stream() -> AsyncGenerator[str, None]: + yield MESSAGE_START_CHUNK + + stream: Final = any_stream() + assert wrap_sse_stream_with_keepalive_pings(stream=stream, ping_interval_seconds=0) is stream + await stream.aclose() + + +@pytest.mark.asyncio +async def test_create_response_streams_ping_first_for_slow_upstream(): + async def slow_start_stream() -> AsyncGenerator[str, None]: + await asyncio.sleep(0.2) + yield MESSAGE_START_CHUNK + + response: Final = await create_response( + generator=wrap_sse_stream_with_keepalive_pings(stream=slow_start_stream(), ping_interval_seconds=0.05), + media_type="text/event-stream", + headers={}, + ) + + assert isinstance(response, StreamingResponse) + collected: Final = [chunk async for chunk in response.body_iterator] + assert collected[0] == ANTHROPIC_PING_SSE_CHUNK + assert collected[-1] == MESSAGE_START_CHUNK From 6ca120a6745b7c979ccd34becff5dba17640955b Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Wed, 5 Aug 2026 17:23:39 -0700 Subject: [PATCH 2/3] fix(proxy): coerce and validate the sse keepalive ping interval from config --- litellm/proxy/common_utils/sse_keepalive.py | 20 +++++++++++++--- .../proxy/common_utils/test_sse_keepalive.py | 24 +++++++++++++++++++ 2 files changed, 41 insertions(+), 3 deletions(-) diff --git a/litellm/proxy/common_utils/sse_keepalive.py b/litellm/proxy/common_utils/sse_keepalive.py index 3ea17b7b1e8..e6d5da03a71 100644 --- a/litellm/proxy/common_utils/sse_keepalive.py +++ b/litellm/proxy/common_utils/sse_keepalive.py @@ -1,5 +1,6 @@ import asyncio import contextlib +import math from collections.abc import AsyncGenerator from typing import Final @@ -8,13 +9,26 @@ import anyio ANTHROPIC_PING_SSE_CHUNK: Final = 'event: ping\ndata: {"type": "ping"}\n\n' +def _coerce_interval(ping_interval_seconds: float | str | None) -> float | None: + if ping_interval_seconds is None: + return None + try: + interval: Final = float(ping_interval_seconds) + except ValueError: + return None + if not math.isfinite(interval) or interval <= 0: + return None + return interval + + def wrap_sse_stream_with_keepalive_pings( stream: AsyncGenerator[str, None], - ping_interval_seconds: float, + ping_interval_seconds: float | str | None, ) -> AsyncGenerator[str, None]: - if ping_interval_seconds <= 0: + interval: Final = _coerce_interval(ping_interval_seconds) + if interval is None: return stream - return _keepalive_ping_stream(stream=stream, ping_interval_seconds=ping_interval_seconds) + return _keepalive_ping_stream(stream=stream, ping_interval_seconds=interval) async def _keepalive_ping_stream( diff --git a/tests/test_litellm/proxy/common_utils/test_sse_keepalive.py b/tests/test_litellm/proxy/common_utils/test_sse_keepalive.py index bc6f4060163..6e98248dd18 100644 --- a/tests/test_litellm/proxy/common_utils/test_sse_keepalive.py +++ b/tests/test_litellm/proxy/common_utils/test_sse_keepalive.py @@ -104,6 +104,30 @@ async def test_non_positive_interval_returns_stream_unwrapped(): await stream.aclose() +@pytest.mark.asyncio +@pytest.mark.parametrize("bad_interval", [None, "abc", "", float("inf"), float("nan"), "-3"]) +async def test_invalid_config_interval_returns_stream_unwrapped(bad_interval: float | str | None): + async def any_stream() -> AsyncGenerator[str, None]: + yield MESSAGE_START_CHUNK + + stream: Final = any_stream() + assert wrap_sse_stream_with_keepalive_pings(stream=stream, ping_interval_seconds=bad_interval) is stream + await stream.aclose() + + +@pytest.mark.asyncio +async def test_numeric_string_interval_from_yaml_config_enables_pings(): + async def slow_start_stream() -> AsyncGenerator[str, None]: + await asyncio.sleep(0.2) + yield MESSAGE_START_CHUNK + + wrapped: Final = wrap_sse_stream_with_keepalive_pings(stream=slow_start_stream(), ping_interval_seconds="0.05") + collected: Final = [chunk async for chunk in wrapped] + + assert collected[0] == ANTHROPIC_PING_SSE_CHUNK + assert collected[-1] == MESSAGE_START_CHUNK + + @pytest.mark.asyncio async def test_create_response_streams_ping_first_for_slow_upstream(): async def slow_start_stream() -> AsyncGenerator[str, None]: From 097c03eebb948e502db639c48e5bf88561f369d6 Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Wed, 5 Aug 2026 17:33:09 -0700 Subject: [PATCH 3/3] fix(proxy): tolerate non-scalar sse keepalive interval config shapes --- litellm/proxy/common_utils/sse_keepalive.py | 2 +- .../proxy/common_utils/test_sse_keepalive.py | 16 ++++++++++++++-- 2 files changed, 15 insertions(+), 3 deletions(-) diff --git a/litellm/proxy/common_utils/sse_keepalive.py b/litellm/proxy/common_utils/sse_keepalive.py index e6d5da03a71..6700700ff7c 100644 --- a/litellm/proxy/common_utils/sse_keepalive.py +++ b/litellm/proxy/common_utils/sse_keepalive.py @@ -14,7 +14,7 @@ def _coerce_interval(ping_interval_seconds: float | str | None) -> float | None: return None try: interval: Final = float(ping_interval_seconds) - except ValueError: + except (TypeError, ValueError): return None if not math.isfinite(interval) or interval <= 0: return None diff --git a/tests/test_litellm/proxy/common_utils/test_sse_keepalive.py b/tests/test_litellm/proxy/common_utils/test_sse_keepalive.py index 6e98248dd18..9cca9bbfe12 100644 --- a/tests/test_litellm/proxy/common_utils/test_sse_keepalive.py +++ b/tests/test_litellm/proxy/common_utils/test_sse_keepalive.py @@ -1,6 +1,6 @@ import asyncio from collections.abc import AsyncGenerator -from typing import Final +from typing import Final, cast import pytest from fastapi.responses import StreamingResponse @@ -105,7 +105,19 @@ async def test_non_positive_interval_returns_stream_unwrapped(): @pytest.mark.asyncio -@pytest.mark.parametrize("bad_interval", [None, "abc", "", float("inf"), float("nan"), "-3"]) +@pytest.mark.parametrize( + "bad_interval", + [ + None, + "abc", + "", + float("inf"), + float("nan"), + "-3", + cast("float | str | None", [15]), + cast("float | str | None", {"seconds": 15}), + ], +) async def test_invalid_config_interval_returns_stream_unwrapped(bad_interval: float | str | None): async def any_stream() -> AsyncGenerator[str, None]: yield MESSAGE_START_CHUNK