[opentelemetry-instrumentation-genai-anthropic] Instrument beta.messages create, stream, and parse (#550)

instrumentation/opentelemetry-instrumentation-genai-anthropic/.changelog/661.added

@@ -0,0 +1 @@
+Add instrumentation for `beta.messages` (`create`, `stream`, and `parse`).

instrumentation/opentelemetry-instrumentation-genai-anthropic/README.rst

@@ -46,6 +46,16 @@ Check out the `manual example <examples/manual>`_ for more details.
     )
 
 
+Supported Operations
+--------------------
+
+The instrumentation supports synchronous and asynchronous calls for:
+
+- ``messages.create``, ``messages.stream``, and ``messages.parse``
+- ``beta.messages.create``, ``beta.messages.stream``, and ``beta.messages.parse``
+- Raw and streaming response helpers (``with_raw_response`` and ``with_streaming_response``)
+
+
 Configuration
 -------------
 

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

@@ -77,6 +77,37 @@ def _is_parse_supported() -> bool:
         return False
 
 
+def _is_beta_messages_supported() -> bool:
+    """Check if beta Messages classes are available on the anthropic SDK."""
+    try:
+        from anthropic.resources.beta.messages import (  # pylint: disable=import-outside-toplevel
+            AsyncMessages,
+            Messages,
+        )
+
+        return (
+            hasattr(Messages, "create")
+            and hasattr(AsyncMessages, "create")
+            and hasattr(Messages, "stream")
+            and hasattr(AsyncMessages, "stream")
+        )
+    except (ImportError, AttributeError):
+        return False
+
+
+def _is_beta_parse_supported() -> bool:
+    """Check if parse() is available on the beta Messages classes."""
+    try:
+        from anthropic.resources.beta.messages import (  # pylint: disable=import-outside-toplevel
+            AsyncMessages,
+            Messages,
+        )
+
+        return hasattr(Messages, "parse") and hasattr(AsyncMessages, "parse")
+    except (ImportError, AttributeError):
+        return False
+
+
 class AnthropicInstrumentor(BaseInstrumentor):
     """An instrumentor for the Anthropic Python SDK.
 
@@ -90,6 +121,8 @@ def __init__(self) -> None:
         self._logger = None
         self._meter = None
         self._parse_supported = _is_parse_supported()
+        self._beta_messages_supported = _is_beta_messages_supported()
+        self._beta_parse_supported = _is_beta_parse_supported()
 
     # pylint: disable=no-self-use
     def instrumentation_dependencies(self) -> Collection[str]:
@@ -109,6 +142,10 @@ def _instrument(self, **kwargs: Any) -> None:
         meter_provider = kwargs.get("meter_provider")
         logger_provider = kwargs.get("logger_provider")
 
+        self._parse_supported = _is_parse_supported()
+        self._beta_messages_supported = _is_beta_messages_supported()
+        self._beta_parse_supported = _is_beta_parse_supported()
+
         handler = TelemetryHandler(
             tracer_provider=tracer_provider,
             meter_provider=meter_provider,
@@ -150,6 +187,39 @@ def _instrument(self, **kwargs: Any) -> None:
             async_response_context_manager_exit,
         )
 
+        if self._beta_messages_supported:
+            wrap_function_wrapper(
+                "anthropic.resources.beta.messages",
+                "Messages.create",
+                messages_create(handler),
+            )
+            wrap_function_wrapper(
+                "anthropic.resources.beta.messages",
+                "AsyncMessages.create",
+                async_messages_create(handler),
+            )
+            wrap_function_wrapper(
+                "anthropic.resources.beta.messages",
+                "Messages.stream",
+                messages_stream(handler),
+            )
+            wrap_function_wrapper(
+                "anthropic.resources.beta.messages",
+                "AsyncMessages.stream",
+                async_messages_stream(handler),
+            )
+            if self._beta_parse_supported:
+                wrap_function_wrapper(
+                    "anthropic.resources.beta.messages",
+                    "Messages.parse",
+                    messages_create(handler),
+                )
+                wrap_function_wrapper(
+                    "anthropic.resources.beta.messages",
+                    "AsyncMessages.parse",
+                    async_messages_create(handler),
+                )
+
         # parse() wraps create() internally in the Anthropic SDK and returns a
         # parsed message whose telemetry-relevant fields match Message, so the
         # existing create() wrappers handle it correctly. It was added in a
@@ -171,37 +241,36 @@ def _uninstrument(self, **kwargs: Any) -> None:
 
         This removes all patches applied during instrumentation.
         """
-        import anthropic  # pylint: disable=import-outside-toplevel
-
-        unwrap(
-            anthropic.resources.messages.Messages,  # pyright: ignore[reportAttributeAccessIssue,reportUnknownMemberType,reportUnknownArgumentType]
-            "create",
-        )
-        unwrap(
-            anthropic.resources.messages.AsyncMessages,  # pyright: ignore[reportAttributeAccessIssue,reportUnknownMemberType,reportUnknownArgumentType]
-            "create",
-        )
-        unwrap(
-            anthropic.resources.messages.Messages,  # pyright: ignore[reportAttributeAccessIssue,reportUnknownMemberType,reportUnknownArgumentType]
-            "stream",
-        )
-        unwrap(
-            anthropic.resources.messages.AsyncMessages,  # pyright: ignore[reportAttributeAccessIssue,reportUnknownMemberType,reportUnknownArgumentType]
-            "stream",
-        )
         from anthropic._response import (  # pylint: disable=import-outside-toplevel
             AsyncResponseContextManager,
             ResponseContextManager,
         )
+        from anthropic.resources.messages import (  # pylint: disable=import-outside-toplevel
+            AsyncMessages,
+            Messages,
+        )
 
+        unwrap(Messages, "create")
+        unwrap(AsyncMessages, "create")
+        unwrap(Messages, "stream")
+        unwrap(AsyncMessages, "stream")
         unwrap(ResponseContextManager, "__exit__")
         unwrap(AsyncResponseContextManager, "__aexit__")
-        if self._parse_supported:
-            unwrap(
-                anthropic.resources.messages.Messages,  # pyright: ignore[reportAttributeAccessIssue,reportUnknownMemberType,reportUnknownArgumentType]
-                "parse",
+        if self._beta_messages_supported:
+            from anthropic.resources.beta.messages import (  # pylint: disable=import-outside-toplevel
+                AsyncMessages as AsyncBetaMessages,
             )
-            unwrap(
-                anthropic.resources.messages.AsyncMessages,  # pyright: ignore[reportAttributeAccessIssue,reportUnknownMemberType,reportUnknownArgumentType]
-                "parse",
+            from anthropic.resources.beta.messages import (
+                Messages as BetaMessages,
             )
+
+            unwrap(BetaMessages, "create")
+            unwrap(AsyncBetaMessages, "create")
+            unwrap(BetaMessages, "stream")
+            unwrap(AsyncBetaMessages, "stream")
+            if self._beta_parse_supported:
+                unwrap(BetaMessages, "parse")
+                unwrap(AsyncBetaMessages, "parse")
+        if self._parse_supported:
+            unwrap(Messages, "parse")
+            unwrap(AsyncMessages, "parse")

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

@@ -18,7 +18,11 @@
 from anthropic._models import construct_type
 from anthropic.types import Message as AnthropicMessage
 
-from .utils import is_anthropic_async_stream, is_anthropic_stream
+from .utils import (
+    AnthropicBetaMessage,
+    is_anthropic_async_stream,
+    is_anthropic_stream,
+)
 from .wrappers import (
     AsyncMessagesStreamWrapper,
     MessagesStreamWrapper,
@@ -86,10 +90,12 @@ def __init__(
         raw_response: Any,
         invocation: InferenceInvocation,
         capture_content: bool,
+        is_beta: bool = False,
     ) -> None:
         super().__init__(raw_response)
         self._self_invocation = invocation
         self._self_capture_content = capture_content
+        self._self_is_beta = is_beta
         # The span is ours to end until a stream wrapper takes it over.
         self._self_span_open = True
         # Response telemetry is settled: whichever path saw the body first
@@ -174,7 +180,8 @@ def _finalize_read_fallback(self) -> None:
         if self._self_parsing or self._self_dispatched:
             return
         message = _message_from_read_body(
-            getattr(self.__wrapped__, "http_response", None)
+            getattr(self.__wrapped__, "http_response", None),
+            is_beta=self._self_is_beta,
         )
         if message is not None:
             MessageWrapper(message, self._self_capture_content).extract_into(
@@ -370,7 +377,7 @@ def _dispatch(self, parsed: Any) -> object:
             # A read fallback already settled and ended this span; the caller
             # still gets the SDK's object, just without a second recording.
             return parsed
-        if isinstance(parsed, AnthropicMessage):
+        if isinstance(parsed, (AnthropicMessage, AnthropicBetaMessage)):
             MessageWrapper(parsed, self._self_capture_content).extract_into(
                 self._self_invocation
             )
@@ -379,7 +386,10 @@ def _dispatch(self, parsed: Any) -> object:
             return parsed
         try:
             wrapped = _wrap_parsed_stream(
-                parsed, self._self_invocation, self._self_capture_content
+                parsed,
+                self._self_invocation,
+                self._self_capture_content,
+                is_beta=self._self_is_beta,
             )
         except Exception:  # pylint: disable=broad-exception-caught
             # Same rule as message extraction: a wrapper we failed to build
@@ -413,6 +423,7 @@ def _wrap_parsed_stream(
     stream: Any,
     invocation: InferenceInvocation,
     capture_content: bool,
+    is_beta: bool = False,
 ) -> object | None:
     """Wrap a parsed stream in the matching instrumented wrapper.
 
@@ -425,12 +436,14 @@ def _wrap_parsed_stream(
             cast("AnthropicAsyncStream[RawMessageStreamEvent]", stream),
             invocation,
             capture_content,
+            is_beta=is_beta,
         )
     if is_anthropic_stream(stream):
         return MessagesStreamWrapper[None](
             cast("AnthropicStream[RawMessageStreamEvent]", stream),
             invocation,
             capture_content,
+            is_beta=is_beta,
         )
     return None
 
@@ -446,7 +459,10 @@ def _body_was_read(http_response: Any) -> bool:
     return True
 
 
-def _message_from_read_body(http_response: Any) -> AnthropicMessage | None:
+def _message_from_read_body(
+    http_response: Any,
+    is_beta: bool = False,
+) -> AnthropicMessage | AnthropicBetaMessage | None:
     """Deserialize an already-read response body into a ``Message``.
 
     Used instead of ``result.parse()`` so telemetry never runs the caller's
@@ -471,9 +487,20 @@ def _message_from_read_body(http_response: Any) -> AnthropicMessage | None:
         if isinstance(body, dict):
             fields = cast("dict[str, object]", body)
             if fields.get("type") == "message":
+                target_type = (
+                    AnthropicBetaMessage
+                    if (
+                        is_beta
+                        and (
+                            hasattr(AnthropicBetaMessage, "model_fields")
+                            or hasattr(AnthropicBetaMessage, "__fields__")
+                        )
+                    )
+                    else AnthropicMessage
+                )
                 return cast(
-                    AnthropicMessage,
-                    construct_type(type_=AnthropicMessage, value=fields),
+                    "AnthropicMessage | AnthropicBetaMessage",
+                    construct_type(type_=target_type, value=fields),
                 )
     except Exception:  # pylint: disable=broad-exception-caught
         _logger.debug(
@@ -493,6 +520,7 @@ def wrap_raw_response(
     result: Any,
     invocation: InferenceInvocation,
     capture_content: bool,
+    is_beta: bool = False,
 ) -> Any:
     """Wrap a ``with_raw_response`` / ``with_streaming_response`` result.
 
@@ -506,10 +534,12 @@ def wrap_raw_response(
     """
     http_response = getattr(result, "http_response", None)
     if getattr(http_response, "is_closed", False):
-        message = _message_from_read_body(http_response)
+        message = _message_from_read_body(http_response, is_beta=is_beta)
         if message is not None:
             MessageWrapper(message, capture_content).extract_into(invocation)
         invocation.stop()
         return result
 
-    return RawResponseProxy(result, invocation, capture_content)
+    return RawResponseProxy(
+        result, invocation, capture_content, is_beta=is_beta
+    )