Fix lazy request iterator capture in Anthropic instrumentation

.github/instructions/instrumentation.instructions.md

@@ -54,6 +54,9 @@ prefer opt-in or additive. Breaking changes need explicit justification in the P
 - Message content, prompts, and tool call arguments must only be set through the util's content
   capture path — never as unconditional span/log attributes.
 - Adding attributes to invocations produced by the util is fine.
+- Flag captured request iterators that bypass `IteratorCapture`, eager consumption or replay,
+  wrapping unrelated arguments, capture when disabled, and snapshots taken before SDK consumption.
+  See [request property capture](../../AGENTS.md#capturing-properties-that-may-contain-iterators).
 - Streaming responses must be instrumented by subclassing the util's `SyncStreamWrapper` /
   `AsyncStreamWrapper` (`opentelemetry.util.genai.stream`). Flag hand-rolled stream wrappers, and
   invocation-backed wrappers that do not pass the invocation to `super().__init__(stream, invocation)`
@@ -109,6 +112,7 @@ Flag, with a link to the rule:
 
 - For every public API instrumented, cover sync/async variants when both exist.
 - Cover streaming and non-streaming variants when both exist.
+- Flag missing iterator tests for nested values, partial reads, errors, or file-backed content.
 - Cover happy path and error scenarios. For error scenarios, at minimum include: provider error /
   endpoint unavailable, stream interrupted by network, stream closed early by the caller, and a
   caller-side exception raised inside the `with …stream(…) as stream:` block before the stream is drained.

AGENTS.md

@@ -233,6 +233,32 @@ as the reference:
   `instrument(completion_hook=…)` override) and ship a `custom_hook.py` example mirroring the
   OpenAI package.
 
+#### Capturing properties that may contain iterators
+
+- Use `opentelemetry.util.genai.iterator_capture.IteratorCapture` for captured request properties
+  that may contain iterators, including nested values. Wrap those properties, not all `kwargs`.
+  Don't drain or replay them for telemetry.
+- Use one instance per named property: `IteratorCapture.from_property(kwargs, "messages", enabled=capture_content)`.
+  It replaces that property in `kwargs`. Use `capture.extract(extractor)`
+  after SDK consumption, including failures and lazy stream entry.
+  The extractor receives a snapshot with `()` for unread iterators and `None`
+  for disabled or missing properties; `capture.snapshot()` exposes the raw snapshot.
+- Preserve file objects. Extend the util for missing support; don't write local wrappers.
+- Test nested generators, partial reads, errors, and files across supported sync/async and
+  streaming variants.
+
+```python
+from opentelemetry.util.genai.iterator_capture import IteratorCapture
+
+capture = IteratorCapture.from_property(
+    kwargs, "messages", enabled=capture_content
+)
+try:
+    response = sdk_call(**kwargs)
+finally:
+    captured_messages = capture.extract(get_input_messages)
+```
+
 #### Streaming responses
 
 A streamed response only finishes once the caller has drained the stream, so the invocation must

instrumentation/opentelemetry-instrumentation-genai-anthropic/.changelog/873.fixed

@@ -0,0 +1 @@
+Capture one-shot message, content, system, and tool iterators lazily as the Anthropic SDK consumes them.

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

@@ -85,4 +85,3 @@ References
 * `OpenTelemetry GenAI semantic conventions <https://opentelemetry.io/docs/specs/semconv/gen-ai/>`_
 * `Anthropic SDK (Python) <https://github.com/anthropics/anthropic-sdk-python>`_
 * `Anthropic Documentation <https://docs.anthropic.com/>`_
-

instrumentation/opentelemetry-instrumentation-genai-anthropic/pyproject.toml

@@ -28,7 +28,7 @@ dependencies = [
   "opentelemetry-api ~= 1.43",
   "opentelemetry-instrumentation >= 0.64b0, <1",
   "opentelemetry-semantic-conventions >= 0.64b0, <1",
-  "opentelemetry-util-genai >= 1.2b0, <2",
+  "opentelemetry-util-genai >= 1.3b0.dev, <2",
 ]
 
 [project.optional-dependencies]

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

@@ -120,6 +120,8 @@ def get_input_messages(
         return []
     result: list[InputMessage] = []
     for message in messages:
+        if "role" not in message or "content" not in message:
+            continue
         role = message["role"]
         parts = convert_content_to_parts(message["content"])
         result.append(InputMessage(role=role, parts=parts))

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

@@ -6,7 +6,7 @@
 from __future__ import annotations
 
 import logging
-from collections.abc import Awaitable, Callable, Iterator
+from collections.abc import Awaitable, Callable
 from typing import TYPE_CHECKING, Any, cast
 
 from anthropic.types import Message as AnthropicMessage
@@ -16,6 +16,7 @@
 )
 from opentelemetry.util.genai.handler import TelemetryHandler
 from opentelemetry.util.genai.invocation import InferenceInvocation
+from opentelemetry.util.genai.iterator_capture import IteratorCapture
 
 from ._raw_response import wrap_raw_response
 from .messages_extractors import (
@@ -96,19 +97,6 @@ def _fail_context_manager_response(
         )
 
 
-def _materialize_tools(kwargs: dict[str, Any]) -> None:
-    """Replace a one-shot ``tools`` iterator with a list.
-
-    ``tools`` is typed as an iterable, so it may be a generator, and both the
-    SDK and the tool definitions read it. Call this before either of them does:
-    ``create`` reads the request after this wrapper, but ``stream`` serializes
-    it while building the manager, before the invocation is created.
-    """
-    tools = kwargs.get("tools")
-    if isinstance(tools, Iterator):
-        kwargs["tools"] = list(cast("Iterator[Any]", tools))
-
-
 def _is_raw_response(result: object) -> bool:
     """Whether ``result`` is a raw-response object to route through the proxy.
 
@@ -144,15 +132,31 @@ def traced_method(
         | AnthropicStream[RawMessageStreamEvent]
         | MessagesStreamWrapper[None]
     ):
-        _materialize_tools(kwargs)
-        invocation = _create_invocation(
-            handler, instance, args, kwargs, capture_content
+        messages_capture = IteratorCapture.from_property(
+            kwargs, "messages", capture_content
+        )
+        system_capture = IteratorCapture.from_property(
+            kwargs, "system", capture_content
         )
+        tools_capture = IteratorCapture.from_property(
+            kwargs, "tools", capture_content
+        )
+        invocation = _create_invocation(handler, instance, args, kwargs)
         try:
-            # ``Any``: through ``with_raw_response`` / ``with_streaming_response``
-            # this returns a response object that ``create``'s annotations do
-            # not describe, so the checks below cannot be narrowed away.
-            result: Any = wrapped(*args, **kwargs)
+            try:
+                # Raw response variants return types outside create's
+                # annotations, so Any keeps the checks below reachable.
+                result: Any = wrapped(*args, **kwargs)
+            finally:
+                invocation.input_messages = (
+                    messages_capture.extract(get_input_messages) or []
+                )
+                invocation.system_instruction = (
+                    system_capture.extract(get_system_instruction) or []
+                )
+                invocation.tool_definitions = tools_capture.extract(
+                    get_tool_definitions
+                )
         except BaseException as exc:
             invocation.fail(exc)
             raise
@@ -203,13 +207,30 @@ async def traced_method(
         | AnthropicAsyncStream[RawMessageStreamEvent]
         | AsyncMessagesStreamWrapper[None]
     ):
-        _materialize_tools(kwargs)
-        invocation = _create_invocation(
-            handler, instance, args, kwargs, capture_content
+        messages_capture = IteratorCapture.from_property(
+            kwargs, "messages", capture_content
+        )
+        system_capture = IteratorCapture.from_property(
+            kwargs, "system", capture_content
         )
+        tools_capture = IteratorCapture.from_property(
+            kwargs, "tools", capture_content
+        )
+        invocation = _create_invocation(handler, instance, args, kwargs)
         try:
-            # See ``messages_create`` on why this is Any.
-            result: Any = await wrapped(*args, **kwargs)
+            try:
+                # See ``messages_create`` on why this is Any.
+                result: Any = await wrapped(*args, **kwargs)
+            finally:
+                invocation.input_messages = (
+                    messages_capture.extract(get_input_messages) or []
+                )
+                invocation.system_instruction = (
+                    system_capture.extract(get_system_instruction) or []
+                )
+                invocation.tool_definitions = tools_capture.extract(
+                    get_tool_definitions
+                )
         except BaseException as exc:
             invocation.fail(exc)
             raise
@@ -239,7 +260,6 @@ def _create_invocation(
     instance: Messages | AsyncMessages,
     args: tuple[Any, ...],
     kwargs: dict[str, Any],
-    capture_content: bool,
 ) -> InferenceInvocation:
     params = extract_params(*args, **kwargs)
     attributes = get_llm_request_attributes(params, instance)
@@ -259,15 +279,6 @@ def _create_invocation(
         server_address=server_address,
         server_port=server_port,
     )
-    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.tool_definitions = (
-        get_tool_definitions(params.tools) if capture_content else None
-    )
     invocation.attributes = attributes
     return invocation
 
@@ -284,13 +295,22 @@ def traced_method(
         args: tuple[Any, ...],
         kwargs: dict[str, Any],
     ) -> MessagesStreamManagerWrapper[Any]:
-        _materialize_tools(kwargs)
+        messages_capture = IteratorCapture.from_property(
+            kwargs, "messages", capture_content
+        )
+        system_capture = IteratorCapture.from_property(
+            kwargs, "system", capture_content
+        )
+        tools_capture = IteratorCapture.from_property(
+            kwargs, "tools", capture_content
+        )
         return MessagesStreamManagerWrapper(
             wrapped(*args, **kwargs),
-            lambda: _create_invocation(
-                handler, instance, args, kwargs, capture_content
-            ),
+            lambda: _create_invocation(handler, instance, args, kwargs),
             capture_content,
+            messages_capture=messages_capture,
+            system_capture=system_capture,
+            tools_capture=tools_capture,
         )
 
     return cast(
@@ -310,13 +330,22 @@ def traced_method(
         args: tuple[Any, ...],
         kwargs: dict[str, Any],
     ) -> AsyncMessagesStreamManagerWrapper[Any]:
-        _materialize_tools(kwargs)
+        messages_capture = IteratorCapture.from_property(
+            kwargs, "messages", capture_content
+        )
+        system_capture = IteratorCapture.from_property(
+            kwargs, "system", capture_content
+        )
+        tools_capture = IteratorCapture.from_property(
+            kwargs, "tools", capture_content
+        )
         return AsyncMessagesStreamManagerWrapper(
             wrapped(*args, **kwargs),
-            lambda: _create_invocation(
-                handler, instance, args, kwargs, capture_content
-            ),
+            lambda: _create_invocation(handler, instance, args, kwargs),
             capture_content,
+            messages_capture=messages_capture,
+            system_capture=system_capture,
+            tools_capture=tools_capture,
         )
 
     return cast(

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

@@ -29,7 +29,12 @@
     finalize_on_close,
 )
 
-from .messages_extractors import set_invocation_response_attributes
+from .messages_extractors import (
+    get_input_messages,
+    get_system_instruction,
+    get_tool_definitions,
+    set_invocation_response_attributes,
+)
 
 _logger = logging.getLogger(__name__)
 
@@ -58,6 +63,7 @@
     from anthropic.types.parsed_message import ParsedMessage
 
     from opentelemetry.util.genai.invocation import InferenceInvocation
+    from opentelemetry.util.genai.iterator_capture import IteratorCapture
 
 
 ResponseFormatT = TypeVar("ResponseFormatT")
@@ -307,9 +313,32 @@ def __init__(
         manager: MessageStreamManager[ResponseFormatT],
         invocation_factory: Callable[[], InferenceInvocation],
         capture_content: bool,
+        *,
+        messages_capture: IteratorCapture,
+        system_capture: IteratorCapture,
+        tools_capture: IteratorCapture,
     ):
         super().__init__(manager, invocation_factory)
         self._self_capture_content = capture_content
+        self._self_messages_capture = messages_capture
+        self._self_system_capture = system_capture
+        self._self_tools_capture = tools_capture
+
+    def _enter_manager(
+        self, invocation: InferenceInvocation
+    ) -> MessageStream[ResponseFormatT]:
+        try:
+            return super()._enter_manager(invocation)
+        finally:
+            invocation.input_messages = (
+                self._self_messages_capture.extract(get_input_messages) or []
+            )
+            invocation.system_instruction = (
+                self._self_system_capture.extract(get_system_instruction) or []
+            )
+            invocation.tool_definitions = self._self_tools_capture.extract(
+                get_tool_definitions
+            )
 
     def _wrap_stream(
         self,
@@ -340,9 +369,32 @@ def __init__(
         manager: AsyncMessageStreamManager[ResponseFormatT],
         invocation_factory: Callable[[], InferenceInvocation],
         capture_content: bool,
+        *,
+        messages_capture: IteratorCapture,
+        system_capture: IteratorCapture,
+        tools_capture: IteratorCapture,
     ):
         super().__init__(manager, invocation_factory)
         self._self_capture_content = capture_content
+        self._self_messages_capture = messages_capture
+        self._self_system_capture = system_capture
+        self._self_tools_capture = tools_capture
+
+    async def _enter_manager(
+        self, invocation: InferenceInvocation
+    ) -> AsyncMessageStream[ResponseFormatT]:
+        try:
+            return await super()._enter_manager(invocation)
+        finally:
+            invocation.input_messages = (
+                self._self_messages_capture.extract(get_input_messages) or []
+            )
+            invocation.system_instruction = (
+                self._self_system_capture.extract(get_system_instruction) or []
+            )
+            invocation.tool_definitions = self._self_tools_capture.extract(
+                get_tool_definitions
+            )
 
     def _wrap_stream(
         self,

instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/requirements.oldest.txt

@@ -20,3 +20,5 @@
 # factor, so they are NOT pinned here — pyproject.toml is the single source of truth. The
 # OpenTelemetry SDK and test utilities come transitively from opentelemetry-test-util-genai, which
 # every oldest env installs. Pin here only test-only deps that nothing else already provides.
+
+-e util/opentelemetry-util-genai