diff --git a/litellm/llms/anthropic/experimental_pass_through/responses_adapters/handler.py b/litellm/llms/anthropic/experimental_pass_through/responses_adapters/handler.py index 198ebe1ff8c..e80323eeb07 100644 --- a/litellm/llms/anthropic/experimental_pass_through/responses_adapters/handler.py +++ b/litellm/llms/anthropic/experimental_pass_through/responses_adapters/handler.py @@ -4,7 +4,7 @@ Handler for the Anthropic v1/messages -> OpenAI Responses API path. Used when the target model is an OpenAI or Azure model. """ -from typing import Any, AsyncIterator, Coroutine, Dict, List, Optional, Union +from typing import Any, AsyncIterator, Coroutine, Dict, Iterator, List, Optional, Union import litellm from litellm.types.llms.anthropic import AnthropicMessagesRequest @@ -178,6 +178,7 @@ class LiteLLMMessagesToResponsesAPIHandler: **kwargs, ) -> Union[ AnthropicMessagesResponse, + Iterator[Any], AsyncIterator[Any], Coroutine[Any, Any, Union[AnthropicMessagesResponse, AsyncIterator[Any]]], ]: @@ -231,7 +232,7 @@ class LiteLLMMessagesToResponsesAPIHandler: wrapper = AnthropicResponsesStreamWrapper( responses_stream=result, model=model ) - return wrapper.async_anthropic_sse_wrapper() + return wrapper.anthropic_sse_wrapper() if not isinstance(result, ResponsesAPIResponse): raise ValueError(f"Expected ResponsesAPIResponse, got {type(result)}") diff --git a/litellm/llms/anthropic/experimental_pass_through/responses_adapters/streaming_iterator.py b/litellm/llms/anthropic/experimental_pass_through/responses_adapters/streaming_iterator.py index aa0738a0719..7b0bafbfda0 100644 --- a/litellm/llms/anthropic/experimental_pass_through/responses_adapters/streaming_iterator.py +++ b/litellm/llms/anthropic/experimental_pass_through/responses_adapters/streaming_iterator.py @@ -3,7 +3,7 @@ import json import traceback from collections import deque -from typing import Any, AsyncIterator, Dict +from typing import Any, AsyncIterator, Dict, Iterator from litellm import verbose_logger from litellm._uuid import uuid @@ -300,6 +300,39 @@ class AnthropicResponsesStreamWrapper: self._sent_message_stop = True return + def __iter__(self) -> "AnthropicResponsesStreamWrapper": + return self + + def __next__(self) -> Dict[str, Any]: + # Return any queued chunks first + if self._chunk_queue: + return self._chunk_queue.popleft() + + # Emit message_start if not yet done (fallback if response.created wasn't fired) + if not self._sent_message_start: + self._sent_message_start = True + self._chunk_queue.append(self._make_message_start()) + return self._chunk_queue.popleft() + + # Consume the upstream stream + try: + for event in self.responses_stream: + self._process_event(event) + if self._chunk_queue: + return self._chunk_queue.popleft() + except StopIteration: + pass + except Exception as e: + verbose_logger.error( + f"AnthropicResponsesStreamWrapper error: {e}\n{traceback.format_exc()}" + ) + + # Drain any remaining queued chunks + if self._chunk_queue: + return self._chunk_queue.popleft() + + raise StopIteration + def __aiter__(self) -> "AnthropicResponsesStreamWrapper": return self @@ -333,6 +366,16 @@ class AnthropicResponsesStreamWrapper: raise StopAsyncIteration + def anthropic_sse_wrapper(self) -> Iterator[bytes]: + """Yield SSE-encoded bytes for each Anthropic event chunk (sync).""" + for chunk in self: + if isinstance(chunk, dict): + event_type: str = str(chunk.get("type", "message")) + payload = f"event: {event_type}\ndata: {json.dumps(chunk)}\n\n" + yield payload.encode() + else: + yield chunk + async def async_anthropic_sse_wrapper(self) -> AsyncIterator[bytes]: """Yield SSE-encoded bytes for each Anthropic event chunk.""" async for chunk in self: