[opentelemetry-instrumentation-genai-anthropic] Record response telemetry for messages.stream() helpers

instrumentation/opentelemetry-instrumentation-genai-anthropic/.changelog/654.fixed

@@ -0,0 +1 @@
+Record response telemetry on ``messages.stream()`` when the stream is consumed through ``text_stream``, ``get_final_message()``, ``get_final_text()`` or ``until_done()``, which previously produced a span with request attributes only.

instrumentation/opentelemetry-instrumentation-genai-anthropic/src/opentelemetry/instrumentation/genai/anthropic/wrappers.py

@@ -100,6 +100,9 @@ class _MessagesStreamMixin(Generic[ResponseFormatT]):
     def _stop(self) -> None:
         if self._self_message_telemetry_finalized:
             return
+        # text_stream and the get_final_* helpers bypass _process_chunk, so the
+        # snapshot can be the only record of the response.
+        self._adopt_sdk_snapshot()
         _set_response_attributes(
             self._self_invocation,
             self._self_message,
@@ -120,19 +123,35 @@ def _on_stream_end(self) -> None:
     def _on_stream_error(self, error: BaseException) -> None:
         self._fail(error)
 
+    def _adopt_sdk_snapshot(self) -> bool:
+        """Adopt the SDK stream's accumulated message, when it keeps one.
+
+        ``MessageStream`` updates ``current_message_snapshot`` as the response
+        arrives, whichever accessor the caller reads it through. A plain
+        ``Stream`` has no snapshot and leaves accumulation to us.
+        """
+        stream = cast(_StreamWrapperWithStream, self).stream
+        try:
+            snapshot = cast(
+                "ParsedMessage[ResponseFormatT] | None",
+                getattr(stream, "current_message_snapshot", None),
+            )
+        except AssertionError:
+            # The property asserts the snapshot is set, so an unconsumed stream
+            # raises rather than answering None.
+            return False
+        if snapshot is None:
+            return False
+        self._self_message = snapshot
+        return True
+
     def _process_chunk(
         self,
         chunk: RawMessageStreamEvent
         | ParsedMessageStreamEvent[ResponseFormatT],
     ) -> None:
         """Accumulate a final message snapshot from a streaming chunk."""
-        stream = cast(_StreamWrapperWithStream, self).stream
-        snapshot = cast(
-            "ParsedMessage[ResponseFormatT] | None",
-            getattr(stream, "current_message_snapshot", None),
-        )
-        if snapshot is not None:
-            self._self_message = snapshot
+        if self._adopt_sdk_snapshot():
             return
         if accumulate_event is None:
             return

instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/test_async_messages.py

@@ -2087,3 +2087,172 @@ def __init__(self):
     assert first_stops == [True]
     assert second_stops == [True]
     assert http_response.close_calls == 1
+
+
+@pytest.mark.asyncio
+@pytest.mark.vcr()
+@pytest.mark.cassette("test_async_messages_stream")
+async def test_async_messages_stream_text_stream_records_response(
+    span_exporter, async_anthropic_client, instrument_with_content
+):
+    """``stream.text_stream`` records the response the same as event iteration."""
+    model = "claude-haiku-4-5"
+
+    text = ""
+    async with async_anthropic_client.messages.stream(
+        model=model,
+        max_tokens=100,
+        messages=[{"role": "user", "content": "Say hello in one word."}],
+    ) as stream:
+        async for chunk in stream.text_stream:
+            text += chunk
+
+    assert text == "Hello."
+
+    spans = span_exporter.get_finished_spans()
+    assert len(spans) == 1
+    span = spans[0]
+    assert span.attributes[GenAIAttributes.GEN_AI_REQUEST_MODEL] == model
+    assert (
+        span.attributes[GenAIAttributes.GEN_AI_RESPONSE_ID]
+        == "msg_01N2EGWxw2zHUcjTjToMivN6"
+    )
+    assert (
+        span.attributes[GenAIAttributes.GEN_AI_RESPONSE_MODEL]
+        == "claude-haiku-4-5-20251001"
+    )
+    assert span.attributes[GenAIAttributes.GEN_AI_USAGE_INPUT_TOKENS] == 13
+    assert span.attributes[GenAIAttributes.GEN_AI_USAGE_OUTPUT_TOKENS] == 5
+    assert span.attributes[GenAIAttributes.GEN_AI_RESPONSE_FINISH_REASONS] == (
+        "stop",
+    )
+    output_messages = _load_span_messages(
+        span, GenAIAttributes.GEN_AI_OUTPUT_MESSAGES
+    )
+    assert output_messages[0]["role"] == "assistant"
+    assert output_messages[0]["parts"] == [
+        {"type": "text", "content": "Hello."}
+    ]
+
+
+@pytest.mark.asyncio
+@pytest.mark.vcr()
+@pytest.mark.cassette("test_async_messages_stream")
+async def test_async_messages_stream_get_final_message_records_response(
+    span_exporter, async_anthropic_client, instrument_no_content
+):
+    """``get_final_message()`` drains the SDK's iterator and still records."""
+    model = "claude-haiku-4-5"
+
+    async with async_anthropic_client.messages.stream(
+        model=model,
+        max_tokens=100,
+        messages=[{"role": "user", "content": "Say hello in one word."}],
+    ) as stream:
+        message = await stream.get_final_message()
+
+    spans = span_exporter.get_finished_spans()
+    assert len(spans) == 1
+    span = spans[0]
+    assert span.attributes[GenAIAttributes.GEN_AI_REQUEST_MODEL] == model
+    assert span.attributes[GenAIAttributes.GEN_AI_RESPONSE_ID] == message.id
+    assert (
+        span.attributes[GenAIAttributes.GEN_AI_RESPONSE_MODEL] == message.model
+    )
+    assert span.attributes[
+        GenAIAttributes.GEN_AI_USAGE_INPUT_TOKENS
+    ] == expected_input_tokens(message.usage)
+    assert (
+        span.attributes[GenAIAttributes.GEN_AI_USAGE_OUTPUT_TOKENS]
+        == message.usage.output_tokens
+    )
+    assert span.attributes[GenAIAttributes.GEN_AI_RESPONSE_FINISH_REASONS] == (
+        normalize_stop_reason(message.stop_reason),
+    )
+
+
+@pytest.mark.asyncio
+@pytest.mark.vcr()
+@pytest.mark.cassette("test_async_messages_stream")
+async def test_async_messages_stream_get_final_text_records_response(
+    span_exporter, async_anthropic_client, instrument_no_content
+):
+    """``get_final_text()`` drains the SDK's iterator and still records."""
+    model = "claude-haiku-4-5"
+
+    async with async_anthropic_client.messages.stream(
+        model=model,
+        max_tokens=100,
+        messages=[{"role": "user", "content": "Say hello in one word."}],
+    ) as stream:
+        text = await stream.get_final_text()
+
+    assert text == "Hello."
+
+    spans = span_exporter.get_finished_spans()
+    assert len(spans) == 1
+    span = spans[0]
+    assert span.attributes[GenAIAttributes.GEN_AI_REQUEST_MODEL] == model
+    assert (
+        span.attributes[GenAIAttributes.GEN_AI_RESPONSE_MODEL]
+        == "claude-haiku-4-5-20251001"
+    )
+    assert span.attributes[GenAIAttributes.GEN_AI_USAGE_INPUT_TOKENS] == 13
+    assert span.attributes[GenAIAttributes.GEN_AI_USAGE_OUTPUT_TOKENS] == 5
+    assert span.attributes[GenAIAttributes.GEN_AI_RESPONSE_FINISH_REASONS] == (
+        "stop",
+    )
+
+
+@pytest.mark.asyncio
+@pytest.mark.vcr()
+@pytest.mark.cassette("test_async_messages_stream")
+async def test_async_messages_stream_until_done_records_response(
+    span_exporter, async_anthropic_client, instrument_no_content
+):
+    """``until_done()`` drains the SDK's iterator and still records."""
+    model = "claude-haiku-4-5"
+
+    async with async_anthropic_client.messages.stream(
+        model=model,
+        max_tokens=100,
+        messages=[{"role": "user", "content": "Say hello in one word."}],
+    ) as stream:
+        await stream.until_done()
+
+    spans = span_exporter.get_finished_spans()
+    assert len(spans) == 1
+    span = spans[0]
+    assert (
+        span.attributes[GenAIAttributes.GEN_AI_RESPONSE_MODEL]
+        == "claude-haiku-4-5-20251001"
+    )
+    assert span.attributes[GenAIAttributes.GEN_AI_USAGE_OUTPUT_TOKENS] == 5
+    assert span.attributes[GenAIAttributes.GEN_AI_RESPONSE_FINISH_REASONS] == (
+        "stop",
+    )
+
+
+@pytest.mark.asyncio
+@pytest.mark.vcr()
+@pytest.mark.cassette("test_async_messages_stream")
+async def test_async_messages_stream_text_stream_user_exception(
+    span_exporter, async_anthropic_client, instrument_no_content
+):
+    """A caller error while reading ``text_stream`` propagates and is recorded."""
+    model = "claude-haiku-4-5"
+
+    with pytest.raises(ValueError, match="caller failed"):
+        async with async_anthropic_client.messages.stream(
+            model=model,
+            max_tokens=100,
+            messages=[{"role": "user", "content": "Say hello in one word."}],
+        ) as stream:
+            async for _ in stream.text_stream:
+                raise ValueError("caller failed")
+
+    spans = span_exporter.get_finished_spans()
+    assert len(spans) == 1
+    span = spans[0]
+    assert span.attributes[GenAIAttributes.GEN_AI_REQUEST_MODEL] == model
+    assert span.attributes[ErrorAttributes.ERROR_TYPE] == "ValueError"

instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/test_sync_messages.py

@@ -2304,3 +2304,157 @@ def test_sync_messages_raw_response_parse_after_exit(
     assert spans[0].attributes[GenAIAttributes.GEN_AI_RESPONSE_MODEL] == model
 
     assert raw_response.parse().model == model
+
+
+@pytest.mark.vcr()
+@pytest.mark.cassette("test_sync_messages_stream")
+def test_sync_messages_stream_text_stream_records_response(
+    span_exporter, anthropic_client, instrument_with_content
+):
+    """``stream.text_stream`` records the response the same as event iteration."""
+    model = "claude-sonnet-4-20250514"
+
+    with anthropic_client.messages.stream(
+        model=model,
+        max_tokens=100,
+        messages=[{"role": "user", "content": "Say hello in one word."}],
+    ) as stream:
+        text = "".join(stream.text_stream)
+
+    assert text == "Hello!"
+
+    spans = span_exporter.get_finished_spans()
+    assert len(spans) == 1
+    span = spans[0]
+    assert_span_attributes(
+        span,
+        request_model=model,
+        response_id="msg_01FpWuSsvRgJp3eYbdHBinNp",
+        response_model=model,
+        input_tokens=13,
+        output_tokens=5,
+        finish_reasons=["stop"],
+    )
+    assert isinstance(span.attributes[GenAIAttributes.GEN_AI_RESPONSE_ID], str)
+    assert isinstance(
+        span.attributes[GenAIAttributes.GEN_AI_USAGE_INPUT_TOKENS], int
+    )
+    assert isinstance(
+        span.attributes[GenAIAttributes.GEN_AI_USAGE_OUTPUT_TOKENS], int
+    )
+    output_messages = _load_span_messages(
+        span, GenAIAttributes.GEN_AI_OUTPUT_MESSAGES
+    )
+    assert output_messages[0]["role"] == "assistant"
+    assert output_messages[0]["parts"] == [
+        {"type": "text", "content": "Hello!"}
+    ]
+
+
+@pytest.mark.vcr()
+@pytest.mark.cassette("test_sync_messages_stream")
+def test_sync_messages_stream_get_final_message_records_response(
+    span_exporter, anthropic_client, instrument_no_content
+):
+    """``get_final_message()`` drains the SDK's iterator and still records."""
+    model = "claude-sonnet-4-20250514"
+
+    with anthropic_client.messages.stream(
+        model=model,
+        max_tokens=100,
+        messages=[{"role": "user", "content": "Say hello in one word."}],
+    ) as stream:
+        message = stream.get_final_message()
+
+    spans = span_exporter.get_finished_spans()
+    assert len(spans) == 1
+    assert_span_attributes(
+        spans[0],
+        request_model=model,
+        response_id=message.id,
+        response_model=message.model,
+        input_tokens=expected_input_tokens(message.usage),
+        output_tokens=message.usage.output_tokens,
+        finish_reasons=[normalize_stop_reason(message.stop_reason)],
+    )
+
+
+@pytest.mark.vcr()
+@pytest.mark.cassette("test_sync_messages_stream")
+def test_sync_messages_stream_get_final_text_records_response(
+    span_exporter, anthropic_client, instrument_no_content
+):
+    """``get_final_text()`` drains the SDK's iterator and still records."""
+    model = "claude-sonnet-4-20250514"
+
+    with anthropic_client.messages.stream(
+        model=model,
+        max_tokens=100,
+        messages=[{"role": "user", "content": "Say hello in one word."}],
+    ) as stream:
+        text = stream.get_final_text()
+
+    assert text == "Hello!"
+
+    spans = span_exporter.get_finished_spans()
+    assert len(spans) == 1
+    assert_span_attributes(
+        spans[0],
+        request_model=model,
+        response_id="msg_01FpWuSsvRgJp3eYbdHBinNp",
+        response_model=model,
+        input_tokens=13,
+        output_tokens=5,
+        finish_reasons=["stop"],
+    )
+
+
+@pytest.mark.vcr()
+@pytest.mark.cassette("test_sync_messages_stream")
+def test_sync_messages_stream_until_done_records_response(
+    span_exporter, anthropic_client, instrument_no_content
+):
+    """``until_done()`` drains the SDK's iterator and still records."""
+    model = "claude-sonnet-4-20250514"
+
+    with anthropic_client.messages.stream(
+        model=model,
+        max_tokens=100,
+        messages=[{"role": "user", "content": "Say hello in one word."}],
+    ) as stream:
+        stream.until_done()
+
+    spans = span_exporter.get_finished_spans()
+    assert len(spans) == 1
+    assert_span_attributes(
+        spans[0],
+        request_model=model,
+        response_model=model,
+        input_tokens=13,
+        output_tokens=5,
+        finish_reasons=["stop"],
+    )
+
+
+@pytest.mark.vcr()
+@pytest.mark.cassette("test_sync_messages_stream")
+def test_sync_messages_stream_text_stream_user_exception(
+    span_exporter, anthropic_client, instrument_no_content
+):
+    """A caller error while reading ``text_stream`` propagates and is recorded."""
+    model = "claude-sonnet-4-20250514"
+
+    with pytest.raises(ValueError, match="caller failed"):
+        with anthropic_client.messages.stream(
+            model=model,
+            max_tokens=100,
+            messages=[{"role": "user", "content": "Say hello in one word."}],
+        ) as stream:
+            for _ in stream.text_stream:
+                raise ValueError("caller failed")
+
+    spans = span_exporter.get_finished_spans()
+    assert len(spans) == 1
+    span = spans[0]
+    assert span.attributes[GenAIAttributes.GEN_AI_REQUEST_MODEL] == model
+    assert span.attributes[ErrorAttributes.ERROR_TYPE] == "ValueError"