v0.4

agentops/CURRENT_FLOW.md

@@ -0,0 +1,67 @@
+## 1. Current Architecture Flow
+
+```mermaid
+flowchart TB
+    subgraph Client["Client Singleton"]
+        Config["Configuration"]
+        direction TB
+        Client_API["Client API Layer"]
+        LLMTracker["LLM Tracker"]
+    end
+
+    subgraph Sessions["Session Management"]
+        Session["Session Class"]
+        SessionManager["SessionManager"]
+        LogCapture["LogCapture"]
+        SessionAPI["SessionApiClient"]
+    end
+
+    subgraph Events["Event System"]
+        Event["Base Event"]
+        LLMEvent["LLMEvent"]
+        ActionEvent["ActionEvent"]
+        ToolEvent["ToolEvent"]
+        ErrorEvent["ErrorEvent"]
+    end
+
+    subgraph Telemetry["Current Telemetry"]
+        SessionTelemetry["SessionTelemetry"]
+        OTELTracer["OTEL Tracer"]
+        SessionExporter["SessionExporter"]
+        BatchProcessor["BatchSpanProcessor"]
+    end
+
+    subgraph Providers["LLM Providers"]
+        InstrumentedProvider["InstrumentedProvider"]
+        OpenAIProvider["OpenAIProvider"]
+        AnthropicProvider["AnthropicProvider"]
+    end
+
+    %% Client Relationships
+    Client_API -->|initializes| Session
+    Client_API -->|configures| LLMTracker
+    LLMTracker -->|instruments| Providers
+
+    %% Session Direct Dependencies
+    Session -->|creates| SessionManager
+    Session -->|creates| SessionTelemetry
+    Session -->|creates| LogCapture
+    Session -->|owns| SessionAPI
+
+    %% Event Flow
+    InstrumentedProvider -->|creates| LLMEvent
+    InstrumentedProvider -->|requires| Session
+    Session -->|records| Event
+    SessionManager -->|processes| Event
+    SessionTelemetry -->|converts to spans| Event
+
+    %% Telemetry Flow
+    SessionTelemetry -->|uses| OTELTracer
+    OTELTracer -->|sends to| BatchProcessor
+    BatchProcessor -->|exports via| SessionExporter
+    SessionExporter -->|uses| SessionAPI
+
+    %% Problem Areas
+    style Session fill:#f77,stroke:#333
+    style InstrumentedProvider fill:#f77,stroke:#333
+    style SessionTelemetry fill:#f77,stroke:#333

agentops/PROPOSED_FLOW.md

@@ -0,0 +1,92 @@
+```mermaid
+flowchart TB
+    subgraph Client["Client Singleton"]
+        Config["Configuration"]
+        direction TB
+        Client_API["Client API Layer"]
+        InstrumentationManager["Instrumentation Manager"]
+    end
+
+    subgraph Sessions["Session Management"]
+        Session["Session Class"]
+        SessionManager["SessionManager"]
+        LogCapture["LogCapture"]
+        SessionAPI["SessionApiClient"]
+    end
+
+    subgraph Events["Event System"]
+        Event["Base Event"]
+        LLMEvent["LLMEvent"]
+        ActionEvent["ActionEvent"]
+        ToolEvent["ToolEvent"]
+        ErrorEvent["ErrorEvent"]
+    end
+
+    subgraph Telemetry["Enhanced Telemetry"]
+        TelemetryManager["TelemetryManager"]
+        OTELTracer["OTEL Tracer"]
+
+        subgraph Exporters["Exporters"]
+            SessionExporter["SessionExporter"]
+            OTLPExporter["OTLP Exporter"]
+        end
+
+        subgraph Processors["Processors"]
+            BatchProcessor["BatchProcessor"]
+            SamplingProcessor["SamplingProcessor"]
+        end
+    end
+
+    subgraph Providers["LLM Providers"]
+        BaseInstrumentation["BaseInstrumentation"]
+        InstrumentedProvider["InstrumentedProvider"]
+        OpenAIProvider["OpenAIProvider"]
+        AnthropicProvider["AnthropicProvider"]
+    end
+
+    subgraph Context["Context Management"]
+        TraceContext["TraceContext"]
+        ContextPropagation["ContextPropagation"]
+    end
+
+    %% Client Relationships
+    Client_API -->|initializes| Session
+    Client_API -->|configures| InstrumentationManager
+    InstrumentationManager -->|manages| BaseInstrumentation
+
+    %% Session Dependencies
+    Session -->|creates| SessionManager
+    Session -->|uses| TelemetryManager
+    Session -->|creates| LogCapture
+    Session -->|owns| SessionAPI
+
+    %% Event Flow
+    InstrumentedProvider -->|creates| LLMEvent
+    InstrumentedProvider -->|requires| Session
+    Session -->|records| Event
+    SessionManager -->|processes| Event
+
+    %% Telemetry Flow
+    TelemetryManager -->|manages| OTELTracer
+    TelemetryManager -->|uses| TraceContext
+    OTELTracer -->|uses| Processors
+    Processors -->|send to| Exporters
+
+    %% Provider Structure
+    BaseInstrumentation -->|extends| InstrumentedProvider
+    InstrumentedProvider -->|implements| OpenAIProvider
+    InstrumentedProvider -->|implements| AnthropicProvider
+
+    %% Context Flow
+    ContextPropagation -->|enriches| Event
+    TraceContext -->|propagates to| SessionAPI
+
+    %% Highlight New/Changed Components
+    style InstrumentationManager fill:#90EE90,stroke:#333
+    style TelemetryManager fill:#90EE90,stroke:#333
+    style BaseInstrumentation fill:#90EE90,stroke:#333
+    style TraceContext fill:#90EE90,stroke:#333
+    style ContextPropagation fill:#90EE90,stroke:#333
+    style OTLPExporter fill:#90EE90,stroke:#333
+    style SamplingProcessor fill:#90EE90,stroke:#333
+```

agentops/api/base.py

@@ -0,0 +1,77 @@
+from typing import Optional, Dict, Any
+import requests
+from requests.adapters import HTTPAdapter
+from urllib3.util import Retry
+
+from ..exceptions import ApiServerException
+
+
+class ApiClient:
+    """Base class for API communication with connection pooling"""
+
+    _session: Optional[requests.Session] = None
+
+    @classmethod
+    def get_session(cls) -> requests.Session:
+        """Get or create the global session with optimized connection pooling"""
+        if cls._session is None:
+            cls._session = requests.Session()
+
+            # Configure connection pooling
+            adapter = HTTPAdapter(
+                pool_connections=15,
+                pool_maxsize=256,
+                max_retries=Retry(total=3, backoff_factor=0.1, status_forcelist=[500, 502, 503, 504]),
+            )
+
+            # Mount adapter for both HTTP and HTTPS
+            cls._session.mount("http://", adapter)
+            cls._session.mount("https://", adapter)
+
+            # Set default headers
+            cls._session.headers.update(
+                {
+                    "Connection": "keep-alive",
+                    "Keep-Alive": "timeout=10, max=1000",
+                    "Content-Type": "application/json",
+                }
+            )
+
+        return cls._session
+
+    def __init__(self, endpoint: str):
+        self.endpoint = endpoint
+
+    def _prepare_headers(
+        self,
+        api_key: Optional[str] = None,
+        parent_key: Optional[str] = None,
+        jwt: Optional[str] = None,
+        custom_headers: Optional[Dict[str, str]] = None,
+    ) -> Dict[str, str]:
+        """Prepare headers for the request"""
+        headers = {"Content-Type": "application/json; charset=UTF-8", "Accept": "*/*"}
+
+        if api_key:
+            headers["X-Agentops-Api-Key"] = api_key
+
+        if parent_key:
+            headers["X-Agentops-Parent-Key"] = parent_key
+
+        if jwt:
+            headers["Authorization"] = f"Bearer {jwt}"
+
+        if custom_headers:
+            # Don't let custom headers override critical headers
+            safe_headers = custom_headers.copy()
+            for protected in ["Authorization", "X-Agentops-Api-Key", "X-Agentops-Parent-Key"]:
+                safe_headers.pop(protected, None)
+            headers.update(safe_headers)
+
+        return headers
+
+    def post(self, path: str, data: Dict[str, Any], headers: Dict[str, str]) -> requests.Response:
+        """Make POST request"""
+        url = f"{self.endpoint}{path}"
+        session = self.get_session()
+        return session.post(url, json=data, headers=headers)

agentops/api/session.py

@@ -0,0 +1,84 @@
+from typing import Dict, List, Optional, Tuple, Union, Any
+from uuid import UUID
+import requests
+
+from .base import ApiClient
+from ..exceptions import ApiServerException
+from ..helpers import safe_serialize
+from ..log_config import logger
+from ..event import Event
+
+
+class SessionApiClient(ApiClient):
+    """Handles API communication for sessions"""
+
+    def __init__(self, endpoint: str, session_id: UUID, api_key: str, jwt: Optional[str] = None):
+        super().__init__(endpoint)
+        self.session_id = session_id
+        self.api_key = api_key
+        self.jwt = jwt
+
+    def create_session(
+        self, session_data: Dict[str, Any], parent_key: Optional[str] = None
+    ) -> Tuple[bool, Optional[str]]:
+        """Create a new session"""
+        try:
+            headers = self._prepare_headers(
+                api_key=self.api_key, parent_key=parent_key, custom_headers={"X-Session-ID": str(self.session_id)}
+            )
+
+            res = self.post("/v2/create_session", {"session": session_data}, headers)
+            jwt = res.json().get("jwt")
+            return bool(jwt), jwt
+
+        except ApiServerException as e:
+            logger.error(f"Could not create session - {e}")
+            return False, None
+
+    def update_session(self, session_data: Optional[Dict[str, Any]] = None) -> Optional[Dict[str, Any]]:
+        """Update session state"""
+        try:
+            headers = self._prepare_headers(
+                api_key=self.api_key, jwt=self.jwt, custom_headers={"X-Session-ID": str(self.session_id)}
+            )
+
+            res = self.post("/v2/update_session", {"session": session_data or {}}, headers)
+            return res.json()
+
+        except ApiServerException as e:
+            logger.error(f"Could not update session - {e}")
+            return None
+
+    def create_events(self, events: List[Dict[str, Any]]) -> bool:
+        """Send events to API"""
+        try:
+            headers = self._prepare_headers(
+                api_key=self.api_key, jwt=self.jwt, custom_headers={"X-Session-ID": str(self.session_id)}
+            )
+
+            res = self.post("/v2/create_events", {"events": events}, headers)
+            return res.status_code == 200
+
+        except ApiServerException as e:
+            logger.error(f"Could not create events - {e}")
+            return False
+
+    def create_agent(self, name: str, agent_id: str) -> bool:
+        """Create a new agent"""
+        try:
+            headers = self._prepare_headers(
+                api_key=self.api_key, jwt=self.jwt, custom_headers={"X-Session-ID": str(self.session_id)}
+            )
+
+            res = self.post("/v2/create_agent", {"id": agent_id, "name": name}, headers)
+            return res.status_code == 200
+
+        except ApiServerException as e:
+            logger.error(f"Could not create agent - {e}")
+            return False
+
+    def _post(self, path: str, data: Dict[str, Any], headers: Dict[str, str]) -> requests.Response:
+        """Make POST request"""
+        url = f"{self.endpoint}{path}"
+        session = self.get_session()
+        return session.post(url, json=data, headers=headers)

agentops/client.py

@@ -179,7 +179,7 @@ def get_default_tags(self) -> List[str]:
         """
         return list(self._config.default_tags)
 
-    def record(self, event: Union[Event, ErrorEvent]) -> None:
+    def record(self, event: Event) -> None:
         """
         Record an event with the AgentOps service.
 
@@ -198,47 +198,31 @@ def start_session(
         self,
         tags: Optional[List[str]] = None,
         inherited_session_id: Optional[str] = None,
-    ) -> Union[Session, None]:
-        """
-        Start a new session for recording events.
-
-        Args:
-            tags (List[str], optional): Tags that can be used for grouping or sorting later.
-                e.g. ["test_run"].
-            config: (Configuration, optional): Client configuration object
-            inherited_session_id (optional, str): assign session id to match existing Session
-        """
+    ) -> Optional[Session]:
+        """Start a new session"""
         if not self.is_initialized:
-            return
-
-        if inherited_session_id is not None:
-            try:
-                session_id = UUID(inherited_session_id)
-            except ValueError:
-                return logger.warning(f"Invalid session id: {inherited_session_id}")
-        else:
-            session_id = uuid4()
+            return None
 
-        session_tags = self._config.default_tags.copy()
-        if tags is not None:
-            session_tags.update(tags)
+        try:
+            session_id = UUID(inherited_session_id) if inherited_session_id else uuid4()
+        except ValueError:
+            return logger.warning(f"Invalid session id: {inherited_session_id}")
 
+        default_tags = list(self._config.default_tags) if self._config.default_tags else []
         session = Session(
             session_id=session_id,
-            tags=list(session_tags),
-            host_env=self.host_env,
             config=self._config,
+            tags=tags or default_tags,
+            host_env=self.host_env,
         )
 
-        if not session.is_running:
-            return logger.error("Failed to start session")
-
-        if self._pre_init_queue["agents"] and len(self._pre_init_queue["agents"]) > 0:
-            for agent_args in self._pre_init_queue["agents"]:
-                session.create_agent(name=agent_args["name"], agent_id=agent_args["agent_id"])
-            self._pre_init_queue["agents"] = []
+        if session.is_running:
+            # Process any queued agents
+            if self._pre_init_queue["agents"]:
+                for agent_args in self._pre_init_queue["agents"]:
+                    session.create_agent(name=agent_args["name"], agent_id=agent_args["agent_id"])
+                self._pre_init_queue["agents"].clear()
 
-        self._sessions.append(session)
         return session
 
     def end_session(