feat(telemetry): support per-request OpenTelemetry configuration

Adds a TelemetryConfig that callers can attach to RunConfig.telemetry to
set OpenTelemetry knobs per request instead of via process-wide OTEL_*
env vars. Each field falls back to its env var when unset, so the default
preserves existing behavior.

PiperOrigin-RevId: 929715433

Author: google-genai-bot

src/google/adk/agents/run_config.py

@@ -29,6 +29,7 @@
 from pydantic import model_validator
 
 from ..sessions.base_session_service import GetSessionConfig
+from ..telemetry.context import TelemetryConfig
 
 logger = logging.getLogger('google_adk.' + __name__)
 
@@ -327,6 +328,19 @@ class RunConfig(BaseModel):
   custom_metadata: Optional[dict[str, Any]] = None
   """Custom metadata for the current invocation."""
 
+  telemetry: TelemetryConfig | None = None
+  """Per-request OpenTelemetry configuration.
+
+  Overrides the process-global telemetry env vars for the duration of this
+  invocation. Each ``None`` field on the
+  :class:`~google.adk.telemetry.TelemetryConfig` falls back to its
+  corresponding env var. Lets multi-tenant hosts toggle telemetry knobs per
+  request without leaking configuration across concurrent invocations.
+
+  .. warning::
+      Experimental; API may change.
+  """
+
   get_session_config: Optional[GetSessionConfig] = None
   """Configuration for controlling which events are fetched when loading
   a session.

src/google/adk/flows/llm_flows/base_llm_flow.py

@@ -28,8 +28,6 @@
 from websockets.exceptions import ConnectionClosed
 from websockets.exceptions import ConnectionClosedOK
 
-from . import _output_schema_processor
-from . import functions
 from ...agents.base_agent import BaseAgent
 from ...agents.callback_context import CallbackContext
 from ...agents.invocation_context import InvocationContext
@@ -52,6 +50,8 @@
 from ...tools.tool_context import ToolContext
 from ...utils import model_name_utils
 from ...utils.context_utils import Aclosing
+from . import _output_schema_processor
+from . import functions
 from .audio_cache_manager import AudioCacheManager
 from .functions import build_auth_request_event
 
@@ -385,7 +385,7 @@ async def _run_on_model_error_callbacks(
     ) as tel_ctx:
       async with Aclosing(response_generator) as agen:
         async for llm_response in agen:
-          tel_ctx.record_llm_response(llm_response)
+          tel_ctx.record_llm_response(invocation_context, llm_response)
           yield llm_response
   except Exception as model_error:
     callback_context = CallbackContext(
@@ -434,8 +434,8 @@ async def _process_agent_tools(
   names to ``BaseTool`` instances ready for function call dispatch.
 
   Args:
-    invocation_context: The invocation context (``agent`` is read
-      from ``invocation_context.agent``).
+    invocation_context: The invocation context (``agent`` is read from
+      ``invocation_context.agent``).
     llm_request: The LLM request to populate with tool declarations.
   """
   agent = invocation_context.agent

src/google/adk/flows/llm_flows/functions.py

@@ -470,6 +470,7 @@ async def handle_function_call_list_async(
       trace_merged_tool_calls(
           response_event_id=merged_event.id,
           function_response_event=merged_event,
+          invocation_context=invocation_context,
       )
   return merged_event
 
@@ -642,7 +643,7 @@ async def _run_with_trace():
     return function_response_event
 
   async with _instrumentation.record_tool_execution(
-      tool, agent, function_args
+      tool, agent, function_args, invocation_context=invocation_context
   ) as tel_ctx:
     tel_ctx.function_response_event = await _run_with_trace()
     tel_ctx.error_type = detected_error_type
@@ -712,6 +713,7 @@ async def handle_function_calls_live(
       trace_merged_tool_calls(
           response_event_id=merged_event.id,
           function_response_event=merged_event,
+          invocation_context=invocation_context,
       )
   return merged_event
 
@@ -889,7 +891,7 @@ async def _run_with_trace():
     return function_response_event
 
   async with _instrumentation.record_tool_execution(
-      tool, agent, function_args
+      tool, agent, function_args, invocation_context=invocation_context
   ) as tel_ctx:
     tel_ctx.function_response_event = await _run_with_trace()
     tel_ctx.error_type = detected_error_type

src/google/adk/telemetry/__init__.py

@@ -12,13 +12,17 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from .context import ContentCapturingMode
+from .context import TelemetryConfig
 from .tracing import trace_call_llm
 from .tracing import trace_merged_tool_calls
 from .tracing import trace_send_data
 from .tracing import trace_tool_call
 from .tracing import tracer
 
 __all__ = [
+    'ContentCapturingMode',
+    'TelemetryConfig',
     'trace_call_llm',
     'trace_merged_tool_calls',
     'trace_send_data',

src/google/adk/telemetry/_experimental_semconv.py

@@ -19,9 +19,7 @@
 
 from collections.abc import Mapping
 from collections.abc import MutableMapping
-import contextvars
 import json
-import os
 import sys
 from typing import Any
 from typing import Literal
@@ -54,15 +52,17 @@
 GEN_AI_OUTPUT_MESSAGES = 'gen_ai.output.messages'
 GEN_AI_SYSTEM_INSTRUCTIONS = 'gen_ai.system_instructions'
 
+from .context import TelemetryConfig
+
 # Use the import symbol once the minimum OpenTelemetry SDK version is updated to 1.39.0
 # from opentelemetry.semconv._incubating.attributes.gen_ai_attributes import GEN_AI_TOOL_DEFINITIONS
 GEN_AI_TOOL_DEFINITIONS = 'gen_ai.tool.definitions'
 
-OTEL_SEMCONV_STABILITY_OPT_IN = 'OTEL_SEMCONV_STABILITY_OPT_IN'
+# Use the import symbol once the minimum OpenTelemetry SDK version is updated to 1.40.0
+# from opentelemetry.semconv._incubating.attributes.gen_ai_attributes import GEN_AI_USAGE_CACHE_READ_INPUT_TOKENS
+GEN_AI_USAGE_CACHE_READ_INPUT_TOKENS = 'gen_ai.usage.cache_read.input_tokens'
 
-OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT = (
-    'OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT'
-)
+GEN_AI_USAGE_REASONING_OUTPUT_TOKENS = 'gen_ai.usage.reasoning.output_tokens'
 
 FUNCTION_TOOL_DEFINITION_TYPE = 'function'
 
@@ -133,7 +133,8 @@ def _safe_json_serialize_no_whitespaces(obj) -> str:
     obj: The object to serialize.
 
   Returns:
-    The JSON-serialized object string or <non-serializable> if the object cannot be serialized.
+    The JSON-serialized object string or <non-serializable> if the object cannot
+    be serialized.
   """
 
   try:
@@ -148,18 +149,43 @@ def _safe_json_serialize_no_whitespaces(obj) -> str:
     return '<not serializable>'
 
 
-def is_experimental_semconv() -> bool:
-  opt_ins = os.getenv(OTEL_SEMCONV_STABILITY_OPT_IN)
-  if not opt_ins:
-    return False
-  opt_ins_list = [s.strip() for s in opt_ins.split(',')]
-  return 'gen_ai_latest_experimental' in opt_ins_list
+def is_experimental_semconv(
+    telemetry_config: TelemetryConfig | None = None,
+) -> bool:
+  """Returns whether to emit experimental Generative AI semconv attributes.
+
+  Thin wrapper over
+  :attr:`TelemetryConfig.should_use_experimental_genai_semconv`, which owns the
+  precedence ladder (admin lock > per-request field > env var > default).
+
+  Args:
+    telemetry_config: The per-request config, or ``None`` for the env-only path
+      (modeled as an empty :class:`TelemetryConfig`).
+
+  Returns:
+    Whether the experimental GenAI semconv attributes should be emitted.
+  """
+  cfg = telemetry_config if telemetry_config is not None else TelemetryConfig()
+  return cfg.should_use_experimental_genai_semconv
 
 
-def get_content_capturing_mode() -> str:
-  return os.getenv(
-      OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT, ''
-  ).upper()
+def get_content_capturing_mode(
+    telemetry_config: TelemetryConfig | None = None,
+) -> str:
+  """Returns the experimental GenAI semconv content-capturing mode string.
+
+  Thin wrapper over :attr:`TelemetryConfig.content_capturing_mode_value`, which
+  owns the precedence ladder and the legacy env-string coercion.
+
+  Args:
+    telemetry_config: The per-request config, or ``None`` for the env-only path
+      (modeled as an empty :class:`TelemetryConfig`).
+
+  Returns:
+    One of ``''`` / ``'EVENT_ONLY'`` / ``'SPAN_ONLY'`` / ``'SPAN_AND_EVENT'``.
+  """
+  cfg = telemetry_config if telemetry_config is not None else TelemetryConfig()
+  return cfg.content_capturing_mode_value
 
 
 def _model_dump_to_tool_definition(tool: Any) -> dict[str, Any]:
@@ -439,12 +465,11 @@ def set_operation_details_common_attributes(
     operation_details_common_attributes: MutableMapping[str, AttributeValue],
     attributes: Mapping[str, AttributeValue],
     log_only_attributes: Mapping[str, AttributeValue] | None = None,
+    telemetry_config: TelemetryConfig | None = None,
 ) -> None:
   operation_details_common_attributes.update(attributes)
-  if log_only_attributes and get_content_capturing_mode() in (
-      'EVENT_ONLY',
-      'SPAN_AND_EVENT',
-  ):
+  cfg = telemetry_config if telemetry_config is not None else TelemetryConfig()
+  if log_only_attributes and cfg.should_add_content_to_logs:
     operation_details_common_attributes.update(log_only_attributes)
 
 
@@ -499,18 +524,19 @@ def maybe_log_completion_details(
     otel_logger: Logger,
     operation_details_attributes: Mapping[str, AttributeValue],
     operation_details_common_attributes: Mapping[str, AttributeValue],
+    telemetry_config: TelemetryConfig | None = None,
 ):
-  """Logs completion details based on the experimental semantic convention capturing mode."""
+  """Logs completion details based on the experimental semconv capturing mode."""
   if span is None:
     return
 
-  if not is_experimental_semconv():
+  cfg = telemetry_config if telemetry_config is not None else TelemetryConfig()
+  if not cfg.should_use_experimental_genai_semconv:
     return
 
-  capturing_mode = get_content_capturing_mode()
   final_attributes = operation_details_common_attributes
 
-  if capturing_mode in ['EVENT_ONLY', 'SPAN_AND_EVENT']:
+  if cfg.should_add_content_to_logs:
     final_attributes = final_attributes | operation_details_attributes
   else:
     final_attributes = (
@@ -525,7 +551,7 @@ def maybe_log_completion_details(
       )
   )
 
-  if capturing_mode in ['SPAN_ONLY', 'SPAN_AND_EVENT']:
+  if cfg.should_add_content_to_experimental_spans:
     for key, value in operation_details_attributes.items():
       span.set_attribute(key, _safe_json_serialize_no_whitespaces(value))
   else:

src/google/adk/telemetry/_instrumentation.py

@@ -26,9 +26,9 @@
 from opentelemetry import trace
 import opentelemetry.context as context_api
 
+from ..events import event as event_lib
 from . import _metrics
 from . import tracing
-from ..events import event as event_lib
 
 if TYPE_CHECKING:
   from ..agents.base_agent import BaseAgent
@@ -84,9 +84,11 @@ class TelemetryContext:
   def llm_responses(self) -> list[LlmResponse]:
     return self._llm_responses
 
-  def record_llm_response(self, response: LlmResponse) -> None:
+  def record_llm_response(
+      self, invocation_context: InvocationContext, response: LlmResponse
+  ) -> None:
     self._llm_responses.append(response)
-    tracing.trace_inference_result(self.span, response)
+    tracing.trace_inference_result(invocation_context, self.span, response)
 
 
 def _record_agent_metrics(
@@ -143,6 +145,7 @@ async def record_tool_execution(
     tool: BaseTool,
     agent: BaseAgent,
     function_args: dict[str, Any],
+    invocation_context: InvocationContext | None = None,
 ) -> AsyncIterator[TelemetryContext]:
   """Unified context manager for consolidated tool execution telemetry."""
   start_time = time.monotonic()
@@ -167,6 +170,7 @@ async def record_tool_execution(
             args=function_args,
             function_response_event=response_event,
             error=caught_error,
+            invocation_context=invocation_context,
             error_type=tel_ctx.error_type,
         )
   finally: