feat(util-genai): add per-signal context-scoped attributes

util/opentelemetry-util-genai/.changelog/337.added

@@ -0,0 +1 @@
+Add ``set_context_scoped_attributes`` for attaching attributes to an OpenTelemetry context so that GenAI telemetry emitted within it carries them, targeted per signal at spans or events.

util/opentelemetry-util-genai/README.rst

@@ -12,6 +12,7 @@ Key Components
 - ``TelemetryHandler`` -- manages LLM invocation lifecycles (spans, metrics, events)
 - ``InferenceInvocation`` and message types (``Text``, ``Reasoning``, ``Blob``, etc.) -- structured data model for GenAI interactions
 - ``CompletionHook`` -- protocol for uploading content to external storage (built-in ``fsspec`` support)
+- ``set_context_scoped_attributes`` -- attach attributes to a context so GenAI telemetry emitted within it carries them
 - Metrics -- ``gen_ai.client.operation.duration`` and ``gen_ai.client.token.usage`` histograms, plus
   the streaming timing histograms ``gen_ai.client.operation.time_to_first_chunk`` and
   ``gen_ai.client.operation.time_per_output_chunk``
@@ -24,6 +25,47 @@ See the module docstring in ``opentelemetry.util.genai.handler`` for usage examp
 including context manager and manual lifecycle patterns.
 
 
+Context-scoped Attributes
+-------------------------
+
+An agentic framework knows which agent, workflow, or conversation is running.
+The model-client instrumentation that emits the inference telemetry sits a layer
+below it and has no way to learn any of it -- there is no shared call path, and
+attributes cannot be read back off a parent span.
+``set_context_scoped_attributes`` bridges the two through the OpenTelemetry
+context.
+
+Each attribute declares which signal it applies to, so content that is unsafe on
+a sampled, widely-read span can still be recorded on the event::
+
+    from opentelemetry import context
+    from opentelemetry.util.genai.context_attributes import (
+        set_context_scoped_attributes,
+    )
+
+    token = context.attach(
+        set_context_scoped_attributes(
+            span_attributes={"gen_ai.agent.name": "trip-planner"},
+            log_attributes={"user.id": user_id},
+        )
+    )
+    try:
+        client.chat.completions.create(...)  # instrumented elsewhere
+    finally:
+        context.detach(token)
+
+- Attributes apply to GenAI telemetry emitted by this package only. Other
+  instrumentation, and telemetry the application emits directly, are unaffected.
+- Attributes are never propagated out of the process.
+- ``span_attributes`` are applied when the span starts, so they are visible to
+  samplers. Attributes an invocation sets itself take precedence.
+- ``log_attributes`` apply to the ``gen_ai.client.inference.operation.details``
+  event, which is the only event this package emits.
+- Nested calls merge, with the inner call taking precedence for keys it sets.
+- An invocation reads the context once, when it starts.
+- Metrics are deliberately not supported, to avoid unbounded cardinality.
+
+
 Environment Variables
 ---------------------
 

util/opentelemetry-util-genai/src/opentelemetry/util/genai/_inference_invocation.py

@@ -246,6 +246,8 @@ def _maybe_create_event(self) -> LogRecord | None:
             return None
 
         attributes = self._get_start_attributes()
+        if self._context_scoped_attributes.log:
+            attributes = {**self._context_scoped_attributes.log, **attributes}
         attributes.update(self._get_attributes())
         attributes.update(self._get_message_attributes(for_span=False))
         attributes.update(self.attributes)

util/opentelemetry-util-genai/src/opentelemetry/util/genai/_invocation.py

@@ -21,6 +21,10 @@
 from opentelemetry.trace import Span, SpanKind, Tracer, set_span_in_context
 from opentelemetry.trace.status import Status, StatusCode
 from opentelemetry.util.genai.completion_hook import CompletionHook
+from opentelemetry.util.genai.context_attributes import (
+    _ContextScopedAttributes,
+    _get_context_scoped_attributes,
+)
 from opentelemetry.util.genai.types import (
     Error,
     ErrorTypeResolver,
@@ -94,6 +98,7 @@ def __init__(
         self._request_stream: bool | None = None
         self._ttfc_seconds: float | None = None
         self._stream_last_chunk_at: float | None = None
+        self._context_scoped_attributes = _ContextScopedAttributes()
 
     def _start(
         self, attributes: dict[str, AttributeValue] | None = None
@@ -103,6 +108,12 @@ def _start(
         Args:
             attributes: Initial span attributes available for sampling decisions.
         """
+        self._context_scoped_attributes = _get_context_scoped_attributes()
+        if self._context_scoped_attributes.span:
+            attributes = {
+                **self._context_scoped_attributes.span,
+                **(attributes or {}),
+            }
         self.span = self._tracer.start_span(
             name=self._span_name,
             kind=self._span_kind,

util/opentelemetry-util-genai/src/opentelemetry/util/genai/context_attributes.py

@@ -0,0 +1,97 @@
+# Copyright The OpenTelemetry Authors
+# SPDX-License-Identifier: Apache-2.0
+
+"""Attach attributes to a context so GenAI telemetry emitted within it carries them.
+
+This lets a caller that knows something the instrumentation cannot -- which
+agent is running, which user made the request -- add it to telemetry emitted
+further down the stack. Each attribute targets a signal, so data that is unsafe
+on a span can still go on the event.
+
+.. code-block:: python
+
+    token = context.attach(
+        set_context_scoped_attributes(
+            span_attributes={"gen_ai.agent.name": "trip-planner"},
+            log_attributes={"user.id": user_id},
+        )
+    )
+    try:
+        client.chat.completions.create(...)  # instrumented elsewhere
+    finally:
+        context.detach(token)
+
+Only telemetry from this package is affected, and attributes never leave the
+process. Metrics are not supported, to avoid unbounded cardinality.
+
+The naming follows `OTEP 4931
+<https://github.com/open-telemetry/opentelemetry-specification/blob/main/oteps/4931-context-scoped-attributes.md>`_,
+but this is narrower: that OTEP has the SDK stamp all telemetry and gates it
+per signal on the provider.
+"""
+
+from __future__ import annotations
+
+from types import MappingProxyType
+from typing import Mapping, NamedTuple
+
+from opentelemetry.context import (
+    Context,
+    create_key,
+    get_current,
+    get_value,
+    set_value,
+)
+from opentelemetry.util.types import AttributeValue
+
+__all__ = ["set_context_scoped_attributes"]
+
+_CONTEXT_SCOPED_ATTRIBUTES_KEY = create_key(
+    "opentelemetry.util.genai.context_scoped_attributes"
+)
+
+_EMPTY: Mapping[str, AttributeValue] = MappingProxyType({})
+
+
+class _ContextScopedAttributes(NamedTuple):
+    """The per-signal attribute bags carried by a single context."""
+
+    span: Mapping[str, AttributeValue] = _EMPTY
+    log: Mapping[str, AttributeValue] = _EMPTY
+
+
+_NO_ATTRIBUTES = _ContextScopedAttributes()
+
+
+def set_context_scoped_attributes(
+    *,
+    span_attributes: Mapping[str, AttributeValue] | None = None,
+    log_attributes: Mapping[str, AttributeValue] | None = None,
+    context: Context | None = None,
+) -> Context:
+    """Return a new context carrying the given attributes. Does not attach it.
+
+    Args:
+        span_attributes: Added to GenAI spans on span start.
+        log_attributes: Added to the GenAI logs emitted.
+        context: Base context. Defaults to the current context.
+    """
+    if not span_attributes and not log_attributes:
+        return context if context is not None else get_current()
+
+    existing = _get_context_scoped_attributes(context)
+    merged = _ContextScopedAttributes(
+        span=MappingProxyType({**existing.span, **(span_attributes or {})}),
+        log=MappingProxyType({**existing.log, **(log_attributes or {})}),
+    )
+    return set_value(_CONTEXT_SCOPED_ATTRIBUTES_KEY, merged, context)
+
+
+def _get_context_scoped_attributes(
+    context: Context | None = None,
+) -> _ContextScopedAttributes:
+    """Return the attribute bags on the given context, or empty ones."""
+    value = get_value(_CONTEXT_SCOPED_ATTRIBUTES_KEY, context)
+    if isinstance(value, _ContextScopedAttributes):
+        return value
+    return _NO_ATTRIBUTES

util/opentelemetry-util-genai/tests/test_context_attributes.py

@@ -0,0 +1,182 @@
+# Copyright The OpenTelemetry Authors
+# SPDX-License-Identifier: Apache-2.0
+
+import os
+import unittest
+from contextlib import contextmanager
+from unittest.mock import patch
+
+from opentelemetry import context as otel_context
+from opentelemetry.sdk._logs import LoggerProvider
+from opentelemetry.sdk._logs.export import (
+    InMemoryLogRecordExporter,
+    SimpleLogRecordProcessor,
+)
+from opentelemetry.sdk.trace import TracerProvider
+from opentelemetry.sdk.trace.export import SimpleSpanProcessor
+from opentelemetry.sdk.trace.export.in_memory_span_exporter import (
+    InMemorySpanExporter,
+)
+from opentelemetry.sdk.trace.sampling import Decision, Sampler, SamplingResult
+from opentelemetry.semconv._incubating.attributes import (
+    gen_ai_attributes as GenAI,
+)
+from opentelemetry.semconv._incubating.attributes import user_attributes
+from opentelemetry.util.genai.context_attributes import (
+    set_context_scoped_attributes,
+)
+from opentelemetry.util.genai.handler import get_telemetry_handler
+
+
+class _RecordingSampler(Sampler):
+    """Samples everything, remembering the attributes it was given."""
+
+    def __init__(self):
+        self.seen = []
+
+    def should_sample(
+        self,
+        parent_context,
+        trace_id,
+        name,
+        kind=None,
+        attributes=None,
+        *_,
+        **__,
+    ):
+        self.seen.append(attributes)
+        return SamplingResult(Decision.RECORD_AND_SAMPLE, attributes)
+
+    def get_description(self):
+        return "RecordingSampler"
+
+
+@patch.dict(
+    os.environ,
+    {
+        "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT": "EVENT_ONLY",
+        "OTEL_INSTRUMENTATION_GENAI_EMIT_EVENT": "true",
+    },
+)
+class TestContextScopedAttributes(unittest.TestCase):
+    def setUp(self):
+        self.sampler = _RecordingSampler()
+        self.span_exporter = InMemorySpanExporter()
+        tracer_provider = TracerProvider(sampler=self.sampler)
+        tracer_provider.add_span_processor(
+            SimpleSpanProcessor(self.span_exporter)
+        )
+        self.log_exporter = InMemoryLogRecordExporter()
+        logger_provider = LoggerProvider()
+        logger_provider.add_log_record_processor(
+            SimpleLogRecordProcessor(self.log_exporter)
+        )
+        self.handler = get_telemetry_handler(
+            tracer_provider=tracer_provider, logger_provider=logger_provider
+        )
+
+    def tearDown(self):
+        if hasattr(get_telemetry_handler, "_default_handler"):
+            delattr(get_telemetry_handler, "_default_handler")
+
+    @contextmanager
+    def _inference(self, context):
+        """Run one inference invocation with `context` attached."""
+        token = otel_context.attach(context)
+        try:
+            with self.handler.inference(
+                "test-provider", request_model="test-model"
+            ) as invocation:
+                yield invocation
+        finally:
+            otel_context.detach(token)
+
+    @property
+    def span_attributes(self):
+        (span,) = self.span_exporter.get_finished_spans()
+        return span.attributes
+
+    @property
+    def event_attributes(self):
+        (log,) = self.log_exporter.get_finished_logs()
+        return log.log_record.attributes
+
+    def test_attributes_go_only_to_the_signal_they_target(self):
+        with self._inference(
+            set_context_scoped_attributes(
+                span_attributes={GenAI.GEN_AI_AGENT_NAME: "trip-planner"},
+                log_attributes={user_attributes.USER_ID: "u-1"},
+            )
+        ):
+            pass
+
+        self.assertEqual(
+            self.span_attributes[GenAI.GEN_AI_AGENT_NAME], "trip-planner"
+        )
+        self.assertNotIn(user_attributes.USER_ID, self.span_attributes)
+        self.assertEqual(self.event_attributes[user_attributes.USER_ID], "u-1")
+        self.assertNotIn(GenAI.GEN_AI_AGENT_NAME, self.event_attributes)
+
+    def test_invocation_attributes_win(self):
+        context = set_context_scoped_attributes(
+            span_attributes={
+                GenAI.GEN_AI_PROVIDER_NAME: "from-context",
+                "custom": "from-context",
+            },
+            log_attributes={GenAI.GEN_AI_PROVIDER_NAME: "from-context"},
+        )
+        with self._inference(context) as invocation:
+            invocation.attributes["custom"] = "from-invocation"
+
+        self.assertEqual(
+            self.span_attributes[GenAI.GEN_AI_PROVIDER_NAME], "test-provider"
+        )
+        self.assertEqual(self.span_attributes["custom"], "from-invocation")
+        self.assertEqual(
+            self.event_attributes[GenAI.GEN_AI_PROVIDER_NAME], "test-provider"
+        )
+
+    def test_nested_contexts_merge_with_the_inner_one_winning(self):
+        outer = set_context_scoped_attributes(
+            span_attributes={"outer": "outer", "shared": "outer"}
+        )
+        with self._inference(
+            set_context_scoped_attributes(
+                span_attributes={"inner": "inner", "shared": "inner"},
+                context=outer,
+            )
+        ):
+            pass
+
+        self.assertEqual(self.span_attributes["outer"], "outer")
+        self.assertEqual(self.span_attributes["inner"], "inner")
+        self.assertEqual(self.span_attributes["shared"], "inner")
+
+    def test_span_attributes_are_visible_to_the_sampler(self):
+        with self._inference(
+            set_context_scoped_attributes(
+                span_attributes={GenAI.GEN_AI_AGENT_NAME: "trip-planner"}
+            )
+        ):
+            pass
+
+        (seen,) = self.sampler.seen
+        self.assertEqual(seen[GenAI.GEN_AI_AGENT_NAME], "trip-planner")
+
+    def test_attributes_do_not_apply_outside_the_attached_context(self):
+        with self._inference(
+            set_context_scoped_attributes(
+                span_attributes={GenAI.GEN_AI_AGENT_NAME: "trip-planner"}
+            )
+        ):
+            pass
+        self.span_exporter.clear()
+
+        with self.handler.inference("test-provider"):
+            pass
+
+        self.assertNotIn(GenAI.GEN_AI_AGENT_NAME, self.span_attributes)
+
+
+if __name__ == "__main__":
+    unittest.main()