From dc8acf9fba1af473f0b17b9b3c279cda46d4cbb5 Mon Sep 17 00:00:00 2001 From: Blair Jordan Date: Mon, 24 Aug 2026 23:02:23 +1000 Subject: [PATCH 1/2] fix(anthropic): close abandoned streaming responses --- litellm/llms/anthropic/chat/handler.py | 9 ++++ .../chat/test_anthropic_chat_handler.py | 54 +++++++++++++++++++ 2 files changed, 63 insertions(+) diff --git a/litellm/llms/anthropic/chat/handler.py b/litellm/llms/anthropic/chat/handler.py index cd47cdd57d6..6d2a5a3678b 100644 --- a/litellm/llms/anthropic/chat/handler.py +++ b/litellm/llms/anthropic/chat/handler.py @@ -114,6 +114,7 @@ async def make_call( speed=speed, tool_name_reverse_map=tool_name_reverse_map, ) + completion_stream.http_response = response # LOGGING logging_obj.post_call( @@ -634,6 +635,7 @@ class ModelResponseIterator: ): self.streaming_response = streaming_response self.response_iterator = self.streaming_response + self.http_response: httpx.Response | None = None self.content_blocks: list[ContentBlockDelta] = [] self.tool_index = -1 self.json_mode = json_mode @@ -687,6 +689,13 @@ class ModelResponseIterator: def accumulated_json(self, value: str) -> None: self._json_buffer.set(value) + async def aclose(self) -> None: + response: Final = self.http_response + self.http_response = None + if response is None: + return + await response.aclose() + def check_empty_tool_call_args(self) -> bool: """ Check if the tool call block so far has been an empty string diff --git a/tests/test_litellm/llms/anthropic/chat/test_anthropic_chat_handler.py b/tests/test_litellm/llms/anthropic/chat/test_anthropic_chat_handler.py index bd750a47f63..7095970fbb9 100644 --- a/tests/test_litellm/llms/anthropic/chat/test_anthropic_chat_handler.py +++ b/tests/test_litellm/llms/anthropic/chat/test_anthropic_chat_handler.py @@ -1,12 +1,17 @@ import json import threading +from collections.abc import AsyncIterator from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from typing import Final from unittest.mock import AsyncMock, MagicMock, patch +import anyio +import httpx import pytest import litellm from litellm.constants import RESPONSE_FORMAT_TOOL_NAME +from litellm.litellm_core_utils.streaming_handler import CustomStreamWrapper from litellm.llms.anthropic.chat.handler import ModelResponseIterator, make_call from litellm.types.llms.openai import ( ChatCompletionToolCallChunk, @@ -15,6 +20,18 @@ from litellm.types.llms.openai import ( from litellm.types.responses.main import OutputCodeInterpreterCall +class _RecordingAsyncByteStream(httpx.AsyncByteStream): + def __init__(self) -> None: + self.aclose_calls: int = 0 + + async def __aiter__(self) -> AsyncIterator[bytes]: + yield b'data: {"type":"message_start"}\n\n' + + async def aclose(self) -> None: + await anyio.sleep(0) + self.aclose_calls += 1 + + @pytest.mark.asyncio async def test_make_call_passes_logging_obj_to_client_post(): """make_call must pass logging_obj to client.post so track_llm_api_timing can set llm_api_duration_ms for litellm_overhead_time_ms.""" @@ -46,6 +63,43 @@ async def test_make_call_passes_logging_obj_to_client_post(): assert call_kwargs.get("logging_obj") is logging_obj +@pytest.mark.asyncio +async def test_make_call_stream_cleanup_closes_http_response_once_under_cancellation(): + stream: Final = _RecordingAsyncByteStream() + response: Final = httpx.Response(200, stream=stream) + mock_client: Final = AsyncMock() + mock_client.post.return_value = response + logging_obj: Final = MagicMock(model_call_details={"litellm_params": {}}) + + completion_stream, _ = await make_call( + client=mock_client, + api_base="https://api.anthropic.com/v1/messages", + headers={}, + data="{}", + model="claude-haiku-4-5", + messages=[{"role": "user", "content": "Hi"}], + logging_obj=logging_obj, + timeout=60.0, + json_mode=False, + ) + wrapper: Final = CustomStreamWrapper( + completion_stream=completion_stream, + model="claude-haiku-4-5", + custom_llm_provider="anthropic", + logging_obj=logging_obj, + ) + + with anyio.CancelScope() as cancel_scope: + cancel_scope.cancel() + await wrapper.aclose() + await wrapper.aclose() + await completion_stream.aclose() + + assert response.is_closed is True + assert completion_stream.http_response is None + assert stream.aclose_calls == 1 + + def test_redacted_thinking_content_block_delta(): chunk = { "type": "content_block_start", From ddb7409151574891238431fddf9a6df724c95f92 Mon Sep 17 00:00:00 2001 From: Blair Jordan Date: Tue, 25 Aug 2026 08:45:35 +1000 Subject: [PATCH 2/2] fix(anthropic): close sync streaming responses --- litellm/llms/anthropic/chat/handler.py | 5 ++ .../chat/test_anthropic_chat_handler.py | 54 ++++++++++++++++++- 2 files changed, 57 insertions(+), 2 deletions(-) diff --git a/litellm/llms/anthropic/chat/handler.py b/litellm/llms/anthropic/chat/handler.py index 6d2a5a3678b..2c3959acd06 100644 --- a/litellm/llms/anthropic/chat/handler.py +++ b/litellm/llms/anthropic/chat/handler.py @@ -183,6 +183,7 @@ def make_sync_call( speed=speed, tool_name_reverse_map=tool_name_reverse_map, ) + completion_stream.http_response = response # LOGGING logging_obj.post_call( @@ -635,6 +636,7 @@ class ModelResponseIterator: ): self.streaming_response = streaming_response self.response_iterator = self.streaming_response + self.sync_stream = sync_stream self.http_response: httpx.Response | None = None self.content_blocks: list[ContentBlockDelta] = [] self.tool_index = -1 @@ -694,6 +696,9 @@ class ModelResponseIterator: self.http_response = None if response is None: return + if self.sync_stream: + response.close() + return await response.aclose() def check_empty_tool_call_args(self) -> bool: diff --git a/tests/test_litellm/llms/anthropic/chat/test_anthropic_chat_handler.py b/tests/test_litellm/llms/anthropic/chat/test_anthropic_chat_handler.py index 7095970fbb9..bd0898a0c76 100644 --- a/tests/test_litellm/llms/anthropic/chat/test_anthropic_chat_handler.py +++ b/tests/test_litellm/llms/anthropic/chat/test_anthropic_chat_handler.py @@ -1,6 +1,6 @@ import json import threading -from collections.abc import AsyncIterator +from collections.abc import AsyncIterator, Iterator from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from typing import Final from unittest.mock import AsyncMock, MagicMock, patch @@ -12,7 +12,11 @@ import pytest import litellm from litellm.constants import RESPONSE_FORMAT_TOOL_NAME from litellm.litellm_core_utils.streaming_handler import CustomStreamWrapper -from litellm.llms.anthropic.chat.handler import ModelResponseIterator, make_call +from litellm.llms.anthropic.chat.handler import ( + ModelResponseIterator, + make_call, + make_sync_call, +) from litellm.types.llms.openai import ( ChatCompletionToolCallChunk, ChatCompletionToolCallFunctionChunk, @@ -32,6 +36,17 @@ class _RecordingAsyncByteStream(httpx.AsyncByteStream): self.aclose_calls += 1 +class _RecordingSyncByteStream(httpx.SyncByteStream): + def __init__(self) -> None: + self.close_calls: int = 0 + + def __iter__(self) -> Iterator[bytes]: + yield b'data: {"type":"message_start"}\n\n' + + def close(self) -> None: + self.close_calls += 1 + + @pytest.mark.asyncio async def test_make_call_passes_logging_obj_to_client_post(): """make_call must pass logging_obj to client.post so track_llm_api_timing can set llm_api_duration_ms for litellm_overhead_time_ms.""" @@ -100,6 +115,41 @@ async def test_make_call_stream_cleanup_closes_http_response_once_under_cancella assert stream.aclose_calls == 1 +@pytest.mark.asyncio +async def test_make_sync_call_stream_cleanup_closes_http_response_once(): + stream: Final = _RecordingSyncByteStream() + response: Final = httpx.Response(200, stream=stream) + mock_client: Final = MagicMock() + mock_client.post.return_value = response + logging_obj: Final = MagicMock(model_call_details={"litellm_params": {}}) + + completion_stream, _ = make_sync_call( + client=mock_client, + api_base="https://api.anthropic.com/v1/messages", + headers={}, + data="{}", + model="claude-haiku-4-5", + messages=[{"role": "user", "content": "Hi"}], + logging_obj=logging_obj, + timeout=60.0, + json_mode=False, + ) + wrapper: Final = CustomStreamWrapper( + completion_stream=completion_stream, + model="claude-haiku-4-5", + custom_llm_provider="anthropic", + logging_obj=logging_obj, + ) + + await wrapper.aclose() + await wrapper.aclose() + await completion_stream.aclose() + + assert response.is_closed is True + assert completion_stream.http_response is None + assert stream.close_calls == 1 + + def test_redacted_thinking_content_block_delta(): chunk = { "type": "content_block_start",