test_messages_retries_async() — anthropic-sdk-python Function Reference

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

Function python AnthropicClient AsyncAPI

Entity Profile

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

Dependency Diagram

graph TD
  c4a35ef0_5fb9_8b6c_0ee9_48f3791bff1f["test_messages_retries_async()"]
  ba53edc6_e72a_dfe0_42d5_c2a06c57d6a1["test_bedrock.py"]
  c4a35ef0_5fb9_8b6c_0ee9_48f3791bff1f -->|defined in| ba53edc6_e72a_dfe0_42d5_c2a06c57d6a1
  style c4a35ef0_5fb9_8b6c_0ee9_48f3791bff1f fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

tests/lib/test_bedrock.py lines 102–132

async def test_messages_retries_async(respx_mock: MockRouter) -> None:
    respx_mock.post(re.compile(r"https://bedrock-runtime\.us-east-1\.amazonaws\.com/model/.*/invoke")).mock(
        side_effect=[
            httpx.Response(500, json={"error": "server error"}, headers={"retry-after-ms": "10"}),
            httpx.Response(200, json={"foo": "bar"}),
        ]
    )

    await async_client.messages.create(
        max_tokens=1024,
        messages=[
            {
                "role": "user",
                "content": "Say hello there!",
            }
        ],
        model="anthropic.claude-3-5-sonnet-20241022-v2:0",
    )

    calls = cast("list[MockRequestCall]", respx_mock.calls)

    assert len(calls) == 2

    assert (
        calls[0].request.url
        == "https://bedrock-runtime.us-east-1.amazonaws.com/model/anthropic.claude-3-5-sonnet-20241022-v2:0/invoke"
    )
    assert (
        calls[1].request.url
        == "https://bedrock-runtime.us-east-1.amazonaws.com/model/anthropic.claude-3-5-sonnet-20241022-v2:0/invoke"
    )