fix(streaming): tolerate message_start without usage

src/anthropic/lib/streaming/_beta_messages.py

@@ -31,7 +31,9 @@
 from ...types.beta import BetaRawMessageStreamEvent
 from ..._utils._utils import is_given
 from .._parse._response import ResponseFormatT, parse_text
+from ...types.beta.beta_usage import BetaUsage
 from ...types.beta.parsed_beta_message import ParsedBetaMessage, ParsedBetaContentBlock
+from ...types.beta.beta_message_delta_usage import BetaMessageDeltaUsage
 
 
 class BetaMessageStream(Generic[ResponseFormatT]):
@@ -586,30 +588,42 @@ def accumulate_event(
         current_snapshot.stop_details = event.delta.stop_details
         if event.delta.container is not None:
             current_snapshot.container = event.delta.container
-        current_snapshot.usage.output_tokens = event.usage.output_tokens
         if event.context_management is not None:
             current_snapshot.context_management = event.context_management
         # only sent on `message_delta` after a mid-stream fallback, in which case it
         # replaces the `message_start` value; otherwise that value must survive
         if event.input_transformations is not None:
             current_snapshot.input_transformations = event.input_transformations
 
-        # Usage counts on a message_delta are cumulative totals, so they overwrite rather
-        # than add; optional ones are omitted when not applicable, in which case the
-        # message_start value must survive.
-        if event.usage.input_tokens is not None:
-            current_snapshot.usage.input_tokens = event.usage.input_tokens
-        if event.usage.cache_creation_input_tokens is not None:
-            current_snapshot.usage.cache_creation_input_tokens = event.usage.cache_creation_input_tokens
-        if event.usage.cache_read_input_tokens is not None:
-            current_snapshot.usage.cache_read_input_tokens = event.usage.cache_read_input_tokens
-        if event.usage.server_tool_use is not None:
-            current_snapshot.usage.server_tool_use = event.usage.server_tool_use
-        if event.usage.output_tokens_details is not None:
-            current_snapshot.usage.output_tokens_details = event.usage.output_tokens_details
-        if event.usage.iterations is not None:
-            current_snapshot.usage.iterations = event.usage.iterations
-        if event.usage.fallback_credit is not None:
-            current_snapshot.usage.fallback_credit = event.usage.fallback_credit
+        # `message_start` may omit `usage` (the streaming docs' thinking example does), and
+        # `construct()` fills the missing field with `None`. Seed it from the first delta that
+        # carries usage, and let a stream that never carries it finish with `usage=None`
+        # rather than raising an `AttributeError` mid-stream.
+        delta_usage = cast("BetaMessageDeltaUsage | None", event.usage)
+        if delta_usage is not None:
+            snapshot_usage = cast("BetaUsage | None", current_snapshot.usage)
+            if snapshot_usage is None:
+                snapshot_usage = BetaUsage.construct(**cast(Any, delta_usage.to_dict()))
+                current_snapshot.usage = snapshot_usage
+
+            snapshot_usage.output_tokens = delta_usage.output_tokens
+
+            # Usage counts on a message_delta are cumulative totals, so they overwrite rather
+            # than add; optional ones are omitted when not applicable, in which case the
+            # message_start value must survive.
+            if delta_usage.input_tokens is not None:
+                snapshot_usage.input_tokens = delta_usage.input_tokens
+            if delta_usage.cache_creation_input_tokens is not None:
+                snapshot_usage.cache_creation_input_tokens = delta_usage.cache_creation_input_tokens
+            if delta_usage.cache_read_input_tokens is not None:
+                snapshot_usage.cache_read_input_tokens = delta_usage.cache_read_input_tokens
+            if delta_usage.server_tool_use is not None:
+                snapshot_usage.server_tool_use = delta_usage.server_tool_use
+            if delta_usage.output_tokens_details is not None:
+                snapshot_usage.output_tokens_details = delta_usage.output_tokens_details
+            if delta_usage.iterations is not None:
+                snapshot_usage.iterations = delta_usage.iterations
+            if delta_usage.fallback_credit is not None:
+                snapshot_usage.fallback_credit = delta_usage.fallback_credit
 
     return current_snapshot

src/anthropic/lib/streaming/_messages.py

@@ -26,9 +26,11 @@
 from ..._models import build, construct_type, construct_type_unchecked
 from .._tracing import share_message_snapshot, withhold_message_snapshot
 from ..._streaming import Stream, AsyncStream
+from ...types.usage import Usage
 from ..._utils._utils import is_given
 from .._parse._response import ResponseFormatT, parse_text
 from ...types.parsed_message import ParsedMessage, ParsedContentBlock
+from ...types.message_delta_usage import MessageDeltaUsage
 
 
 class MessageStream(Generic[ResponseFormatT]):
@@ -540,20 +542,32 @@ def accumulate_event(
         current_snapshot.stop_details = event.delta.stop_details
         if event.delta.container is not None:
             current_snapshot.container = event.delta.container
-        current_snapshot.usage.output_tokens = event.usage.output_tokens
-
-        # Usage counts on a message_delta are cumulative totals, so they overwrite rather
-        # than add; optional ones are omitted when not applicable, in which case the
-        # message_start value must survive.
-        if event.usage.input_tokens is not None:
-            current_snapshot.usage.input_tokens = event.usage.input_tokens
-        if event.usage.cache_creation_input_tokens is not None:
-            current_snapshot.usage.cache_creation_input_tokens = event.usage.cache_creation_input_tokens
-        if event.usage.cache_read_input_tokens is not None:
-            current_snapshot.usage.cache_read_input_tokens = event.usage.cache_read_input_tokens
-        if event.usage.server_tool_use is not None:
-            current_snapshot.usage.server_tool_use = event.usage.server_tool_use
-        if event.usage.output_tokens_details is not None:
-            current_snapshot.usage.output_tokens_details = event.usage.output_tokens_details
+
+        # `message_start` may omit `usage` (the streaming docs' thinking example does), and
+        # `construct()` fills the missing field with `None`. Seed it from the first delta that
+        # carries usage, and let a stream that never carries it finish with `usage=None`
+        # rather than raising an `AttributeError` mid-stream.
+        delta_usage = cast("MessageDeltaUsage | None", event.usage)
+        if delta_usage is not None:
+            snapshot_usage = cast("Usage | None", current_snapshot.usage)
+            if snapshot_usage is None:
+                snapshot_usage = Usage.construct(**cast(Any, delta_usage.to_dict()))
+                current_snapshot.usage = snapshot_usage
+
+            snapshot_usage.output_tokens = delta_usage.output_tokens
+
+            # Usage counts on a message_delta are cumulative totals, so they overwrite rather
+            # than add; optional ones are omitted when not applicable, in which case the
+            # message_start value must survive.
+            if delta_usage.input_tokens is not None:
+                snapshot_usage.input_tokens = delta_usage.input_tokens
+            if delta_usage.cache_creation_input_tokens is not None:
+                snapshot_usage.cache_creation_input_tokens = delta_usage.cache_creation_input_tokens
+            if delta_usage.cache_read_input_tokens is not None:
+                snapshot_usage.cache_read_input_tokens = delta_usage.cache_read_input_tokens
+            if delta_usage.server_tool_use is not None:
+                snapshot_usage.server_tool_use = delta_usage.server_tool_use
+            if delta_usage.output_tokens_details is not None:
+                snapshot_usage.output_tokens_details = delta_usage.output_tokens_details
 
     return current_snapshot

tests/lib/streaming/fixtures/message_start_omitted_usage_delta_usage_response.txt

@@ -0,0 +1,17 @@
+event: message_start
+data: {"type":"message_start","message":{"id":"msg_01MessageStartOmitted000001","type":"message","role":"assistant","content":[],"model":"claude-sonnet-4-5","stop_reason":null,"stop_sequence":null}}
+
+event: content_block_start
+data: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}
+
+event: content_block_delta
+data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Hello there!"}}
+
+event: content_block_stop
+data: {"type":"content_block_stop","index":0}
+
+event: message_delta
+data: {"type":"message_delta","delta":{"stop_reason":"end_turn","stop_sequence":null},"usage":{"output_tokens":7}}
+
+event: message_stop
+data: {"type":"message_stop"}

tests/lib/streaming/fixtures/message_start_omitted_usage_no_usage_response.txt

@@ -0,0 +1,17 @@
+event: message_start
+data: {"type":"message_start","message":{"id":"msg_01MessageStartOmitted000002","type":"message","role":"assistant","content":[],"model":"claude-sonnet-4-5","stop_reason":null,"stop_sequence":null}}
+
+event: content_block_start
+data: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}
+
+event: content_block_delta
+data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Hello there!"}}
+
+event: content_block_stop
+data: {"type":"content_block_stop","index":0}
+
+event: message_delta
+data: {"type":"message_delta","delta":{"stop_reason":"end_turn","stop_sequence":null}}
+
+event: message_stop
+data: {"type":"message_stop"}

tests/lib/streaming/test_beta_messages.py

@@ -34,6 +34,11 @@
 sync_client = Anthropic(base_url=base_url, api_key=api_key, _strict_response_validation=True)
 async_client = AsyncAnthropic(base_url=base_url, api_key=api_key, _strict_response_validation=True)
 
+# a `message_start` without `usage` fails strict response validation, so the regression tests
+# for streams that omit it use default clients, which is what streaming users get
+nonstrict_sync_client = Anthropic(base_url=base_url, api_key=api_key)
+nonstrict_async_client = AsyncAnthropic(base_url=base_url, api_key=api_key)
+
 _T = TypeVar("_T")
 
 
@@ -411,6 +416,23 @@ def assert_input_transformations_response(message: BetaMessage, expected: List[D
     assert [entry.to_dict() for entry in message.input_transformations] == expected
 
 
+def assert_message_start_omitted_usage_seeded_from_delta(message: BetaMessage) -> None:
+    # `message_start` omitted `usage`, so the first delta that carries it seeds the snapshot;
+    # counts the stream never reported stay unset instead of being invented as zero
+    assert message.usage.output_tokens == 7
+    assert message.to_dict()["usage"] == {"output_tokens": 7}
+
+
+def assert_message_start_omitted_usage_never_supplied(message: BetaMessage) -> None:
+    # the streaming docs' thinking sequence omits `usage` from both events: the stream must
+    # finish without raising and leave `usage` unset
+    assert "usage" not in message.to_dict()
+    assert message.stop_reason == "end_turn"
+    content = message.content[0]
+    assert content.type == "text"
+    assert content.text == "Hello there!"
+
+
 class TestSyncMessages:
     @pytest.mark.respx(base_url=base_url)
     def test_basic_response(self, respx_mock: MockRouter) -> None:
@@ -684,6 +706,34 @@ def test_input_transformations_propagated(
         ) as stream:
             assert_input_transformations_response(stream.get_final_message(), expected)
 
+    @pytest.mark.respx(base_url=base_url)
+    def test_message_start_omitted_usage_seeded_from_delta(self, respx_mock: MockRouter) -> None:
+        respx_mock.post("/v1/messages").mock(
+            return_value=httpx2.Response(
+                200, content=get_response("message_start_omitted_usage_delta_usage_response.txt")
+            )
+        )
+
+        with nonstrict_sync_client.beta.messages.stream(
+            max_tokens=1024,
+            messages=[{"role": "user", "content": "Say hello there!"}],
+            model="claude-sonnet-5-5",
+        ) as stream:
+            assert_message_start_omitted_usage_seeded_from_delta(stream.get_final_message())
+
+    @pytest.mark.respx(base_url=base_url)
+    def test_message_start_omitted_usage_never_supplied(self, respx_mock: MockRouter) -> None:
+        respx_mock.post("/v1/messages").mock(
+            return_value=httpx2.Response(200, content=get_response("message_start_omitted_usage_no_usage_response.txt"))
+        )
+
+        with nonstrict_sync_client.beta.messages.stream(
+            max_tokens=1024,
+            messages=[{"role": "user", "content": "Say hello there!"}],
+            model="claude-sonnet-5-5",
+        ) as stream:
+            assert_message_start_omitted_usage_never_supplied(stream.get_final_message())
+
 
 class TestAsyncMessages:
     @pytest.mark.asyncio
@@ -1008,6 +1058,38 @@ async def test_input_transformations_propagated(
         ) as stream:
             assert_input_transformations_response(await stream.get_final_message(), expected)
 
+    @pytest.mark.asyncio
+    @pytest.mark.respx(base_url=base_url)
+    async def test_message_start_omitted_usage_seeded_from_delta(self, respx_mock: MockRouter) -> None:
+        respx_mock.post("/v1/messages").mock(
+            return_value=httpx2.Response(
+                200, content=to_async_iter(get_response("message_start_omitted_usage_delta_usage_response.txt"))
+            )
+        )
+
+        async with nonstrict_async_client.beta.messages.stream(
+            max_tokens=1024,
+            messages=[{"role": "user", "content": "Say hello there!"}],
+            model="claude-sonnet-5-5",
+        ) as stream:
+            assert_message_start_omitted_usage_seeded_from_delta(await stream.get_final_message())
+
+    @pytest.mark.asyncio
+    @pytest.mark.respx(base_url=base_url)
+    async def test_message_start_omitted_usage_never_supplied(self, respx_mock: MockRouter) -> None:
+        respx_mock.post("/v1/messages").mock(
+            return_value=httpx2.Response(
+                200, content=to_async_iter(get_response("message_start_omitted_usage_no_usage_response.txt"))
+            )
+        )
+
+        async with nonstrict_async_client.beta.messages.stream(
+            max_tokens=1024,
+            messages=[{"role": "user", "content": "Say hello there!"}],
+            model="claude-sonnet-5-5",
+        ) as stream:
+            assert_message_start_omitted_usage_never_supplied(await stream.get_final_message())
+
 
 def test_message_delta_fields_are_all_accumulated() -> None:
     # tripwire: handle a new field in accumulate_event, then list it here

tests/lib/streaming/test_messages.py

@@ -29,6 +29,11 @@
 sync_client = Anthropic(base_url=base_url, api_key=api_key, _strict_response_validation=True)
 async_client = AsyncAnthropic(base_url=base_url, api_key=api_key, _strict_response_validation=True)
 
+# a `message_start` without `usage` fails strict response validation, so the regression tests
+# for streams that omit it use default clients, which is what streaming users get
+nonstrict_sync_client = Anthropic(base_url=base_url, api_key=api_key)
+nonstrict_async_client = AsyncAnthropic(base_url=base_url, api_key=api_key)
+
 _T = TypeVar("_T")
 
 # the accumulator must wrap the raw parser error with context and echo the offending JSON
@@ -253,6 +258,23 @@ def assert_message_delta_omitted_usage_response(message: Message) -> None:
     assert message.container is None
 
 
+def assert_message_start_omitted_usage_seeded_from_delta(message: Message) -> None:
+    # `message_start` omitted `usage`, so the first delta that carries it seeds the snapshot;
+    # counts the stream never reported stay unset instead of being invented as zero
+    assert message.usage.output_tokens == 7
+    assert message.to_dict()["usage"] == {"output_tokens": 7}
+
+
+def assert_message_start_omitted_usage_never_supplied(message: Message) -> None:
+    # the streaming docs' thinking sequence omits `usage` from both events: the stream must
+    # finish without raising and leave `usage` unset
+    assert "usage" not in message.to_dict()
+    assert message.stop_reason == "end_turn"
+    content = message.content[0]
+    assert content.type == "text"
+    assert content.text == "Hello there!"
+
+
 class TestSyncMessages:
     @pytest.mark.respx(base_url=base_url)
     def test_basic_response(self, respx_mock: MockRouter) -> None:
@@ -508,6 +530,34 @@ def test_speed_is_forwarded(self, respx_mock: MockRouter) -> None:
         request_body = json.loads(route.calls.last.request.content)
         assert request_body["speed"] == "fast"
 
+    @pytest.mark.respx(base_url=base_url)
+    def test_message_start_omitted_usage_seeded_from_delta(self, respx_mock: MockRouter) -> None:
+        respx_mock.post("/v1/messages").mock(
+            return_value=httpx2.Response(
+                200, content=get_response("message_start_omitted_usage_delta_usage_response.txt")
+            )
+        )
+
+        with nonstrict_sync_client.messages.stream(
+            max_tokens=1024,
+            messages=[{"role": "user", "content": "Say hello there!"}],
+            model="claude-sonnet-5-5",
+        ) as stream:
+            assert_message_start_omitted_usage_seeded_from_delta(stream.get_final_message())
+
+    @pytest.mark.respx(base_url=base_url)
+    def test_message_start_omitted_usage_never_supplied(self, respx_mock: MockRouter) -> None:
+        respx_mock.post("/v1/messages").mock(
+            return_value=httpx2.Response(200, content=get_response("message_start_omitted_usage_no_usage_response.txt"))
+        )
+
+        with nonstrict_sync_client.messages.stream(
+            max_tokens=1024,
+            messages=[{"role": "user", "content": "Say hello there!"}],
+            model="claude-sonnet-5-5",
+        ) as stream:
+            assert_message_start_omitted_usage_never_supplied(stream.get_final_message())
+
 
 class TestAsyncMessages:
     @pytest.mark.asyncio
@@ -788,6 +838,38 @@ async def test_speed_is_forwarded(self, respx_mock: MockRouter) -> None:
         request_body = json.loads(route.calls.last.request.content)
         assert request_body["speed"] == "fast"
 
+    @pytest.mark.asyncio
+    @pytest.mark.respx(base_url=base_url)
+    async def test_message_start_omitted_usage_seeded_from_delta(self, respx_mock: MockRouter) -> None:
+        respx_mock.post("/v1/messages").mock(
+            return_value=httpx2.Response(
+                200, content=to_async_iter(get_response("message_start_omitted_usage_delta_usage_response.txt"))
+            )
+        )
+
+        async with nonstrict_async_client.messages.stream(
+            max_tokens=1024,
+            messages=[{"role": "user", "content": "Say hello there!"}],
+            model="claude-sonnet-5-5",
+        ) as stream:
+            assert_message_start_omitted_usage_seeded_from_delta(await stream.get_final_message())
+
+    @pytest.mark.asyncio
+    @pytest.mark.respx(base_url=base_url)
+    async def test_message_start_omitted_usage_never_supplied(self, respx_mock: MockRouter) -> None:
+        respx_mock.post("/v1/messages").mock(
+            return_value=httpx2.Response(
+                200, content=to_async_iter(get_response("message_start_omitted_usage_no_usage_response.txt"))
+            )
+        )
+
+        async with nonstrict_async_client.messages.stream(
+            max_tokens=1024,
+            messages=[{"role": "user", "content": "Say hello there!"}],
+            model="claude-sonnet-5-5",
+        ) as stream:
+            assert_message_start_omitted_usage_never_supplied(await stream.get_final_message())
+
 
 def test_message_delta_fields_are_all_accumulated() -> None:
     # tripwire: handle a new field in accumulate_event (src/anthropic/lib/streaming/_messages.py), then list it here