test_stream_with_raw_schema() — anthropic-sdk-python Function Reference

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

Function python AnthropicClient AsyncAPI

Entity Profile

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

Dependency Diagram

graph TD
  a53a8817_6e66_31ac_b782_0e5a157e3580["test_stream_with_raw_schema()"]
  af4d287a_1a7a_7175_0a72_07c984020b6c["TestAsyncStream"]
  a53a8817_6e66_31ac_b782_0e5a157e3580 -->|defined in| af4d287a_1a7a_7175_0a72_07c984020b6c
  style a53a8817_6e66_31ac_b782_0e5a157e3580 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

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

    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]")