diff --git a/src/anthropic/lib/streaming/_beta_messages.py b/src/anthropic/lib/streaming/_beta_messages.py index 8101b4a32..c4ce89985 100644 --- a/src/anthropic/lib/streaming/_beta_messages.py +++ b/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,7 +588,6 @@ 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 @@ -594,22 +595,35 @@ def accumulate_event( 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 diff --git a/src/anthropic/lib/streaming/_messages.py b/src/anthropic/lib/streaming/_messages.py index 8ef8721d3..25007f706 100644 --- a/src/anthropic/lib/streaming/_messages.py +++ b/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 diff --git a/tests/lib/streaming/fixtures/message_start_omitted_usage_delta_usage_response.txt b/tests/lib/streaming/fixtures/message_start_omitted_usage_delta_usage_response.txt new file mode 100644 index 000000000..87fbcd01b --- /dev/null +++ b/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"} diff --git a/tests/lib/streaming/fixtures/message_start_omitted_usage_no_usage_response.txt b/tests/lib/streaming/fixtures/message_start_omitted_usage_no_usage_response.txt new file mode 100644 index 000000000..0a54e5f02 --- /dev/null +++ b/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"} diff --git a/tests/lib/streaming/test_beta_messages.py b/tests/lib/streaming/test_beta_messages.py index 6a0b3838f..5f030a122 100644 --- a/tests/lib/streaming/test_beta_messages.py +++ b/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 diff --git a/tests/lib/streaming/test_messages.py b/tests/lib/streaming/test_messages.py index 90d95ee37..5ebf09f01 100644 --- a/tests/lib/streaming/test_messages.py +++ b/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