Instrument OpenAI Responses.retrieve and AsyncResponses.retrieve

instrumentation/opentelemetry-instrumentation-genai-openai/.changelog/184.added

@@ -0,0 +1 @@
+Add instrumentation for the OpenAI Responses ``retrieve`` API (sync and async), reported as a ``fetch_response`` operation with no token usage.

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

@@ -62,9 +62,15 @@ Check out the `manual example <examples/manual>`_ for more details.
 Instrumenting all clients
 *************************
 
-When using the instrumentor, all clients will automatically trace OpenAI operations including chat completions and embeddings.
+When using the instrumentor, all clients will automatically trace OpenAI operations including chat completions, the Responses API, and embeddings.
 You can also optionally capture prompts and completions as log events.
 
+Fetching a stored response with ``client.responses.retrieve(...)`` performs no
+inference, so it is reported as a ``fetch_response`` operation rather than as an
+inference call. Its span carries the fetched response's identifier, model,
+status, and finish reasons, but no token usage — those counts belong to the
+operation that originally generated the response.
+
 Make sure to configure OpenTelemetry tracing, logging, and events to capture all telemetry emitted by the instrumentation.
 
 .. code-block:: python
@@ -82,6 +88,14 @@ Make sure to configure OpenTelemetry tracing, logging, and events to capture all
         ],
     )
 
+    # Responses API example, fetching a stored response back by its id
+    created = client.responses.create(
+        model="gpt-4o-mini",
+        input="Write a short poem on open telemetry.",
+        store=True,
+    )
+    fetched = client.responses.retrieve(created.id)
+
     # Embeddings example
     embedding_response = client.embeddings.create(
         model="text-embedding-3-small",

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

@@ -79,8 +79,10 @@
 )
 from .patch_responses import (
     async_responses_create,
+    async_responses_retrieve,
     async_responses_stream,
     responses_create,
+    responses_retrieve,
     responses_stream,
 )
 
@@ -188,6 +190,20 @@ def _instrument(self, **kwargs):
                 async_responses_stream(handler),
             )
 
+            # retrieve() fetches a stored response by id. No inference happens
+            # and no tokens are consumed, so it is traced as a fetch_response
+            # operation rather than through the create wrappers.
+            wrap_function_wrapper(
+                "openai.resources.responses.responses",
+                "Responses.retrieve",
+                responses_retrieve(handler),
+            )
+            wrap_function_wrapper(
+                "openai.resources.responses.responses",
+                "AsyncResponses.retrieve",
+                async_responses_retrieve(handler),
+            )
+
     def _uninstrument(self, **kwargs):
         import openai  # pylint: disable=import-outside-toplevel
 
@@ -204,6 +220,8 @@ def _uninstrument(self, **kwargs):
             unwrap(responses_module.Responses, "stream")
             unwrap(responses_module.AsyncResponses, "create")
             unwrap(responses_module.AsyncResponses, "stream")
+            unwrap(responses_module.Responses, "retrieve")
+            unwrap(responses_module.AsyncResponses, "retrieve")
 
 
 def _get_responses_module():

instrumentation/opentelemetry-instrumentation-genai-openai/src/opentelemetry/instrumentation/genai/openai/patch_responses.py

@@ -6,24 +6,33 @@
 from collections.abc import Awaitable, Callable
 from typing import TYPE_CHECKING, Any, Union, cast
 
+from opentelemetry.semconv._incubating.attributes import (
+    openai_attributes as OpenAIAttributes,
+)
 from opentelemetry.util.genai.handler import TelemetryHandler
+from opentelemetry.util.genai.invocation import FetchResponseInvocation
 
 from ._raw_response import wrap_stream_result
 from .response_extractors import (
     apply_request_attributes,
     extract_params,
+    get_fetch_response_creation_kwargs,
     get_inference_creation_kwargs,
     get_response_error,
+    is_streamed_raw_response,
+    set_fetch_response_attributes,
     set_invocation_response_attributes,
 )
 from .response_wrappers import (
+    AsyncFetchResponseStreamWrapper,
     AsyncResponseStreamManagerWrapper,
     AsyncResponseStreamWrapper,
+    FetchResponseStreamWrapper,
     ResponseStreamManagerWrapper,
     ResponseStreamWrapper,
     responses_stream_context,
 )
-from .utils import is_streaming
+from .utils import is_streaming, value_is_set
 
 if TYPE_CHECKING:
     from openai import AsyncStream as OpenAIAsyncStream
@@ -91,9 +100,9 @@ def traced_method(
                 )
 
             set_invocation_response_attributes(
-                invocation, result, capture_content
+                invocation, result, capture_content, kwargs
             )
-            error = get_response_error(result)
+            error = get_response_error(result, kwargs)
             if error is not None:
                 invocation.fail(error)
             else:
@@ -164,9 +173,9 @@ async def traced_method(
                 )
 
             set_invocation_response_attributes(
-                invocation, result, capture_content
+                invocation, result, capture_content, kwargs
             )
-            error = get_response_error(result)
+            error = get_response_error(result, kwargs)
             if error is not None:
                 invocation.fail(error)
             else:
@@ -182,6 +191,181 @@ async def traced_method(
     )
 
 
+def _get_retrieve_response_id(
+    args: tuple[Any, ...], kwargs: dict[str, Any]
+) -> str | None:
+    """Return the ``response_id`` a ``responses.retrieve`` call was made with."""
+    response_id = args[0] if args else kwargs.get("response_id")
+    return response_id if isinstance(response_id, str) else None
+
+
+def _get_stream_cursor(kwargs: dict[str, Any]) -> str | None:
+    """Return the ``starting_after`` cursor a streamed fetch resumes from."""
+    starting_after = kwargs.get("starting_after")
+    if not value_is_set(starting_after):
+        return None
+    return str(starting_after)
+
+
+def _start_fetch_response_invocation(
+    handler: TelemetryHandler,
+    instance: Responses | AsyncResponses,
+    response_id: str,
+    kwargs: dict[str, Any],
+) -> FetchResponseInvocation:
+    invocation = handler.fetch_response(
+        **get_fetch_response_creation_kwargs(response_id, instance),
+        request_stream=(
+            True
+            if is_streaming(kwargs) or is_streamed_raw_response(kwargs)
+            else None
+        ),
+    )
+    invocation.stream_cursor = _get_stream_cursor(kwargs)
+    invocation.attributes[OpenAIAttributes.OPENAI_API_TYPE] = (
+        OpenAIAttributes.OpenaiApiTypeValues.RESPONSES.value
+    )
+    return invocation
+
+
+def responses_retrieve(
+    handler: TelemetryHandler,
+) -> Callable[
+    ...,
+    Union[
+        ResponseResult,
+        ResponseStreamResult,
+        FetchResponseStreamWrapper[Any],
+    ],
+]:
+    """Wrap ``Responses.retrieve`` to trace fetching a stored response by id.
+
+    Traces :meth:`openai.resources.responses.responses.Responses.retrieve`.
+    Fetching a stored response performs no inference and consumes no tokens, so
+    it is reported as a ``fetch_response`` operation rather than as an
+    inference call.
+    """
+
+    capture_content = handler.should_capture_content()
+
+    def traced_method(
+        wrapped: Callable[..., Union[ResponseResult, ResponseStreamResult]],
+        instance: Responses,
+        args: tuple[Any, ...],
+        kwargs: dict[str, Any],
+    ) -> Union[
+        ResponseResult,
+        ResponseStreamResult,
+        FetchResponseStreamWrapper[Any],
+    ]:
+        response_id = _get_retrieve_response_id(args, kwargs)
+        if response_id is None:
+            # gen_ai.response.id is required on a fetch_response span and the
+            # SDK rejects the call without it, so leave it untraced.
+            return wrapped(*args, **kwargs)
+
+        invocation = _start_fetch_response_invocation(
+            handler, instance, response_id, kwargs
+        )
+
+        try:
+            result = wrapped(*args, **kwargs)
+            if is_streaming(kwargs):
+                return wrap_stream_result(
+                    FetchResponseStreamWrapper,
+                    result,
+                    invocation,
+                    capture_content,
+                )
+
+            set_fetch_response_attributes(
+                invocation,
+                result,
+                capture_content,
+                kwargs,
+            )
+            invocation.stop()
+            return result
+        except Exception as error:
+            invocation.fail(error)
+            raise
+
+    return cast(
+        'Callable[..., Union["ResponseResult", "ResponseStreamResult", FetchResponseStreamWrapper[Any]]]',
+        traced_method,
+    )
+
+
+def async_responses_retrieve(
+    handler: TelemetryHandler,
+) -> Callable[
+    ...,
+    Awaitable[
+        Union[
+            ResponseResult,
+            AsyncResponseStreamResult,
+            AsyncFetchResponseStreamWrapper[Any],
+        ]
+    ],
+]:
+    """Wrap ``AsyncResponses.retrieve`` to trace fetching a stored response by id.
+
+    Traces :meth:`openai.resources.responses.responses.AsyncResponses.retrieve`.
+    Fetching a stored response performs no inference and consumes no tokens, so
+    it is reported as a ``fetch_response`` operation rather than as an
+    inference call.
+    """
+
+    capture_content = handler.should_capture_content()
+
+    async def traced_method(
+        wrapped: Callable[
+            ...,
+            Awaitable[Union[ResponseResult, AsyncResponseStreamResult]],
+        ],
+        instance: AsyncResponses,
+        args: tuple[Any, ...],
+        kwargs: dict[str, Any],
+    ) -> Union[
+        ResponseResult,
+        AsyncResponseStreamResult,
+        AsyncFetchResponseStreamWrapper[Any],
+    ]:
+        response_id = _get_retrieve_response_id(args, kwargs)
+        if response_id is None:
+            # gen_ai.response.id is required on a fetch_response span and the
+            # SDK rejects the call without it, so leave it untraced.
+            return await wrapped(*args, **kwargs)
+
+        invocation = _start_fetch_response_invocation(
+            handler, instance, response_id, kwargs
+        )
+
+        try:
+            result = await wrapped(*args, **kwargs)
+            if is_streaming(kwargs):
+                return wrap_stream_result(
+                    AsyncFetchResponseStreamWrapper,
+                    result,
+                    invocation,
+                    capture_content,
+                )
+
+            set_fetch_response_attributes(
+                invocation, result, capture_content, kwargs
+            )
+            invocation.stop()
+            return result
+        except Exception as error:
+            invocation.fail(error)
+            raise
+
+    return cast(
+        'Callable[..., Awaitable[Union["ResponseResult", "AsyncResponseStreamResult", AsyncFetchResponseStreamWrapper[Any]]]]',
+        traced_method,
+    )
+
+
 def responses_stream(
     handler: TelemetryHandler,
 ) -> Callable[..., ResponseStreamManagerWrapper[Any]]: