feat: Add sync streaming support for Anthropic instrumentation

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

@@ -9,6 +9,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Add sync streaming support for `Messages.create(stream=True)` and `Messages.stream()`
+  ([#4155](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4155))
+  - `StreamWrapper` for handling `Messages.create(stream=True)` telemetry
+  - `MessageStreamManagerWrapper` for handling `Messages.stream()` telemetry
+  - `MessageWrapper` for non-streaming response telemetry extraction
 - Initial implementation of Anthropic instrumentation
   ([#3978](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3978))
 - Implement sync `Messages.create` instrumentation with GenAI semantic convention attributes

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

@@ -54,7 +54,9 @@
 )
 
 from opentelemetry.instrumentation.anthropic.package import _instruments
-from opentelemetry.instrumentation.anthropic.patch import messages_create
+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
@@ -89,11 +91,12 @@ def _instrument(self, **kwargs: Any) -> None:
         # Get providers from kwargs
         tracer_provider = kwargs.get("tracer_provider")
         meter_provider = kwargs.get("meter_provider")
+        logger_provider = kwargs.get("logger_provider")
 
-        # TODO: Add logger_provider to TelemetryHandler to capture content events.
         handler = TelemetryHandler(
             tracer_provider=tracer_provider,
             meter_provider=meter_provider,
+            logger_provider=logger_provider,
         )
 
         # Patch Messages.create

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

@@ -0,0 +1,202 @@
+# Copyright The OpenTelemetry Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Get/extract helpers for Anthropic Messages instrumentation."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Any, Optional, Sequence, cast
+from urllib.parse import urlparse
+
+from opentelemetry.semconv._incubating.attributes import (
+    gen_ai_attributes as GenAIAttributes,
+)
+from opentelemetry.semconv._incubating.attributes import (
+    server_attributes as ServerAttributes,
+)
+from opentelemetry.util.genai.types import (
+    InputMessage,
+    MessagePart,
+    OutputMessage,
+)
+from opentelemetry.util.types import AttributeValue
+
+from .utils import (
+    _as_str,
+    _get_field,
+    as_int,
+    convert_content_to_parts,
+    normalize_finish_reason,
+)
+
+if TYPE_CHECKING:
+    from anthropic.resources.messages import Messages
+
+
+@dataclass
+class MessageRequestParams:
+    model: str | None = None
+    max_tokens: int | None = None
+    temperature: float | None = None
+    top_k: int | None = None
+    top_p: float | None = None
+    stop_sequences: Sequence[str] | None = None
+    messages: Any | None = None
+    system: Any | None = None
+
+
+GEN_AI_USAGE_CACHE_CREATION_INPUT_TOKENS = (
+    "gen_ai.usage.cache_creation.input_tokens"
+)
+GEN_AI_USAGE_CACHE_READ_INPUT_TOKENS = "gen_ai.usage.cache_read.input_tokens"
+
+
+def extract_usage_tokens(
+    usage: Any | None,
+) -> tuple[int | None, int | None, int | None, int | None]:
+    if usage is None:
+        return None, None, None, None
+
+    input_tokens = as_int(getattr(usage, "input_tokens", None))
+    cache_creation_input_tokens = as_int(
+        getattr(usage, "cache_creation_input_tokens", None)
+    )
+    cache_read_input_tokens = as_int(
+        getattr(usage, "cache_read_input_tokens", None)
+    )
+    output_tokens = as_int(getattr(usage, "output_tokens", None))
+
+    if (
+        input_tokens is None
+        and cache_creation_input_tokens is None
+        and cache_read_input_tokens is None
+    ):
+        total_input_tokens = None
+    else:
+        total_input_tokens = (
+            (input_tokens or 0)
+            + (cache_creation_input_tokens or 0)
+            + (cache_read_input_tokens or 0)
+        )
+
+    return (
+        total_input_tokens,
+        output_tokens,
+        cache_creation_input_tokens,
+        cache_read_input_tokens,
+    )
+
+
+def get_input_messages(messages: Any) -> list[InputMessage]:
+    if not isinstance(messages, list):
+        return []
+
+    result: list[InputMessage] = []
+    for message in cast(list[Any], messages):
+        role = _as_str(_get_field(message, "role")) or "user"
+        parts = convert_content_to_parts(_get_field(message, "content"))
+        result.append(InputMessage(role=role, parts=parts))
+    return result
+
+
+def get_system_instruction(system: Any) -> list[MessagePart]:
+    return convert_content_to_parts(system)
+
+
+def get_output_messages_from_message(message: Any) -> list[OutputMessage]:
+    if message is None:
+        return []
+
+    parts = convert_content_to_parts(_get_field(message, "content"))
+    finish_reason = normalize_finish_reason(_get_field(message, "stop_reason"))
+    return [
+        OutputMessage(
+            role=_as_str(_get_field(message, "role")) or "assistant",
+            parts=parts,
+            finish_reason=finish_reason or "",
+        )
+    ]
+
+
+def extract_params(  # pylint: disable=too-many-locals
+    *,
+    max_tokens: int | None = None,
+    messages: Any | None = None,
+    model: str | None = None,
+    metadata: Any | None = None,
+    service_tier: Any | None = None,
+    stop_sequences: Sequence[str] | None = None,
+    stream: Any | None = None,
+    system: Any | None = None,
+    temperature: float | None = None,
+    thinking: Any | None = None,
+    tool_choice: Any | None = None,
+    tools: Any | None = None,
+    top_k: int | None = None,
+    top_p: float | None = None,
+    extra_headers: Any | None = None,
+    extra_query: Any | None = None,
+    extra_body: Any | None = None,
+    timeout: Any | None = None,
+    **_kwargs: Any,
+) -> MessageRequestParams:
+    return MessageRequestParams(
+        model=model,
+        max_tokens=max_tokens,
+        temperature=temperature,
+        top_p=top_p,
+        top_k=top_k,
+        stop_sequences=stop_sequences,
+        messages=messages,
+        system=system,
+    )
+
+
+def _set_server_address_and_port(
+    client_instance: "Messages", attributes: dict[str, Any]
+) -> None:
+    base_client = getattr(client_instance, "_client", None)
+    base_url = getattr(base_client, "base_url", None)
+    if not base_url:
+        return
+
+    port: Optional[int] = None
+    if hasattr(base_url, "host"):
+        attributes[ServerAttributes.SERVER_ADDRESS] = base_url.host
+        port = getattr(base_url, "port", None)
+    elif isinstance(base_url, str):
+        url = urlparse(base_url)
+        attributes[ServerAttributes.SERVER_ADDRESS] = url.hostname
+        port = url.port
+
+    if port and port != 443 and port > 0:
+        attributes[ServerAttributes.SERVER_PORT] = port
+
+
+def get_llm_request_attributes(
+    params: MessageRequestParams, client_instance: "Messages"
+) -> dict[str, AttributeValue]:
+    attributes = {
+        GenAIAttributes.GEN_AI_OPERATION_NAME: GenAIAttributes.GenAiOperationNameValues.CHAT.value,
+        GenAIAttributes.GEN_AI_SYSTEM: GenAIAttributes.GenAiSystemValues.ANTHROPIC.value,  # pyright: ignore[reportDeprecated]
+        GenAIAttributes.GEN_AI_REQUEST_MODEL: params.model,
+        GenAIAttributes.GEN_AI_REQUEST_MAX_TOKENS: params.max_tokens,
+        GenAIAttributes.GEN_AI_REQUEST_TEMPERATURE: params.temperature,
+        GenAIAttributes.GEN_AI_REQUEST_TOP_P: params.top_p,
+        GenAIAttributes.GEN_AI_REQUEST_TOP_K: params.top_k,
+        GenAIAttributes.GEN_AI_REQUEST_STOP_SEQUENCES: params.stop_sequences,
+    }
+    _set_server_address_and_port(client_instance, attributes)
+    return {k: v for k, v in attributes.items() if v is not None}

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

@@ -14,65 +14,95 @@
 
 """Patching functions for Anthropic instrumentation."""
 
-from typing import TYPE_CHECKING, Any, Callable
+from typing import TYPE_CHECKING, Any, Callable, Union, cast
 
 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 opentelemetry.util.genai.utils import should_capture_content
 
-from .utils import (
+from .messages_extractors import (
     extract_params,
+    get_input_messages,
     get_llm_request_attributes,
+    get_system_instruction,
+)
+from .wrappers import (
+    MessageWrapper,
+    StreamWrapper,
 )
 
 if TYPE_CHECKING:
+    from anthropic._streaming import Stream
     from anthropic.resources.messages import Messages
-    from anthropic.types import Message
+    from anthropic.types import Message, RawMessageStreamEvent
+
+
+ANTHROPIC = "anthropic"
 
 
 def messages_create(
     handler: TelemetryHandler,
-) -> Callable[..., "Message"]:
+) -> Callable[..., Union["Message", "Stream[RawMessageStreamEvent]"]]:
     """Wrap the `create` method of the `Messages` class to trace it."""
+    capture_content = should_capture_content()
 
     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(
-            attributes.get(GenAIAttributes.GEN_AI_REQUEST_MODEL)
-            or params.model
-            or "unknown"
+        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 = LLMInvocation(
             request_model=request_model,
-            provider="anthropic",
+            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,
         )
 
-        with handler.llm(invocation) as invocation:
-            result = wrapped(*args, **kwargs)
-
-            if result.model:
-                invocation.response_model_name = result.model
+        is_streaming = kwargs.get("stream", False)
 
-            if result.id:
-                invocation.response_id = result.id
-
-            if result.stop_reason:
-                invocation.finish_reasons = [result.stop_reason]
-
-            if result.usage:
-                invocation.input_tokens = result.usage.input_tokens
-                invocation.output_tokens = result.usage.output_tokens
-
-            return result
-
-    return traced_method
+        # Use manual lifecycle management for both streaming and non-streaming
+        handler.start_llm(invocation)
+        try:
+            result = wrapped(*args, **kwargs)
+            if is_streaming:
+                stream_result = cast("Stream[RawMessageStreamEvent]", result)
+                return StreamWrapper(
+                    stream_result, handler, invocation, capture_content
+                )
+            message_result = cast("Message", result)
+            wrapper = MessageWrapper(message_result, capture_content)
+            wrapper.extract_into(invocation)
+            handler.stop_llm(invocation)
+            return wrapper.message
+        except Exception as exc:
+            handler.fail_llm(
+                invocation, Error(message=str(exc), type=type(exc))
+            )
+            raise
+
+    return cast(
+        Callable[..., Union["Message", "Stream[RawMessageStreamEvent]"]],
+        traced_method,
+    )