Add agentops integration

python/composio/tools/toolset.py

@@ -10,6 +10,8 @@
 import os
 import time
 import typing as t
+from functools import wraps
+from importlib.util import find_spec
 
 import typing_extensions as te
 from pydantic import BaseModel
@@ -69,6 +71,28 @@ class ProcessorsType(te.TypedDict):
     """Response processors."""
 
 
+def _check_agentops() -> bool:
+    """Check if AgentOps is installed and initialized."""
+    if find_spec("agentops") is None:
+        return False
+    import agentops  # pylint: disable=import-outside-toplevel
+
+    return agentops.get_api_key() is not None
+
+
+def _record_action_if_available(func: t.Callable) -> t.Callable:
+    @wraps(func)
+    def wrapper(self, *args, **kwargs):
+        if _check_agentops():
+            import agentops  # pylint: disable=import-outside-toplevel
+
+            action_name = str(kwargs.get("action", "unknown_action"))
+            return agentops.record_action(action_name)(func)(self, *args, **kwargs)
+        return func(self, *args, **kwargs)
+
+    return wrapper
+
+
 class ComposioToolSet(WithLogger):
     """Composio toolset."""
 
@@ -486,6 +510,7 @@ def _process_respone(self, action: Action, response: t.Dict) -> t.Dict:
             type_="post",
         )
 
+    @_record_action_if_available
     def execute_action(
         self,
         action: ActionType,