fix(anthropic): instrument beta stream managers (fixes #4388)

packages/opentelemetry-instrumentation-anthropic/opentelemetry/instrumentation/anthropic/__init__.py

@@ -170,22 +170,51 @@ def is_streaming_response(response):
     return isinstance(response, Stream) or isinstance(response, AsyncStream)
 
 
+ASYNC_STREAM_MANAGER_CLASS_NAMES = (
+    "AsyncMessageStreamManager",
+    "BetaAsyncMessageStreamManager",
+)
+
+
+def is_async_stream_manager(response):
+    """Check if response is an (async) stream manager, including the beta variant"""
+    return response.__class__.__name__ in ASYNC_STREAM_MANAGER_CLASS_NAMES
+
+
 def is_stream_manager(response):
-    """Check if response is a MessageStreamManager or AsyncMessageStreamManager"""
+    """Check if response is a (Beta)MessageStreamManager or (Beta)AsyncMessageStreamManager"""
+    stream_manager_types = ()
     try:
         from anthropic.lib.streaming._messages import (
             MessageStreamManager,
             AsyncMessageStreamManager,
         )
 
-        return isinstance(response, (MessageStreamManager, AsyncMessageStreamManager))
+        stream_manager_types += (MessageStreamManager, AsyncMessageStreamManager)
     except ImportError:
-        # Check by class name as fallback
-        return (
-            response.__class__.__name__ == "MessageStreamManager"
-            or response.__class__.__name__ == "AsyncMessageStreamManager"
+        pass
+
+    try:
+        from anthropic.lib.streaming._beta_messages import (
+            BetaMessageStreamManager,
+            BetaAsyncMessageStreamManager,
         )
 
+        stream_manager_types += (BetaMessageStreamManager, BetaAsyncMessageStreamManager)
+    except ImportError:
+        pass
+
+    if stream_manager_types:
+        return isinstance(response, stream_manager_types)
+
+    # Check by class name as fallback
+    return response.__class__.__name__ in (
+        "MessageStreamManager",
+        "AsyncMessageStreamManager",
+        "BetaMessageStreamManager",
+        "BetaAsyncMessageStreamManager",
+    )
+
 
 @dont_throw
 async def _aset_token_usage(
@@ -599,7 +628,7 @@ def _wrap(
             kwargs,
         )
     elif is_stream_manager(response):
-        if response.__class__.__name__ == "AsyncMessageStreamManager":
+        if is_async_stream_manager(response):
             return WrappedAsyncMessageStreamManager(
                 response,
                 span,
@@ -729,7 +758,7 @@ async def _awrap(
             kwargs,
         )
     elif is_stream_manager(response):
-        if response.__class__.__name__ == "AsyncMessageStreamManager":
+        if is_async_stream_manager(response):
             return WrappedAsyncMessageStreamManager(
                 response,
                 span,

packages/opentelemetry-instrumentation-anthropic/tests/test_stream_manager_detection.py

@@ -0,0 +1,42 @@
+from anthropic.lib.streaming._beta_messages import (
+    BetaAsyncMessageStreamManager,
+    BetaMessageStreamManager,
+)
+from anthropic.lib.streaming._messages import (
+    AsyncMessageStreamManager,
+    MessageStreamManager,
+)
+
+from opentelemetry.instrumentation.anthropic import (
+    is_async_stream_manager,
+    is_stream_manager,
+)
+
+
+def _make(cls):
+    # The manager classes take a client-bound request callable; we only need an
+    # instance for type/name detection, so bypass __init__.
+    return cls.__new__(cls)
+
+
+def test_is_stream_manager_recognizes_all_managers():
+    for cls in (
+        MessageStreamManager,
+        AsyncMessageStreamManager,
+        BetaMessageStreamManager,
+        BetaAsyncMessageStreamManager,
+    ):
+        assert is_stream_manager(_make(cls)), cls.__name__
+
+
+def test_is_stream_manager_rejects_plain_objects():
+    assert not is_stream_manager(object())
+
+
+def test_is_async_stream_manager_matches_async_variants_only():
+    assert is_async_stream_manager(_make(AsyncMessageStreamManager))
+    # Regression for #4388: beta async streams must route to the async wrapper.
+    assert is_async_stream_manager(_make(BetaAsyncMessageStreamManager))
+
+    assert not is_async_stream_manager(_make(MessageStreamManager))
+    assert not is_async_stream_manager(_make(BetaMessageStreamManager))