extract_stream_chunk_type() — anthropic-sdk-python Function Reference

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

Function python AnthropicClient AsyncAPI called by 2

Entity Profile

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

Dependency Diagram

graph TD
  941ce251_c4b2_51eb_76d8_a3a9127c15e5["extract_stream_chunk_type()"]
  8d141d22_ab1c_b4a1_744c_99e460d07454["_streaming.py"]
  941ce251_c4b2_51eb_76d8_a3a9127c15e5 -->|defined in| 8d141d22_ab1c_b4a1_744c_99e460d07454
  58686c44_d8b1_cf13_381b_6cff1300f441["_parse()"]
  58686c44_d8b1_cf13_381b_6cff1300f441 -->|calls| 941ce251_c4b2_51eb_76d8_a3a9127c15e5
  368da297_37b7_ab67_d177_ee29db8d5086["_parse()"]
  368da297_37b7_ab67_d177_ee29db8d5086 -->|calls| 941ce251_c4b2_51eb_76d8_a3a9127c15e5
  style 941ce251_c4b2_51eb_76d8_a3a9127c15e5 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

src/anthropic/_streaming.py lines 421–443

def extract_stream_chunk_type(
    stream_cls: type,
    *,
    failure_message: str | None = None,
) -> type:
    """Given a type like `Stream[T]`, returns the generic type variable `T`.

    This also handles the case where a concrete subclass is given, e.g.
    ```py
    class MyStream(Stream[bytes]):
        ...

    extract_stream_chunk_type(MyStream) -> bytes
    ```
    """
    from ._base_client import Stream, AsyncStream

    return extract_type_var_from_base(
        stream_cls,
        index=0,
        generic_bases=cast("tuple[type, ...]", (Stream, AsyncStream)),
        failure_message=failure_message,
    )