Implementation:BerriAI Litellm Responses Streaming Iterator

Property	Value
sources	litellm/responses/streaming_iterator.py
domains	Streaming, Responses API, SSE, Logging
last_updated	2026-02-15 16:00 GMT

Overview

The Responses Streaming Iterator module implements a class hierarchy of streaming iterators that process Server-Sent Events (SSE) chunks from the Responses API, providing async, sync, and mock streaming variants with integrated logging and callback support.

Description

This module defines a base class BaseResponsesAPIStreamingIterator that contains shared logic for parsing SSE chunks, handling completed response events, triggering logging callbacks, and managing failure handlers. Three concrete subclasses extend this base: ResponsesAPIStreamingIterator (async, using __aiter__/__anext__), SyncResponsesAPIStreamingIterator (sync, using __iter__/__next__), and MockResponsesAPIStreamingIterator (fakes streaming by slicing a complete response into 5-character delta chunks for models like o1-pro that do not natively support streaming). Each iterator processes raw HTTP response lines, delegates transformation to the provider config, updates response IDs with LiteLLM metadata, runs post-streaming deployment hooks, and logs completed responses through both async and sync success handlers.

Usage

Import this module when you need to consume streaming responses from the Responses API. The iterators are created internally by BaseLLMHTTPHandler and returned from litellm.responses() when stream=True.

Code Reference

Source Location

Property	Value
Repository	github.com/BerriAI/litellm
File	litellm/responses/streaming_iterator.py
Lines	635
Module	litellm.responses.streaming_iterator

Signature

class BaseResponsesAPIStreamingIterator:
    def __init__(
        self,
        response: httpx.Response,
        model: str,
        responses_api_provider_config: BaseResponsesAPIConfig,
        logging_obj: LiteLLMLoggingObj,
        litellm_metadata: Optional[Dict[str, Any]] = None,
        custom_llm_provider: Optional[str] = None,
        request_data: Optional[Dict[str, Any]] = None,
        call_type: Optional[str] = None,
    )
    def _process_chunk(self, chunk) -> Optional[ResponsesAPIStreamingResponse]
    def _handle_logging_completed_response(self)
    def _run_post_success_hooks(self, end_time: datetime)
    def _handle_failure(self, exception: Exception)

class ResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator):
    # Async: __aiter__ / __anext__

class SyncResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator):
    # Sync: __iter__ / __next__

class MockResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator):
    CHUNK_SIZE = 5
    # Both __aiter__/__anext__ and __iter__/__next__

Import

from litellm.responses.streaming_iterator import (
    BaseResponsesAPIStreamingIterator,
    ResponsesAPIStreamingIterator,
    SyncResponsesAPIStreamingIterator,
    MockResponsesAPIStreamingIterator,
)

I/O Contract

Inputs

Parameter	Type	Required	Description
response	httpx.Response	Yes	The raw HTTP response object from the provider
model	str	Yes	The model name used for the request
responses_api_provider_config	BaseResponsesAPIConfig	Yes	Provider-specific config for transforming streaming chunks
logging_obj	LiteLLMLoggingObj	Yes	The logging object for recording success/failure events
litellm_metadata	Optional[Dict[str, Any]]	No	Metadata containing model_info for response ID encoding
custom_llm_provider	Optional[str]	No	The provider name for response ID encoding
request_data	Optional[Dict[str, Any]]	No	The original request data for hooks
call_type	Optional[str]	No	The call type (e.g., "responses") for callback routing

Outputs

Output	Type	Description
Each iteration yields	ResponsesAPIStreamingResponse	A parsed streaming event (delta, completed, etc.)
Final event	ResponseCompletedEvent	The completed response event with full response and usage

Usage Examples

import litellm

# The streaming iterator is returned when stream=True
stream = litellm.responses(
    input="Write a story about a dragon.",
    model="openai/gpt-4",
    stream=True,
)

# Sync iteration
for chunk in stream:
    if hasattr(chunk, "delta"):
        print(chunk.delta, end="")

import asyncio
import litellm

async def main():
    stream = await litellm.aresponses(
        input="Tell me a joke.",
        model="openai/gpt-4",
        stream=True,
    )
    async for chunk in stream:
        if hasattr(chunk, "delta"):
            print(chunk.delta, end="")

asyncio.run(main())

Related Pages

• Implementation:BerriAI_Litellm_Responses_API -- The main Responses API module that creates and returns these streaming iterators
• Implementation:BerriAI_Litellm_Responses_Utils -- Utility methods used by the iterator for response ID encoding