feat(streaming): carry final response cost on streamed usage by default

Streamed responses through the proxy previously exposed no usable cost:
the x-litellm-response-cost header is unreadable mid-stream and the final
usage chunk carried only tokens, priced against an alias model name the
client cannot resolve. The include_cost_in_streaming_usage flag existed
but was off by default and only fixed the wire, not SDK clients.

Stamp usage.cost into the joined streaming response by default wherever a
final usage object is built: the chat-completions stream_chunk_builder,
the native /v1/responses RESPONSE_COMPLETED event, and synthetic response
events. Provider-reported cost always wins over the computed value, and
only positive computed costs are stamped so unpriceable alias responses
keep deferring to the logging object's own calculation. Per-chunk SSE
cost injection (/v1/messages, generateContent, passthrough) stays behind
the flag.

Also normalize non-litellm usage objects in stream_chunk_builder: openai
CompletionUsage lacks Usage.__contains__, so membership probes silently
returned False and client-side rebuilds dropped the wire cost and
recounted token usage locally. Wire token counts and cost now survive.

Resolves LIT-6427

litellm/litellm_core_utils/streaming_chunk_builder_utils.py

@@ -36,6 +36,8 @@ from litellm.types.utils import (
 from litellm.utils import print_verbose, token_counter
 
 if TYPE_CHECKING:
+    from openai.types.completion_usage import CompletionUsage
+
     from litellm.litellm_core_utils.litellm_logging import Logging
     from litellm.types.litellm_core_utils.streaming_chunk_builder_utils import (
         UsagePerChunk,
@@ -782,7 +784,7 @@ class ChunkProcessor:
 
     @staticmethod
     def _extract_usage_chunk(chunk: "_UsageBearingChunk | ModelResponse | ModelResponseStream") -> Usage | None:
-        usage_chunk: Usage | None = None
+        usage_chunk: Usage | CompletionUsage | None = None
         if hasattr(chunk, "usage") and chunk.usage is not None:
             usage_chunk = chunk.usage
         elif "usage" in chunk:
@@ -794,7 +796,9 @@ class ChunkProcessor:
 
         if isinstance(usage_chunk, dict):
             return Usage(**usage_chunk)
-        return usage_chunk
+        if usage_chunk is None or isinstance(usage_chunk, Usage):
+            return usage_chunk
+        return Usage(**usage_chunk.model_dump())
 
     def _calculate_usage_per_chunk(
         self,

litellm/main.py

@@ -8634,6 +8634,16 @@ def _set_stream_builder_response_cost(response: ModelResponse, logging_obj: Opti
     hidden_params["response_cost"] = response_cost
 
 
+def _stamp_streaming_usage_cost(usage: Usage, response: ModelResponse, logging_obj: Optional["Logging"]) -> None:
+    if logging_obj is None:
+        return
+    if isinstance(getattr(usage, "cost", None), (int, float)):
+        return
+    computed_cost: Final = logging_obj._response_cost_calculator(result=response)
+    if isinstance(computed_cost, (int, float)) and computed_cost > 0:
+        setattr(usage, "cost", computed_cost)
+
+
 def stream_chunk_builder(
     chunks: list,
     messages: list | None = None,
@@ -8728,12 +8738,7 @@ def stream_chunk_builder(
                     )
                     break
 
-            if litellm.include_cost_in_streaming_usage and logging_obj is not None:
-                setattr(
-                    usage,
-                    "cost",
-                    logging_obj._response_cost_calculator(result=response),
-                )
+            _stamp_streaming_usage_cost(usage, response, logging_obj)
             _set_stream_builder_response_cost(response, logging_obj)
 
             processor.apply_provider_assembled_streaming_metadata(response, chunks, logging_obj)
@@ -8912,10 +8917,7 @@ def stream_chunk_builder(
                 )
                 break
 
-        # Add cost to usage object if include_cost_in_streaming_usage is True
-        if litellm.include_cost_in_streaming_usage and logging_obj is not None:
-            setattr(usage, "cost", logging_obj._response_cost_calculator(result=response))
-
+        _stamp_streaming_usage_cost(usage, response, logging_obj)
         _set_stream_builder_response_cost(response, logging_obj)
 
         processor.apply_provider_assembled_streaming_metadata(response, chunks, logging_obj)

litellm/responses/litellm_completion_transformation/streaming_iterator.py

@@ -1164,16 +1164,6 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
 
     def _emit_response_completed_event(self, litellm_model_response: ModelResponse) -> ResponseCompletedEvent | None:
         if litellm_model_response:
-            # Add cost to usage object if include_cost_in_streaming_usage is True
-            if litellm.include_cost_in_streaming_usage and self.litellm_logging_obj is not None:
-                usage: Final[object] = getattr(litellm_model_response, "usage", None)
-                if usage is not None:
-                    setattr(
-                        usage,
-                        "cost",
-                        self.litellm_logging_obj._response_cost_calculator(result=litellm_model_response),
-                    )
-
             # Transform the response
             responses_api_response: Final = (
                 LiteLLMCompletionResponsesConfig.transform_chat_completion_response_to_responses_api_response(

litellm/responses/streaming_iterator.py

@@ -405,23 +405,7 @@ class BaseResponsesAPIStreamingIterator:
                     openai_types.ResponsesAPIStreamEvents.RESPONSE_FAILED,
                 ):
                     self.completed_response = openai_responses_api_chunk
-                    # Add cost to usage object if include_cost_in_streaming_usage is True
-                    if litellm.include_cost_in_streaming_usage and self.logging_obj is not None:
-                        response_obj: Final[ResponsesAPIResponse | None] = getattr(
-                            openai_responses_api_chunk, "response", None
-                        )
-                        if response_obj:
-                            usage_obj: Final[ResponseAPIUsage | None] = getattr(response_obj, "usage", None)
-                            if usage_obj is not None:
-                                try:
-                                    cost: Final[float | None] = self.logging_obj._response_cost_calculator(
-                                        result=response_obj
-                                    )
-                                    if cost is not None:
-                                        setattr(usage_obj, "cost", cost)
-                                except Exception:
-                                    # Best-effort usage cost annotation should not break stream replay.
-                                    pass
+                    _stamp_responses_usage_cost(getattr(openai_responses_api_chunk, "response", None), self.logging_obj)
 
                     if _chunk_type == openai_types.ResponsesAPIStreamEvents.RESPONSE_FAILED:
                         self._handle_logging_failed_response()
@@ -1272,6 +1256,24 @@ def _add_text_like_part_events(
         )
 
 
+def _stamp_responses_usage_cost(
+    response_obj: ResponsesAPIResponse | None, logging_obj: LiteLLMLoggingObj | None
+) -> None:
+    if response_obj is None or logging_obj is None:
+        return
+    usage_obj: Final[ResponseAPIUsage | None] = getattr(response_obj, "usage", None)
+    if usage_obj is None:
+        return
+    if isinstance(getattr(usage_obj, "cost", None), (int, float)):
+        return
+    try:
+        cost: Final[float | None] = logging_obj._response_cost_calculator(result=response_obj)
+    except Exception:
+        return
+    if isinstance(cost, (int, float)) and cost > 0:
+        setattr(usage_obj, "cost", cost)
+
+
 def _build_synthetic_response_events(
     *,
     transformed: ResponsesAPIResponse,
@@ -1279,15 +1281,7 @@ def _build_synthetic_response_events(
     chunk_size: int,
 ) -> list[ResponsesAPIStreamingResponse]:
     openai_types: Final = _get_openai_response_types()
-    if litellm.include_cost_in_streaming_usage and logging_obj is not None:
-        usage_obj: Final = transformed.usage if hasattr(transformed, "usage") else None
-        if usage_obj is not None:
-            try:
-                cost: Final[float | None] = logging_obj._response_cost_calculator(result=transformed)
-                if cost is not None:
-                    setattr(usage_obj, "cost", cost)
-            except Exception:
-                pass
+    _stamp_responses_usage_cost(transformed, logging_obj)
 
     events: Final[list[ResponsesAPIStreamingResponse]] = [
         _build_response_status_event(openai_types.ResponsesAPIStreamEvents.RESPONSE_CREATED, transformed),

tests/test_litellm/litellm_core_utils/test_streaming_chunk_builder_utils.py

@@ -592,6 +592,59 @@ def test_stream_chunk_builder_litellm_usage_chunks():
     assert usage.total_tokens == 77
 
 
+def test_calculate_usage_honors_openai_sdk_completion_usage_chunks():
+    from openai.types.completion_usage import CompletionUsage
+
+    content_chunk = ModelResponseStream(
+        id="chatcmpl-sdk-usage-1",
+        created=1745513206,
+        model="mantle-claude",
+        object="chat.completion.chunk",
+        system_fingerprint=None,
+        choices=[
+            StreamingChoices(
+                finish_reason="stop",
+                index=0,
+                delta=Delta(
+                    provider_specific_fields=None,
+                    content="ok",
+                    role=None,
+                    function_call=None,
+                    tool_calls=None,
+                    audio=None,
+                ),
+                logprobs=None,
+            )
+        ],
+        provider_specific_fields=None,
+        stream_options={"include_usage": True},
+    )
+    usage_chunk = ModelResponseStream(
+        id="chatcmpl-sdk-usage-1",
+        created=1745513207,
+        model="mantle-claude",
+        object="chat.completion.chunk",
+        system_fingerprint=None,
+        choices=[],
+        provider_specific_fields=None,
+        stream_options={"include_usage": True},
+    )
+    usage_chunk.usage = CompletionUsage(
+        prompt_tokens=20, completion_tokens=60, total_tokens=80, cost=0.000704
+    )
+    assert type(usage_chunk.usage) is CompletionUsage
+
+    chunks = [content_chunk, usage_chunk]
+    usage = ChunkProcessor(chunks=chunks).calculate_usage(
+        chunks=chunks, model="mantle-claude", completion_output=""
+    )
+
+    assert usage.prompt_tokens == 20
+    assert usage.completion_tokens == 60
+    assert usage.total_tokens == 80
+    assert getattr(usage, "cost", None) == pytest.approx(0.000704)
+
+
 def test_get_model_from_chunks_azure_model_router():
     """
     Test that _get_model_from_chunks finds the actual model from Azure Model Router chunks.

tests/test_litellm/responses/test_streaming_iterator.py

@@ -326,3 +326,55 @@ def test_run_post_success_hooks_does_not_report_generation_time_as_overhead():
 
     assert iterator.completed_response._hidden_params["_response_ms"] == 10000.0
     assert "litellm_overhead_time_ms" not in iterator.completed_response._hidden_params
+
+
+def _responses_api_response_with_usage() -> ResponsesAPIResponse:
+    from litellm.types.llms.openai import ResponseAPIUsage
+
+    return ResponsesAPIResponse(
+        id="resp_lit6427",
+        created_at=int(datetime(2025, 1, 1).timestamp()),
+        status="completed",
+        model="mantle-claude",
+        object="response",
+        output=[],
+        usage=ResponseAPIUsage(input_tokens=20, output_tokens=60, total_tokens=80),
+    )
+
+
+def test_stamp_responses_usage_cost_stamps_computed_cost():
+    from litellm.responses.streaming_iterator import _stamp_responses_usage_cost
+
+    response = _responses_api_response_with_usage()
+    logging_obj = Mock(spec=LiteLLMLoggingObj)
+    logging_obj._response_cost_calculator.return_value = 0.000704
+
+    _stamp_responses_usage_cost(response, logging_obj)
+
+    assert getattr(response.usage, "cost", None) == pytest.approx(0.000704)
+    logging_obj._response_cost_calculator.assert_called_once_with(result=response)
+
+
+def test_stamp_responses_usage_cost_keeps_provider_reported_cost():
+    from litellm.responses.streaming_iterator import _stamp_responses_usage_cost
+
+    response = _responses_api_response_with_usage()
+    setattr(response.usage, "cost", 0.5)
+    logging_obj = Mock(spec=LiteLLMLoggingObj)
+
+    _stamp_responses_usage_cost(response, logging_obj)
+
+    assert getattr(response.usage, "cost", None) == pytest.approx(0.5)
+    logging_obj._response_cost_calculator.assert_not_called()
+
+
+def test_stamp_responses_usage_cost_survives_calculator_failure():
+    from litellm.responses.streaming_iterator import _stamp_responses_usage_cost
+
+    response = _responses_api_response_with_usage()
+    logging_obj = Mock(spec=LiteLLMLoggingObj)
+    logging_obj._response_cost_calculator.side_effect = RuntimeError("cost map unavailable")
+
+    _stamp_responses_usage_cost(response, logging_obj)
+
+    assert getattr(response.usage, "cost", None) is None

tests/test_litellm/test_main.py

@@ -3150,8 +3150,8 @@ def _stream_builder_logging_obj() -> LiteLLMLogging:
     return logging_obj
 
 
-def test_stream_chunk_builder_reports_streaming_usage_cost_when_enabled(monkeypatch: pytest.MonkeyPatch):
-    monkeypatch.setattr(litellm, "include_cost_in_streaming_usage", True)
+def test_stream_chunk_builder_stamps_streaming_usage_cost_by_default(monkeypatch: pytest.MonkeyPatch):
+    monkeypatch.setattr(litellm, "include_cost_in_streaming_usage", False)
     chunks: Final = [
         _stream_builder_text_chunk("gpt-4o", "Hello "),
         _stream_builder_text_chunk("gpt-4o", "world.", finish_reason="stop"),
@@ -3168,11 +3168,45 @@ def test_stream_chunk_builder_reports_streaming_usage_cost_when_enabled(monkeypa
     assert response._hidden_params["response_cost"] == pytest.approx(usage_cost)
 
 
-def test_stream_chunk_builder_defers_cost_to_logging_obj_when_usage_cost_absent(monkeypatch: pytest.MonkeyPatch):
-    monkeypatch.setattr(litellm, "include_cost_in_streaming_usage", False)
+def test_stream_chunk_builder_skips_stamp_when_cost_is_unpriceable():
+    import time as time_module
+
+    from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLogging
+
+    logging_obj: Final = LiteLLMLogging(
+        model="us.anthropic.claude-opus-5",
+        messages=[{"role": "user", "content": "hi"}],
+        stream=True,
+        call_type="completion",
+        start_time=time_module.time(),
+        litellm_call_id="stream-builder-alias-unpriceable",
+        function_id="1",
+    )
+    logging_obj.model_call_details["custom_llm_provider"] = "bedrock"
+    logging_obj.optional_params = {}
+    usage_chunk: Final = _stream_builder_text_chunk("bedrock-claude-opus-5", "")
+    usage_chunk.usage = Usage(prompt_tokens=40, completion_tokens=5, total_tokens=45)
+    chunks: Final = [
+        _stream_builder_text_chunk("bedrock-claude-opus-5", "Hello ", finish_reason="stop"),
+        usage_chunk,
+    ]
+
+    response: Final = litellm.stream_chunk_builder(
+        chunks=chunks, messages=[{"role": "user", "content": "hi"}], logging_obj=logging_obj
+    )
+
+    assert response is not None
+    assert getattr(response.usage, "cost", None) is None
+    assert response._hidden_params.get("response_cost") is None
+
+
+def test_stream_chunk_builder_keeps_provider_reported_usage_cost():
+    usage_chunk: Final = _stream_builder_text_chunk("gpt-4o", "")
+    usage_chunk.usage = Usage(prompt_tokens=10, completion_tokens=5, total_tokens=15, cost=0.5)
     chunks: Final = [
         _stream_builder_text_chunk("gpt-4o", "Hello "),
         _stream_builder_text_chunk("gpt-4o", "world.", finish_reason="stop"),
+        usage_chunk,
     ]
 
     response: Final = litellm.stream_chunk_builder(
@@ -3180,4 +3214,26 @@ def test_stream_chunk_builder_defers_cost_to_logging_obj_when_usage_cost_absent(
     )
 
     assert response is not None
-    assert response._hidden_params.get("response_cost") is None
+    assert getattr(response.usage, "cost", None) == pytest.approx(0.5)
+    assert response._hidden_params["response_cost"] == pytest.approx(0.5)
+
+
+def test_stream_chunk_builder_prices_alias_from_openai_sdk_usage_chunk():
+    from openai.types.completion_usage import CompletionUsage
+
+    usage_chunk: Final = _stream_builder_text_chunk("mantle-claude", "")
+    usage_chunk.usage = CompletionUsage(prompt_tokens=20, completion_tokens=60, total_tokens=80, cost=0.000704)
+    assert type(usage_chunk.usage) is CompletionUsage
+    chunks: Final = [
+        _stream_builder_text_chunk("mantle-claude", "Hello "),
+        _stream_builder_text_chunk("mantle-claude", "world.", finish_reason="stop"),
+        usage_chunk,
+    ]
+
+    response: Final = litellm.stream_chunk_builder(chunks=chunks, messages=[{"role": "user", "content": "hi"}])
+
+    assert response is not None
+    assert response.usage.prompt_tokens == 20
+    assert response.usage.completion_tokens == 60
+    assert getattr(response.usage, "cost", None) == pytest.approx(0.000704)
+    assert response._hidden_params["response_cost"] == pytest.approx(0.000704)