stream() — anthropic-sdk-python Function Reference

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

Function python AnthropicClient SyncAPI calls 4

Entity Profile

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

Dependency Diagram

graph TD
  e1c1dc57_89ae_cea1_a4a6_084c63e1e1fa["__stream__()"]
  3c15a36b_3b82_93a8_fe82_3d955b1934ca["Stream"]
  e1c1dc57_89ae_cea1_a4a6_084c63e1e1fa -->|defined in| 3c15a36b_3b82_93a8_fe82_3d955b1934ca
  945cdcce_6566_7ff1_9ea3_601246f05c72["_iter_events()"]
  e1c1dc57_89ae_cea1_a4a6_084c63e1e1fa -->|calls| 945cdcce_6566_7ff1_9ea3_601246f05c72
  961352c5_6ade_542b_ef50_ed2ed6ed356b["json()"]
  e1c1dc57_89ae_cea1_a4a6_084c63e1e1fa -->|calls| 961352c5_6ade_542b_ef50_ed2ed6ed356b
  4eb2e348_281f_08ee_df0f_a3cb13e8d3ce["_make_status_error()"]
  e1c1dc57_89ae_cea1_a4a6_084c63e1e1fa -->|calls| 4eb2e348_281f_08ee_df0f_a3cb13e8d3ce
  ae6ec5e9_59f7_69b5_d385_025941be2db1["close()"]
  e1c1dc57_89ae_cea1_a4a6_084c63e1e1fa -->|calls| ae6ec5e9_59f7_69b5_d385_025941be2db1
  style e1c1dc57_89ae_cea1_a4a6_084c63e1e1fa fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

src/anthropic/_streaming.py lines 74–118

    def __stream__(self) -> Iterator[_T]:
        cast_to = cast(Any, self._cast_to)
        response = self.response
        process_data = self._client._process_response_data
        iterator = self._iter_events()

        try:
            for sse in iterator:
                if sse.event == "completion":
                    yield process_data(data=sse.json(), cast_to=cast_to, response=response)

                if (
                    sse.event == "message_start"
                    or sse.event == "message_delta"
                    or sse.event == "message_stop"
                    or sse.event == "content_block_start"
                    or sse.event == "content_block_delta"
                    or sse.event == "content_block_stop"
                ):
                    data = sse.json()
                    if is_dict(data) and "type" not in data:
                        data["type"] = sse.event

                    yield process_data(data=data, cast_to=cast_to, response=response)

                if sse.event == "ping":
                    continue

                if sse.event == "error":
                    body = sse.data

                    try:
                        body = sse.json()
                        err_msg = f"{body}"
                    except Exception:
                        err_msg = sse.data or f"Error code: {response.status_code}"

                    raise self._client._make_status_error(
                        err_msg,
                        body=body,
                        response=self.response,
                    )
        finally:
            # Ensure the response is closed even if the consumer doesn't read all data
            response.close()