test_stream_with_raw_schema() — anthropic-sdk-python Function Reference

Architecture documentation for the test_stream_with_raw_schema() function in test_beta_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
  dd54031b_584a_4852_c5cc_c514cb424766["test_stream_with_raw_schema()"]
  692db1db_e9a7_ed62_b129_14081098a7ae["TestAsyncMessages"]
  dd54031b_584a_4852_c5cc_c514cb424766 -->|defined in| 692db1db_e9a7_ed62_b129_14081098a7ae
  style dd54031b_584a_4852_c5cc_c514cb424766 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

tests/lib/_parse/test_beta_messages.py lines 18–48

    async def test_stream_with_raw_schema(self, async_client: AsyncAnthropic, respx_mock: MockRouter) -> None:
        async def async_stream_parse(client: AsyncAnthropic) -> ParsedBetaMessage[None]:
            async with client.beta.messages.stream(
                model="claude-sonnet-4-5",
                messages=[
                    {
                        "role": "user",
                        "content": "Extract order IDs from the following text:\n\nOrder 12345\nOrder 67890",
                    }
                ],
                output_format={
                    "type": "json_schema",
                    "schema": {
                        "type": "array",
                        "items": {"type": "integer"},
                    },
                },
                betas=["structured-outputs-2025-12-15"],
                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:48aac7c3-f271-47b3-854b-af4ed31e10bb.json"),
            respx_mock=respx_mock,
            mock_client=async_client,
            path="/v1/messages?beta=true",
        )

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