to_streamed_response_wrapper() — anthropic-sdk-python Function Reference

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

Function python AnthropicClient SyncClient calls 1

Entity Profile

AnthropicClient→ SyncClient→ to_streamed_response_wrapper() — anthropic-sdk-python Function Reference

Dependency Diagram

graph TD
  940b1f4d_092c_500d_97d2_2257b8a209c6["to_streamed_response_wrapper()"]
  1a90c143_a412_238a_409b_ed09811c4cc9["_response.py"]
  940b1f4d_092c_500d_97d2_2257b8a209c6 -->|defined in| 1a90c143_a412_238a_409b_ed09811c4cc9
  bad13882_92c8_5b27_0a6d_fade7a911293["get()"]
  940b1f4d_092c_500d_97d2_2257b8a209c6 -->|calls| bad13882_92c8_5b27_0a6d_fade7a911293
  style 940b1f4d_092c_500d_97d2_2257b8a209c6 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

src/anthropic/_response.py lines 687–703

def to_streamed_response_wrapper(func: Callable[P, R]) -> Callable[P, ResponseContextManager[APIResponse[R]]]:
    """Higher order function that takes one of our bound API methods and wraps it
    to support streaming and returning the raw `APIResponse` object directly.
    """

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

        kwargs["extra_headers"] = extra_headers

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

        return ResponseContextManager(cast(Callable[[], APIResponse[R]], make_request))

    return wrapped