Improve UX for action processor (ComposioHQ#709)

Adds a `processors` field on the `execute_action` and `get_tools`
methods, and deprecates the use of `processors` field in the
`ComposioToolSet` constructor.


<!-- ELLIPSIS_HIDDEN -->


----

> [!IMPORTANT]
> Adds `processors` parameter to `execute_action` and `get_tools`,
deprecating its use in `ComposioToolSet` constructor, with tests for new
functionality.
> 
>   - **Behavior**:
> - Adds `processors` parameter to `execute_action` and `get_tools`
methods in `ComposioToolSet` and derived classes.
> - Deprecates `processors` parameter in `ComposioToolSet` constructor.
>   - **Methods**:
> - Updates `execute_action` in `toolset.py` to merge processors if
provided.
> - Updates `get_tools` in `toolset.py` and derived classes to handle
processors.
>   - **Deprecation**:
> - Adds deprecation warning for `processors` in `ComposioToolSet`
constructor.
>   - **Tests**:
> - Adds tests in `test_toolset.py` for new `processors` handling in
`execute_action` and `get_tools` methods.
> 
> <sup>This description was created by </sup>[<img alt="Ellipsis"
src="https://img.shields.io/badge/Ellipsis-blue?color=175173">](https://www.ellipsis.dev?ref=SamparkAI%2Fcomposio&utm_source=github&utm_medium=referral)<sup>
for fe574c6. It will automatically
update as commits are pushed.</sup>


<!-- ELLIPSIS_HIDDEN -->

python/composio/tools/toolset.py

@@ -65,22 +65,28 @@
 P = te.ParamSpec("P")
 
 _KeyType = t.Union[AppType, ActionType]
-_ProcessorType = t.Callable[[t.Dict], t.Dict]
+_CallableType = t.Callable[[t.Dict], t.Dict]
 
 MetadataType = t.Dict[_KeyType, t.Dict]
 ParamType = t.TypeVar("ParamType")
 
+# Enable deprecation warnings
+warnings.simplefilter("always", DeprecationWarning)
+
+
+ProcessorType = te.Literal["pre", "post", "schema"]
+
 
 class ProcessorsType(te.TypedDict):
     """Request and response processors."""
 
-    pre: te.NotRequired[t.Dict[_KeyType, _ProcessorType]]
+    pre: te.NotRequired[t.Dict[_KeyType, _CallableType]]
     """Request processors."""
 
-    post: te.NotRequired[t.Dict[_KeyType, _ProcessorType]]
+    post: te.NotRequired[t.Dict[_KeyType, _CallableType]]
     """Response processors."""
 
-    schema: te.NotRequired[t.Dict[_KeyType, _ProcessorType]]
+    schema: te.NotRequired[t.Dict[_KeyType, _CallableType]]
     """Schema processors"""
 
 
@@ -249,18 +255,24 @@ def _limit_file_search_response(response: t.Dict) -> t.Dict:
             self._api_key = None
             self.logger.debug("`api_key` is not set when initializing toolset.")
 
-        self._processors = (
-            processors
-            if processors is not None
-            else {"post": {}, "pre": {}, "schema": {}}
-        )
+        if processors is not None:
+            warnings.warn(
+                "Setting 'processors' on the ToolSet is deprecated, they should"
+                "be provided to the 'get_tools()' method instead.",
+                DeprecationWarning,
+                stacklevel=2,
+            )
+            self._processors: ProcessorsType = processors
+        else:
+            self._processors = {"post": {}, "pre": {}, "schema": {}}
+
         self._metadata = metadata or {}
         self._workspace_id = workspace_id
         self._workspace_config = workspace_config
         self._local_client = LocalClient()
 
         if len(kwargs) > 0:
-            self.logger.info(f"Extra kwards while initializing toolset: {kwargs}")
+            self.logger.info(f"Extra kwargs while initializing toolset: {kwargs}")
 
         self.logger.debug("Loading local tools")
         load_local_tools()
@@ -540,7 +552,10 @@ def _serialize_execute_params(self, param: ParamType) -> ParamType:
             return [self._serialize_execute_params(p) for p in param]  # type: ignore
 
         if isinstance(param, dict):
-            return {key: self._serialize_execute_params(val) for key, val in param.items()}  # type: ignore
+            return {
+                key: self._serialize_execute_params(val)  # type: ignore
+                for key, val in param.items()
+            }
 
         raise ValueError(
             "Invalid value found for execute parameters"
@@ -567,7 +582,7 @@ def _get_processor(
         self,
         key: _KeyType,
         type_: te.Literal["post", "pre", "schema"],
-    ) -> t.Optional[_ProcessorType]:
+    ) -> t.Optional[_CallableType]:
         """Get processor for given app or action"""
         processor = self._processors.get(type_, {}).get(key)  # type: ignore
         if processor is not None:
@@ -591,6 +606,13 @@ def _process(
                 f" through: {processor.__name__}"
             )
             data = processor(data)
+            # Users may not respect our type annotations and return something that isn't a dict.
+            # If that happens we should show a friendly error message.
+            if not isinstance(data, t.Dict):
+                warnings.warn(
+                    f"Expected {type_}-processor to return 'dict', got {type(data).__name__!r}",
+                    stacklevel=2,
+                )
         return data
 
     def _process_request(self, action: Action, request: t.Dict) -> t.Dict:
@@ -626,6 +648,22 @@ def _process_schema_properties(self, action: Action, properties: t.Dict) -> t.Di
             type_="schema",
         )
 
+    def _merge_processors(self, processors: ProcessorsType) -> None:
+        for processor_type in self._processors.keys():
+            if processor_type not in processors:
+                continue
+
+            processor_type = t.cast(ProcessorType, processor_type)
+            new_processors = processors[processor_type]
+
+            if processor_type in self._processors:
+                existing_processors = self._processors[processor_type]
+            else:
+                existing_processors = {}
+                self._processors[processor_type] = existing_processors
+
+            existing_processors.update(new_processors)
+
     @_record_action_if_available
     def execute_action(
         self,
@@ -635,6 +673,8 @@ def execute_action(
         entity_id: t.Optional[str] = None,
         connected_account_id: t.Optional[str] = None,
         text: t.Optional[str] = None,
+        *,
+        processors: t.Optional[ProcessorsType] = None,
     ) -> t.Dict:
         """
         Execute an action on a given entity.
@@ -649,6 +689,9 @@ def execute_action(
         """
         action = Action(action)
         params = self._serialize_execute_params(param=params)
+        if processors is not None:
+            self._merge_processors(processors)
+
         if not action.is_runtime:
             params = self._process_request(action=action, request=params)
             metadata = self._add_metadata(action=action, metadata=metadata)

python/plugins/camel/composio_camel/toolset.py

@@ -13,6 +13,7 @@
 from composio.constants import DEFAULT_ENTITY_ID
 from composio.tools import ComposioToolSet as BaseComposioToolSet
 from composio.tools.schema import OpenAISchema, SchemaType
+from composio.tools.toolset import ProcessorsType
 
 
 # pylint: enable=E0611
@@ -151,6 +152,8 @@ def get_tools(
         apps: t.Optional[t.Sequence[AppType]] = None,
         tags: t.Optional[t.List[TagType]] = None,
         entity_id: t.Optional[str] = None,
+        *,
+        processors: t.Optional[ProcessorsType] = None,
     ) -> t.List[OpenAIFunction]:
         """
         Get composio tools wrapped as Camel `OpenAIFunction` objects.
@@ -163,6 +166,8 @@ def get_tools(
         :return: Composio tools wrapped as `OpenAIFunction` objects
         """
         self.validate_tools(apps=apps, actions=actions, tags=tags)
+        if processors is not None:
+            self._merge_processors(processors)
         return [
             self._wrap_tool(  # type: ignore
                 t.cast(

python/plugins/claude/composio_claude/toolset.py

@@ -15,6 +15,7 @@
 from composio.constants import DEFAULT_ENTITY_ID
 from composio.tools import ComposioToolSet as BaseComposioToolSet
 from composio.tools.schema import ClaudeSchema, SchemaType
+from composio.tools.toolset import ProcessorsType
 
 
 class ComposioToolSet(
@@ -94,6 +95,8 @@ def get_tools(
         actions: t.Optional[t.Sequence[ActionType]] = None,
         apps: t.Optional[t.Sequence[AppType]] = None,
         tags: t.Optional[t.List[TagType]] = None,
+        *,
+        processors: t.Optional[ProcessorsType] = None,
     ) -> t.List[ToolParam]:
         """
         Get composio tools wrapped as OpenAI `ChatCompletionToolParam` objects.
@@ -105,6 +108,8 @@ def get_tools(
         :return: Composio tools wrapped as `ChatCompletionToolParam` objects
         """
         self.validate_tools(apps=apps, actions=actions, tags=tags)
+        if processors is not None:
+            self._merge_processors(processors)
         return [
             ToolParam(
                 **t.cast(

python/plugins/griptape/composio_griptape/toolset.py

@@ -8,6 +8,7 @@
 
 from composio import Action, ActionType, AppType, TagType
 from composio.tools import ComposioToolSet as BaseComposioToolSet
+from composio.tools.toolset import ProcessorsType
 from composio.utils.shared import PYDANTIC_TYPE_TO_PYTHON_TYPE
 
 
@@ -135,6 +136,8 @@ def get_tools(
         apps: t.Optional[t.Sequence[AppType]] = None,
         tags: t.Optional[t.List[TagType]] = None,
         entity_id: t.Optional[str] = None,
+        *,
+        processors: t.Optional[ProcessorsType] = None,
     ) -> t.List[BaseTool]:
         """
         Get composio tools wrapped as GripTape `BaseTool` type objects.
@@ -147,6 +150,8 @@ def get_tools(
         :return: Composio tools wrapped as `BaseTool` objects
         """
         self.validate_tools(apps=apps, actions=actions, tags=tags)
+        if processors is not None:
+            self._merge_processors(processors)
         return [
             self._wrap_tool(
                 schema=tool.model_dump(

python/plugins/langchain/composio_langchain/toolset.py

@@ -10,6 +10,7 @@
 
 from composio import Action, ActionType, AppType, TagType
 from composio.tools import ComposioToolSet as BaseComposioToolSet
+from composio.tools.toolset import ProcessorsType
 from composio.utils.pydantic import parse_pydantic_error
 from composio.utils.shared import (
     get_signature_format_from_schema_params,
@@ -154,6 +155,8 @@ def get_tools(
         apps: t.Optional[t.Sequence[AppType]] = None,
         tags: t.Optional[t.List[TagType]] = None,
         entity_id: t.Optional[str] = None,
+        *,
+        processors: t.Optional[ProcessorsType] = None,
     ) -> t.Sequence[StructuredTool]:
         """
         Get composio tools wrapped as Langchain StructuredTool objects.
@@ -166,6 +169,8 @@ def get_tools(
         :return: Composio tools wrapped as `StructuredTool` objects
         """
         self.validate_tools(apps=apps, actions=actions, tags=tags)
+        if processors is not None:
+            self._merge_processors(processors)
         return [
             self._wrap_tool(
                 schema=tool.model_dump(

python/plugins/llamaindex/composio_llamaindex/toolset.py

@@ -8,6 +8,7 @@
 from composio import Action, ActionType, AppType
 from composio import ComposioToolSet as BaseComposioToolSet
 from composio import TagType
+from composio.tools.toolset import ProcessorsType
 from composio.utils.shared import get_pydantic_signature_format_from_schema_params
 
 
@@ -131,6 +132,8 @@ def get_tools(
         apps: t.Optional[t.Sequence[AppType]] = None,
         tags: t.Optional[t.List[TagType]] = None,
         entity_id: t.Optional[str] = None,
+        *,
+        processors: t.Optional[ProcessorsType] = None,
     ) -> t.Sequence[FunctionTool]:
         """
         Get composio tools wrapped as LlamaIndex FunctionTool objects.
@@ -143,6 +146,8 @@ def get_tools(
         :return: Composio tools wrapped as `StructuredTool` objects
         """
         self.validate_tools(apps=apps, actions=actions, tags=tags)
+        if processors is not None:
+            self._merge_processors(processors)
         return [
             self._wrap_tool(
                 schema=tool.model_dump(

python/plugins/lyzr/composio_lyzr/toolset.py

@@ -11,6 +11,7 @@
 
 from composio import Action, ActionType, AppType, TagType
 from composio.tools import ComposioToolSet as BaseComposioToolSet
+from composio.tools.toolset import ProcessorsType
 from composio.utils.shared import (
     get_signature_format_from_schema_params,
     json_schema_to_model,
@@ -92,6 +93,8 @@ def get_tools(
         apps: t.Optional[t.Sequence[AppType]] = None,
         tags: t.Optional[t.List[TagType]] = None,
         entity_id: t.Optional[str] = None,
+        *,
+        processors: t.Optional[ProcessorsType] = None,
     ) -> t.List[Tool]:
         """
         Get composio tools wrapped as Lyzr `Tool` objects.
@@ -104,6 +107,8 @@ def get_tools(
         :return: Composio tools wrapped as `Tool` objects
         """
         self.validate_tools(apps=apps, actions=actions, tags=tags)
+        if processors is not None:
+            self._merge_processors(processors)
         return [
             self._wrap_tool(
                 schema=schema.model_dump(exclude_none=True),

python/plugins/openai/composio_openai/toolset.py

@@ -20,6 +20,7 @@
 from composio.constants import DEFAULT_ENTITY_ID
 from composio.tools import ComposioToolSet as BaseComposioToolSet
 from composio.tools.schema import OpenAISchema, SchemaType
+from composio.tools.toolset import ProcessorsType
 
 
 class ComposioToolSet(
@@ -101,6 +102,8 @@ def get_tools(
         actions: t.Optional[t.Sequence[ActionType]] = None,
         apps: t.Optional[t.Sequence[AppType]] = None,
         tags: t.Optional[t.List[TagType]] = None,
+        *,
+        processors: t.Optional[ProcessorsType] = None,
     ) -> t.List[ChatCompletionToolParam]:
         """
         Get composio tools wrapped as OpenAI `ChatCompletionToolParam` objects.
@@ -112,6 +115,8 @@ def get_tools(
         :return: Composio tools wrapped as `ChatCompletionToolParam` objects
         """
         self.validate_tools(apps=apps, actions=actions, tags=tags)
+        if processors is not None:
+            self._merge_processors(processors)
         return [
             ChatCompletionToolParam(  # type: ignore
                 **t.cast(

python/plugins/phidata/composio_phidata/toolset.py

@@ -10,6 +10,7 @@
 from pydantic import validate_call
 
 from composio import Action, ActionType, AppType, TagType
+from composio.tools.toolset import ProcessorsType
 
 from composio_openai import ComposioToolSet as BaseComposioToolSet
 
@@ -68,6 +69,8 @@ def get_tools(
         actions: t.Optional[t.Sequence[ActionType]] = None,
         apps: t.Optional[t.Sequence[AppType]] = None,
         tags: t.Optional[t.List[TagType]] = None,
+        *,
+        processors: t.Optional[ProcessorsType] = None,
     ) -> t.List[Function]:
         """
         Get composio tools wrapped as Lyzr `Function` objects.
@@ -79,6 +82,8 @@ def get_tools(
         :return: Composio tools wrapped as `Function` objects
         """
         self.validate_tools(apps=apps, actions=actions, tags=tags)
+        if processors is not None:
+            self._merge_processors(processors)
         return [
             self._wrap_tool(
                 schema=schema.model_dump(

python/plugins/praisonai/composio_praisonai/toolset.py

@@ -6,6 +6,7 @@
 from composio import Action, ActionType, AppType
 from composio import ComposioToolSet as BaseComposioToolSet
 from composio import TagType
+from composio.tools.toolset import ProcessorsType
 
 
 _openapi_to_python = {
@@ -187,6 +188,8 @@ def get_tools(
         apps: t.Optional[t.Sequence[AppType]] = None,
         tags: t.Optional[t.List[TagType]] = None,
         entity_id: t.Optional[str] = None,
+        *,
+        processors: t.Optional[ProcessorsType] = None,
     ) -> t.List[str]:
         """
         Get composio tools written as ParisonAi supported tools.
@@ -199,6 +202,8 @@ def get_tools(
         :return: Name of the tools written
         """
         self.validate_tools(apps=apps, actions=actions, tags=tags)
+        if processors is not None:
+            self._merge_processors(processors)
         return [
             self._write_tool(
                 schema=tool.model_dump(exclude_none=True),