to_custom_streamed_response_wrapper() — anthropic-sdk-python Function Reference

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

Function python AnthropicClient SyncAPI calls 1

Entity Profile

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

Dependency Diagram

graph TD
  71704806_f10e_ef1b_aafe_385b7cbe8672["to_custom_streamed_response_wrapper()"]
  2e9c15bb_de03_aa91_93be_0e05db0a552e["_response.py"]
  71704806_f10e_ef1b_aafe_385b7cbe8672 -->|defined in| 2e9c15bb_de03_aa91_93be_0e05db0a552e
  afc80c40_7ec9_e8a7_8063_f2ab74bc2ee7["get()"]
  71704806_f10e_ef1b_aafe_385b7cbe8672 -->|calls| afc80c40_7ec9_e8a7_8063_f2ab74bc2ee7
  style 71704806_f10e_ef1b_aafe_385b7cbe8672 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

src/anthropic/_response.py lines 727–749

def to_custom_streamed_response_wrapper(
    func: Callable[P, object],
    response_cls: type[_APIResponseT],
) -> Callable[P, ResponseContextManager[_APIResponseT]]:
    """Higher order function that takes one of our bound API methods and an `APIResponse` class
    and wraps the method to support streaming and returning the given response class directly.

    Note: the given `response_cls` *must* be concrete, e.g. `class BinaryAPIResponse(APIResponse[bytes])`
    """

    @functools.wraps(func)
    def wrapped(*args: P.args, **kwargs: P.kwargs) -> ResponseContextManager[_APIResponseT]:
        extra_headers: dict[str, Any] = {**(cast(Any, kwargs.get("extra_headers")) or {})}
        extra_headers[RAW_RESPONSE_HEADER] = "stream"
        extra_headers[OVERRIDE_CAST_TO_HEADER] = response_cls

        kwargs["extra_headers"] = extra_headers

        make_request = functools.partial(func, *args, **kwargs)

        return ResponseContextManager(cast(Callable[[], _APIResponseT], make_request))

    return wrapped