make_stream_snapshot_request() — anthropic-sdk-python Function Reference

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

Function python AnthropicClient AsyncAPI

Entity Profile

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

Dependency Diagram

graph TD
  943089fb_d9a5_efa9_680d_85739098ff0b["make_stream_snapshot_request()"]
  e7250a77_f5e0_e4de_3e1b_f2b040399c23["snapshots.py"]
  943089fb_d9a5_efa9_680d_85739098ff0b -->|defined in| e7250a77_f5e0_e4de_3e1b_f2b040399c23
  style 943089fb_d9a5_efa9_680d_85739098ff0b fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

tests/lib/snapshots.py lines 81–137

def make_stream_snapshot_request(
    func: Callable[[Anthropic], _T],
    *,
    content_snapshot: Any,
    respx_mock: MockRouter,
    mock_client: Anthropic,
    path: str,
) -> _T:
    live = os.environ.get("ANTHROPIC_LIVE") == "1"
    collected: list[str] = []
    if live:

        def _on_response(response: httpx.Response) -> None:
            response.read()
            collected.append(response.text)

        respx_mock.stop()

        client = Anthropic(
            http_client=httpx.Client(
                event_hooks={
                    "response": [_on_response],
                }
            )
        )
    else:
        response_contents = get_snapshot_value(content_snapshot)
        assert is_list(response_contents)

        response_contents = cast(
            List[str],
            response_contents,
        )

        curr = 0

        def get_response(_request: httpx.Request) -> httpx.Response:
            nonlocal curr
            content = response_contents[curr]
            assert isinstance(content, str)

            curr += 1
            return httpx.Response(
                200,
                content=content.encode("utf-8"),
                headers={"content-type": "text/event-stream"},
            )

        respx_mock.post(path).mock(side_effect=get_response)
        client = mock_client

    result = func(client)
    if not live:
        return result

    assert outsource(collected) == content_snapshot
    return result