TestAsyncMessages Class — anthropic-sdk-python Architecture

Architecture documentation for the TestAsyncMessages class in test_messages.py from the anthropic-sdk-python codebase.

Class python AnthropicClient

Entity Profile

AnthropicClient→ TestAsyncMessages Class — anthropic-sdk-python Architecture

Dependency Diagram

graph TD
  58e51afa_63b8_b81d_14f4_b90e7b76512c["TestAsyncMessages"]
  03fc0a8b_1c63_1aee_ef30_754aeebc2ff6["AsyncStream"]
  58e51afa_63b8_b81d_14f4_b90e7b76512c -->|extends| 03fc0a8b_1c63_1aee_ef30_754aeebc2ff6
  8cc8cd20_a242_314f_b59d_2d554c624e7c["test_messages.py"]
  58e51afa_63b8_b81d_14f4_b90e7b76512c -->|defined in| 8cc8cd20_a242_314f_b59d_2d554c624e7c
  f6c26296_9449_f4e1_1c5a_123a02758efb["test_basic_response()"]
  58e51afa_63b8_b81d_14f4_b90e7b76512c -->|method| f6c26296_9449_f4e1_1c5a_123a02758efb
  fd7d72e5_d0f6_7fe7_a90d_08126c6c65d9["test_context_manager()"]
  58e51afa_63b8_b81d_14f4_b90e7b76512c -->|method| fd7d72e5_d0f6_7fe7_a90d_08126c6c65d9
  f42894a9_1e28_d352_5d76_aa4d35f3716f["test_deprecated_model_warning_stream()"]
  58e51afa_63b8_b81d_14f4_b90e7b76512c -->|method| f42894a9_1e28_d352_5d76_aa4d35f3716f
  5ebbac27_1dab_4a90_da3e_df522b071a97["test_tool_use()"]
  58e51afa_63b8_b81d_14f4_b90e7b76512c -->|method| 5ebbac27_1dab_4a90_da3e_df522b071a97

Relationship Graph

Source Code

tests/lib/streaming/test_messages.py lines 189–271

class TestAsyncMessages:
    @pytest.mark.asyncio
    @pytest.mark.respx(base_url=base_url)
    async def test_basic_response(self, respx_mock: MockRouter) -> None:
        respx_mock.post("/v1/messages").mock(
            return_value=httpx.Response(200, content=to_async_iter(get_response("basic_response.txt")))
        )

        async with async_client.messages.stream(
            max_tokens=1024,
            messages=[
                {
                    "role": "user",
                    "content": "Say hello there!",
                }
            ],
            model="claude-3-opus-latest",
        ) as stream:
            with pytest.warns(DeprecationWarning):
                assert isinstance(cast(Any, stream), AsyncStream)

            assert_basic_response([event async for event in stream], await stream.get_final_message())

    @pytest.mark.asyncio
    @pytest.mark.respx(base_url=base_url)
    async def test_context_manager(self, respx_mock: MockRouter) -> None:
        respx_mock.post("/v1/messages").mock(
            return_value=httpx.Response(200, content=to_async_iter(get_response("basic_response.txt")))
        )

        async with async_client.messages.stream(
            max_tokens=1024,
            messages=[
                {
                    "role": "user",
                    "content": "Say hello there!",
                }
            ],
            model="claude-3-opus-latest",
        ) as stream:
            assert not stream.response.is_closed

        # response should be closed even if the body isn't read
        assert stream.response.is_closed

    @pytest.mark.asyncio
    @pytest.mark.respx(base_url=base_url)
    async def test_deprecated_model_warning_stream(self, respx_mock: MockRouter) -> None:
        for deprecated_model in DEPRECATED_MODELS:
            respx_mock.post("/v1/messages").mock(
                return_value=httpx.Response(200, content=to_async_iter(get_response("basic_response.txt")))
            )

            with pytest.warns(DeprecationWarning, match=f"The model '{deprecated_model}' is deprecated"):
                async with async_client.messages.stream(
                    max_tokens=1024,
                    messages=[{"role": "user", "content": "Hello"}],
                    model=deprecated_model,
                ) as stream:
                    # Consume the stream to ensure the warning is triggered
                    await stream.get_final_message()

    @pytest.mark.asyncio
    @pytest.mark.respx(base_url=base_url)
    async def test_tool_use(self, respx_mock: MockRouter) -> None:
        respx_mock.post("/v1/messages").mock(
            return_value=httpx.Response(200, content=to_async_iter(get_response("tool_use_response.txt")))
        )

        async with async_client.messages.stream(
            max_tokens=1024,
            messages=[
                {
                    "role": "user",
                    "content": "Say hello there!",
                }
            ],
            model="claude-sonnet-4-20250514",
        ) as stream:
            with pytest.warns(DeprecationWarning):
                assert isinstance(cast(Any, stream), AsyncStream)