honeyhive.tracer.core.context

Context and baggage management for HoneyHive tracer.

This module provides dynamic context management, baggage operations, and session enrichment capabilities. It uses dynamic logic for flexible context handling and robust state management.

TracerContextInterface

Bases: ABC

Abstract interface for tracer context operations. This ABC defines the required methods that must be implemented by any class that uses TracerContextMixin. Provides explicit type safety and clear contracts.

Note: too-few-public-methods disabled - ABC interface defines only abstract methods, concrete implementations in TracerContextMixin provide public methods.

Source code in src/honeyhive/tracer/core/context.py

class TracerContextInterface(ABC):  # pylint: disable=too-few-public-methods
    """Abstract interface for tracer context operations.
    This ABC defines the required methods that must be implemented by any class
    that uses TracerContextMixin. Provides explicit type safety and clear contracts.

    Note: too-few-public-methods disabled - ABC interface defines only abstract methods,
    concrete implementations in TracerContextMixin provide public methods.
    """

    @abstractmethod
    def _normalize_attribute_key_dynamically(self, key: str) -> str:
        """Normalize attribute key dynamically for OpenTelemetry compatibility.
        Args:
            key: The attribute key to normalize

        Returns:
            Normalized key string
        """

    @abstractmethod
    def _normalize_attribute_value_dynamically(self, value: Any) -> Any:
        """Normalize attribute value dynamically for OpenTelemetry compatibility.

        Args:
            value: The attribute value to normalize

        Returns:
            Normalized value
        """

TracerContextMixin

Bases: TracerContextInterface

Mixin providing dynamic context and baggage management for HoneyHive tracer.

This mixin uses dynamic logic for baggage operations, context propagation, and session enrichment with comprehensive error handling and thread safety.

This mixin requires implementation of TracerContextInterface abstract methods.

Source code in src/honeyhive/tracer/core/context.py

class TracerContextMixin(TracerContextInterface):
    """Mixin providing dynamic context and baggage management for HoneyHive tracer.

    This mixin uses dynamic logic for baggage operations, context propagation,
    and session enrichment with comprehensive error handling and thread safety.

    This mixin requires implementation of TracerContextInterface abstract methods.
    """

    # Type hint for mypy - these attributes will be provided by the composed class
    if TYPE_CHECKING:
        client: Optional[Any]
        _session_id: Optional[str]
        _baggage_lock: Any

    def force_flush(self, timeout_millis: float = 30000) -> bool:
        """Force flush tracer data with dynamic timeout handling.

        Args:
            timeout_millis: Timeout in milliseconds

        Returns:
            True if flush successful, False otherwise
        """
        return force_flush_tracer(self, timeout_millis)

    def shutdown(self) -> None:
        """Shutdown tracer with dynamic cleanup including cache management."""
        # Clean up cache manager first to prevent resource leaks
        if hasattr(self, "_cache_manager") and self._cache_manager:
            try:
                self._cache_manager.close_all()
                safe_log(self, "debug", "Cache manager closed successfully")
            except Exception as e:
                # Graceful degradation - cache cleanup should not break shutdown
                safe_log(
                    self, "warning", f"Error closing cache manager during shutdown: {e}"
                )

        # Proceed with standard tracer shutdown
        shutdown_tracer(self)

    # pylint: disable=too-many-arguments,too-many-positional-arguments
    # Justification: Session enrichment requires multiple optional parameters
    # for comprehensive session data (inputs, outputs, metadata, config, etc.).
    def enrich_session(
        self,
        metadata: Optional[Dict[str, Any]] = None,
        inputs: Optional[Dict[str, Any]] = None,
        outputs: Optional[Dict[str, Any]] = None,
        config: Optional[Dict[str, Any]] = None,
        feedback: Optional[Dict[str, Any]] = None,
        metrics: Optional[Dict[str, Any]] = None,
        user_properties: Optional[Dict[str, Any]] = None,
        session_id: Optional[str] = None,
        **kwargs: Any,
    ) -> None:
        """Enrich current session with dynamic metadata management.

        **PRIMARY PATTERN (v1.0+):** This instance method is the recommended way
        to enrich sessions. It provides explicit tracer reference and works seamlessly
        in multi-instance environments.

        This method uses dynamic logic to update session metadata with
        flexible parameter handling and automatic session detection. Use it to
        add user properties, feedback, metrics, or custom metadata to sessions.

        Args:
            session_id: Optional explicit session ID to enrich.
                        If not provided, uses tracer's current session ID.
                        (Provided for backwards compatibility)
            metadata: Additional metadata for the session
            inputs: Session input data (captured at session start)
            outputs: Session output data (captured at session end)
            config: Configuration data used during session
            feedback: User feedback or evaluation results
            metrics: Performance metrics (latency, token count, etc.)
            user_properties: User-specific properties (user_id, plan, etc.)
                             Automatically prefixed with 'user_properties.'
            **kwargs: Additional dynamic parameters

        Examples:
            Basic session enrichment::

                tracer = HoneyHiveTracer.init(api_key="...", project="...")

                # Enrich with user properties
                tracer.enrich_session(
                    user_properties={"user_id": "user-123", "plan": "premium"}
                )

            Enrichment with feedback and metrics::

                # After processing user request
                tracer.enrich_session(
                    inputs={"query": "What is AI?"},
                    outputs={"response": "AI is..."},
                    feedback={"rating": 5, "helpful": True},
                    metrics={"latency_ms": 250, "tokens": 150}
                )

            Multiple enrichments throughout session::

                # At session start
                tracer.enrich_session(
                    metadata={"source": "web-app"},
                    user_properties={"user_id": "user-456"}
                )

                # During processing
                tracer.enrich_session(
                    metrics={"api_calls": 3}
                )

                # At session end
                tracer.enrich_session(
                    outputs={"final_result": "success"},
                    feedback={"satisfaction": "high"}
                )

        Note:
            **Backwards Compatibility:** This method maintains compatibility
            with v0.2.x signature. The free function ``enrich_session()``
            is also available but will be deprecated in v2.0.
            See :func:`honeyhive.tracer.integration.compatibility.enrich_session`

        See Also:
            - :meth:`enrich_span` - Enrich individual spans with metadata
            - :meth:`session_start` - Start a new session
            - :meth:`session_end` - End current session

        .. versionadded:: 1.0
            Instance method pattern introduced as primary API.
        """
        if not self._can_enrich_session_dynamically():
            return

        try:
            # Build session update parameters dynamically
            # user_properties should be passed directly to API, not merged into metadata
            update_params = self._build_session_update_params_dynamically(
                inputs=inputs,
                outputs=outputs,
                metadata=metadata,
                config=config,
                feedback=feedback,
                metrics=metrics,
                user_properties=user_properties,
                **kwargs,
            )

            # Get target session ID - use explicit session_id if provided
            # (backwards compat). Otherwise fall back to dynamic detection
            target_session_id: Optional[str]
            if session_id:
                target_session_id = session_id
            else:
                target_session_id = self._get_session_id_for_enrichment_dynamically()

            if target_session_id and update_params:
                # Update session via EventsAPI (sessions are events in the backend)
                if self.client is not None and hasattr(self.client, "events"):
                    # Build update data dict with event_id and update params
                    self.client.events.update(
                        data=UpdateEventRequest(
                            event_id=target_session_id,
                            metadata=update_params.get("metadata"),
                            feedback=update_params.get("feedback"),
                            metrics=update_params.get("metrics"),
                            outputs=update_params.get("outputs"),
                            config=update_params.get("config"),
                            user_properties=update_params.get("user_properties"),
                            duration=update_params.get("duration"),
                        )
                    )
                else:
                    safe_log(self, "warning", "Events API not available for update")

                safe_log(
                    self,
                    "debug",
                    "Session enriched successfully",
                    honeyhive_data={
                        "session_id": target_session_id,
                        "update_fields": list(update_params.keys()),
                    },
                )

        except Exception as e:
            safe_log(
                self,
                "error",
                f"Failed to enrich session: {e}",
                honeyhive_data={"error_type": type(e).__name__},
            )

    def session_start(self) -> Optional[str]:
        """Start a new session and return session ID.

        Creates a new session using the tracer's configuration and returns
        the session ID. This provides backward compatibility with the original
        SDK's session_start() method.

        .. note::
            This method stores session_id on the tracer instance, which is NOT
            safe for concurrent requests. For multi-session handling in web
            servers, use :meth:`create_session` instead.

        Returns:
            Session ID if successful, None otherwise

        Example:
            >>> tracer = HoneyHiveTracer(api_key="...", project="...")
            >>> session_id = tracer.session_start()
            >>> print(f"Created session: {session_id}")
        """
        if not self.client:
            safe_log(self, "warning", "No client available for session creation")
            return None

        try:
            # Use existing session creation logic from base class
            if hasattr(self, "_create_session_dynamically"):
                self._create_session_dynamically()  # type: ignore[attr-defined]
                return getattr(self, "_session_id", None)

            # Fallback: create session directly
            safe_log(self, "error", "Session creation method not available")
            return None
        except Exception as e:
            safe_log(
                self,
                "error",
                "Failed to start session",
                honeyhive_data={"error": str(e), "error_type": type(e).__name__},
            )
            return None

    # pylint: disable=too-many-arguments,too-many-positional-arguments
    # Justification: Session creation supports multiple optional parameters
    # for flexibility in different use cases.
    def create_session(
        self,
        session_name: Optional[str] = None,
        session_id: Optional[str] = None,
        inputs: Optional[Dict[str, Any]] = None,
        metadata: Optional[Dict[str, Any]] = None,
        user_properties: Optional[Dict[str, Any]] = None,
        source: Optional[str] = None,
        skip_api_call: bool = False,
    ) -> Optional[str]:
        """Create a new session and set it in the current request context.

        **RECOMMENDED FOR WEB SERVERS:** This method creates a session via the
        API and stores the session_id in OpenTelemetry baggage (ContextVar-based),
        enabling proper request-scoped session isolation. It does NOT modify
        the tracer instance's session_id, making it safe for concurrent requests.

        The session_id is stored in baggage, which means:
        - Each async task/thread gets its own isolated session context
        - The span processor reads from baggage first (priority over instance)
        - No race conditions between concurrent requests

        Args:
            session_name: Name for the session. Auto-generated if not provided.
            session_id: Custom session ID. If provided along with skip_api_call=True,
                       sets this ID in baggage WITHOUT making an API call
                       (bring-your-own-session-id pattern for linking to existing
                       sessions). If skip_api_call=False (default), creates the
                       session via API with this ID.
            inputs: Input data for the session (e.g., user query, request data)
            metadata: Additional metadata for the session
            user_properties: User-specific properties (user_id, plan, etc.)
            source: Source environment override. Uses tracer's source if not provided.
            skip_api_call: If True and session_id is provided, skip API call and just
                          set session_id in baggage. Useful for linking to sessions
                          that were already created. Defaults to False.

        Returns:
            Session ID if successful, None otherwise

        Example:
            FastAPI middleware for per-request sessions::

                from fastapi import FastAPI, Request
                from honeyhive import HoneyHiveTracer, trace

                tracer = HoneyHiveTracer.init(api_key="...", project="my-api")
                app = FastAPI()

                @app.middleware("http")
                async def session_middleware(request: Request, call_next):
                    # Creates session, sets session_id in baggage (not on tracer)
                    session_id = tracer.create_session(
                        session_name=f"api-{request.url.path}",
                        inputs={"method": request.method, "path": str(request.url)}
                    )

                    response = await call_next(request)

                    # enrich_session reads session_id from baggage
                    tracer.enrich_session(outputs={"status_code": response.status_code})
                    return response

                @app.post("/chat")
                @trace(tracer=tracer, event_type="chain")
                async def chat(message: str):
                    # Span automatically uses session_id from baggage
                    return await process_message(message)

        See Also:
            - :meth:`acreate_session` - Async version for async frameworks
            - :meth:`with_session` - Context manager for automatic cleanup
            - :meth:`session_start` - Legacy method (stores on instance, not baggage)

        .. versionadded:: 1.0.0rc8
            Added for multi-session handling with global tracer pattern.
        """
        try:
            # If session_id provided with skip_api_call, just set in baggage
            if session_id and skip_api_call:
                current_ctx = context.get_current()
                new_ctx = baggage.set_baggage("session_id", session_id, current_ctx)
                if session_name:
                    new_ctx = baggage.set_baggage("session_name", session_name, new_ctx)
                context.attach(new_ctx)

                safe_log(
                    self,
                    "info",
                    f"Set provided session_id in baggage (no API call): {session_id}",
                    honeyhive_data={
                        "session_id": session_id,
                        "session_name": session_name,
                        "storage": "baggage",
                        "source": "provided",
                        "api_call": False,
                    },
                )
                return session_id

            # Create session via API
            if not self.client:
                safe_log(
                    self, "warning", "No API client available for session creation"
                )
                return None

            # Build session parameters
            effective_session_name = session_name or f"session-{uuid.uuid4().hex[:8]}"
            effective_source = source or getattr(self, "source_environment", "dev")

            session_params: Dict[str, Any] = {
                "project": getattr(self, "project_name", None),
                "source": effective_source,
                "session_name": effective_session_name,
            }

            # Include customer-provided session_id if specified
            if session_id:
                session_params["session_id"] = session_id

            if inputs:
                session_params["inputs"] = inputs
            if metadata:
                session_params["metadata"] = metadata
            if user_properties:
                session_params["user_properties"] = user_properties

            # Create session via API using the sessions service
            response = self.client.sessions.start(session_params)
            new_session_id = response.session_id

            # Set session_id in baggage (ContextVar-based, request-scoped)
            # CRITICAL: Do NOT set self._session_id - that would break concurrency
            current_ctx = context.get_current()
            new_ctx = baggage.set_baggage("session_id", new_session_id, current_ctx)
            context.attach(new_ctx)

            safe_log(
                self,
                "info",
                f"Created session in baggage: {new_session_id}",
                honeyhive_data={
                    "session_id": new_session_id,
                    "session_name": effective_session_name,
                    "storage": "baggage",
                },
            )

            return new_session_id

        except Exception as e:
            safe_log(
                self,
                "error",
                f"Failed to create session: {e}",
                honeyhive_data={"error_type": type(e).__name__},
            )
            return None

    async def acreate_session(
        self,
        session_name: Optional[str] = None,
        session_id: Optional[str] = None,
        inputs: Optional[Dict[str, Any]] = None,
        metadata: Optional[Dict[str, Any]] = None,
        user_properties: Optional[Dict[str, Any]] = None,
        source: Optional[str] = None,
        skip_api_call: bool = False,
    ) -> Optional[str]:
        """Async version of create_session for async frameworks like FastAPI.

        Creates a session via async API call and stores session_id in baggage.
        This is the recommended method for async web servers.

        Args:
            session_name: Name for the session. Auto-generated if not provided.
            session_id: Custom session ID. If provided along with skip_api_call=True,
                       sets this ID in baggage WITHOUT making an API call.
                       If skip_api_call=False (default), creates session via API
                       with this ID.
            inputs: Input data for the session
            metadata: Additional metadata for the session
            user_properties: User-specific properties
            source: Source environment override
            skip_api_call: If True and session_id is provided, skip API call.

        Returns:
            Session ID if successful, None otherwise

        Example:
            FastAPI async middleware::

                @app.middleware("http")
                async def session_middleware(request: Request, call_next):
                    session_id = await tracer.acreate_session(
                        session_name=f"api-{request.url.path}",
                        inputs={"method": request.method}
                    )
                    response = await call_next(request)
                    tracer.enrich_session(outputs={"status_code": response.status_code})
                    return response

        See Also:
            - :meth:`create_session` - Sync version

        .. versionadded:: 1.0.0rc8
        """
        try:
            # If session_id provided with skip_api_call, just set in baggage
            if session_id and skip_api_call:
                current_ctx = context.get_current()
                new_ctx = baggage.set_baggage("session_id", session_id, current_ctx)
                if session_name:
                    new_ctx = baggage.set_baggage("session_name", session_name, new_ctx)
                context.attach(new_ctx)

                safe_log(
                    self,
                    "info",
                    f"Set provided session_id in baggage (async, no API): {session_id}",
                    honeyhive_data={
                        "session_id": session_id,
                        "session_name": session_name,
                        "storage": "baggage",
                        "source": "provided",
                        "api_call": False,
                    },
                )
                return session_id

            # Create session via API
            if not self.client:
                safe_log(
                    self, "warning", "No API client available for session creation"
                )
                return None

            # Build session parameters
            effective_session_name = session_name or f"session-{uuid.uuid4().hex[:8]}"
            effective_source = source or getattr(self, "source_environment", "dev")

            session_params: Dict[str, Any] = {
                "project": getattr(self, "project_name", None),
                "source": effective_source,
                "session_name": effective_session_name,
            }

            # Include customer-provided session_id if specified
            if session_id:
                session_params["session_id"] = session_id

            if inputs:
                session_params["inputs"] = inputs
            if metadata:
                session_params["metadata"] = metadata
            if user_properties:
                session_params["user_properties"] = user_properties

            # Create session via async API using the sessions service
            response = await self.client.sessions.start_async(session_params)
            new_session_id = response.session_id

            # Set session_id in baggage (ContextVar-based, request-scoped)
            current_ctx = context.get_current()
            new_ctx = baggage.set_baggage("session_id", new_session_id, current_ctx)
            context.attach(new_ctx)

            safe_log(
                self,
                "info",
                f"Created session in baggage (async): {new_session_id}",
                honeyhive_data={
                    "session_id": new_session_id,
                    "session_name": effective_session_name,
                    "storage": "baggage",
                },
            )

            return new_session_id

        except Exception as e:
            safe_log(
                self,
                "error",
                f"Failed to create session (async): {e}",
                honeyhive_data={"error_type": type(e).__name__},
            )
            return None

    @contextmanager
    def with_session(
        self,
        session_name: Optional[str] = None,
        inputs: Optional[Dict[str, Any]] = None,
        metadata: Optional[Dict[str, Any]] = None,
        user_properties: Optional[Dict[str, Any]] = None,
        source: Optional[str] = None,
    ) -> Iterator[Optional[str]]:
        """Context manager that creates a session for the enclosed scope.

        Creates a session and yields the session_id. All spans created within
        the context will use this session. The session context is automatically
        managed via OpenTelemetry baggage.

        Args:
            session_name: Name for the session
            inputs: Input data for the session
            metadata: Additional metadata
            user_properties: User-specific properties
            source: Source environment override

        Yields:
            Session ID if successful, None otherwise

        Example:
            Using with_session for scoped tracing::

                tracer = HoneyHiveTracer.init(api_key="...", project="my-app")

                with tracer.with_session("user-req", inputs={"q": query}) as sid:
                    # All spans here use this session
                    result = process_query(query)
                    tracer.enrich_session(outputs={"result": result})

        See Also:
            - :meth:`create_session` - Direct session creation

        .. versionadded:: 1.0.0rc8
        """
        session_id = self.create_session(
            session_name=session_name,
            inputs=inputs,
            metadata=metadata,
            user_properties=user_properties,
            source=source,
        )
        try:
            yield session_id
        finally:
            # Context cleanup happens automatically when ContextVar scope ends
            # No explicit detach needed - baggage is scoped to this context
            pass

    def _can_enrich_session_dynamically(self) -> bool:
        """Dynamically check if session enrichment is possible."""
        # Check if client with events API is available (for session updates)
        if not self.client or not hasattr(self.client, "events"):
            safe_log(self, "debug", "No session API available for enrichment")
            return False

        if not self._get_session_id_for_enrichment_dynamically():
            safe_log(self, "debug", "No session ID available for enrichment")
            return False

        return True

    def _get_session_id_for_enrichment_dynamically(self) -> Optional[str]:
        """Dynamically get session ID for enrichment operations.

        Priority order (matches span processor behavior):
        1. Baggage session_id (request-scoped, from create_session())
        2. Instance session_id (tracer._session_id, from session_start())

        This order ensures multi-session handling works correctly with
        a global tracer, while maintaining backwards compatibility.
        """
        # Priority 1: Check baggage first (for multi-session / concurrent requests)
        try:
            current_baggage = get_current_baggage()
            baggage_session = current_baggage.get("session_id")
            if baggage_session:
                return baggage_session
        except Exception as e:
            # Graceful degradation - never crash host
            safe_log(
                self,
                "debug",
                "Failed to get session from baggage",
                honeyhive_data={"error_type": type(e).__name__},
            )

        # Priority 2: Fallback to instance session_id (for single-session scripts)
        if self._session_id:
            return str(self._session_id)

        return None

    # pylint: disable=too-many-arguments
    # Justification: Session parameter building requires multiple optional parameters
    # for flexible session update configuration.
    def _build_session_update_params_dynamically(
        self,
        *,
        inputs: Optional[Dict[str, Any]] = None,
        outputs: Optional[Dict[str, Any]] = None,
        metadata: Optional[Dict[str, Any]] = None,
        config: Optional[Dict[str, Any]] = None,
        feedback: Optional[Dict[str, Any]] = None,
        metrics: Optional[Dict[str, Any]] = None,
        user_properties: Optional[Dict[str, Any]] = None,
        **kwargs: Any,
    ) -> Dict[str, Any]:
        """Dynamically build session update parameters.

        Maps parameters to UpdateEventRequest supported fields only.
        Unsupported fields (inputs, unrecognized kwargs) are merged into metadata.

        UpdateEventRequest supports: metadata, feedback, metrics, outputs,
        config, user_properties, duration (see src/honeyhive/api/events.py:45)
        """
        # Fields supported by UpdateEventRequest
        # pylint: disable=invalid-name  # SUPPORTED_FIELDS is semantically a constant
        SUPPORTED_FIELDS = {
            "metadata",
            "feedback",
            "metrics",
            "outputs",
            "config",
            "user_properties",
            "duration",
        }

        # Start with provided metadata (or empty dict)
        merged_metadata = dict(metadata) if metadata else {}

        # Map inputs to metadata (NOT supported by UpdateEventRequest)
        if inputs:
            merged_metadata["inputs"] = inputs
            safe_log(
                self,
                "debug",
                "Mapped 'inputs' to metadata (not supported by UpdateEventRequest)",
            )

        # Map unsupported kwargs to metadata
        unsupported_kwargs = {
            k: v
            for k, v in kwargs.items()
            if k not in SUPPORTED_FIELDS and v is not None
        }
        if unsupported_kwargs:
            merged_metadata.update(unsupported_kwargs)
            safe_log(
                self,
                "debug",
                "Mapped unsupported kwargs to metadata: %s",
                list(unsupported_kwargs.keys()),
            )

        # Build update params with only supported fields
        update_params = {}

        if merged_metadata:
            update_params["metadata"] = merged_metadata

        if outputs:
            update_params["outputs"] = outputs

        if config:
            update_params["config"] = config

        if feedback:
            update_params["feedback"] = feedback

        if metrics:
            update_params["metrics"] = metrics

        if user_properties:
            update_params["user_properties"] = user_properties

        # Handle duration from kwargs if present (supported field)
        if "duration" in kwargs and kwargs["duration"] is not None:
            update_params["duration"] = kwargs["duration"]

        return update_params

    def enrich_span(
        self,
        attributes: Optional[Dict[str, Any]] = None,
        metadata: Optional[Dict[str, Any]] = None,
        metrics: Optional[Dict[str, Any]] = None,
        feedback: Optional[Dict[str, Any]] = None,
        inputs: Optional[Dict[str, Any]] = None,
        outputs: Optional[Dict[str, Any]] = None,
        config: Optional[Dict[str, Any]] = None,
        user_properties: Optional[Dict[str, Any]] = None,
        error: Optional[str] = None,
        event_id: Optional[str] = None,
        update_event_id: Optional[str] = None,
        **kwargs: Any,
    ) -> bool:
        """Enrich current span with dynamic attribute management.

        **PRIMARY PATTERN (v1.0+):** This instance method is the recommended way
        to enrich spans. It provides explicit tracer reference and works seamlessly
        in multi-instance environments.

        This method uses dynamic logic to add attributes to the current span
        with flexible parameter handling and automatic span detection. It enriches
        the currently active span with metadata, metrics, or custom attributes.

        Args:
            attributes: Span attributes to add directly (dict of key-value pairs)
            metadata: Metadata to add (automatically prefixed with
                'honeyhive_metadata.')
            metrics: Metrics to add (automatically prefixed with 'honeyhive_metrics.')
            feedback: Feedback to add (automatically prefixed with
                'honeyhive_feedback.')
            inputs: Inputs to add (automatically prefixed with 'honeyhive_inputs.')
            outputs: Outputs to add (automatically prefixed with 'honeyhive_outputs.')
            config: Config to add (automatically prefixed with 'honeyhive_config.')
            user_properties: User properties to add (automatically prefixed with
                'honeyhive_user_properties.' for spans)
            error: Error message (stored as 'honeyhive_error')
            event_id: If provided, update an existing event with this ID
                via PUT /events API instead of enriching the current span
            update_event_id: Event ID to override the default event ID on the span
                (stored as 'honeyhive_event_id' span attribute)
            **kwargs: Additional dynamic attributes (routed to metadata namespace)

        Returns:
            True if enrichment succeeded, False otherwise

        Examples:
            Basic enrichment with metadata::

                from honeyhive import trace
                tracer = HoneyHiveTracer.init(api_key="...", project="...")

                @trace(tracer=tracer, event_type="tool")
                def process_data(input_text):
                    result = transform(input_text)

                    # Enrich with metadata and metrics
                    tracer.enrich_span(
                        metadata={"input": input_text, "result": result},
                        metrics={"processing_time_ms": 150}
                    )

                    return result

            Enrichment with user_properties and metrics::

                tracer.enrich_span(
                    user_properties={"user_id": "user-123", "plan": "premium"},
                    metrics={"score": 0.95, "latency_ms": 150}
                )

        Note:
            For backward compatibility, the free function ``enrich_span()``
            is also available but will be deprecated in v2.0.
            See :func:`honeyhive.tracer.integration.compatibility.enrich_span`

        See Also:
            - :meth:`enrich_session` - Enrich session with metadata
            - :meth:`start_span` - Create and manage spans manually
            - :meth:`trace` - Decorator for automatic span creation

        .. versionadded:: 1.0
            Instance method pattern introduced as primary API.
        """
        try:
            # Use the enrichment core logic which handles reserved parameters correctly
            # Import here to avoid circular dependency
            from ..instrumentation.enrichment import (  # pylint: disable=import-outside-toplevel
                enrich_span_core,
            )

            # enrich_span_core handles both:
            # - update_event_id: Updates an existing event via PUT /events API
            # - event_id: Overrides the default event ID on the span attribute
            result = enrich_span_core(
                attributes=attributes,
                metadata=metadata,
                metrics=metrics,
                feedback=feedback,
                inputs=inputs,
                outputs=outputs,
                config=config,
                error=error,
                event_id=event_id,
                update_event_id=update_event_id,
                tracer_instance=self,
                verbose=False,
                # Handle user_properties specially - for spans, it goes to a namespace
                user_properties=user_properties,
                **kwargs,
            )

            if result.get("success"):
                safe_log(
                    self,
                    "debug",
                    "Span enriched successfully",
                    honeyhive_data={
                        "attribute_count": result.get("attribute_count", 0)
                    },
                )

            return bool(result.get("success", False))

        except Exception as e:
            safe_log(
                self,
                "error",
                f"Failed to enrich span: {e}",
                honeyhive_data={"error_type": type(e).__name__},
            )
            return False

    def _enrich_existing_event(
        self,
        event_id: str,
        metadata: Optional[Dict[str, Any]] = None,
        metrics: Optional[Dict[str, Any]] = None,
        feedback: Optional[Dict[str, Any]] = None,
        inputs: Optional[Dict[str, Any]] = None,
        outputs: Optional[Dict[str, Any]] = None,
        config: Optional[Dict[str, Any]] = None,
        user_properties: Optional[Dict[str, Any]] = None,
        error: Optional[str] = None,
        attributes: Optional[Dict[str, Any]] = None,
        **kwargs: Any,
    ) -> bool:
        """Enrich an existing event by event_id via PUT /events API.

        This method is called when enrich_span() is invoked with an event_id,
        allowing users to update a specific existing event with new enrichment data.

        Args:
            event_id: The ID of the existing event to update.
            metadata: Metadata to add/update on the event.
            metrics: Metrics to add/update on the event.
            feedback: Feedback to add/update on the event.
            inputs: Inputs to add/update on the event.
            outputs: Outputs to add/update on the event.
            config: Config to add/update on the event.
            user_properties: User properties to add/update on the event.
            error: Error message to set on the event.
            attributes: Additional attributes to merge into metadata.
            **kwargs: Additional kwargs to merge into metadata.

        Returns:
            True if the event was successfully updated, False otherwise.
        """
        try:
            # Check if we have a client available
            if not hasattr(self, "client") or self.client is None:
                safe_log(
                    self,
                    "warning",
                    "No API client available to update event by event_id",
                    honeyhive_data={"event_id": event_id},
                )
                return False

            # Build update data for the event
            update_data: Dict[str, Any] = {"event_id": event_id}

            # Add enrichment fields if provided
            if metadata:
                update_data["metadata"] = metadata
            if metrics:
                update_data["metrics"] = metrics
            if feedback:
                update_data["feedback"] = feedback
            if inputs:
                update_data["inputs"] = inputs
            if outputs:
                update_data["outputs"] = outputs
            if config:
                update_data["config"] = config
            if user_properties:
                update_data["user_properties"] = user_properties
            if error:
                update_data["error"] = error

            # Merge attributes and kwargs into metadata
            extra_metadata: Dict[str, Any] = {}
            if attributes:
                extra_metadata.update(attributes)
            if kwargs:
                extra_metadata.update(kwargs)
            if extra_metadata:
                if "metadata" in update_data:
                    update_data["metadata"].update(extra_metadata)
                else:
                    update_data["metadata"] = extra_metadata

            # Call the Events API to update the event
            if hasattr(self.client, "events") and hasattr(self.client.events, "update"):
                self.client.events.update(
                    data=UpdateEventRequest(
                        event_id=event_id,
                        metadata=update_data.get("metadata"),
                        feedback=update_data.get("feedback"),
                        metrics=update_data.get("metrics"),
                        outputs=update_data.get("outputs"),
                        config=update_data.get("config"),
                        user_properties=update_data.get("user_properties"),
                    )
                )
                safe_log(
                    self,
                    "debug",
                    "Successfully updated event via API",
                    honeyhive_data={
                        "event_id": event_id,
                        "updated_fields": list(update_data.keys()),
                    },
                )
                return True
            else:
                safe_log(
                    self,
                    "warning",
                    "Events API update method not available",
                    honeyhive_data={"event_id": event_id},
                )
                return False

        except Exception as e:
            safe_log(
                self,
                "error",
                f"Failed to update event {event_id}: {e}",
                honeyhive_data={"event_id": event_id, "error_type": type(e).__name__},
            )
            return False

    def _get_current_span_dynamically(self) -> Any:
        """Dynamically get the current active span."""
        try:
            return trace.get_current_span()
        except Exception as e:
            # Graceful degradation - never crash host
            safe_log(
                self,
                "debug",
                "Failed to get current span",
                honeyhive_data={"error_type": type(e).__name__},
            )
            return None

    def _build_enrichment_attributes_dynamically(
        self,
        attributes: Optional[Dict[str, Any]] = None,
        metadata: Optional[Dict[str, Any]] = None,
        **kwargs: Any,
    ) -> Dict[str, Any]:
        """Dynamically build enrichment attributes from multiple sources."""
        enrichment_attrs = {}

        # Add direct attributes
        if attributes:
            enrichment_attrs.update(attributes)

        # Add metadata with prefix
        if metadata:
            for key, value in metadata.items():
                prefixed_key = f"honeyhive_metadata.{key}"
                enrichment_attrs[prefixed_key] = value

        # Add kwargs dynamically
        for key, value in kwargs.items():
            if value is not None:
                # Normalize key for OpenTelemetry
                normalized_key = self._normalize_attribute_key_dynamically(key)
                enrichment_attrs[normalized_key] = value

        return enrichment_attrs

    def _apply_attributes_to_span_dynamically(
        self, span: Any, attributes: Dict[str, Any]
    ) -> None:
        """Dynamically apply attributes to span with error handling."""
        for key, value in attributes.items():
            try:
                # Normalize value for OpenTelemetry compatibility
                normalized_value = self._normalize_attribute_value_dynamically(value)
                if normalized_value is not None:
                    span.set_attribute(key, normalized_value)
            except Exception as e:
                safe_log(
                    self,
                    "warning",
                    f"Failed to set span attribute '{key}': {e}",
                    honeyhive_data={"attribute_key": key},
                )

    def get_baggage(self, key: str) -> Optional[str]:
        """Get baggage value using dynamic context access.

        Args:
            key: Baggage key to retrieve

        Returns:
            Baggage value if found, None otherwise
        """
        try:
            # Use dynamic baggage access with error handling
            current_baggage = get_current_baggage()

            # Dynamic key lookup with normalization
            normalized_key = self._normalize_baggage_key_dynamically(key)

            # Try multiple key formats dynamically
            key_variants = [key, normalized_key, key.lower(), key.upper()]

            for variant in key_variants:
                if variant in current_baggage:
                    value = current_baggage[variant]
                    safe_log(
                        self,
                        "debug",
                        f"Retrieved baggage: {key}",
                        honeyhive_data={"key": key, "found_as": variant},
                    )
                    return value

            return None

        except Exception as e:
            safe_log(
                self,
                "warning",
                f"Failed to get baggage '{key}': {e}",
                honeyhive_data={"error_type": type(e).__name__},
            )
            return None

    def _normalize_baggage_key_dynamically(self, key: str) -> str:
        """Dynamically normalize baggage key for consistent access."""
        # Replace common separators with underscores
        normalized = key.replace("-", "_").replace(".", "_").replace(" ", "_")
        return normalized.lower()

    def set_baggage(self, key: str, value: str) -> None:
        """Set baggage value using dynamic context management.

        Args:
            key: Baggage key to set
            value: Baggage value to set
        """
        if not key or value is None:
            return

        try:
            with self._baggage_lock:
                # Dynamic baggage setting with context management
                current_ctx = context.get_current()

                # Normalize key and value dynamically
                normalized_key = self._normalize_baggage_key_dynamically(key)
                normalized_value = str(value) if value is not None else ""

                # Set baggage in current context
                new_ctx = baggage.set_baggage(
                    normalized_key, normalized_value, current_ctx
                )

                # Attach context (implementation depends on usage pattern)
                context.attach(new_ctx)

                safe_log(
                    self,
                    "debug",
                    f"Set baggage: {key}",
                    honeyhive_data={
                        "key": key,
                        "normalized_key": normalized_key,
                        "value_length": len(normalized_value),
                    },
                )

        except Exception as e:
            safe_log(
                self,
                "error",
                f"Failed to set baggage '{key}': {e}",
                honeyhive_data={"error_type": type(e).__name__},
            )

    def inject_context(self, carrier: Dict[str, str]) -> None:
        """Inject current context into carrier using dynamic propagation.

        Args:
            carrier: Dictionary to inject context into
        """
        try:
            # Dynamic context injection with error handling
            if inject_context_into_carrier is not None:
                inject_context_into_carrier(carrier, cast("HoneyHiveTracer", self))
            else:
                safe_log(self, "warning", "Context injection not available")

            safe_log(
                self,
                "debug",
                "Context injected into carrier",
                honeyhive_data={
                    "carrier_keys": list(carrier.keys()),
                    "injection_count": len(carrier),
                },
            )

        except Exception as e:
            safe_log(
                self,
                "error",
                f"Failed to inject context: {e}",
                honeyhive_data={"error_type": type(e).__name__},
            )

    def extract_context(self, carrier: Dict[str, str]) -> Optional["Context"]:
        """Extract context from carrier using dynamic propagation.

        Args:
            carrier: Dictionary to extract context from

        Returns:
            Extracted context if successful, None otherwise
        """
        try:
            # Dynamic context extraction with validation
            if extract_context_from_carrier is not None:
                extracted_context = extract_context_from_carrier(
                    carrier, cast("HoneyHiveTracer", self)
                )
            else:
                extracted_context = None

            if extracted_context:
                safe_log(
                    self,
                    "debug",
                    "Context extracted from carrier",
                    honeyhive_data={
                        "carrier_keys": list(carrier.keys()),
                        "extraction_successful": True,
                    },
                )
                return extracted_context

            safe_log(
                self,
                "debug",
                "No context found in carrier",
                honeyhive_data={"carrier_keys": list(carrier.keys())},
            )
            return None

        except Exception as e:
            safe_log(
                self,
                "error",
                f"Failed to extract context: {e}",
                honeyhive_data={"error_type": type(e).__name__},
            )
            return None

client instance-attribute

client: Optional[Any]

force_flush

force_flush(timeout_millis: float = 30000) -> bool

Force flush tracer data with dynamic timeout handling.

Parameters:

Name	Type	Description	Default
timeout_millis	float	Timeout in milliseconds	30000

Returns:

Type	Description
bool	True if flush successful, False otherwise

Source code in src/honeyhive/tracer/core/context.py

def force_flush(self, timeout_millis: float = 30000) -> bool:
    """Force flush tracer data with dynamic timeout handling.

    Args:
        timeout_millis: Timeout in milliseconds

    Returns:
        True if flush successful, False otherwise
    """
    return force_flush_tracer(self, timeout_millis)

shutdown

shutdown() -> None

Shutdown tracer with dynamic cleanup including cache management.

Source code in src/honeyhive/tracer/core/context.py

def shutdown(self) -> None:
    """Shutdown tracer with dynamic cleanup including cache management."""
    # Clean up cache manager first to prevent resource leaks
    if hasattr(self, "_cache_manager") and self._cache_manager:
        try:
            self._cache_manager.close_all()
            safe_log(self, "debug", "Cache manager closed successfully")
        except Exception as e:
            # Graceful degradation - cache cleanup should not break shutdown
            safe_log(
                self, "warning", f"Error closing cache manager during shutdown: {e}"
            )

    # Proceed with standard tracer shutdown
    shutdown_tracer(self)

enrich_session

enrich_session(
    metadata: Optional[Dict[str, Any]] = None,
    inputs: Optional[Dict[str, Any]] = None,
    outputs: Optional[Dict[str, Any]] = None,
    config: Optional[Dict[str, Any]] = None,
    feedback: Optional[Dict[str, Any]] = None,
    metrics: Optional[Dict[str, Any]] = None,
    user_properties: Optional[Dict[str, Any]] = None,
    session_id: Optional[str] = None,
    **kwargs: Any
) -> None

Enrich current session with dynamic metadata management.

PRIMARY PATTERN (v1.0+): This instance method is the recommended way to enrich sessions. It provides explicit tracer reference and works seamlessly in multi-instance environments.

This method uses dynamic logic to update session metadata with flexible parameter handling and automatic session detection. Use it to add user properties, feedback, metrics, or custom metadata to sessions.

Parameters:

Name	Type	Description	Default
session_id	Optional[str]	Optional explicit session ID to enrich. If not provided, uses tracer's current session ID. (Provided for backwards compatibility)	None
metadata	Optional[Dict[str, Any]]	Additional metadata for the session	None
inputs	Optional[Dict[str, Any]]	Session input data (captured at session start)	None
outputs	Optional[Dict[str, Any]]	Session output data (captured at session end)	None
config	Optional[Dict[str, Any]]	Configuration data used during session	None
feedback	Optional[Dict[str, Any]]	User feedback or evaluation results	None
metrics	Optional[Dict[str, Any]]	Performance metrics (latency, token count, etc.)	None
user_properties	Optional[Dict[str, Any]]	User-specific properties (user_id, plan, etc.) Automatically prefixed with 'user_properties.'	None
**kwargs	Any	Additional dynamic parameters	{}

Examples:

Basic session enrichment::

tracer = HoneyHiveTracer.init(api_key="...", project="...")

# Enrich with user properties
tracer.enrich_session(
    user_properties={"user_id": "user-123", "plan": "premium"}
)

Enrichment with feedback and metrics::

# After processing user request
tracer.enrich_session(
    inputs={"query": "What is AI?"},
    outputs={"response": "AI is..."},
    feedback={"rating": 5, "helpful": True},
    metrics={"latency_ms": 250, "tokens": 150}
)

Multiple enrichments throughout session::

# At session start
tracer.enrich_session(
    metadata={"source": "web-app"},
    user_properties={"user_id": "user-456"}
)

# During processing
tracer.enrich_session(
    metrics={"api_calls": 3}
)

# At session end
tracer.enrich_session(
    outputs={"final_result": "success"},
    feedback={"satisfaction": "high"}
)

Note

Backwards Compatibility: This method maintains compatibility with v0.2.x signature. The free function enrich_session() is also available but will be deprecated in v2.0. See :func:honeyhive.tracer.integration.compatibility.enrich_session

See Also

• :meth:enrich_span - Enrich individual spans with metadata
• :meth:session_start - Start a new session
• :meth:session_end - End current session

.. versionadded:: 1.0 Instance method pattern introduced as primary API.

Source code in src/honeyhive/tracer/core/context.py

def enrich_session(
    self,
    metadata: Optional[Dict[str, Any]] = None,
    inputs: Optional[Dict[str, Any]] = None,
    outputs: Optional[Dict[str, Any]] = None,
    config: Optional[Dict[str, Any]] = None,
    feedback: Optional[Dict[str, Any]] = None,
    metrics: Optional[Dict[str, Any]] = None,
    user_properties: Optional[Dict[str, Any]] = None,
    session_id: Optional[str] = None,
    **kwargs: Any,
) -> None:
    """Enrich current session with dynamic metadata management.

    **PRIMARY PATTERN (v1.0+):** This instance method is the recommended way
    to enrich sessions. It provides explicit tracer reference and works seamlessly
    in multi-instance environments.

    This method uses dynamic logic to update session metadata with
    flexible parameter handling and automatic session detection. Use it to
    add user properties, feedback, metrics, or custom metadata to sessions.

    Args:
        session_id: Optional explicit session ID to enrich.
                    If not provided, uses tracer's current session ID.
                    (Provided for backwards compatibility)
        metadata: Additional metadata for the session
        inputs: Session input data (captured at session start)
        outputs: Session output data (captured at session end)
        config: Configuration data used during session
        feedback: User feedback or evaluation results
        metrics: Performance metrics (latency, token count, etc.)
        user_properties: User-specific properties (user_id, plan, etc.)
                         Automatically prefixed with 'user_properties.'
        **kwargs: Additional dynamic parameters

    Examples:
        Basic session enrichment::

            tracer = HoneyHiveTracer.init(api_key="...", project="...")

            # Enrich with user properties
            tracer.enrich_session(
                user_properties={"user_id": "user-123", "plan": "premium"}
            )

        Enrichment with feedback and metrics::

            # After processing user request
            tracer.enrich_session(
                inputs={"query": "What is AI?"},
                outputs={"response": "AI is..."},
                feedback={"rating": 5, "helpful": True},
                metrics={"latency_ms": 250, "tokens": 150}
            )

        Multiple enrichments throughout session::

            # At session start
            tracer.enrich_session(
                metadata={"source": "web-app"},
                user_properties={"user_id": "user-456"}
            )

            # During processing
            tracer.enrich_session(
                metrics={"api_calls": 3}
            )

            # At session end
            tracer.enrich_session(
                outputs={"final_result": "success"},
                feedback={"satisfaction": "high"}
            )

    Note:
        **Backwards Compatibility:** This method maintains compatibility
        with v0.2.x signature. The free function ``enrich_session()``
        is also available but will be deprecated in v2.0.
        See :func:`honeyhive.tracer.integration.compatibility.enrich_session`

    See Also:
        - :meth:`enrich_span` - Enrich individual spans with metadata
        - :meth:`session_start` - Start a new session
        - :meth:`session_end` - End current session

    .. versionadded:: 1.0
        Instance method pattern introduced as primary API.
    """
    if not self._can_enrich_session_dynamically():
        return

    try:
        # Build session update parameters dynamically
        # user_properties should be passed directly to API, not merged into metadata
        update_params = self._build_session_update_params_dynamically(
            inputs=inputs,
            outputs=outputs,
            metadata=metadata,
            config=config,
            feedback=feedback,
            metrics=metrics,
            user_properties=user_properties,
            **kwargs,
        )

        # Get target session ID - use explicit session_id if provided
        # (backwards compat). Otherwise fall back to dynamic detection
        target_session_id: Optional[str]
        if session_id:
            target_session_id = session_id
        else:
            target_session_id = self._get_session_id_for_enrichment_dynamically()

        if target_session_id and update_params:
            # Update session via EventsAPI (sessions are events in the backend)
            if self.client is not None and hasattr(self.client, "events"):
                # Build update data dict with event_id and update params
                self.client.events.update(
                    data=UpdateEventRequest(
                        event_id=target_session_id,
                        metadata=update_params.get("metadata"),
                        feedback=update_params.get("feedback"),
                        metrics=update_params.get("metrics"),
                        outputs=update_params.get("outputs"),
                        config=update_params.get("config"),
                        user_properties=update_params.get("user_properties"),
                        duration=update_params.get("duration"),
                    )
                )
            else:
                safe_log(self, "warning", "Events API not available for update")

            safe_log(
                self,
                "debug",
                "Session enriched successfully",
                honeyhive_data={
                    "session_id": target_session_id,
                    "update_fields": list(update_params.keys()),
                },
            )

    except Exception as e:
        safe_log(
            self,
            "error",
            f"Failed to enrich session: {e}",
            honeyhive_data={"error_type": type(e).__name__},
        )

session_start

session_start() -> Optional[str]

Start a new session and return session ID.

Creates a new session using the tracer's configuration and returns the session ID. This provides backward compatibility with the original SDK's session_start() method.

.. note:: This method stores session_id on the tracer instance, which is NOT safe for concurrent requests. For multi-session handling in web servers, use :meth:create_session instead.

Returns:

Type	Description
Optional[str]	Session ID if successful, None otherwise

Example

tracer = HoneyHiveTracer(api_key="...", project="...") session_id = tracer.session_start() print(f"Created session: {session_id}")

Source code in src/honeyhive/tracer/core/context.py

def session_start(self) -> Optional[str]:
    """Start a new session and return session ID.

    Creates a new session using the tracer's configuration and returns
    the session ID. This provides backward compatibility with the original
    SDK's session_start() method.

    .. note::
        This method stores session_id on the tracer instance, which is NOT
        safe for concurrent requests. For multi-session handling in web
        servers, use :meth:`create_session` instead.

    Returns:
        Session ID if successful, None otherwise

    Example:
        >>> tracer = HoneyHiveTracer(api_key="...", project="...")
        >>> session_id = tracer.session_start()
        >>> print(f"Created session: {session_id}")
    """
    if not self.client:
        safe_log(self, "warning", "No client available for session creation")
        return None

    try:
        # Use existing session creation logic from base class
        if hasattr(self, "_create_session_dynamically"):
            self._create_session_dynamically()  # type: ignore[attr-defined]
            return getattr(self, "_session_id", None)

        # Fallback: create session directly
        safe_log(self, "error", "Session creation method not available")
        return None
    except Exception as e:
        safe_log(
            self,
            "error",
            "Failed to start session",
            honeyhive_data={"error": str(e), "error_type": type(e).__name__},
        )
        return None

create_session

create_session(
    session_name: Optional[str] = None,
    session_id: Optional[str] = None,
    inputs: Optional[Dict[str, Any]] = None,
    metadata: Optional[Dict[str, Any]] = None,
    user_properties: Optional[Dict[str, Any]] = None,
    source: Optional[str] = None,
    skip_api_call: bool = False,
) -> Optional[str]

Create a new session and set it in the current request context.

RECOMMENDED FOR WEB SERVERS: This method creates a session via the API and stores the session_id in OpenTelemetry baggage (ContextVar-based), enabling proper request-scoped session isolation. It does NOT modify the tracer instance's session_id, making it safe for concurrent requests.

The session_id is stored in baggage, which means: - Each async task/thread gets its own isolated session context - The span processor reads from baggage first (priority over instance) - No race conditions between concurrent requests

Parameters:

Name	Type	Description	Default
session_name	Optional[str]	Name for the session. Auto-generated if not provided.	None
session_id	Optional[str]	Custom session ID. If provided along with skip_api_call=True, sets this ID in baggage WITHOUT making an API call (bring-your-own-session-id pattern for linking to existing sessions). If skip_api_call=False (default), creates the session via API with this ID.	None
inputs	Optional[Dict[str, Any]]	Input data for the session (e.g., user query, request data)	None
metadata	Optional[Dict[str, Any]]	Additional metadata for the session	None
user_properties	Optional[Dict[str, Any]]	User-specific properties (user_id, plan, etc.)	None
source	Optional[str]	Source environment override. Uses tracer's source if not provided.	None
skip_api_call	bool	If True and session_id is provided, skip API call and just set session_id in baggage. Useful for linking to sessions that were already created. Defaults to False.	False

Returns:

Type	Description
Optional[str]	Session ID if successful, None otherwise

Example

FastAPI middleware for per-request sessions::

from fastapi import FastAPI, Request
from honeyhive import HoneyHiveTracer, trace

tracer = HoneyHiveTracer.init(api_key="...", project="my-api")
app = FastAPI()

@app.middleware("http")
async def session_middleware(request: Request, call_next):
    # Creates session, sets session_id in baggage (not on tracer)
    session_id = tracer.create_session(
        session_name=f"api-{request.url.path}",
        inputs={"method": request.method, "path": str(request.url)}
    )

    response = await call_next(request)

    # enrich_session reads session_id from baggage
    tracer.enrich_session(outputs={"status_code": response.status_code})
    return response

@app.post("/chat")
@trace(tracer=tracer, event_type="chain")
async def chat(message: str):
    # Span automatically uses session_id from baggage
    return await process_message(message)

See Also

• :meth:acreate_session - Async version for async frameworks
• :meth:with_session - Context manager for automatic cleanup
• :meth:session_start - Legacy method (stores on instance, not baggage)

.. versionadded:: 1.0.0rc8 Added for multi-session handling with global tracer pattern.

Source code in src/honeyhive/tracer/core/context.py

def create_session(
    self,
    session_name: Optional[str] = None,
    session_id: Optional[str] = None,
    inputs: Optional[Dict[str, Any]] = None,
    metadata: Optional[Dict[str, Any]] = None,
    user_properties: Optional[Dict[str, Any]] = None,
    source: Optional[str] = None,
    skip_api_call: bool = False,
) -> Optional[str]:
    """Create a new session and set it in the current request context.

    **RECOMMENDED FOR WEB SERVERS:** This method creates a session via the
    API and stores the session_id in OpenTelemetry baggage (ContextVar-based),
    enabling proper request-scoped session isolation. It does NOT modify
    the tracer instance's session_id, making it safe for concurrent requests.

    The session_id is stored in baggage, which means:
    - Each async task/thread gets its own isolated session context
    - The span processor reads from baggage first (priority over instance)
    - No race conditions between concurrent requests

    Args:
        session_name: Name for the session. Auto-generated if not provided.
        session_id: Custom session ID. If provided along with skip_api_call=True,
                   sets this ID in baggage WITHOUT making an API call
                   (bring-your-own-session-id pattern for linking to existing
                   sessions). If skip_api_call=False (default), creates the
                   session via API with this ID.
        inputs: Input data for the session (e.g., user query, request data)
        metadata: Additional metadata for the session
        user_properties: User-specific properties (user_id, plan, etc.)
        source: Source environment override. Uses tracer's source if not provided.
        skip_api_call: If True and session_id is provided, skip API call and just
                      set session_id in baggage. Useful for linking to sessions
                      that were already created. Defaults to False.

    Returns:
        Session ID if successful, None otherwise

    Example:
        FastAPI middleware for per-request sessions::

            from fastapi import FastAPI, Request
            from honeyhive import HoneyHiveTracer, trace

            tracer = HoneyHiveTracer.init(api_key="...", project="my-api")
            app = FastAPI()

            @app.middleware("http")
            async def session_middleware(request: Request, call_next):
                # Creates session, sets session_id in baggage (not on tracer)
                session_id = tracer.create_session(
                    session_name=f"api-{request.url.path}",
                    inputs={"method": request.method, "path": str(request.url)}
                )

                response = await call_next(request)

                # enrich_session reads session_id from baggage
                tracer.enrich_session(outputs={"status_code": response.status_code})
                return response

            @app.post("/chat")
            @trace(tracer=tracer, event_type="chain")
            async def chat(message: str):
                # Span automatically uses session_id from baggage
                return await process_message(message)

    See Also:
        - :meth:`acreate_session` - Async version for async frameworks
        - :meth:`with_session` - Context manager for automatic cleanup
        - :meth:`session_start` - Legacy method (stores on instance, not baggage)

    .. versionadded:: 1.0.0rc8
        Added for multi-session handling with global tracer pattern.
    """
    try:
        # If session_id provided with skip_api_call, just set in baggage
        if session_id and skip_api_call:
            current_ctx = context.get_current()
            new_ctx = baggage.set_baggage("session_id", session_id, current_ctx)
            if session_name:
                new_ctx = baggage.set_baggage("session_name", session_name, new_ctx)
            context.attach(new_ctx)

            safe_log(
                self,
                "info",
                f"Set provided session_id in baggage (no API call): {session_id}",
                honeyhive_data={
                    "session_id": session_id,
                    "session_name": session_name,
                    "storage": "baggage",
                    "source": "provided",
                    "api_call": False,
                },
            )
            return session_id

        # Create session via API
        if not self.client:
            safe_log(
                self, "warning", "No API client available for session creation"
            )
            return None

        # Build session parameters
        effective_session_name = session_name or f"session-{uuid.uuid4().hex[:8]}"
        effective_source = source or getattr(self, "source_environment", "dev")

        session_params: Dict[str, Any] = {
            "project": getattr(self, "project_name", None),
            "source": effective_source,
            "session_name": effective_session_name,
        }

        # Include customer-provided session_id if specified
        if session_id:
            session_params["session_id"] = session_id

        if inputs:
            session_params["inputs"] = inputs
        if metadata:
            session_params["metadata"] = metadata
        if user_properties:
            session_params["user_properties"] = user_properties

        # Create session via API using the sessions service
        response = self.client.sessions.start(session_params)
        new_session_id = response.session_id

        # Set session_id in baggage (ContextVar-based, request-scoped)
        # CRITICAL: Do NOT set self._session_id - that would break concurrency
        current_ctx = context.get_current()
        new_ctx = baggage.set_baggage("session_id", new_session_id, current_ctx)
        context.attach(new_ctx)

        safe_log(
            self,
            "info",
            f"Created session in baggage: {new_session_id}",
            honeyhive_data={
                "session_id": new_session_id,
                "session_name": effective_session_name,
                "storage": "baggage",
            },
        )

        return new_session_id

    except Exception as e:
        safe_log(
            self,
            "error",
            f"Failed to create session: {e}",
            honeyhive_data={"error_type": type(e).__name__},
        )
        return None

acreate_session async

acreate_session(
    session_name: Optional[str] = None,
    session_id: Optional[str] = None,
    inputs: Optional[Dict[str, Any]] = None,
    metadata: Optional[Dict[str, Any]] = None,
    user_properties: Optional[Dict[str, Any]] = None,
    source: Optional[str] = None,
    skip_api_call: bool = False,
) -> Optional[str]

Async version of create_session for async frameworks like FastAPI.

Creates a session via async API call and stores session_id in baggage. This is the recommended method for async web servers.

Parameters:

Name	Type	Description	Default
session_name	Optional[str]	Name for the session. Auto-generated if not provided.	None
session_id	Optional[str]	Custom session ID. If provided along with skip_api_call=True, sets this ID in baggage WITHOUT making an API call. If skip_api_call=False (default), creates session via API with this ID.	None
inputs	Optional[Dict[str, Any]]	Input data for the session	None
metadata	Optional[Dict[str, Any]]	Additional metadata for the session	None
user_properties	Optional[Dict[str, Any]]	User-specific properties	None
source	Optional[str]	Source environment override	None
skip_api_call	bool	If True and session_id is provided, skip API call.	False

Returns:

Type	Description
Optional[str]	Session ID if successful, None otherwise

Example

FastAPI async middleware::

@app.middleware("http")
async def session_middleware(request: Request, call_next):
    session_id = await tracer.acreate_session(
        session_name=f"api-{request.url.path}",
        inputs={"method": request.method}
    )
    response = await call_next(request)
    tracer.enrich_session(outputs={"status_code": response.status_code})
    return response

See Also

• :meth:create_session - Sync version

.. versionadded:: 1.0.0rc8

Source code in src/honeyhive/tracer/core/context.py

async def acreate_session(
    self,
    session_name: Optional[str] = None,
    session_id: Optional[str] = None,
    inputs: Optional[Dict[str, Any]] = None,
    metadata: Optional[Dict[str, Any]] = None,
    user_properties: Optional[Dict[str, Any]] = None,
    source: Optional[str] = None,
    skip_api_call: bool = False,
) -> Optional[str]:
    """Async version of create_session for async frameworks like FastAPI.

    Creates a session via async API call and stores session_id in baggage.
    This is the recommended method for async web servers.

    Args:
        session_name: Name for the session. Auto-generated if not provided.
        session_id: Custom session ID. If provided along with skip_api_call=True,
                   sets this ID in baggage WITHOUT making an API call.
                   If skip_api_call=False (default), creates session via API
                   with this ID.
        inputs: Input data for the session
        metadata: Additional metadata for the session
        user_properties: User-specific properties
        source: Source environment override
        skip_api_call: If True and session_id is provided, skip API call.

    Returns:
        Session ID if successful, None otherwise

    Example:
        FastAPI async middleware::

            @app.middleware("http")
            async def session_middleware(request: Request, call_next):
                session_id = await tracer.acreate_session(
                    session_name=f"api-{request.url.path}",
                    inputs={"method": request.method}
                )
                response = await call_next(request)
                tracer.enrich_session(outputs={"status_code": response.status_code})
                return response

    See Also:
        - :meth:`create_session` - Sync version

    .. versionadded:: 1.0.0rc8
    """
    try:
        # If session_id provided with skip_api_call, just set in baggage
        if session_id and skip_api_call:
            current_ctx = context.get_current()
            new_ctx = baggage.set_baggage("session_id", session_id, current_ctx)
            if session_name:
                new_ctx = baggage.set_baggage("session_name", session_name, new_ctx)
            context.attach(new_ctx)

            safe_log(
                self,
                "info",
                f"Set provided session_id in baggage (async, no API): {session_id}",
                honeyhive_data={
                    "session_id": session_id,
                    "session_name": session_name,
                    "storage": "baggage",
                    "source": "provided",
                    "api_call": False,
                },
            )
            return session_id

        # Create session via API
        if not self.client:
            safe_log(
                self, "warning", "No API client available for session creation"
            )
            return None

        # Build session parameters
        effective_session_name = session_name or f"session-{uuid.uuid4().hex[:8]}"
        effective_source = source or getattr(self, "source_environment", "dev")

        session_params: Dict[str, Any] = {
            "project": getattr(self, "project_name", None),
            "source": effective_source,
            "session_name": effective_session_name,
        }

        # Include customer-provided session_id if specified
        if session_id:
            session_params["session_id"] = session_id

        if inputs:
            session_params["inputs"] = inputs
        if metadata:
            session_params["metadata"] = metadata
        if user_properties:
            session_params["user_properties"] = user_properties

        # Create session via async API using the sessions service
        response = await self.client.sessions.start_async(session_params)
        new_session_id = response.session_id

        # Set session_id in baggage (ContextVar-based, request-scoped)
        current_ctx = context.get_current()
        new_ctx = baggage.set_baggage("session_id", new_session_id, current_ctx)
        context.attach(new_ctx)

        safe_log(
            self,
            "info",
            f"Created session in baggage (async): {new_session_id}",
            honeyhive_data={
                "session_id": new_session_id,
                "session_name": effective_session_name,
                "storage": "baggage",
            },
        )

        return new_session_id

    except Exception as e:
        safe_log(
            self,
            "error",
            f"Failed to create session (async): {e}",
            honeyhive_data={"error_type": type(e).__name__},
        )
        return None

with_session

with_session(
    session_name: Optional[str] = None,
    inputs: Optional[Dict[str, Any]] = None,
    metadata: Optional[Dict[str, Any]] = None,
    user_properties: Optional[Dict[str, Any]] = None,
    source: Optional[str] = None,
) -> Iterator[Optional[str]]

Context manager that creates a session for the enclosed scope.

Creates a session and yields the session_id. All spans created within the context will use this session. The session context is automatically managed via OpenTelemetry baggage.

Parameters:

Name	Type	Description	Default
session_name	Optional[str]	Name for the session	None
inputs	Optional[Dict[str, Any]]	Input data for the session	None
metadata	Optional[Dict[str, Any]]	Additional metadata	None
user_properties	Optional[Dict[str, Any]]	User-specific properties	None
source	Optional[str]	Source environment override	None

Yields:

Type	Description
Optional[str]	Session ID if successful, None otherwise

Example

Using with_session for scoped tracing::

tracer = HoneyHiveTracer.init(api_key="...", project="my-app")

with tracer.with_session("user-req", inputs={"q": query}) as sid:
    # All spans here use this session
    result = process_query(query)
    tracer.enrich_session(outputs={"result": result})

See Also

• :meth:create_session - Direct session creation

.. versionadded:: 1.0.0rc8

Source code in src/honeyhive/tracer/core/context.py

@contextmanager
def with_session(
    self,
    session_name: Optional[str] = None,
    inputs: Optional[Dict[str, Any]] = None,
    metadata: Optional[Dict[str, Any]] = None,
    user_properties: Optional[Dict[str, Any]] = None,
    source: Optional[str] = None,
) -> Iterator[Optional[str]]:
    """Context manager that creates a session for the enclosed scope.

    Creates a session and yields the session_id. All spans created within
    the context will use this session. The session context is automatically
    managed via OpenTelemetry baggage.

    Args:
        session_name: Name for the session
        inputs: Input data for the session
        metadata: Additional metadata
        user_properties: User-specific properties
        source: Source environment override

    Yields:
        Session ID if successful, None otherwise

    Example:
        Using with_session for scoped tracing::

            tracer = HoneyHiveTracer.init(api_key="...", project="my-app")

            with tracer.with_session("user-req", inputs={"q": query}) as sid:
                # All spans here use this session
                result = process_query(query)
                tracer.enrich_session(outputs={"result": result})

    See Also:
        - :meth:`create_session` - Direct session creation

    .. versionadded:: 1.0.0rc8
    """
    session_id = self.create_session(
        session_name=session_name,
        inputs=inputs,
        metadata=metadata,
        user_properties=user_properties,
        source=source,
    )
    try:
        yield session_id
    finally:
        # Context cleanup happens automatically when ContextVar scope ends
        # No explicit detach needed - baggage is scoped to this context
        pass

enrich_span

enrich_span(
    attributes: Optional[Dict[str, Any]] = None,
    metadata: Optional[Dict[str, Any]] = None,
    metrics: Optional[Dict[str, Any]] = None,
    feedback: Optional[Dict[str, Any]] = None,
    inputs: Optional[Dict[str, Any]] = None,
    outputs: Optional[Dict[str, Any]] = None,
    config: Optional[Dict[str, Any]] = None,
    user_properties: Optional[Dict[str, Any]] = None,
    error: Optional[str] = None,
    event_id: Optional[str] = None,
    update_event_id: Optional[str] = None,
    **kwargs: Any
) -> bool

Enrich current span with dynamic attribute management.

PRIMARY PATTERN (v1.0+): This instance method is the recommended way to enrich spans. It provides explicit tracer reference and works seamlessly in multi-instance environments.

This method uses dynamic logic to add attributes to the current span with flexible parameter handling and automatic span detection. It enriches the currently active span with metadata, metrics, or custom attributes.

Parameters:

Name	Type	Description	Default
attributes	Optional[Dict[str, Any]]	Span attributes to add directly (dict of key-value pairs)	None
metadata	Optional[Dict[str, Any]]	Metadata to add (automatically prefixed with 'honeyhive_metadata.')	None
metrics	Optional[Dict[str, Any]]	Metrics to add (automatically prefixed with 'honeyhive_metrics.')	None
feedback	Optional[Dict[str, Any]]	Feedback to add (automatically prefixed with 'honeyhive_feedback.')	None
inputs	Optional[Dict[str, Any]]	Inputs to add (automatically prefixed with 'honeyhive_inputs.')	None
outputs	Optional[Dict[str, Any]]	Outputs to add (automatically prefixed with 'honeyhive_outputs.')	None
config	Optional[Dict[str, Any]]	Config to add (automatically prefixed with 'honeyhive_config.')	None
user_properties	Optional[Dict[str, Any]]	User properties to add (automatically prefixed with 'honeyhive_user_properties.' for spans)	None
error	Optional[str]	Error message (stored as 'honeyhive_error')	None
event_id	Optional[str]	If provided, update an existing event with this ID via PUT /events API instead of enriching the current span	None
update_event_id	Optional[str]	Event ID to override the default event ID on the span (stored as 'honeyhive_event_id' span attribute)	None
**kwargs	Any	Additional dynamic attributes (routed to metadata namespace)	{}

Returns:

Type	Description
bool	True if enrichment succeeded, False otherwise

Examples:

Basic enrichment with metadata::

from honeyhive import trace
tracer = HoneyHiveTracer.init(api_key="...", project="...")

@trace(tracer=tracer, event_type="tool")
def process_data(input_text):
    result = transform(input_text)

    # Enrich with metadata and metrics
    tracer.enrich_span(
        metadata={"input": input_text, "result": result},
        metrics={"processing_time_ms": 150}
    )

    return result

Enrichment with user_properties and metrics::

tracer.enrich_span(
    user_properties={"user_id": "user-123", "plan": "premium"},
    metrics={"score": 0.95, "latency_ms": 150}
)

Note

For backward compatibility, the free function enrich_span() is also available but will be deprecated in v2.0. See :func:honeyhive.tracer.integration.compatibility.enrich_span

See Also

• :meth:enrich_session - Enrich session with metadata
• :meth:start_span - Create and manage spans manually
• :meth:trace - Decorator for automatic span creation

.. versionadded:: 1.0 Instance method pattern introduced as primary API.

Source code in src/honeyhive/tracer/core/context.py

def enrich_span(
    self,
    attributes: Optional[Dict[str, Any]] = None,
    metadata: Optional[Dict[str, Any]] = None,
    metrics: Optional[Dict[str, Any]] = None,
    feedback: Optional[Dict[str, Any]] = None,
    inputs: Optional[Dict[str, Any]] = None,
    outputs: Optional[Dict[str, Any]] = None,
    config: Optional[Dict[str, Any]] = None,
    user_properties: Optional[Dict[str, Any]] = None,
    error: Optional[str] = None,
    event_id: Optional[str] = None,
    update_event_id: Optional[str] = None,
    **kwargs: Any,
) -> bool:
    """Enrich current span with dynamic attribute management.

    **PRIMARY PATTERN (v1.0+):** This instance method is the recommended way
    to enrich spans. It provides explicit tracer reference and works seamlessly
    in multi-instance environments.

    This method uses dynamic logic to add attributes to the current span
    with flexible parameter handling and automatic span detection. It enriches
    the currently active span with metadata, metrics, or custom attributes.

    Args:
        attributes: Span attributes to add directly (dict of key-value pairs)
        metadata: Metadata to add (automatically prefixed with
            'honeyhive_metadata.')
        metrics: Metrics to add (automatically prefixed with 'honeyhive_metrics.')
        feedback: Feedback to add (automatically prefixed with
            'honeyhive_feedback.')
        inputs: Inputs to add (automatically prefixed with 'honeyhive_inputs.')
        outputs: Outputs to add (automatically prefixed with 'honeyhive_outputs.')
        config: Config to add (automatically prefixed with 'honeyhive_config.')
        user_properties: User properties to add (automatically prefixed with
            'honeyhive_user_properties.' for spans)
        error: Error message (stored as 'honeyhive_error')
        event_id: If provided, update an existing event with this ID
            via PUT /events API instead of enriching the current span
        update_event_id: Event ID to override the default event ID on the span
            (stored as 'honeyhive_event_id' span attribute)
        **kwargs: Additional dynamic attributes (routed to metadata namespace)

    Returns:
        True if enrichment succeeded, False otherwise

    Examples:
        Basic enrichment with metadata::

            from honeyhive import trace
            tracer = HoneyHiveTracer.init(api_key="...", project="...")

            @trace(tracer=tracer, event_type="tool")
            def process_data(input_text):
                result = transform(input_text)

                # Enrich with metadata and metrics
                tracer.enrich_span(
                    metadata={"input": input_text, "result": result},
                    metrics={"processing_time_ms": 150}
                )

                return result

        Enrichment with user_properties and metrics::

            tracer.enrich_span(
                user_properties={"user_id": "user-123", "plan": "premium"},
                metrics={"score": 0.95, "latency_ms": 150}
            )

    Note:
        For backward compatibility, the free function ``enrich_span()``
        is also available but will be deprecated in v2.0.
        See :func:`honeyhive.tracer.integration.compatibility.enrich_span`

    See Also:
        - :meth:`enrich_session` - Enrich session with metadata
        - :meth:`start_span` - Create and manage spans manually
        - :meth:`trace` - Decorator for automatic span creation

    .. versionadded:: 1.0
        Instance method pattern introduced as primary API.
    """
    try:
        # Use the enrichment core logic which handles reserved parameters correctly
        # Import here to avoid circular dependency
        from ..instrumentation.enrichment import (  # pylint: disable=import-outside-toplevel
            enrich_span_core,
        )

        # enrich_span_core handles both:
        # - update_event_id: Updates an existing event via PUT /events API
        # - event_id: Overrides the default event ID on the span attribute
        result = enrich_span_core(
            attributes=attributes,
            metadata=metadata,
            metrics=metrics,
            feedback=feedback,
            inputs=inputs,
            outputs=outputs,
            config=config,
            error=error,
            event_id=event_id,
            update_event_id=update_event_id,
            tracer_instance=self,
            verbose=False,
            # Handle user_properties specially - for spans, it goes to a namespace
            user_properties=user_properties,
            **kwargs,
        )

        if result.get("success"):
            safe_log(
                self,
                "debug",
                "Span enriched successfully",
                honeyhive_data={
                    "attribute_count": result.get("attribute_count", 0)
                },
            )

        return bool(result.get("success", False))

    except Exception as e:
        safe_log(
            self,
            "error",
            f"Failed to enrich span: {e}",
            honeyhive_data={"error_type": type(e).__name__},
        )
        return False

get_baggage

get_baggage(key: str) -> Optional[str]

Get baggage value using dynamic context access.

Parameters:

Name	Type	Description	Default
key	str	Baggage key to retrieve	required

Returns:

Type	Description
Optional[str]	Baggage value if found, None otherwise

Source code in src/honeyhive/tracer/core/context.py

def get_baggage(self, key: str) -> Optional[str]:
    """Get baggage value using dynamic context access.

    Args:
        key: Baggage key to retrieve

    Returns:
        Baggage value if found, None otherwise
    """
    try:
        # Use dynamic baggage access with error handling
        current_baggage = get_current_baggage()

        # Dynamic key lookup with normalization
        normalized_key = self._normalize_baggage_key_dynamically(key)

        # Try multiple key formats dynamically
        key_variants = [key, normalized_key, key.lower(), key.upper()]

        for variant in key_variants:
            if variant in current_baggage:
                value = current_baggage[variant]
                safe_log(
                    self,
                    "debug",
                    f"Retrieved baggage: {key}",
                    honeyhive_data={"key": key, "found_as": variant},
                )
                return value

        return None

    except Exception as e:
        safe_log(
            self,
            "warning",
            f"Failed to get baggage '{key}': {e}",
            honeyhive_data={"error_type": type(e).__name__},
        )
        return None

set_baggage

set_baggage(key: str, value: str) -> None

Set baggage value using dynamic context management.

Parameters:

Name	Type	Description	Default
key	str	Baggage key to set	required
value	str	Baggage value to set	required

Source code in src/honeyhive/tracer/core/context.py

def set_baggage(self, key: str, value: str) -> None:
    """Set baggage value using dynamic context management.

    Args:
        key: Baggage key to set
        value: Baggage value to set
    """
    if not key or value is None:
        return

    try:
        with self._baggage_lock:
            # Dynamic baggage setting with context management
            current_ctx = context.get_current()

            # Normalize key and value dynamically
            normalized_key = self._normalize_baggage_key_dynamically(key)
            normalized_value = str(value) if value is not None else ""

            # Set baggage in current context
            new_ctx = baggage.set_baggage(
                normalized_key, normalized_value, current_ctx
            )

            # Attach context (implementation depends on usage pattern)
            context.attach(new_ctx)

            safe_log(
                self,
                "debug",
                f"Set baggage: {key}",
                honeyhive_data={
                    "key": key,
                    "normalized_key": normalized_key,
                    "value_length": len(normalized_value),
                },
            )

    except Exception as e:
        safe_log(
            self,
            "error",
            f"Failed to set baggage '{key}': {e}",
            honeyhive_data={"error_type": type(e).__name__},
        )

inject_context

inject_context(carrier: Dict[str, str]) -> None

Inject current context into carrier using dynamic propagation.

Parameters:

Name	Type	Description	Default
carrier	Dict[str, str]	Dictionary to inject context into	required

Source code in src/honeyhive/tracer/core/context.py

def inject_context(self, carrier: Dict[str, str]) -> None:
    """Inject current context into carrier using dynamic propagation.

    Args:
        carrier: Dictionary to inject context into
    """
    try:
        # Dynamic context injection with error handling
        if inject_context_into_carrier is not None:
            inject_context_into_carrier(carrier, cast("HoneyHiveTracer", self))
        else:
            safe_log(self, "warning", "Context injection not available")

        safe_log(
            self,
            "debug",
            "Context injected into carrier",
            honeyhive_data={
                "carrier_keys": list(carrier.keys()),
                "injection_count": len(carrier),
            },
        )

    except Exception as e:
        safe_log(
            self,
            "error",
            f"Failed to inject context: {e}",
            honeyhive_data={"error_type": type(e).__name__},
        )

extract_context

extract_context(
    carrier: Dict[str, str],
) -> Optional[Context]

Extract context from carrier using dynamic propagation.

Parameters:

Name	Type	Description	Default
carrier	Dict[str, str]	Dictionary to extract context from	required

Returns:

Type	Description
Optional[Context]	Extracted context if successful, None otherwise

Source code in src/honeyhive/tracer/core/context.py

def extract_context(self, carrier: Dict[str, str]) -> Optional["Context"]:
    """Extract context from carrier using dynamic propagation.

    Args:
        carrier: Dictionary to extract context from

    Returns:
        Extracted context if successful, None otherwise
    """
    try:
        # Dynamic context extraction with validation
        if extract_context_from_carrier is not None:
            extracted_context = extract_context_from_carrier(
                carrier, cast("HoneyHiveTracer", self)
            )
        else:
            extracted_context = None

        if extracted_context:
            safe_log(
                self,
                "debug",
                "Context extracted from carrier",
                honeyhive_data={
                    "carrier_keys": list(carrier.keys()),
                    "extraction_successful": True,
                },
            )
            return extracted_context

        safe_log(
            self,
            "debug",
            "No context found in carrier",
            honeyhive_data={"carrier_keys": list(carrier.keys())},
        )
        return None

    except Exception as e:
        safe_log(
            self,
            "error",
            f"Failed to extract context: {e}",
            honeyhive_data={"error_type": type(e).__name__},
        )
        return None