"""Attribute propagation utilities for Langfuse OpenTelemetry integration. This module provides the `propagate_attributes` context manager for setting trace-level attributes (user_id, session_id, metadata, environment, etc.) that automatically propagate to all child spans within the context. """ import re from typing import ( Any, Dict, Generator, List, Literal, Mapping, Optional, Tuple, TypedDict, Union, cast, ) from opentelemetry import ( baggage, ) from opentelemetry import ( baggage as otel_baggage_api, ) from opentelemetry import ( context as otel_context_api, ) from opentelemetry import ( trace as otel_trace_api, ) from opentelemetry.context import _RUNTIME_CONTEXT from opentelemetry.util._decorator import ( _AgnosticContextManager, _agnosticcontextmanager, ) from langfuse._client.attributes import LangfuseOtelSpanAttributes from langfuse._client.constants import LANGFUSE_SDK_EXPERIMENT_ENVIRONMENT from langfuse.logger import langfuse_logger from langfuse.model import PromptClient PropagatedKeys = Literal[ "user_id", "session_id", "metadata", "version", "tags", "trace_name", "environment", "prompt_name", "prompt_version", ] InternalPropagatedKeys = Literal[ "experiment_id", "experiment_name", "experiment_metadata", "experiment_dataset_id", "experiment_item_id", "experiment_item_metadata", "experiment_item_root_observation_id", ] propagated_keys: List[Union[PropagatedKeys, InternalPropagatedKeys]] = [ "user_id", "session_id", "metadata", "version", "tags", "trace_name", "environment", "prompt_name", "prompt_version", "experiment_id", "experiment_name", "experiment_metadata", "experiment_dataset_id", "experiment_item_id", "experiment_item_metadata", "experiment_item_root_observation_id", ] class PropagatedExperimentAttributes(TypedDict): experiment_id: str experiment_name: str experiment_metadata: Optional[Dict[str, str]] experiment_dataset_id: Optional[str] experiment_item_id: str experiment_item_metadata: Optional[Dict[str, str]] experiment_item_root_observation_id: str def _detach_context_token_safely(token: Any) -> None: """Detach a context token without emitting noisy async teardown errors. OpenTelemetry tokens are backed by ``contextvars`` and must be detached in the same execution context where they were attached. Async frameworks can legitimately end spans or unwind context managers in a different task/context, in which case detach raises and the public OpenTelemetry helper logs an error. At that point the observation is already completed, so the mismatch is safe to ignore. """ try: _RUNTIME_CONTEXT.detach(token) except Exception: pass def propagate_attributes( *, user_id: Optional[str] = None, session_id: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None, version: Optional[str] = None, tags: Optional[List[str]] = None, trace_name: Optional[str] = None, environment: Optional[str] = None, prompt: Optional[Union[PromptClient, Mapping[str, Any]]] = None, as_baggage: bool = False, ) -> _AgnosticContextManager[Any]: """Propagate trace-level attributes to all spans created within this context. This context manager sets attributes on the currently active span AND automatically propagates them to all new child spans created within the context. This is the recommended way to set trace-level attributes like user_id, session_id, environment, and metadata dimensions that should be consistently applied across all observations in a trace. This is a module-level function, not a method on the Langfuse client: import it with `from langfuse import propagate_attributes`. **IMPORTANT**: Call this as early as possible within your trace/workflow — ideally wrapping the creation of your root span, or immediately inside it. Only the currently active span and spans created after entering this context will have these attributes. Pre-existing spans will NOT be retroactively updated. **Why this matters**: Langfuse aggregation queries (e.g., total cost by user_id, filtering by session_id) only include observations that have the attribute set. If you call `propagate_attributes` late in your workflow, earlier spans won't be included in aggregations for that attribute. Args: user_id: User identifier to associate with all spans in this context. Must be US-ASCII string, ≤200 characters. Use this to track which user generated each trace and enable e.g. per-user cost/performance analysis. session_id: Session identifier to associate with all spans in this context. Must be US-ASCII string, ≤200 characters. Use this to group related traces within a user session (e.g., a conversation thread, multi-turn interaction). metadata: Additional key-value metadata to propagate to all spans. - Keys must be US-ASCII strings - Values are coerced to strings - Coerced values must be ≤200 characters - Use for dimensions like internal correlating identifiers - AVOID: large payloads or sensitive data version: Version identfier for parts of your application that are independently versioned, e.g. agents tags: List of tags to categorize the group of observations trace_name: Name to assign to the trace. Must be US-ASCII string, ≤200 characters. Use this to set a consistent trace name for all spans created within this context. prompt: Langfuse prompt to link to generations created within this context. Accepts a `PromptClient` returned by `langfuse.get_prompt(...)` or any object/dict exposing `name` (string) and `version` (integer) — e.g. `{"name": "my-prompt", "version": 3}`. This is the recommended way to link prompts to generations emitted by auto-instrumentation libraries (e.g. LiteLLM's `langfuse_otel`, OpenAI Agents SDK, OpenInference) where you don't create the generation via the Langfuse SDK yourself. The prompt link is only applied to generation-type observations by the Langfuse backend. Fallback prompts are never linked. An explicit `prompt` passed to `start_observation` / `update_current_generation` takes precedence over the propagated one. environment: Langfuse environment to assign to spans created in this context. Must be a lowercase alphanumeric string with optional hyphens or underscores, must be ≤40 characters, and must not start with "langfuse". This maps to the first-class `langfuse.environment` attribute, not to trace metadata. Use it for request-scoped environments, for example when one shared proxy handles calls from dev, staging, qa, and prod. A propagated environment takes precedence over the local client default configured via `Langfuse(environment=...)` or `LANGFUSE_TRACING_ENVIRONMENT` for spans created while this propagation context is active. as_baggage: If True, propagates attributes using OpenTelemetry baggage for cross-process/service propagation. **Security warning**: When enabled, attribute values are added to HTTP headers on ALL outbound requests. This includes `environment` as the `langfuse_environment` baggage entry. Only enable if values are safe to transmit via HTTP headers and you need cross-service tracing. Default: False. Returns: Context manager that propagates attributes to all child spans. Example: Basic usage with user and session tracking (note: `propagate_attributes` is a top-level import, not a client method): ```python from langfuse import Langfuse, propagate_attributes langfuse = Langfuse() # Set attributes early: wrap everything inside the root span with langfuse.start_as_current_observation(name="user_workflow") as span: with propagate_attributes( user_id="user_123", session_id="session_abc", environment="production", metadata={"experiment": "variant_a"} ): # All spans created here will have user_id, session_id, environment, and metadata with langfuse.start_as_current_observation(name="llm_call") as llm_span: # This span inherits user_id, session_id, environment, and experiment metadata ... with langfuse.start_as_current_observation( name="completion", as_type="generation" ) as gen: # This span also inherits all attributes ... ``` Prompt linking with auto-instrumented libraries: ```python from langfuse import Langfuse, propagate_attributes langfuse = Langfuse() prompt = langfuse.get_prompt("my-prompt") with propagate_attributes(prompt=prompt): # Generations emitted by auto-instrumentation (LiteLLM langfuse_otel, # OpenAI Agents SDK, OpenInference, ...) within this context are # linked to the prompt version. completion = litellm.completion( model="gpt-4o", messages=prompt.compile(topic="chickens"), ) ``` Late propagation (anti-pattern): ```python with langfuse.start_as_current_observation(name="workflow") as span: # These spans WON'T have user_id early_span = langfuse.start_observation(name="early_work") early_span.end() # Set attributes in the middle with propagate_attributes(user_id="user_123"): # Only spans created AFTER this point will have user_id late_span = langfuse.start_observation(name="late_work") late_span.end() # Result: Aggregations by user_id will miss "early_work" span ``` Cross-service propagation with baggage (advanced): ```python # Service A - originating service with langfuse.start_as_current_observation(name="api_request"): with propagate_attributes( user_id="user_123", session_id="session_abc", environment="staging", as_baggage=True # Propagate via HTTP headers ): # Make HTTP request to Service B response = requests.get("https://service-b.example.com/api") # user_id, session_id, and environment are now in HTTP headers # Service B - downstream service # OpenTelemetry will automatically extract baggage from HTTP headers # and propagate attributes to spans in Service B. If Service B has a local # Langfuse environment configured, the propagated environment wins for # spans created within this context. ``` Note: - **Validation**: Attribute values (user_id, session_id, version, tags, trace_name) must be strings ≤200 characters. Environment must also match Langfuse's environment format: lowercase alphanumeric with optional hyphens or underscores, must be ≤40 characters, and it must not start with "langfuse". Metadata values are coerced to strings before the 200 character limit is applied. Invalid values will be dropped with a warning logged. - **OpenTelemetry**: This uses OpenTelemetry context propagation under the hood, making it compatible with other OTel-instrumented libraries. Raises: No exceptions are raised. Invalid values are logged as warnings and dropped. See also: `Langfuse.start_as_current_observation` (create the root span this wraps), https://langfuse.com/docs/observability/features/sessions, https://langfuse.com/docs/observability/features/users, https://langfuse.com/docs/observability/features/environments """ return _propagate_attributes( user_id=user_id, session_id=session_id, metadata=metadata, version=version, tags=tags, trace_name=trace_name, environment=environment, prompt=prompt, as_baggage=as_baggage, ) @_agnosticcontextmanager def _propagate_attributes( *, user_id: Optional[str] = None, session_id: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None, version: Optional[str] = None, tags: Optional[List[str]] = None, trace_name: Optional[str] = None, environment: Optional[str] = None, prompt: Optional[Union[PromptClient, Mapping[str, Any]]] = None, as_baggage: bool = False, experiment: Optional[PropagatedExperimentAttributes] = None, ) -> Generator[Any, Any, Any]: context = otel_context_api.get_current() current_span = otel_trace_api.get_current_span() propagated_string_attributes: Dict[str, Optional[Union[str, List[str]]]] = { "user_id": user_id, "session_id": session_id, "version": version, "tags": tags, "trace_name": trace_name, "environment": environment, } prompt_info = _extract_propagated_prompt(prompt) if prompt is not None else None if prompt_info is not None: prompt_name, prompt_version = prompt_info context = _set_propagated_attribute( key="prompt_name", value=prompt_name, context=context, span=current_span, as_baggage=as_baggage, ) context = _set_propagated_attribute( key="prompt_version", value=prompt_version, context=context, span=current_span, as_baggage=as_baggage, ) propagated_metadata_attributes: Dict[str, Optional[Dict[str, Any]]] = { "metadata": metadata, } if experiment: for key, value in experiment.items(): if key in ("experiment_metadata", "experiment_item_metadata"): propagated_metadata_attributes[key] = cast( Optional[Dict[str, str]], value ) else: propagated_string_attributes[key] = cast( Optional[Union[str, List[str]]], value ) # Filter out None values propagated_string_attributes = { k: v for k, v in propagated_string_attributes.items() if v is not None } for key, value in propagated_string_attributes.items(): validated_value = _validate_propagated_value(value=value, key=key) if validated_value is not None: context = _set_propagated_attribute( key=key, value=validated_value, context=context, span=current_span, as_baggage=as_baggage, ) for metadata_key, metadata_value in propagated_metadata_attributes.items(): if metadata_value is None: continue validated_metadata: Dict[str, str] = {} for key, value in metadata_value.items(): coerced_value = value if isinstance(value, str) else str(value) if _validate_string_value(value=coerced_value, key=f"{metadata_key}.{key}"): validated_metadata[key] = coerced_value if validated_metadata: context = _set_propagated_attribute( key=metadata_key, value=validated_metadata, context=context, span=current_span, as_baggage=as_baggage, ) # Activate context, execute, and detach context token = otel_context_api.attach(context=context) try: yield finally: _detach_context_token_safely(token) def _extract_propagated_prompt( prompt: Union[PromptClient, Mapping[str, Any]], ) -> Optional[Tuple[str, int]]: """Extract and validate (name, version) from a prompt-like value. Accepts a PromptClient or any mapping/object exposing `name` and `version`. Returns None (with a warning) if the value is invalid or a fallback prompt. """ if isinstance(prompt, Mapping): name = prompt.get("name") version = prompt.get("version") is_fallback = bool(prompt.get("is_fallback", False)) else: name = getattr(prompt, "name", None) version = getattr(prompt, "version", None) is_fallback = bool(getattr(prompt, "is_fallback", False)) if is_fallback: langfuse_logger.debug( "Propagated prompt is a fallback prompt. Skipping prompt linking." ) return None if not isinstance(name, str) or not name: langfuse_logger.warning( "Propagated 'prompt' has no valid 'name' (non-empty string required). Dropping prompt link." ) return None if isinstance(version, str) and version.isdigit(): version = int(version) if not isinstance(version, int) or isinstance(version, bool): langfuse_logger.warning( "Propagated 'prompt' has no valid 'version' (integer required). Dropping prompt link." ) return None return name, version def _get_propagated_attributes_from_context( context: otel_context_api.Context, ) -> Dict[str, Union[str, int, List[str]]]: propagated_attributes: Dict[str, Union[str, int, List[str]]] = {} # Handle baggage baggage_entries = baggage.get_all(context=context) for baggage_key, baggage_value in baggage_entries.items(): if baggage_key == LANGFUSE_TRACE_ID_BAGGAGE_KEY: continue if baggage_key.startswith(LANGFUSE_BAGGAGE_PREFIX): span_key = _get_span_key_from_baggage_key(baggage_key) if span_key: if span_key == LangfuseOtelSpanAttributes.ENVIRONMENT: validated_environment = _validate_environment_value( value=baggage_value ) if validated_environment is None: continue propagated_attributes[span_key] = validated_environment continue if ( span_key == LangfuseOtelSpanAttributes.OBSERVATION_PROMPT_VERSION and isinstance(baggage_value, str) and baggage_value.isdigit() ): propagated_attributes[span_key] = int(baggage_value) continue propagated_attributes[span_key] = ( baggage_value if isinstance(baggage_value, (str, list)) else str(baggage_value) ) # Handle OTEL context for key in propagated_keys: context_key = _get_propagated_context_key(key) value = otel_context_api.get_value(key=context_key, context=context) if value is None: continue if key == "environment": validated_environment = _validate_environment_value(value=value) if validated_environment is None: continue propagated_attributes[LangfuseOtelSpanAttributes.ENVIRONMENT] = ( validated_environment ) continue if isinstance(value, dict): # Handle metadata span_key = _get_propagated_span_key(key) for k, v in value.items(): propagated_attributes[f"{span_key}.{k}"] = v else: span_key = _get_propagated_span_key(key) propagated_attributes[span_key] = ( value if isinstance(value, (str, int, list)) else str(value) ) if ( LangfuseOtelSpanAttributes.EXPERIMENT_ITEM_ROOT_OBSERVATION_ID in propagated_attributes ): propagated_attributes[LangfuseOtelSpanAttributes.ENVIRONMENT] = ( LANGFUSE_SDK_EXPERIMENT_ENVIRONMENT ) return propagated_attributes def _set_propagated_attribute( *, key: str, value: Union[str, int, List[str], Dict[str, str]], context: otel_context_api.Context, span: otel_trace_api.Span, as_baggage: bool, ) -> otel_context_api.Context: # Get key names context_key = _get_propagated_context_key(key) span_key = _get_propagated_span_key(key) baggage_key = _get_propagated_baggage_key(key) # Merge metadata with previously set metadata keys if isinstance(value, dict): existing_metadata_in_context = cast( dict, otel_context_api.get_value(context_key) or {} ) value = existing_metadata_in_context | value # Merge tags with previously set tags if isinstance(value, list): existing_tags_in_context = cast( list, otel_context_api.get_value(context_key) or [] ) merged_tags = list(existing_tags_in_context) merged_tags.extend(tag for tag in value if tag not in existing_tags_in_context) value = merged_tags # Set in context context = otel_context_api.set_value( key=context_key, value=value, context=context, ) # Set on current span if span is not None and span.is_recording(): if isinstance(value, dict): # Handle metadata for k, v in value.items(): span.set_attribute( key=f"{span_key}.{k}", value=v, ) else: span.set_attribute(key=span_key, value=value) # Set on baggage if as_baggage: if isinstance(value, dict): # Handle metadata for k, v in value.items(): context = otel_baggage_api.set_baggage( name=f"{baggage_key}_{k}", value=v, context=context ) else: context = otel_baggage_api.set_baggage( name=baggage_key, value=str(value) if isinstance(value, int) else value, context=context, ) return context def _validate_propagated_value( *, value: Any, key: str ) -> Optional[Union[str, List[str]]]: if key == "environment": return _validate_environment_value(value=value) if isinstance(value, list): validated_values = [ v for v in value if _validate_string_value(key=key, value=v) ] return validated_values if len(validated_values) > 0 else None if not isinstance(value, str): langfuse_logger.warning( # type: ignore f"Propagated attribute '{key}' value is not a string. Dropping value." ) return None if len(value) > 200: langfuse_logger.warning( f"Propagated attribute '{key}' value is over 200 characters ({len(value)} chars). Dropping value." ) return None return value def _validate_string_value(*, value: str, key: str) -> bool: if not isinstance(value, str): langfuse_logger.warning( # type: ignore f"Propagated attribute '{key}' value is not a string. Dropping value." ) return False if len(value) > 200: langfuse_logger.warning( f"Propagated attribute '{key}' value is over 200 characters ({len(value)} chars). Dropping value." ) return False return True _ENVIRONMENT_VALUE_PATTERN = re.compile(r"^(?!langfuse)[a-z0-9_-]+$") def _validate_environment_value(*, value: Any) -> Optional[str]: key = "environment" if not isinstance(value, str): langfuse_logger.warning( # type: ignore f"Propagated attribute '{key}' value is not a string. Dropping value." ) return None if len(value) > 40: langfuse_logger.warning( f"Propagated attribute '{key}' value is over 40 characters ({len(value)} chars). Dropping value." ) return None if not _ENVIRONMENT_VALUE_PATTERN.fullmatch(value): langfuse_logger.warning( "Propagated attribute 'environment' must be a lowercase alphanumeric " "string with optional hyphens or underscores and must not start with " "'langfuse'. Dropping value." ) return None return value def _get_propagated_context_key(key: str) -> str: return f"langfuse.propagated.{key}" LANGFUSE_BAGGAGE_PREFIX = "langfuse_" LANGFUSE_TRACE_ID_BAGGAGE_KEY = "langfuse_trace_id" def _get_propagated_baggage_key(key: str) -> str: return f"{LANGFUSE_BAGGAGE_PREFIX}{key}" def _get_langfuse_trace_id_from_baggage( context: otel_context_api.Context, ) -> Optional[str]: value = otel_baggage_api.get_baggage( name=LANGFUSE_TRACE_ID_BAGGAGE_KEY, context=context, ) if value is None: return None return str(value).lower() def _set_langfuse_trace_id_in_baggage( *, trace_id: str, context: otel_context_api.Context, ) -> otel_context_api.Context: normalized_trace_id = trace_id.lower() if _get_langfuse_trace_id_from_baggage(context) == normalized_trace_id: return context return otel_baggage_api.set_baggage( name=LANGFUSE_TRACE_ID_BAGGAGE_KEY, value=normalized_trace_id, context=context, ) def _get_span_key_from_baggage_key(key: str) -> Optional[str]: if not key.startswith(LANGFUSE_BAGGAGE_PREFIX): return None # Remove prefix to get the actual key name suffix = key[len(LANGFUSE_BAGGAGE_PREFIX) :] for metadata_key in ("metadata", "experiment_metadata", "experiment_item_metadata"): baggage_metadata_prefix = f"{metadata_key}_" if suffix.startswith(baggage_metadata_prefix): return ( f"{_get_propagated_span_key(metadata_key)}." f"{suffix[len(baggage_metadata_prefix) :]}" ) return _get_propagated_span_key(suffix) def _get_propagated_span_key(key: str) -> str: return { "session_id": LangfuseOtelSpanAttributes.TRACE_SESSION_ID, "user_id": LangfuseOtelSpanAttributes.TRACE_USER_ID, "version": LangfuseOtelSpanAttributes.VERSION, "tags": LangfuseOtelSpanAttributes.TRACE_TAGS, "trace_name": LangfuseOtelSpanAttributes.TRACE_NAME, "environment": LangfuseOtelSpanAttributes.ENVIRONMENT, "metadata": LangfuseOtelSpanAttributes.TRACE_METADATA, "prompt_name": LangfuseOtelSpanAttributes.OBSERVATION_PROMPT_NAME, "prompt_version": LangfuseOtelSpanAttributes.OBSERVATION_PROMPT_VERSION, "experiment_id": LangfuseOtelSpanAttributes.EXPERIMENT_ID, "experiment_name": LangfuseOtelSpanAttributes.EXPERIMENT_NAME, "experiment_metadata": LangfuseOtelSpanAttributes.EXPERIMENT_METADATA, "experiment_dataset_id": LangfuseOtelSpanAttributes.EXPERIMENT_DATASET_ID, "experiment_item_id": LangfuseOtelSpanAttributes.EXPERIMENT_ITEM_ID, "experiment_item_metadata": LangfuseOtelSpanAttributes.EXPERIMENT_ITEM_METADATA, "experiment_item_root_observation_id": LangfuseOtelSpanAttributes.EXPERIMENT_ITEM_ROOT_OBSERVATION_ID, }.get(key) or f"{LangfuseOtelSpanAttributes.TRACE_METADATA}.{key}"