gen-ai instrumentation(feat): anthropic messages stream method instrumentation

instrumentation-genai/opentelemetry-instrumentation-anthropic/CHANGELOG.md

@@ -14,6 +14,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Add instrumentation for Anthropic `Messages.stream()` helper method
+  ([#4499](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4499))
 - Add async Anthropic message stream wrappers and manager wrappers, with wrapper
   tests ([#4346](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4346))
   - `AsyncMessagesStreamWrapper` for async message stream telemetry

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

@@ -53,14 +53,16 @@
     wrap_function_wrapper,  # pyright: ignore[reportUnknownVariableType]
 )
 
-from opentelemetry.instrumentation.anthropic.package import _instruments
-from opentelemetry.instrumentation.anthropic.patch import (
-    messages_create,
-)
 from opentelemetry.instrumentation.instrumentor import BaseInstrumentor
 from opentelemetry.instrumentation.utils import unwrap
 from opentelemetry.util.genai.handler import TelemetryHandler
 
+from .package import _instruments
+from .patch import (
+    messages_create,
+    messages_stream,
+)
+
 
 class AnthropicInstrumentor(BaseInstrumentor):
     """An instrumentor for the Anthropic Python SDK.
@@ -99,12 +101,17 @@ def _instrument(self, **kwargs: Any) -> None:
             logger_provider=logger_provider,
         )
 
-        # Patch Messages.create
+        # Patch Messages.create and Messages.stream
         wrap_function_wrapper(
             "anthropic.resources.messages",
             "Messages.create",
             messages_create(handler),
         )
+        wrap_function_wrapper(
+            "anthropic.resources.messages",
+            "Messages.stream",
+            messages_stream(handler),
+        )
 
     def _uninstrument(self, **kwargs: Any) -> None:
         """Disable Anthropic instrumentation.
@@ -117,3 +124,7 @@ def _uninstrument(self, **kwargs: Any) -> None:
             anthropic.resources.messages.Messages,  # pyright: ignore[reportAttributeAccessIssue,reportUnknownMemberType,reportUnknownArgumentType]
             "create",
         )
+        unwrap(
+            anthropic.resources.messages.Messages,  # pyright: ignore[reportAttributeAccessIssue,reportUnknownMemberType,reportUnknownArgumentType]
+            "stream",
+        )

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

@@ -27,9 +27,9 @@
 from opentelemetry.semconv._incubating.attributes import (
     server_attributes as ServerAttributes,
 )
+from opentelemetry.util.genai.invocation import InferenceInvocation
 from opentelemetry.util.genai.types import (
     InputMessage,
-    LLMInvocation,
     MessagePart,
     OutputMessage,
 )
@@ -155,7 +155,7 @@ def get_output_messages_from_message(
 
 
 def set_invocation_response_attributes(
-    invocation: LLMInvocation,
+    invocation: InferenceInvocation,
     message: Message | None,
     capture_content: bool,
 ) -> None:

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

@@ -26,13 +26,7 @@
     gen_ai_attributes as GenAIAttributes,
 )
 from opentelemetry.util.genai.handler import TelemetryHandler
-from opentelemetry.util.genai.types import (
-    Error,
-    LLMInvocation,  # TODO: migrate to InferenceInvocation
-)
-from opentelemetry.util.genai.utils import (
-    should_capture_content_on_spans_in_experimental_mode,
-)
+from opentelemetry.util.genai.invocation import InferenceInvocation
 
 from .messages_extractors import (
     extract_params,
@@ -41,11 +35,15 @@
     get_system_instruction,
 )
 from .wrappers import (
+    MessagesStreamManagerWrapper,
     MessagesStreamWrapper,
     MessageWrapper,
 )
 
 if TYPE_CHECKING:
+    from anthropic.lib.streaming._messages import (  # pylint: disable=no-name-in-module
+        MessageStreamManager,
+    )
     from anthropic.resources.messages import Messages
     from anthropic.types import RawMessageStreamEvent
 
@@ -65,7 +63,6 @@ def messages_create(
     ],
 ]:
     """Wrap the `create` method of the `Messages` class to trace it."""
-    capture_content = should_capture_content_on_spans_in_experimental_mode()
 
     def traced_method(
         wrapped: Callable[
@@ -83,49 +80,88 @@ def traced_method(
         "AnthropicStream[RawMessageStreamEvent]",
         MessagesStreamWrapper[None],
     ]:
-        params = extract_params(*args, **kwargs)
-        attributes = get_llm_request_attributes(params, instance)
-        request_model_attribute = attributes.get(
-            GenAIAttributes.GEN_AI_REQUEST_MODEL
+        invocation, capture_content = _create_invocation(
+            handler, instance, args, kwargs
         )
-        request_model = (
-            request_model_attribute
-            if isinstance(request_model_attribute, str)
-            else params.model
-        )
-
-        invocation = LLMInvocation(
-            request_model=request_model,
-            provider=ANTHROPIC,
-            input_messages=get_input_messages(params.messages)
-            if capture_content
-            else [],
-            system_instruction=get_system_instruction(params.system)
-            if capture_content
-            else [],
-            attributes=attributes,
-        )
-
-        # Use manual lifecycle management for both streaming and non-streaming
-        handler.start_llm(invocation)
         try:
             result = wrapped(*args, **kwargs)
             if isinstance(result, AnthropicStream):
                 return MessagesStreamWrapper(
-                    result, handler, invocation, capture_content
+                    result, invocation, capture_content
                 )
 
             wrapper = MessageWrapper(result, capture_content)
             wrapper.extract_into(invocation)
-            handler.stop_llm(invocation)
+            invocation.stop()
             return wrapper.message
         except Exception as exc:
-            handler.fail_llm(
-                invocation, Error(message=str(exc), type=type(exc))
-            )
+            invocation.fail(exc)
             raise
 
     return cast(
         'Callable[..., Union["AnthropicMessage", "AnthropicStream[RawMessageStreamEvent]", MessagesStreamWrapper[None]]]',
         traced_method,
     )
+
+
+def _create_invocation(
+    handler: TelemetryHandler,
+    instance: "Messages",
+    args: tuple[Any, ...],
+    kwargs: dict[str, Any],
+) -> tuple[InferenceInvocation, bool]:
+    should_capture_content = cast(
+        "Callable[[], bool]", getattr(handler, "should_capture_content")
+    )
+    capture_content = should_capture_content()
+    params = extract_params(*args, **kwargs)
+    attributes = get_llm_request_attributes(params, instance)
+    request_model_attribute = attributes.get(
+        GenAIAttributes.GEN_AI_REQUEST_MODEL
+    )
+    request_model = (
+        request_model_attribute
+        if isinstance(request_model_attribute, str)
+        else params.model
+    )
+
+    invocation = handler.start_inference(
+        provider=ANTHROPIC,
+        request_model=request_model,
+    )
+    invocation.input_messages = (
+        get_input_messages(params.messages) if capture_content else []
+    )
+    invocation.system_instruction = (
+        get_system_instruction(params.system) if capture_content else []
+    )
+    invocation.attributes = attributes
+    return invocation, capture_content
+
+
+def messages_stream(
+    handler: TelemetryHandler,
+) -> Callable[..., MessagesStreamManagerWrapper[Any]]:
+    """Wrap the sync `stream` method of the `Messages` class."""
+
+    def traced_method(
+        wrapped: Callable[..., "MessageStreamManager"],
+        instance: "Messages",
+        args: tuple[Any, ...],
+        kwargs: dict[str, Any],
+    ) -> MessagesStreamManagerWrapper[Any]:
+        invocation, capture_content = _create_invocation(
+            handler, instance, args, kwargs
+        )
+
+        try:
+            return MessagesStreamManagerWrapper(
+                wrapped(*args, **kwargs), invocation, capture_content
+            )
+        except Exception as exc:
+            invocation.fail(exc)
+            raise
+
+    return cast(
+        "Callable[..., MessagesStreamManagerWrapper[Any]]", traced_method
+    )

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

@@ -19,7 +19,7 @@
 import base64
 import json
 from dataclasses import dataclass
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, cast
 
 from anthropic.types import (
     InputJSONDelta,
@@ -160,12 +160,9 @@ def _convert_content_block_to_part(
             id=block.tool_use_id,
         )
 
-    # ContentBlockParam variants are TypedDicts (dicts at runtime);
-    # newer SDK versions may add Pydantic block types not handled above.
-    if isinstance(block, dict):
-        return _convert_dict_block_to_part(block)
-
-    return None
+    if not hasattr(block, "get"):
+        return None
+    return _convert_dict_block_to_part(cast("Mapping[str, Any]", block))
 
 
 def convert_content_to_parts(