Implementation:Googleapis Python genai Interactions Streaming

Knowledge Sources	Googleapis_Python_genai
Domains	Streaming, SSE
Last Updated	2026-02-15 14:00 GMT

Overview

Concrete tool for Server-Sent Events (SSE) streaming response handling with sync and async support for the _interactions subsystem.

Description

The _streaming module provides Stream and AsyncStream classes that decode SSE byte streams into typed events. The SSEDecoder handles raw byte parsing into ServerSentEvent objects, which are then deserialized into typed model objects.

Usage

Internal infrastructure. Users interact with streaming indirectly through generate_content_stream and Live API methods.

Code Reference

Source Location

• Repository: Googleapis_Python_genai
• File: google/genai/_interactions/_streaming.py
• Lines: 1-354

Signature

class Stream(Generic[_T]):
    def __init__(
        self,
        *,
        cast_to: type[_T],
        response: httpx.Response,
        client: GeminiNextGenAPIClient,
    ) -> None: ...

    def __next__(self) -> _T: ...
    def __iter__(self) -> Iterator[_T]: ...
    def close(self) -> None: ...

class AsyncStream(Generic[_T]):
    async def __anext__(self) -> _T: ...
    async def __aiter__(self) -> AsyncIterator[_T]: ...
    async def close(self) -> None: ...

class ServerSentEvent:
    @property
    def event(self) -> str | None: ...
    @property
    def data(self) -> str: ...
    def json(self) -> Any: ...

class SSEDecoder:
    def iter_bytes(self, iterator: Iterator[bytes]) -> Iterator[ServerSentEvent]: ...
    async def aiter_bytes(self, iterator: AsyncIterator[bytes]) -> AsyncIterator[ServerSentEvent]: ...

Import

from google.genai._interactions._streaming import Stream, AsyncStream

I/O Contract

Inputs

Name	Type	Required	Description
cast_to	type[_T]	Yes	Target type for each event
response	httpx.Response	Yes	HTTP response with SSE body

Outputs

Name	Type	Description
__iter__/__aiter__ yields	_T	Typed events parsed from SSE stream

Usage Examples

# Internal usage - streaming is used by generate_content_stream
# and the Interactions API
from google.genai._interactions._streaming import Stream

# Stream objects are created internally and returned to users
# as iterators over typed events
for event in stream:
    print(event)

Related Pages

• Environment:Googleapis_Python_genai_Python_3_10_SDK_Runtime