Class Session

agentops/legacy/__init__.py:22–61

This class provides compatibility with CrewAI >= 0.105.0, which uses an event-based integration pattern where it calls methods directly on the Session object: - create_agent(): Called when a CrewAI agent is created - record(): Called when a CrewAI tool is used - end_session():

class Session:
    """
    This class provides compatibility with CrewAI >= 0.105.0, which uses an event-based
    integration pattern where it calls methods directly on the Session object:

    - create_agent(): Called when a CrewAI agent is created
    - record(): Called when a CrewAI tool is used
    - end_session(): Called when a CrewAI run completes
    """

    def __init__(self, trace_context: Optional[TraceContext]):
        self.trace_context = trace_context

    @property
    def span(self) -> Optional[Any]:
        return self.trace_context.span if self.trace_context else None

    @property
    def token(self) -> Optional[Any]:
        return self.trace_context.token if self.trace_context else None

    def __del__(self):
        if self.trace_context and self.trace_context.span and self.trace_context.span.is_recording():
            if not self.trace_context.is_init_trace:
                logger.warning(
                    f"Legacy Session (trace ID: {self.trace_context.span.get_span_context().span_id}) \
was garbage collected but its trace might still be recording. Ensure legacy sessions are ended with end_session()."
                )

    def create_agent(self, name: Optional[str] = None, agent_id: Optional[str] = None, **kwargs: Any):
        """Method for CrewAI >= 0.105.0 compatibility. Currently a no-op."""
        pass

    def record(self, event: Any = None):
        """Method for CrewAI >= 0.105.0 compatibility. Currently a no-op."""
        pass

    def end_session(self, **kwargs: Any):
        """Ends the session for CrewAI >= 0.105.0 compatibility. Calls the global legacy end_session."""
        end_session(session_or_status=self, **kwargs)


@deprecated("Use agentops.start_trace() instead.")