From 4d86da80c3516e151d332d8d4a7a888f158818b7 Mon Sep 17 00:00:00 2001 From: lorenzejay Date: Thu, 24 Apr 2025 14:20:40 -0700 Subject: [PATCH 1/2] feat: add EnterpriseActionKitToolAdapter and EnterpriseActionTool for enterprise action execution - Introduced EnterpriseActionTool to execute specific enterprise actions with dynamic parameter validation. - Added EnterpriseActionKitToolAdapter to manage and create tool instances for available enterprise actions. - Implemented methods for fetching action schemas from the API and creating corresponding tools. - Enhanced error handling and provided detailed descriptions for tool parameters. - Included a main execution block for testing the adapter with a sample agent and task setup. --- .../adapters/enterprise_adapter.py | 259 ++++++++++++++++++ 1 file changed, 259 insertions(+) create mode 100644 src/crewai_tools/adapters/enterprise_adapter.py diff --git a/src/crewai_tools/adapters/enterprise_adapter.py b/src/crewai_tools/adapters/enterprise_adapter.py new file mode 100644 index 000000000..9a3d1860a --- /dev/null +++ b/src/crewai_tools/adapters/enterprise_adapter.py @@ -0,0 +1,259 @@ +import requests +from pydantic import Field, create_model +from typing import List, Any, Dict, Optional +import json +from crewai.tools import BaseTool +from crewai import Agent, Task, Crew + + +ENTERPRISE_ACTION_KIT_PROJECT_ID = "dd525517-df22-49d2-a69e-6a0eed211166" + + +class EnterpriseActionTool(BaseTool): + """A tool that executes a specific enterprise action.""" + + enterprise_action_token: str = Field( + default="", description="The enterprise action token" + ) + action_name: str = Field(default="", description="The name of the action") + action_schema: Dict[str, Any] = Field( + default={}, description="The schema of the action" + ) + + def __init__( + self, + name: str, + description: str, + enterprise_action_token: str, + action_name: str, + action_schema: Dict[str, Any], + ): + schema_props = ( + action_schema.get("function", {}) + .get("parameters", {}) + .get("properties", {}) + ) + required = ( + action_schema.get("function", {}).get("parameters", {}).get("required", []) + ) + + # Define field definitions for the model + field_definitions = {} + for param_name, param_details in schema_props.items(): + param_type = str # Default to string type + param_desc = param_details.get("description", "") + is_required = param_name in required + + # Basic type mapping (can be extended) + if param_details.get("type") == "integer": + param_type = int + elif param_details.get("type") == "number": + param_type = float + elif param_details.get("type") == "boolean": + param_type = bool + + # Create field with appropriate type and config + field_definitions[param_name] = ( + param_type if is_required else Optional[param_type], + Field(description=param_desc), + ) + + # Create the model + if field_definitions: + args_schema = create_model( + f"{name.capitalize()}Schema", **field_definitions + ) + else: + # Fallback for empty schema + args_schema = create_model( + f"{name.capitalize()}Schema", + input_text=(str, Field(description="Input for the action")), + ) + + super().__init__(name=name, description=description, args_schema=args_schema) + self.enterprise_action_token = enterprise_action_token + self.action_name = action_name + self.action_schema = action_schema + + def _run(self, **kwargs) -> str: + """Execute the specific enterprise action with validated parameters.""" + try: + params = {k: v for k, v in kwargs.items() if v is not None} + + api_url = f"https://worker-actionkit.tools.crewai.com/projects/{ENTERPRISE_ACTION_KIT_PROJECT_ID}/actions" + headers = { + "Authorization": f"Bearer {self.enterprise_action_token}", + "Content-Type": "application/json", + } + payload = {"action": self.action_name, "parameters": params} + + response = requests.post( + url=api_url, headers=headers, json=payload, timeout=60 + ) + + data = response.json() + if not response.ok: + error_message = data.get("error", {}).get("message", json.dumps(data)) + return f"API request failed: {error_message}" + + return json.dumps(data, indent=2) + + except Exception as e: + return f"Error executing action {self.action_name}: {str(e)}" + + +class EnterpriseActionKitToolAdapter: + """Adapter that creates BaseTool instances for enterprise actions.""" + + def __init__(self, enterprise_action_token: str): + """Initialize the adapter with an enterprise action token.""" + if not enterprise_action_token: + raise ValueError("enterprise_action_token is required") + + self.enterprise_action_token = enterprise_action_token + self._actions_schema = {} + self._tools = None + + def tools(self) -> List[BaseTool]: + """Get the list of tools created from enterprise actions. + + Returns: + List of BaseTool instances, one for each enterprise action. + """ + if self._tools is None: + self._fetch_actions() + self._create_tools() + return self._tools + + def _fetch_actions(self): + """Fetch available actions from the API.""" + try: + actions_url = f"https://worker-actionkit.tools.crewai.com/projects/{ENTERPRISE_ACTION_KIT_PROJECT_ID}/actions" + headers = {"Authorization": f"Bearer {self.enterprise_action_token}"} + params = {"format": "json_schema"} + + response = requests.get( + actions_url, headers=headers, params=params, timeout=30 + ) + response.raise_for_status() + + raw_data = response.json() + if "actions" not in raw_data: + print(f"Unexpected API response structure: {raw_data}") + return + + # Parse the actions schema + parsed_schema = {} + action_categories = raw_data["actions"] + + for category, action_list in action_categories.items(): + if isinstance(action_list, list): + for action in action_list: + func_details = action.get("function") + if func_details and "name" in func_details: + action_name = func_details["name"] + parsed_schema[action_name] = action + + self._actions_schema = parsed_schema + + except Exception as e: + print(f"Error fetching actions: {e}") + import traceback + + traceback.print_exc() + + def _create_tools(self): + """Create BaseTool instances for each action.""" + tools = [] + + for action_name, action_schema in self._actions_schema.items(): + function_details = action_schema.get("function", {}) + description = function_details.get("description", f"Execute {action_name}") + + # Get parameter info for a better description + parameters = function_details.get("parameters", {}).get("properties", {}) + param_info = [] + for param_name, param_details in parameters.items(): + param_desc = param_details.get("description", "") + required = param_name in function_details.get("parameters", {}).get( + "required", [] + ) + param_info.append( + f"- {param_name}: {param_desc} {'(required)' if required else '(optional)'}" + ) + + full_description = f"{description}\n\nParameters:\n" + "\n".join(param_info) + + tool = EnterpriseActionTool( + name=action_name.lower().replace(" ", "_"), + description=full_description, + action_name=action_name, + action_schema=action_schema, + enterprise_action_token=self.enterprise_action_token, + ) + + tools.append(tool) + + self._tools = tools + + # Adding context manager support for convenience, but direct usage is also supported + def __enter__(self): + return self.tools() + + def __exit__(self, exc_type, exc_val, exc_tb): + pass + + +if __name__ == "__main__": + # IMPORTANT: Replace 'YOUR_TOKEN_HERE' with your actual valid token + # You can also load it from an environment variable for better security + import os + + token = os.environ.get( + "ENTERPRISE_TOOL_TOKEN", + ) # Replace YOUR_TOKEN_HERE if not using env var + + if token == "YOUR_TOKEN_HERE" or not token: + print("Error: ENTERPRISE_TOOL_TOKEN is not set.") + print( + "Please replace 'YOUR_TOKEN_HERE' in the code or set the ENTERPRISE_TOOL_TOKEN environment variable." + ) + else: + try: + print("Initializing EnterpriseActionKitTool...") + adapter = EnterpriseActionKitToolAdapter(enterprise_action_token=token) + available_tools = adapter.tools() + + agent = Agent( + model="gpt-4o", + tools=available_tools, + role="You are are expert at google sheets", + goal="Get the sheet with the data x", + backstory="You are a expert at google sheets", + verbose=True, + ) + + task = Task( + description="return data from the sheet with the id: {spreadsheetId}, with the limit: {limit}", + expected_output="The data from the sheet with the id: {spreadsheetId} with the limit: {limit}", + agent=agent, + ) + crew = Crew( + agents=[agent], + tasks=[task], + verbose=True, + ) + result = crew.kickoff( + inputs={ + "spreadsheetId": "1DHDIWGdhUXqXeYOO8yA44poiY222qHPQEUu28olipKs", + "limit": 2, + } + ) + + except ValueError as e: + print(f"\nConfiguration Error: {e}") + except Exception as e: + print(f"\nAn unexpected error occurred during execution: {e}") + import traceback + + traceback.print_exc() From 40dd22ce2ce3b40a42c9f03c1155af2f904bc0d9 Mon Sep 17 00:00:00 2001 From: lorenzejay Date: Thu, 24 Apr 2025 14:39:55 -0700 Subject: [PATCH 2/2] refactor: remove main execution block from EnterpriseActionKitToolAdapter - Removed the main execution block that included token validation and agent/task setup for testing. - This change streamlines the adapter's code, focusing on its core functionality without execution logic. --- .../adapters/enterprise_adapter.py | 56 ------------------- 1 file changed, 56 deletions(-) diff --git a/src/crewai_tools/adapters/enterprise_adapter.py b/src/crewai_tools/adapters/enterprise_adapter.py index 9a3d1860a..1d93894b7 100644 --- a/src/crewai_tools/adapters/enterprise_adapter.py +++ b/src/crewai_tools/adapters/enterprise_adapter.py @@ -3,7 +3,6 @@ from pydantic import Field, create_model from typing import List, Any, Dict, Optional import json from crewai.tools import BaseTool -from crewai import Agent, Task, Crew ENTERPRISE_ACTION_KIT_PROJECT_ID = "dd525517-df22-49d2-a69e-6a0eed211166" @@ -202,58 +201,3 @@ class EnterpriseActionKitToolAdapter: def __exit__(self, exc_type, exc_val, exc_tb): pass - - -if __name__ == "__main__": - # IMPORTANT: Replace 'YOUR_TOKEN_HERE' with your actual valid token - # You can also load it from an environment variable for better security - import os - - token = os.environ.get( - "ENTERPRISE_TOOL_TOKEN", - ) # Replace YOUR_TOKEN_HERE if not using env var - - if token == "YOUR_TOKEN_HERE" or not token: - print("Error: ENTERPRISE_TOOL_TOKEN is not set.") - print( - "Please replace 'YOUR_TOKEN_HERE' in the code or set the ENTERPRISE_TOOL_TOKEN environment variable." - ) - else: - try: - print("Initializing EnterpriseActionKitTool...") - adapter = EnterpriseActionKitToolAdapter(enterprise_action_token=token) - available_tools = adapter.tools() - - agent = Agent( - model="gpt-4o", - tools=available_tools, - role="You are are expert at google sheets", - goal="Get the sheet with the data x", - backstory="You are a expert at google sheets", - verbose=True, - ) - - task = Task( - description="return data from the sheet with the id: {spreadsheetId}, with the limit: {limit}", - expected_output="The data from the sheet with the id: {spreadsheetId} with the limit: {limit}", - agent=agent, - ) - crew = Crew( - agents=[agent], - tasks=[task], - verbose=True, - ) - result = crew.kickoff( - inputs={ - "spreadsheetId": "1DHDIWGdhUXqXeYOO8yA44poiY222qHPQEUu28olipKs", - "limit": 2, - } - ) - - except ValueError as e: - print(f"\nConfiguration Error: {e}") - except Exception as e: - print(f"\nAn unexpected error occurred during execution: {e}") - import traceback - - traceback.print_exc()