Add explicit custom baggage propagation

A365_DOCUMENTATION.md

@@ -254,6 +254,45 @@ with (
         ...
 ```
 
+To copy an application-specific baggage value onto Agent365 GenAI spans,
+explicitly opt in each key with `custom_attribute()` or `custom_attributes()`:
+
+```python
+with (
+    BaggageBuilder()
+    .tenant_id("contoso-tenant")
+    .agent_id("weather-agent-001")
+    .custom_attribute("customer.tier", "gold")
+    .custom_attributes({"customer.region": "west"})
+    .build()
+):
+    with InvokeAgentScope.start(...) as scope:
+        ...
+```
+
+`set_pairs()` only sets baggage. It does not opt arbitrary baggage keys into
+span attributes; use `custom_attribute()` for any custom key that should appear
+on recognized Agent365 GenAI spans. Baggage propagation never overwrites
+attributes already present on the current span. Baggage-propagated values are
+applied when the span starts, so they precede later `record_attributes()` calls
+when duplicate-key protection is also present; direct/current span attributes
+remain authoritative.
+
+A span is recognized as GenAI at span start by evaluating these signals in
+order: a supported `gen_ai.operation.name` attribute; if that attribute is
+present but unrecognized (`chain`, `embeddings`, `text_completion`,
+`generate_content`, `create_agent`, ...) it is authoritative, so baggage and
+span-name inference are skipped and only the instrumentation scope can still
+classify the span; otherwise a recognized `gen_ai.operation.name` baggage entry,
+then a span name matching a supported operation (`invoke_agent ...`,
+`chat ...`, ...) or a known pre-rename name (`chat.completions ...`), then a
+supported GenAI instrumentation scope (`Agent365Sdk`, `semantic_kernel.*`,
+`agent_framework`, `microsoft.opentelemetry._genai.*`,
+`opentelemetry.instrumentation.openai_v2`,
+`opentelemetry.instrumentation.openai_agents`). Spans classified only by
+instrumentation scope are GenAI with an unknown operation: opted-in custom
+baggage applies to them, but `invoke_agent`-only attributes never do.
+
 ### From TurnContext (Hosting Framework)
 
 ```python

CHANGELOG.md

@@ -1,6 +1,7 @@
 # Release History
 # Unreleased
 ### Features Added
+- Add explicit custom baggage APIs and propagate opted-in attributes to supported GenAI spans, including unmodeled operations. ([#264](https://github.com/microsoft/opentelemetry-distro-python/pull/264))
 - Add Python-native `InvokeAgentScope` request and response parameter models
   that emit OpenTelemetry GenAI semantic attributes, including structured
   system instructions and cache read/write token counts, introduced by .NET

samples/langchain/validate_traces.py

@@ -153,9 +153,7 @@ def check(label, condition, detail=""):
     print(f"\n{'='*60}")
     print("RESPONSES-API FAKE SPAN CHECKS")
     print(f"{'='*60}")
-    responses_spans = [
-        s for s in llm_spans if s.attributes.get("gen_ai.response.id") == "resp_abc123"
-    ]
+    responses_spans = [s for s in llm_spans if s.attributes.get("gen_ai.response.id") == "resp_abc123"]
     check("Responses-API fake LLM span found", len(responses_spans) == 1, f"found {len(responses_spans)}")
     if responses_spans:
         attrs = responses_spans[0].attributes

samples/microsoft_agent_framework/sample_maf_agent.py

@@ -60,4 +60,4 @@ async def main():
 
 
 if __name__ == "__main__":
-    asyncio.run(main())
+    asyncio.run(main())

src/microsoft/opentelemetry/a365/README.md

@@ -20,7 +20,7 @@ Core tracing primitives — scopes, configuration, data models, and internal uti
 | `__init__.py` | Public API surface. Re-exports scope classes, data models, enums, and related core types. |
 | `agent_details.py` | `AgentDetails` dataclass — metadata about an AI agent (ID, name, description, blueprint/platform IDs, tenant, version). |
 | `channel.py` | `Channel` dataclass — channel context (name, link) for agent execution. |
-| `constants.py` | Core-level constants for span operations, OTel conventions, feature switches, and error types. |
+| `constants.py` | Core-level constants for span operations, GenAI processor operation names, baggage metadata, OTel conventions, feature switches, and error types. |
 | `gen_ai_request_parameters.py` | GenAI request parameter models for generation settings and related semantic attributes. |
 | `gen_ai_response_parameters.py` | GenAI response parameter models for finish reasons and token usage attributes. |
 | `execute_tool_scope.py` | `ExecuteToolScope` — tracing scope for AI tool executions. Records tool name, arguments, call ID, type, and endpoint. |
@@ -47,15 +47,15 @@ Span export pipeline — processors and exporters for Agent365 and Spectra backe
 | `agent365_exporter_options.py` | `Agent365ExporterOptions` — configuration for the Agent365 exporter (cluster category, token resolver, endpoint flags, batch settings). |
 | `enriched_span.py` | `EnrichedReadableSpan` — wrapper allowing extra attributes on immutable `ReadableSpan` objects. |
 | `enriching_span_processor.py` | Span enrichment support with registration for platform instrumentors (LangChain, Semantic Kernel, OpenAI Agents). `_EnrichingBatchSpanProcessor` applies enrichers before batching. |
-| `span_processor.py` | `A365SpanProcessor` — propagates OpenTelemetry baggage entries onto spans as attributes, with special handling for invoke_agent spans. |
+| `span_processor.py` | `A365SpanProcessor` — propagates documented OpenTelemetry baggage entries onto spans, keeps invoke_agent-specific handling, and copies only opted-in custom baggage keys onto recognized GenAI spans. |
 | `spectra_exporter_options.py` | `SpectraExporterOptions` — configuration for OTLP export to a Spectra Collector sidecar (gRPC or HTTP, tuned for Kubernetes). |
 | `utils.py` | Exporter utilities: hex encoding for trace/span IDs, span size truncation, span partitioning, environment variable handling, payload building helpers. |
 
 ### `core/middleware/`
 
 | File | Description |
 |------|-------------|
-| `baggage_builder.py` | `BaggageBuilder` — fluent API for setting per-request baggage values (tenant ID, agent ID, caller/user details, session/conversation IDs, channel, endpoints). Provides context manager for baggage scope. |
+| `baggage_builder.py` | `BaggageBuilder` — fluent API for setting per-request baggage values (tenant ID, agent ID, caller/user details, session/conversation IDs, channel, endpoints) plus `custom_attribute()` / `custom_attributes()` opt-in for custom GenAI span attributes. Provides context manager for baggage scope. |
 
 ### `core/models/`
 
@@ -150,7 +150,7 @@ below are available via `from microsoft.opentelemetry.a365.core import ...`.
 
 | Symbol | Kind | Description |
 |--------|------|-------------|
-| `BaggageBuilder` | class | Fluent API for setting per-request baggage (tenant, agent, user, channel, session, conversation). Call `.build()` to get a context manager. |
+| `BaggageBuilder` | class | Fluent API for setting per-request baggage (tenant, agent, user, channel, session, conversation) and opting custom baggage keys into recognized GenAI span attributes. Call `.build()` to get a context manager. |
 
 #### Data Classes
 

src/microsoft/opentelemetry/a365/core/constants.py

@@ -7,20 +7,58 @@
 shared across the Agent365 core scopes and exporters.
 """
 
+from microsoft.opentelemetry.a365.core.inference_operation_type import InferenceOperationType
+
 # --- Span operation names ---
 INVOKE_AGENT_OPERATION_NAME = "invoke_agent"
 EXECUTE_TOOL_OPERATION_NAME = "execute_tool"
 OUTPUT_MESSAGES_OPERATION_NAME = "output_messages"
 CHAT_OPERATION_NAME = "chat"
 APPLY_GUARDRAIL_OPERATION_NAME = "apply_guardrail"
 
+GEN_AI_PROCESSOR_OPERATION_NAMES: frozenset[str] = frozenset(
+    {
+        INVOKE_AGENT_OPERATION_NAME,
+        EXECUTE_TOOL_OPERATION_NAME,
+        OUTPUT_MESSAGES_OPERATION_NAME,
+        CHAT_OPERATION_NAME,
+        APPLY_GUARDRAIL_OPERATION_NAME,
+    }
+    | {operation.value for operation in InferenceOperationType}
+)
+
+# --- Baggage metadata ---
+CUSTOM_KEYS_BAGGAGE_KEY = "_internal.custom_keys"
+
 # --- OpenTelemetry semantic conventions ---
 ERROR_TYPE_KEY = "error.type"
 ERROR_MESSAGE_KEY = "error.message"
 AZ_NAMESPACE_KEY = "az.namespace"
 AZURE_RP_NAMESPACE_VALUE = "Microsoft.CognitiveServices"
 SOURCE_NAME = "Agent365Sdk"
 
+# Identify GenAI spans before instrumentations set ``gen_ai.operation.name``.
+# Scope matching accepts an exact root or dotted child.
+GEN_AI_INSTRUMENTATION_SCOPE_ROOTS: tuple[str, ...] = (
+    SOURCE_NAME,
+    "agent_framework",
+    "semantic_kernel",
+    "microsoft.opentelemetry._genai",
+    "opentelemetry.instrumentation.openai_v2",
+    "opentelemetry.instrumentation.openai_agents",
+)
+
+# Initial names used by supported GenAI instrumentations before span renaming.
+GEN_AI_INITIAL_SPAN_NAMES: frozenset[str] = frozenset(
+    {
+        "chat.completions",
+        "chat.streaming_completions",
+        "text.completions",
+        "text.streaming_completions",
+        "text_completions",
+    }
+)
+
 # --- Feature switches ---
 ENABLE_OPENTELEMETRY_SWITCH = "Azure.Experimental.EnableActivitySource"
 TRACE_CONTENTS_SWITCH = "Azure.Experimental.TraceGenAIMessageContent"

src/microsoft/opentelemetry/a365/core/exporters/_gen_ai_span_classifier.py

@@ -0,0 +1,106 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+"""Classify spans for Agent365 GenAI baggage propagation."""
+
+from collections.abc import Mapping
+from dataclasses import dataclass
+from typing import Any
+
+from microsoft.opentelemetry.a365.core.constants import (
+    GEN_AI_INITIAL_SPAN_NAMES,
+    GEN_AI_INSTRUMENTATION_SCOPE_ROOTS,
+    GEN_AI_OPERATION_NAME_KEY,
+    GEN_AI_PROCESSOR_OPERATION_NAMES,
+)
+
+
+@dataclass(frozen=True)
+class _GenAISpanClassification:
+    """Whether a span is GenAI and, when identifiable, the operation it represents."""
+
+    is_gen_ai_span: bool
+    operation_name: str | None = None
+
+
+def _recognized_operation_name(value: object | None) -> str | None:
+    return value if isinstance(value, str) and value in GEN_AI_PROCESSOR_OPERATION_NAMES else None
+
+
+def _span_name(span: Any) -> str | None:
+    span_name = getattr(span, "name", None)
+    return span_name if isinstance(span_name, str) else None
+
+
+def _operation_name_from_span_name(span: Any) -> str | None:
+    span_name = _span_name(span)
+    if span_name is None:
+        return None
+
+    for operation_name in GEN_AI_PROCESSOR_OPERATION_NAMES:
+        if span_name == operation_name or span_name.startswith(f"{operation_name} "):
+            return operation_name
+    return None
+
+
+def _has_known_initial_span_name(span: Any) -> bool:
+    """Match span names supported instrumentations use before renaming the span."""
+    span_name = _span_name(span)
+    if span_name is None:
+        return False
+
+    return any(span_name == known or span_name.startswith(f"{known} ") for known in GEN_AI_INITIAL_SPAN_NAMES)
+
+
+def _instrumentation_scope_name(span: Any) -> str | None:
+    """Read the tracer (source) name recorded on a ReadWriteSpan."""
+    # pylint: disable=broad-exception-caught
+    for attribute_name in ("instrumentation_scope", "instrumentation_info"):
+        try:
+            scope = getattr(span, attribute_name, None)
+            scope_name = getattr(scope, "name", None) if scope is not None else None
+        except Exception:
+            continue
+        if isinstance(scope_name, str) and scope_name:
+            return scope_name
+    return None
+
+
+def _is_supported_gen_ai_scope(span: Any) -> bool:
+    scope_name = _instrumentation_scope_name(span)
+    if scope_name is None:
+        return False
+
+    return any(scope_name == root or scope_name.startswith(f"{root}.") for root in GEN_AI_INSTRUMENTATION_SCOPE_ROOTS)
+
+
+def _classify_gen_ai_span(
+    span: Any,
+    existing_attributes: Mapping[str, object],
+    baggage_map: Mapping[str, object],
+) -> _GenAISpanClassification:
+    """Resolve whether a span is GenAI and which operation it represents.
+
+    An explicit ``gen_ai.operation.name`` attribute is authoritative over the
+    baggage and span-name inference signals, including when it holds a value
+    this processor does not model (``chain``, ``embeddings``,
+    ``text_completion``, ``generate_content``, ``create_agent``). Such a span is
+    still GenAI when a supported instrumentation emitted it, but its operation
+    stays unknown so invoke_agent-only attributes are withheld.
+
+    Without an explicit operation attribute, a recognized span-name operation
+    takes precedence over inherited operation baggage.
+    """
+    if GEN_AI_OPERATION_NAME_KEY in existing_attributes:
+        explicit_operation_name = _recognized_operation_name(existing_attributes.get(GEN_AI_OPERATION_NAME_KEY))
+        if explicit_operation_name is not None:
+            return _GenAISpanClassification(True, explicit_operation_name)
+        return _GenAISpanClassification(_is_supported_gen_ai_scope(span))
+
+    operation_name = _operation_name_from_span_name(span)
+    if operation_name is None:
+        operation_name = _recognized_operation_name(baggage_map.get(GEN_AI_OPERATION_NAME_KEY))
+    if operation_name is not None:
+        return _GenAISpanClassification(True, operation_name)
+
+    return _GenAISpanClassification(_has_known_initial_span_name(span) or _is_supported_gen_ai_scope(span))

src/microsoft/opentelemetry/a365/core/exporters/span_processor.py

@@ -12,16 +12,43 @@
   * For each documented key with a truthy value not already present as a span
     attribute, add it via span.set_attribute
   * Never overwrites existing attributes
+
+Custom baggage is propagated only to recognized GenAI spans. A span is
+recognized as GenAI by evaluating these signals in order at ``on_start``:
+
+  1. An explicit ``gen_ai.operation.name`` attribute holding a recognized
+     operation: GenAI with a known operation.
+  2. An explicit but *unrecognized* ``gen_ai.operation.name`` attribute: the
+     attribute is authoritative, so the baggage and span-name inference of
+     signals 3 and 4 is skipped. The span is still GenAI when a supported
+     instrumentation emitted it (signal 5), with an unknown operation.
+  3. A recognized ``gen_ai.operation.name`` baggage entry.
+  4. A span name that is (or starts with) a recognized operation name, or a
+     name a supported instrumentation is known to use before it renames the
+     span (Semantic Kernel ``chat.completions <model>``).
+  5. The instrumentation scope (source) name of a supported GenAI
+     instrumentation: GenAI with an unknown operation.
+
+Signals 4 and 5 exist because most GenAI instrumentations apply
+``gen_ai.operation.name`` *after* the span starts: LangChain chat spans start
+as ``ChatOpenAI`` and the OpenAI Agents processor starts workflow spans as
+``Agent workflow``. Signal 5 also keeps spans whose operation this processor
+does not model (``chain``, ``embeddings``, ``text_completion``,
+``generate_content``, ``create_agent``) from being dropped. Only signals 1, 3
+and 4 identify *which* operation a span represents, which is what gates the
+invoke_agent-only attributes.
 """
 
 from __future__ import annotations
 
 from opentelemetry import baggage, context
 from opentelemetry.sdk.trace import SpanProcessor as BaseSpanProcessor
 
-from microsoft.opentelemetry.a365.constants import (
+from microsoft.opentelemetry.a365.core.exporters._gen_ai_span_classifier import _classify_gen_ai_span
+from microsoft.opentelemetry.a365.core.constants import (
     CHANNEL_LINK_KEY,
     CHANNEL_NAME_KEY,
+    CUSTOM_KEYS_BAGGAGE_KEY,
     CUSTOM_PARENT_SPAN_ID_KEY,
     CUSTOM_SPAN_NAME_KEY,
     GEN_AI_AGENT_AUID_KEY,
@@ -98,6 +125,21 @@
 ]
 
 
+def _custom_baggage_keys(baggage_map) -> list[str]:
+    metadata = baggage_map.get(CUSTOM_KEYS_BAGGAGE_KEY)
+    if not metadata:
+        return []
+
+    keys: list[str] = []
+    for raw_key in str(metadata).split(","):
+        key = raw_key.strip()
+        if not key or key == CUSTOM_KEYS_BAGGAGE_KEY:
+            continue
+        if key not in keys:
+            keys.append(key)
+    return keys
+
+
 # pylint: disable=broad-exception-caught, too-many-branches, useless-parent-delegation
 # pylint: disable=global-statement
 class A365SpanProcessor(BaseSpanProcessor):
@@ -151,18 +193,17 @@ def on_start(self, span, parent_context=None):  # type: ignore[override]
         except Exception:
             baggage_map = {}
 
-        operation_name = existing.get(GEN_AI_OPERATION_NAME_KEY)
-        is_invoke_agent = False
-        if operation_name == INVOKE_AGENT_OPERATION_NAME:
-            is_invoke_agent = True
-        elif isinstance(getattr(span, "name", None), str) and span.name.startswith(INVOKE_AGENT_OPERATION_NAME):
-            is_invoke_agent = True
+        classification = _classify_gen_ai_span(span, existing, baggage_map)
 
         target_keys = list(COMMON_ATTRIBUTES)
-        if is_invoke_agent:
+        if classification.operation_name == INVOKE_AGENT_OPERATION_NAME:
             for k in INVOKE_AGENT_ATTRIBUTES:
                 if k not in target_keys:
                     target_keys.append(k)
+        if classification.is_gen_ai_span:
+            for k in _custom_baggage_keys(baggage_map):
+                if k not in target_keys:
+                    target_keys.append(k)
 
         for key in target_keys:
             if key in existing: