From bdc341cdee9f2237d8f9982869a5b8dc2989c10f Mon Sep 17 00:00:00 2001 From: Zening Chen Date: Tue, 8 Sep 2026 07:25:44 +0000 Subject: [PATCH 1/2] [`opentelemetry-instrumentation-genai-anthropic`] Record response telemetry for messages.stream() helpers A chat span from messages.stream() carried request attributes only whenever the caller read the stream through text_stream, get_final_message(), get_final_text() or until_done(). MessageStream builds its own iterator in its constructor and each of those helpers drives that one, so no chunk reached the instrumented wrapper and nothing was recorded. Read the SDK's current_message_snapshot at finalization instead of relying on having seen every chunk. The SDK keeps it updated whichever accessor the caller uses, so one read covers every helper. Assisted-by: Claude Opus 5 --- .../.changelog/654.fixed | 1 + .../genai/anthropic/wrappers.py | 33 ++++- .../tests/test_async_messages.py | 136 ++++++++++++++++++ .../tests/test_sync_messages.py | 124 ++++++++++++++++ 4 files changed, 287 insertions(+), 7 deletions(-) create mode 100644 instrumentation/opentelemetry-instrumentation-genai-anthropic/.changelog/654.fixed diff --git a/instrumentation/opentelemetry-instrumentation-genai-anthropic/.changelog/654.fixed b/instrumentation/opentelemetry-instrumentation-genai-anthropic/.changelog/654.fixed new file mode 100644 index 000000000..3805f1637 --- /dev/null +++ b/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. diff --git a/instrumentation/opentelemetry-instrumentation-genai-anthropic/src/opentelemetry/instrumentation/genai/anthropic/wrappers.py b/instrumentation/opentelemetry-instrumentation-genai-anthropic/src/opentelemetry/instrumentation/genai/anthropic/wrappers.py index 5f9dbd0f8..457759a18 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-anthropic/src/opentelemetry/instrumentation/genai/anthropic/wrappers.py +++ b/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 diff --git a/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/test_async_messages.py b/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/test_async_messages.py index cd1637f84..f4b261931 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/test_async_messages.py +++ b/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/test_async_messages.py @@ -2071,3 +2071,139 @@ 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_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" diff --git a/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/test_sync_messages.py b/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/test_sync_messages.py index 821a236b4..60269ea3a 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/test_sync_messages.py +++ b/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/test_sync_messages.py @@ -2286,3 +2286,127 @@ 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_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" From 3d2d9efcb64c3265af21de64be4c245bbd7e819f Mon Sep 17 00:00:00 2001 From: Zening Chen Date: Wed, 9 Sep 2026 23:58:21 +0000 Subject: [PATCH 2/2] Cover get_final_text() on both clients The changelog listed the helper alongside text_stream, get_final_message() and until_done(), but only those three had tests. get_final_text() reaches get_final_message() internally today, so it was covered only by accident. Both new tests fail without the snapshot read. Assisted-by: Claude Opus 5 --- .../tests/test_async_messages.py | 33 +++++++++++++++++++ .../tests/test_sync_messages.py | 30 +++++++++++++++++ 2 files changed, 63 insertions(+) diff --git a/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/test_async_messages.py b/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/test_async_messages.py index f4b261931..ccdeb94e7 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/test_async_messages.py +++ b/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/test_async_messages.py @@ -2155,6 +2155,39 @@ async def test_async_messages_stream_get_final_message_records_response( ) +@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") diff --git a/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/test_sync_messages.py b/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/test_sync_messages.py index 60269ea3a..d8ecab629 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/test_sync_messages.py +++ b/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/test_sync_messages.py @@ -2361,6 +2361,36 @@ def test_sync_messages_stream_get_final_message_records_response( ) +@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(