iter_bytes() — anthropic-sdk-python Function Reference

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

Function python AnthropicClient SyncAPI calls 1

Entity Profile

AnthropicClient→ SyncAPI→ iter_bytes() — anthropic-sdk-python Function Reference

Dependency Diagram

graph TD
  b59cc883_d079_ab3c_26db_78fb47b9bde5["iter_bytes()"]
  582aef90_700e_ea04_755a_159b23a192d2["AWSEventStreamDecoder"]
  b59cc883_d079_ab3c_26db_78fb47b9bde5 -->|defined in| 582aef90_700e_ea04_755a_159b23a192d2
  21cf148e_e403_27d0_043f_c464fcb60144["_parse_message_from_event()"]
  b59cc883_d079_ab3c_26db_78fb47b9bde5 -->|calls| 21cf148e_e403_27d0_043f_c464fcb60144
  style b59cc883_d079_ab3c_26db_78fb47b9bde5 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

src/anthropic/lib/bedrock/_stream_decoder.py lines 30–40

    def iter_bytes(self, iterator: Iterator[bytes]) -> Iterator[ServerSentEvent]:
        """Given an iterator that yields lines, iterate over it & yield every event encountered"""
        from botocore.eventstream import EventStreamBuffer

        event_stream_buffer = EventStreamBuffer()
        for chunk in iterator:
            event_stream_buffer.add_data(chunk)
            for event in event_stream_buffer:
                message = self._parse_message_from_event(event)
                if message:
                    yield ServerSentEvent(data=message, event="completion")