TestAsyncStream Class — anthropic-sdk-python Architecture

Source Code

tests/lib/_parse/test_messages.py lines 163–199

class TestAsyncStream:
    async def test_stream_with_raw_schema(self, async_client: AsyncAnthropic, respx_mock: MockRouter) -> None:
        """Test async messages.stream() with raw JSON schema via output_config."""

        async def async_stream_parse(client: AsyncAnthropic) -> Message:
            async with client.messages.stream(
                model="claude-sonnet-4-5",
                messages=[
                    {
                        "role": "user",
                        "content": "Extract order IDs from the following text:\n\nOrder 12345\nOrder 67890",
                    }
                ],
                output_config={
                    "format": {
                        "type": "json_schema",
                        "schema": {
                            "type": "array",
                            "items": {"type": "integer"},
                        },
                    },
                },
                max_tokens=1024,
            ) as stream:
                return await stream.get_final_message()

        response = await make_async_stream_snapshot_request(
            async_stream_parse,
            content_snapshot=external("uuid:b2c4d6e8-f012-4a56-8b90-1234567890ab.json"),
            respx_mock=respx_mock,
            mock_client=async_client,
            path="/v1/messages",
        )

        content_block = response.content[0]
        assert content_block.type == "text"
        assert content_block.text == snapshot("[12345,67890]")