From bf3f332a838c19b129538ee822b3af55a367058e Mon Sep 17 00:00:00 2001 From: Alex Castillo Date: Fri, 16 Oct 2026 20:33:22 -0400 Subject: [PATCH 1/2] fix(anthropic): take cumulative usage from message_delta when streaming Anthropic's message_delta event reports cumulative usage. With server tools (web search/fetch), its input and cache token counts exceed message_start's, but stream_merge_chunks only copied output_tokens, so streamed turns kept message_start's input and cache counts and under-reported tokens and cost compared with the non-streaming response (the web search recording has 17,269 cache-creation tokens on message_delta and 0 on message_start). Copy input_tokens, the cache counts and server_tool_use when message_delta carries them, as the Anthropic SDK's own accumulator does. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 1 + chatlas/_provider_anthropic.py | 12 ++++ tests/test_provider_anthropic.py | 94 +++++++++++++++++++++++++++++++- 3 files changed, 104 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index bd0cbb14..ed1096b4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,6 +20,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Bug fixes +* When streaming, `ChatAnthropic()` (and `ChatBedrockAnthropic()`) now takes the input and cache token counts from the final `message_delta` event, which reports cumulative usage, as the non-streaming response does. With server tools such as web search, those counts grow during the response, so streamed turns under-reported their tokens and cost (e.g. missing the cache writes made while searching). * `ChatBedrockAnthropic()` now uses native structured outputs for `data_model` on Claude 4.5 and later, as `ChatAnthropic()` does. Its Bedrock model IDs (`anthropic.claude-...`, `us.anthropic.claude-...`) weren't recognized, so it fell back to forcing a tool call, which Claude Opus 5.5 and Sonnet 5.5 reject with a 400. * A tool written as a generator (see "Progressive results" in the tool-calling docs) now reports each yielded value in `.chat_async()` and `.stream_async()`, as it does in `.chat()`. A synchronous generator tool used to send the generator object itself (``) as its result. An `async def` generator tool now counts as an async tool, so `.chat()` raises the usual "Cannot use async tools in a synchronous chat" error instead of sending `` as the result. * MCP tools registered with a `namespace` (in `.register_mcp_tools_http_stream_async()` or `.register_mcp_tools_stdio_async()`) can be called successfully. `Tool.from_mcp()` gains a `name` argument for setting the tool's name independently of the MCP tool's name. (#458) diff --git a/chatlas/_provider_anthropic.py b/chatlas/_provider_anthropic.py index 3a0b4e0e..a8a6adfd 100644 --- a/chatlas/_provider_anthropic.py +++ b/chatlas/_provider_anthropic.py @@ -704,6 +704,18 @@ def stream_merge_chunks(self, completion, chunk): completion.stop_reason = chunk.delta.stop_reason completion.stop_sequence = chunk.delta.stop_sequence completion.usage.output_tokens = chunk.usage.output_tokens + # Usage on message_delta is cumulative and, with server tools, can + # exceed message_start's input and cache counts. Fields are omitted + # when not applicable, in which case message_start's value stands. + for field in ( + "input_tokens", + "cache_creation_input_tokens", + "cache_read_input_tokens", + "server_tool_use", + ): + value = getattr(chunk.usage, field, None) + if value is not None: + setattr(completion.usage, field, value) return completion diff --git a/tests/test_provider_anthropic.py b/tests/test_provider_anthropic.py index d904029f..4bace956 100644 --- a/tests/test_provider_anthropic.py +++ b/tests/test_provider_anthropic.py @@ -9,10 +9,13 @@ CitationsWebSearchResultLocation, DocumentBlock, Message, + MessageDeltaUsage, RawContentBlockDeltaEvent, RawContentBlockStartEvent, RawContentBlockStopEvent, + RawMessageDeltaEvent, RawMessageStartEvent, + ServerToolUsage, TextBlock, TextDelta, Usage, @@ -302,6 +305,93 @@ def test_anthropic_web_fetch_citation_uses_document_index_for_source(): assert citation.source.title == "Second document" +def test_anthropic_stream_usage_takes_cumulative_message_delta_counts(): + # With server tools, message_delta carries the cumulative input and cache + # counts, which differ from message_start (values from the + # test_anthropic_web_search_streaming recording). + provider = AnthropicProvider(model="claude-haiku-4-5", api_key="dummy", kwargs=None) + start_usage = Usage( + input_tokens=2247, + cache_creation_input_tokens=0, + cache_read_input_tokens=0, + output_tokens=26, + ) + start = RawMessageStartEvent( + type="message_start", + message=Message( + id="msg", + type="message", + role="assistant", + model="claude-haiku-4-5", + content=[], + stop_reason=None, + stop_sequence=None, + usage=start_usage, + ), + ) + delta = RawMessageDeltaEvent.model_validate( + { + "type": "message_delta", + "delta": {"stop_reason": "end_turn", "stop_sequence": None}, + "usage": MessageDeltaUsage( + input_tokens=2254, + cache_creation_input_tokens=17269, + cache_read_input_tokens=0, + output_tokens=110, + server_tool_use=ServerToolUsage( + web_search_requests=1, web_fetch_requests=0 + ), + ).model_dump(), + } + ) + + completion = provider.stream_merge_chunks(None, start) + completion = provider.stream_merge_chunks(completion, delta) + + assert completion.usage.input_tokens == 2254 + assert completion.usage.cache_creation_input_tokens == 17269 + assert completion.usage.cache_read_input_tokens == 0 + assert completion.usage.output_tokens == 110 + assert completion.usage.server_tool_use == ServerToolUsage( + web_search_requests=1, web_fetch_requests=0 + ) + assert provider.value_tokens(completion) == (2254 + int(17269 * 1.25), 110, 0) + + +def test_anthropic_stream_usage_keeps_message_start_counts_omitted_by_delta(): + provider = AnthropicProvider(model="claude-haiku-4-5", api_key="dummy", kwargs=None) + start = RawMessageStartEvent( + type="message_start", + message=Message( + id="msg", + type="message", + role="assistant", + model="claude-haiku-4-5", + content=[], + stop_reason=None, + stop_sequence=None, + usage=Usage( + input_tokens=3, + cache_creation_input_tokens=37, + cache_read_input_tokens=17269, + output_tokens=1, + ), + ), + ) + delta = RawMessageDeltaEvent.model_validate( + { + "type": "message_delta", + "delta": {"stop_reason": "end_turn", "stop_sequence": None}, + "usage": {"output_tokens": 10}, + } + ) + + completion = provider.stream_merge_chunks(None, start) + completion = provider.stream_merge_chunks(completion, delta) + + assert provider.value_tokens(completion) == (3 + int(37 * 1.25), 10, 17269) + + def test_anthropic_web_fetch_citation_stream_uses_document_index_for_source(): provider = AnthropicProvider( model="claude-sonnet-4-5", api_key="dummy", kwargs=None @@ -1166,9 +1256,7 @@ def test_anthropic_message_params_convert_unsigned_thinking_blocks(): turns = [ UserTurn("hi"), - AssistantTurn( - [ContentThinking(thinking="hmm"), ContentText(text="hello")] - ), + AssistantTurn([ContentThinking(thinking="hmm"), ContentText(text="hello")]), UserTurn("again"), ] messages = provider._as_message_params(turns) From 19d279e3337fd7bd9a3a3a2b7fe9027977372c49 Mon Sep 17 00:00:00 2001 From: Alex Castillo Date: Fri, 16 Oct 2026 20:34:05 -0400 Subject: [PATCH 2/2] Add PR number to the changelog entry --- CHANGELOG.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ed1096b4..d02026a3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,7 +20,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Bug fixes -* When streaming, `ChatAnthropic()` (and `ChatBedrockAnthropic()`) now takes the input and cache token counts from the final `message_delta` event, which reports cumulative usage, as the non-streaming response does. With server tools such as web search, those counts grow during the response, so streamed turns under-reported their tokens and cost (e.g. missing the cache writes made while searching). +* When streaming, `ChatAnthropic()` (and `ChatBedrockAnthropic()`) now takes the input and cache token counts from the final `message_delta` event, which reports cumulative usage, as the non-streaming response does. With server tools such as web search, those counts grow during the response, so streamed turns under-reported their tokens and cost (e.g. missing the cache writes made while searching). (#466) * `ChatBedrockAnthropic()` now uses native structured outputs for `data_model` on Claude 4.5 and later, as `ChatAnthropic()` does. Its Bedrock model IDs (`anthropic.claude-...`, `us.anthropic.claude-...`) weren't recognized, so it fell back to forcing a tool call, which Claude Opus 5.5 and Sonnet 5.5 reject with a 400. * A tool written as a generator (see "Progressive results" in the tool-calling docs) now reports each yielded value in `.chat_async()` and `.stream_async()`, as it does in `.chat()`. A synchronous generator tool used to send the generator object itself (``) as its result. An `async def` generator tool now counts as an async tool, so `.chat()` raises the usual "Cannot use async tools in a synchronous chat" error instead of sending `` as the result. * MCP tools registered with a `namespace` (in `.register_mcp_tools_http_stream_async()` or `.register_mcp_tools_stdio_async()`) can be called successfully. `Tool.from_mcp()` gains a `name` argument for setting the tool's name independently of the MCP tool's name. (#458)