From ce9fa96315b0a451118a46327975e7c6e6e6ca95 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C4=B0lker=20O=C4=9Fulcan=20Say=C4=B1n?= <47698920+iosayin@users.noreply.github.com> Date: Wed, 7 Oct 2026 03:58:48 +0300 Subject: [PATCH] fix(streaming): guard stop fields in message_delta against being overwritten by None (#1940) --- src/anthropic/lib/streaming/_beta_messages.py | 9 ++-- src/anthropic/lib/streaming/_messages.py | 9 ++-- tests/lib/streaming/test_beta_messages.py | 49 +++++++++++++++++++ tests/lib/streaming/test_messages.py | 43 ++++++++++++++++ 4 files changed, 104 insertions(+), 6 deletions(-) diff --git a/src/anthropic/lib/streaming/_beta_messages.py b/src/anthropic/lib/streaming/_beta_messages.py index c7d3d6c2c..d545ae86c 100644 --- a/src/anthropic/lib/streaming/_beta_messages.py +++ b/src/anthropic/lib/streaming/_beta_messages.py @@ -546,9 +546,12 @@ def accumulate_event( if content_block.type == "text" and is_given(output_format): content_block.parsed_output = parse_text(content_block.text, output_format) elif event.type == "message_delta": - current_snapshot.stop_reason = event.delta.stop_reason - current_snapshot.stop_sequence = event.delta.stop_sequence - current_snapshot.stop_details = event.delta.stop_details + if event.delta.stop_reason is not None: + current_snapshot.stop_reason = event.delta.stop_reason + if event.delta.stop_sequence is not None: + current_snapshot.stop_sequence = event.delta.stop_sequence + if event.delta.stop_details is not None: + 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 diff --git a/src/anthropic/lib/streaming/_messages.py b/src/anthropic/lib/streaming/_messages.py index d036a04d1..53da3553a 100644 --- a/src/anthropic/lib/streaming/_messages.py +++ b/src/anthropic/lib/streaming/_messages.py @@ -512,9 +512,12 @@ def accumulate_event( if content_block.type == "text" and is_given(output_format): content_block.parsed_output = parse_text(content_block.text, output_format) elif event.type == "message_delta": - current_snapshot.stop_reason = event.delta.stop_reason - current_snapshot.stop_sequence = event.delta.stop_sequence - current_snapshot.stop_details = event.delta.stop_details + if event.delta.stop_reason is not None: + current_snapshot.stop_reason = event.delta.stop_reason + if event.delta.stop_sequence is not None: + current_snapshot.stop_sequence = event.delta.stop_sequence + if event.delta.stop_details is not None: + 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 diff --git a/tests/lib/streaming/test_beta_messages.py b/tests/lib/streaming/test_beta_messages.py index fc2af491d..155264f14 100644 --- a/tests/lib/streaming/test_beta_messages.py +++ b/tests/lib/streaming/test_beta_messages.py @@ -895,6 +895,55 @@ async def test_input_transformations_propagated( ) as stream: assert_input_transformations_response(await stream.get_final_message(), expected) + def test_message_delta_subsequent_delta_preserves_stop_fields(self) -> None: + from anthropic.types.beta import ( + BetaUsage, + BetaMessage, + BetaMessageDeltaUsage, + BetaRawMessageDeltaEvent, + BetaRawMessageStartEvent, + ) + from anthropic.lib.streaming._beta_messages import accumulate_event + from anthropic.types.beta.beta_raw_message_delta_event import Delta as BetaMessageDelta + + start_event = BetaRawMessageStartEvent( + type="message_start", + message=BetaMessage( + id="msg_123", + type="message", + role="assistant", + content=[], + model="claude-3-7-sonnet-20250219", + stop_reason=None, + stop_sequence=None, + usage=BetaUsage(input_tokens=10, output_tokens=0), + ), + ) + snapshot = accumulate_event(event=start_event, current_snapshot=None, json_bufs={}, request_headers=httpx2.Headers()) + + delta_1 = BetaRawMessageDeltaEvent( + type="message_delta", + delta=BetaMessageDelta( + stop_reason="end_turn", + stop_sequence=None, + ), + usage=BetaMessageDeltaUsage(output_tokens=5), + ) + snapshot = accumulate_event(event=delta_1, current_snapshot=snapshot, json_bufs={}, request_headers=httpx2.Headers()) + assert snapshot.stop_reason == "end_turn" + + delta_2 = BetaRawMessageDeltaEvent( + type="message_delta", + delta=BetaMessageDelta( + stop_reason=None, + stop_sequence=None, + ), + usage=BetaMessageDeltaUsage(output_tokens=8), + ) + snapshot = accumulate_event(event=delta_2, current_snapshot=snapshot, json_bufs={}, request_headers=httpx2.Headers()) + assert snapshot.stop_reason == "end_turn" + assert snapshot.usage.output_tokens == 8 + def test_message_delta_fields_are_all_accumulated() -> None: # tripwire: handle a new field in accumulate_event, then list it here diff --git a/tests/lib/streaming/test_messages.py b/tests/lib/streaming/test_messages.py index 6f3bb0cc3..4a854418b 100644 --- a/tests/lib/streaming/test_messages.py +++ b/tests/lib/streaming/test_messages.py @@ -641,6 +641,49 @@ async def test_message_stop_event_serialization(self, respx_mock: MockRouter) -> stop_event.model_dump() stop_event.model_dump_json() + def test_message_delta_subsequent_delta_preserves_stop_fields(self) -> None: + from anthropic.types import Usage, Message, MessageDeltaUsage, RawMessageDeltaEvent, RawMessageStartEvent + from anthropic.lib.streaming._messages import accumulate_event + from anthropic.types.raw_message_delta_event import Delta as MessageDelta + + start_event = RawMessageStartEvent( + type="message_start", + message=Message( + id="msg_123", + type="message", + role="assistant", + content=[], + model="claude-3-7-sonnet-20250219", + stop_reason=None, + stop_sequence=None, + usage=Usage(input_tokens=10, output_tokens=0), + ), + ) + snapshot = accumulate_event(event=start_event, current_snapshot=None, json_bufs={}) + + delta_1 = RawMessageDeltaEvent( + type="message_delta", + delta=MessageDelta( + stop_reason="end_turn", + stop_sequence=None, + ), + usage=MessageDeltaUsage(output_tokens=5), + ) + snapshot = accumulate_event(event=delta_1, current_snapshot=snapshot, json_bufs={}) + assert snapshot.stop_reason == "end_turn" + + delta_2 = RawMessageDeltaEvent( + type="message_delta", + delta=MessageDelta( + stop_reason=None, + stop_sequence=None, + ), + usage=MessageDeltaUsage(output_tokens=8), + ) + snapshot = accumulate_event(event=delta_2, current_snapshot=snapshot, json_bufs={}) + assert snapshot.stop_reason == "end_turn" + assert snapshot.usage.output_tokens == 8 + 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