stream() — anthropic-sdk-python Function Reference

Architecture documentation for the __stream__() function in _beta_messages.py from the anthropic-sdk-python codebase.

Function python StreamProtocol AsyncStreaming calls 2

Entity Profile

StreamProtocol→ AsyncStreaming→ __stream__() — anthropic-sdk-python Function Reference

Dependency Diagram

graph TD
  0e363261_728a_0dbe_df13_19e318699314["__stream__()"]
  fa27ef0f_31e3_cd5e_ffa3_12becd4b426b["BetaMessageStream"]
  0e363261_728a_0dbe_df13_19e318699314 -->|defined in| fa27ef0f_31e3_cd5e_ffa3_12becd4b426b
  a902fa3e_1643_6ad3_983d_fb2f08bc9b41["accumulate_event()"]
  0e363261_728a_0dbe_df13_19e318699314 -->|calls| a902fa3e_1643_6ad3_983d_fb2f08bc9b41
  3df97661_6baf_7419_93d1_ca34ae93ba17["build_events()"]
  0e363261_728a_0dbe_df13_19e318699314 -->|calls| 3df97661_6baf_7419_93d1_ca34ae93ba17
  style 0e363261_728a_0dbe_df13_19e318699314 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

src/anthropic/lib/streaming/_beta_messages.py lines 131–142

    def __stream__(self) -> Iterator[ParsedBetaMessageStreamEvent[ResponseFormatT]]:
        for sse_event in self._raw_stream:
            self.__final_message_snapshot = accumulate_event(
                event=sse_event,
                current_snapshot=self.__final_message_snapshot,
                request_headers=self.response.request.headers,
                output_format=self.__output_format,
            )

            events_to_fire = build_events(event=sse_event, message_snapshot=self.current_message_snapshot)
            for event in events_to_fire:
                yield event