Logcap

agentops/helpers.py

@@ -1,3 +1,4 @@
+from enum import Enum
 import inspect
 import json
 from datetime import datetime, timezone
@@ -67,6 +68,10 @@ def default(o):
         try:
             if isinstance(o, UUID):
                 return str(o)
+            # Handle Enum types
+            elif isinstance(o, Enum):
+                return o.value
+            # Handle objects with attributes property that's dict-like
             elif hasattr(o, "model_dump_json"):
                 return str(o.model_dump_json())
             elif hasattr(o, "to_json"):

agentops/http_client.py

@@ -112,21 +112,27 @@ def _prepare_headers(
         return headers
 
     @classmethod
-    def post(
+    def _make_request(
         cls,
+        method: str,
         url: str,
-        payload: bytes,
         api_key: Optional[str] = None,
         parent_key: Optional[str] = None,
         jwt: Optional[str] = None,
         header: Optional[Dict[str, str]] = None,
+        payload: Optional[bytes] = None,
     ) -> Response:
-        """Make HTTP POST request using connection pooling"""
+        """Make HTTP request using connection pooling"""
         result = Response()
         try:
             headers = cls._prepare_headers(api_key, parent_key, jwt, header)
             session = cls.get_session()
-            res = session.post(url, data=payload, headers=headers, timeout=20)
+
+            kwargs = {"headers": headers, "timeout": 20}
+            if payload is not None:
+                kwargs["data"] = payload
+
+            res = getattr(session, method.lower())(url, **kwargs)
             result.parse(res)
 
         except requests.exceptions.Timeout:
@@ -168,41 +174,43 @@ def get(
         jwt: Optional[str] = None,
         header: Optional[Dict[str, str]] = None,
     ) -> Response:
-        """Make HTTP GET request using connection pooling"""
-        result = Response()
-        try:
-            headers = cls._prepare_headers(api_key, None, jwt, header)
-            session = cls.get_session()
-            res = session.get(url, headers=headers, timeout=20)
-            result.parse(res)
+        """Make HTTP GET request"""
+        return cls._make_request("GET", url, api_key=api_key, jwt=jwt, header=header)
 
-        except requests.exceptions.Timeout:
-            result.code = 408
-            result.status = HttpStatus.TIMEOUT
-            raise ApiServerException("Could not reach API server - connection timed out")
-        except requests.exceptions.HTTPError as e:
-            try:
-                result.parse(e.response)
-            except Exception:
-                result = Response()
-                result.code = e.response.status_code
-                result.status = Response.get_status(e.response.status_code)
-                result.body = {"error": str(e)}
-                raise ApiServerException(f"HTTPError: {e}")
-        except requests.exceptions.RequestException as e:
-            result.body = {"error": str(e)}
-            raise ApiServerException(f"RequestException: {e}")
+    @classmethod
+    def post(
+        cls,
+        url: str,
+        payload: bytes,
+        api_key: Optional[str] = None,
+        parent_key: Optional[str] = None,
+        jwt: Optional[str] = None,
+        header: Optional[Dict[str, str]] = None,
+    ) -> Response:
+        """Make HTTP POST request"""
+        return cls._make_request(
+            "POST", url, api_key=api_key, parent_key=parent_key, jwt=jwt, header=header, payload=payload
+        )
 
-        if result.code == 401:
-            raise ApiServerException(
-                f"API server: invalid API key: {api_key}. Find your API key at https://app.agentops.ai/settings/projects"
-            )
-        if result.code == 400:
-            if "message" in result.body:
-                raise ApiServerException(f"API server: {result.body['message']}")
-            else:
-                raise ApiServerException(f"API server: {result.body}")
-        if result.code == 500:
-            raise ApiServerException("API server: - internal server error")
+    @classmethod
+    def put(
+        cls,
+        url: str,
+        payload: bytes,
+        api_key: Optional[str] = None,
+        jwt: Optional[str] = None,
+        header: Optional[Dict[str, str]] = None,
+    ) -> Response:
+        """Make HTTP PUT request"""
+        return cls._make_request("PUT", url, api_key=api_key, jwt=jwt, header=header, payload=payload)
 
-        return result
+    @classmethod
+    def delete(
+        cls,
+        url: str,
+        api_key: Optional[str] = None,
+        jwt: Optional[str] = None,
+        header: Optional[Dict[str, str]] = None,
+    ) -> Response:
+        """Make HTTP DELETE request"""
+        return cls._make_request("DELETE", url, api_key=api_key, jwt=jwt, header=header)

agentops/instrumentation.py

@@ -0,0 +1,138 @@
+from __future__ import annotations
+
+import logging
+import sys
+from typing import TYPE_CHECKING, Dict, List, Optional
+from uuid import UUID
+
+from opentelemetry import trace
+from opentelemetry.sdk._logs import LoggerProvider, LoggingHandler
+from opentelemetry.sdk._logs.export import BatchLogRecordProcessor
+from opentelemetry.sdk.resources import SERVICE_NAME, Resource
+from opentelemetry.sdk.trace import SpanProcessor, TracerProvider
+from opentelemetry.sdk.trace.export import BatchSpanProcessor, SpanExporter
+from opentelemetry.sdk.trace.sampling import ParentBased, Sampler, TraceIdRatioBased
+
+if TYPE_CHECKING:
+    from agentops.client import Client
+
+"""
+This module handles OpenTelemetry instrumentation setup for AgentOps sessions.
+
+Each AgentOps session requires its own telemetry setup to:
+1. Track session-specific logs
+2. Export logs to the AgentOps backend
+3. Maintain isolation between different sessions running concurrently
+
+The module provides functions to:
+- Set up logging telemetry components for a new session
+- Clean up telemetry components when a session ends
+"""
+
+# Map of session_id to LoggingHandler
+_session_handlers: Dict[UUID, LoggingHandler] = {}
+
+
+def get_session_handler(session_id: UUID) -> Optional[LoggingHandler]:
+    """Get the logging handler for a specific session.
+
+    Args:
+        session_id: The UUID of the session
+
+    Returns:
+        The session's LoggingHandler if it exists, None otherwise
+    """
+    return _session_handlers.get(session_id)
+
+
+def set_session_handler(session_id: UUID, handler: Optional[LoggingHandler]) -> None:
+    """Set or remove the logging handler for a session.
+
+    Args:
+        session_id: The UUID of the session
+        handler: The handler to set, or None to remove
+    """
+    if handler is None:
+        _session_handlers.pop(session_id, None)
+    else:
+        _session_handlers[session_id] = handler
+
+
+def setup_session_telemetry(session, log_exporter) -> tuple[LoggingHandler, BatchLogRecordProcessor]:
+    """Set up OpenTelemetry logging components for a new session.
+
+    This function creates the necessary components to capture and export logs for a specific session:
+    - A LoggerProvider with session-specific resource attributes
+    - A BatchLogRecordProcessor to batch and export logs
+    - A LoggingHandler to capture logs and forward them to the processor
+
+    Args:
+        session_id: Unique identifier for the session, used to tag telemetry data
+        log_exporter: SessionLogExporter instance that handles sending logs to AgentOps backend
+
+    Returns:
+        Tuple containing:
+        - LoggingHandler: Handler that should be added to the logger
+        - BatchLogRecordProcessor: Processor that batches and exports logs
+    """
+    # Create logging components
+    resource = Resource.create({SERVICE_NAME: f"agentops.session.{session.session_id}"})
+    logger_provider = LoggerProvider(resource=resource)
+
+    # Create processor and handler
+    log_processor = BatchLogRecordProcessor(log_exporter)
+    logger_provider.add_log_record_processor(log_processor)  # Add processor to provider
+
+    from agentops.log_capture import LogCapture
+
+    logcap = LogCapture(
+        session,
+    )
+
+    logcap.start()
+
+    # log_handler = LoggingHandler(
+    #     level=logging.INFO,
+    #     logger_provider=logger_provider,
+    # )
+    #
+    # # Register handler with session
+    # set_session_handler(session_id, log_handler)
+    #
+    # return log_handler, log_processor
+
+
+def cleanup_session_telemetry(log_handler: LoggingHandler, log_processor: BatchLogRecordProcessor) -> None:
+    """Clean up OpenTelemetry logging components when a session ends.
+
+    This function ensures proper cleanup by:
+    1. Removing the handler from the logger
+    2. Closing the handler to free resources
+    3. Flushing any pending logs in the processor
+    4. Shutting down the processor
+
+    Args:
+        log_handler: The session's LoggingHandler to be removed and closed
+        log_processor: The session's BatchLogRecordProcessor to be flushed and shutdown
+
+    Used by:
+        Session.end_session() to clean up logging components when the session ends
+    """
+    from agentops.log_config import logger
+
+    try:
+        # Remove and close handler
+        logger.removeHandler(log_handler)
+        log_handler.close()
+
+        # Remove from session handlers
+        for session_id, handler in list(_session_handlers.items()):
+            if handler is log_handler:
+                set_session_handler(session_id, None)
+                break
+
+        # Shutdown processor
+        log_processor.force_flush(timeout_millis=5000)
+        log_processor.shutdown()
+    except Exception as e:
+        logger.warning(f"Error during logging cleanup: {e}")