chore(internal): simplify http snapshots

Author: karpetrosyan

.inline-snapshot/external/cd8d3d185e7a993935ab650e0e5c6a7970758bac7d0487f9b1afaad2cc095f3c.json

@@ -99,7 +99,6 @@ $ ./scripts/test
 ### Snapshots
 
 Some tests use [inline-snapshot](https://15r10nk.github.io/inline-snapshot/latest/). To update them after making changes, rerun the tests with the `--inline-snapshot=fix` and `-n0` options:
-
 ```bash
 ./scripts/test --inline-snapshot=fix -n0
 ```
@@ -108,15 +107,14 @@ Some tests use [inline-snapshot](https://15r10nk.github.io/inline-snapshot/lates
 > `inline-snapshot` is incompatible with [pytest-xdist](https://github.com/pytest-dev/pytest-xdist), so you need to disable parallel execution `(-n0)` when using the `--inline-snapshot` option.
 
 In addition, some tests capture snapshots of the HTTP requests they make.
-To refresh these snapshots, run the tests with the `ANTHROPIC_LIVE=1` environment variable enabled.
-
+To refresh these snapshots, run the tests with the `--http-record` flag:
 ```bash
-ANTHROPIC_LIVE=1 ./scripts/test --inline-snapshot=fix
+./scripts/test --inline-snapshot=fix --http-record
 ```
 
 > [!NOTE]
-> Sometimes it makes sense to update only the inline snapshots `(--inline-snapshot=fix)` without refreshing the HTTP snapshots `(ANTHROPIC_LIVE=1)`.
-> This is useful when the endpoint hasn’t changed, but your code handles the response differently and the assertions need updating.
+> Sometimes it makes sense to update only the inline snapshots `(--inline-snapshot=fix)` without refreshing the HTTP snapshots `(--http-record)`.
+> This is useful when the endpoint hasn't changed, but your code handles the response differently and the assertions need updating.
 
 ## Linting and formatting
 

CONTRIBUTING.md

@@ -74,6 +74,7 @@ dev = [
     "pytest-xdist>=3.6.1",
     "inline-snapshot>=0.28.0",
     "griffe>=1",
+    "http-snapshot[httpx]==0.1.4",
 ]
 pydantic-v1 = [
     "pydantic>=1.9.0,<2",

pyproject.toml

@@ -58,6 +58,22 @@ def client(request: FixtureRequest) -> Iterator[Anthropic]:
         yield client
 
 
+@pytest.fixture(scope="function")
+def snapshot_client(snapshot_sync_httpx_client: httpx.Client, is_recording: bool) -> Iterator[Anthropic]:
+    with Anthropic(http_client=snapshot_sync_httpx_client, api_key=None if is_recording else api_key) as client:
+        yield client
+
+
+@pytest.fixture(scope="function")
+async def async_snapshot_client(
+    snapshot_async_httpx_client: httpx.AsyncClient, is_recording: bool
+) -> AsyncIterator[AsyncAnthropic]:
+    async with AsyncAnthropic(
+        http_client=snapshot_async_httpx_client, api_key=None if is_recording else api_key
+    ) as client:
+        yield client
+
+
 @pytest.fixture(scope="session")
 async def async_client(request: FixtureRequest) -> AsyncIterator[AsyncAnthropic]:
     param = getattr(request, "param", True)

tests/conftest.py

@@ -0,0 +1,22 @@
+[
+  {
+    "response": {
+      "status_code": 200,
+      "headers": {
+        "date": "Fri, 05 Dec 2025 15:06:08 GMT",
+        "content-type": "text/event-stream; charset=utf-8",
+        "connection": "keep-alive",
+        "cf-ray": "9a947204ac019fb2-AMS",
+        "cache-control": "no-cache",
+        "request-id": "req_011CVokJu9sXWCZuA7nVS1g3",
+        "strict-transport-security": "max-age=31536000; includeSubDomains; preload",
+        "anthropic-organization-id": "5576611f-a1ee-427c-9800-298d9579899c",
+        "x-envoy-upstream-service-time": "3888",
+        "cf-cache-status": "DYNAMIC",
+        "x-robots-tag": "none",
+        "server": "cloudflare"
+      },
+      "body": "event: message_start\ndata: {\"type\":\"message_start\",\"message\":{\"model\":\"claude-sonnet-4-5-20250929\",\"id\":\"msg_015jm362jgYXWyFRvq8NHxPr\",\"type\":\"message\",\"role\":\"assistant\",\"content\":[],\"stop_reason\":null,\"stop_sequence\":null,\"usage\":{\"input_tokens\":135,\"cache_creation_input_tokens\":0,\"cache_read_input_tokens\":0,\"cache_creation\":{\"ephemeral_5m_input_tokens\":0,\"ephemeral_1h_input_tokens\":0},\"output_tokens\":1,\"service_tier\":\"standard\"}}            }\n\nevent: content_block_start\ndata: {\"type\":\"content_block_start\",\"index\":0,\"content_block\":{\"type\":\"text\",\"text\":\"\"}               }\n\nevent: content_block_delta\ndata: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"text_delta\",\"text\":\"[\"}             }\n\nevent: content_block_delta\ndata: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"text_delta\",\"text\":\"12\"}           }\n\nevent: ping\ndata: {\"type\": \"ping\"}\n\nevent: content_block_delta\ndata: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"text_delta\",\"text\":\"345,\"}             }\n\nevent: content_block_delta\ndata: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"text_delta\",\"text\":\"67890]\"}    }\n\nevent: content_block_stop\ndata: {\"type\":\"content_block_stop\",\"index\":0  }\n\nevent: message_delta\ndata: {\"type\":\"message_delta\",\"delta\":{\"stop_reason\":\"end_turn\",\"stop_sequence\":null},\"usage\":{\"input_tokens\":135,\"cache_creation_input_tokens\":0,\"cache_read_input_tokens\":0,\"output_tokens\":10}         }\n\nevent: message_stop\ndata: {\"type\":\"message_stop\"  }\n\n"
+    }
+  }
+]

tests/lib/_parse/__inline_snapshot__/test_beta_messages/TestAsyncMessages.test_stream_with_raw_schema/48aac7c3-f271-47b3-854b-af4ed31e10bb.json

@@ -1,16 +1,21 @@
+from typing import Any, cast
+
 import pytest
-from respx import MockRouter
 from inline_snapshot import external, snapshot
 
 from anthropic import AsyncAnthropic, _compat
 from anthropic.types.beta.parsed_beta_message import ParsedBetaMessage
 
-from ..snapshots import make_async_stream_snapshot_request
-
 
 @pytest.mark.skipif(_compat.PYDANTIC_V1, reason="tool runner not supported with pydantic v1")
 class TestAsyncMessages:
-    async def test_stream_with_raw_schema(self, async_client: AsyncAnthropic, respx_mock: MockRouter) -> None:
+    @pytest.mark.parametrize(
+        "http_snapshot",
+        [
+            cast(Any, external("uuid:606342ef-f614-4bc1-b5e7-2c3305a48bf3.json")),
+        ],
+    )
+    async def test_stream_with_raw_schema(self, async_snapshot_client: AsyncAnthropic) -> None:
         async def async_stream_parse(client: AsyncAnthropic) -> ParsedBetaMessage[None]:
             async with client.beta.messages.stream(
                 model="claude-sonnet-4-5",
@@ -32,12 +37,6 @@ async def async_stream_parse(client: AsyncAnthropic) -> ParsedBetaMessage[None]:
             ) as stream:
                 return await stream.get_final_message()
 
-        response = await make_async_stream_snapshot_request(
-            async_stream_parse,
-            content_snapshot=external("uuid:48aac7c3-f271-47b3-854b-af4ed31e10bb.json"),
-            respx_mock=respx_mock,
-            mock_client=async_client,
-            path="/v1/messages?beta=true",
-        )
+        response = await async_stream_parse(async_snapshot_client)
 
         assert response.content[0].text == snapshot("[12345,67890]")