aiter_bytes() — anthropic-sdk-python Function Reference

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

Function python AnthropicClient AsyncAPI calls 1

Entity Profile

AnthropicClient→ AsyncAPI→ aiter_bytes() — anthropic-sdk-python Function Reference

Dependency Diagram

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

Relationship Graph

Source Code

src/anthropic/lib/bedrock/_stream_decoder.py lines 42–52

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

        event_stream_buffer = EventStreamBuffer()
        async 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")