Add sync streaming support for Anthropic instrumentation

- Add support for Messages.create(stream=True) with StreamWrapper
- Add support for Messages.stream() with MessageStreamManagerWrapper
- Add MessageWrapper for non-streaming response telemetry
- Rename MessageCreateParams to MessageRequestParams
- Add comprehensive tests for sync streaming functionality

Author: vasantteja

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

@@ -54,7 +54,10 @@
 )
 
 from opentelemetry.instrumentation.anthropic.package import _instruments
-from opentelemetry.instrumentation.anthropic.patch import messages_create
+from opentelemetry.instrumentation.anthropic.patch import (
+    messages_create,
+    messages_stream,
+)
 from opentelemetry.instrumentation.instrumentor import BaseInstrumentor
 from opentelemetry.instrumentation.utils import unwrap
 from opentelemetry.util.genai.handler import TelemetryHandler
@@ -103,6 +106,13 @@ def _instrument(self, **kwargs: Any) -> None:
             wrapper=messages_create(handler),
         )
 
+        # Patch Messages.stream
+        wrap_function_wrapper(
+            module="anthropic.resources.messages",
+            name="Messages.stream",
+            wrapper=messages_stream(handler),
+        )
+
     def _uninstrument(self, **kwargs: Any) -> None:
         """Disable Anthropic instrumentation.
 
@@ -114,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/patch.py

@@ -14,35 +14,40 @@
 
 """Patching functions for Anthropic instrumentation."""
 
-from typing import TYPE_CHECKING, Any, Callable
+from typing import TYPE_CHECKING, Any, Callable, Union
 
 from opentelemetry.semconv._incubating.attributes import (
     gen_ai_attributes as GenAIAttributes,
 )
 from opentelemetry.util.genai.handler import TelemetryHandler
-from opentelemetry.util.genai.types import LLMInvocation
+from opentelemetry.util.genai.types import Error, LLMInvocation
 
 from .utils import (
+    MessageStreamManagerWrapper,
+    MessageWrapper,
+    StreamWrapper,
     extract_params,
     get_llm_request_attributes,
 )
 
 if TYPE_CHECKING:
+    from anthropic._streaming import Stream
+    from anthropic.lib.streaming import MessageStreamManager
     from anthropic.resources.messages import Messages
-    from anthropic.types import Message
+    from anthropic.types import Message, RawMessageStreamEvent
 
 
 def messages_create(
     handler: TelemetryHandler,
-) -> Callable[..., "Message"]:
+) -> Callable[..., Union["Message", "Stream[RawMessageStreamEvent]"]]:
     """Wrap the `create` method of the `Messages` class to trace it."""
 
     def traced_method(
-        wrapped: Callable[..., "Message"],
+        wrapped: Callable[..., Union["Message", "Stream[RawMessageStreamEvent]"]],
         instance: "Messages",
         args: tuple[Any, ...],
         kwargs: dict[str, Any],
-    ) -> "Message":
+    ) -> Union["Message", StreamWrapper]:
         params = extract_params(*args, **kwargs)
         attributes = get_llm_request_attributes(params, instance)
         request_model = str(
@@ -57,22 +62,60 @@ def traced_method(
             attributes=attributes,
         )
 
-        with handler.llm(invocation) as invocation:
+        is_streaming = kwargs.get("stream", False)
+
+        # Use manual lifecycle management for both streaming and non-streaming
+        handler.start_llm(invocation)
+        try:
             result = wrapped(*args, **kwargs)
+            if is_streaming:
+                return StreamWrapper(result, handler, invocation)
+            wrapper = MessageWrapper(result, handler, invocation)
+            return wrapper.message
+        except Exception as exc:
+            handler.fail_llm(
+                invocation, Error(message=str(exc), type=type(exc))
+            )
+            raise
 
-            if result.model:
-                invocation.response_model_name = result.model
+    return traced_method
 
-            if result.id:
-                invocation.response_id = result.id
 
-            if result.stop_reason:
-                invocation.finish_reasons = [result.stop_reason]
+def messages_stream(
+    handler: TelemetryHandler,
+) -> Callable[..., "MessageStreamManager"]:
+    """Wrap the `stream` method of the `Messages` class to trace it."""
 
-            if result.usage:
-                invocation.input_tokens = result.usage.input_tokens
-                invocation.output_tokens = result.usage.output_tokens
+    def traced_method(
+        wrapped: Callable[..., "MessageStreamManager"],
+        instance: "Messages",
+        args: tuple[Any, ...],
+        kwargs: dict[str, Any],
+    ) -> MessageStreamManagerWrapper:
+        params = extract_params(*args, **kwargs)
+        attributes = get_llm_request_attributes(params, instance)
+        request_model = str(
+            attributes.get(GenAIAttributes.GEN_AI_REQUEST_MODEL)
+            or params.model
+            or "unknown"
+        )
+
+        invocation = LLMInvocation(
+            request_model=request_model,
+            provider="anthropic",
+            attributes=attributes,
+        )
 
-            return result
+        # Start the span before calling the wrapped method
+        handler.start_llm(invocation)
+        try:
+            result = wrapped(*args, **kwargs)
+            # Return wrapped MessageStreamManager
+            return MessageStreamManagerWrapper(result, handler, invocation)
+        except Exception as exc:
+            handler.fail_llm(
+                invocation, Error(message=str(exc), type=type(exc))
+            )
+            raise
 
     return traced_method