Add CrewAIPlatformTools (#449)

* chore: add deprecation warning in CrewaiEnterpriseTools

* feat: add CrewAI Platform Tool

* feat: drop support to oldest env-var token
This commit is contained in:
Lucas Gomide
2025-09-12 13:04:26 -03:00
committed by GitHub
parent 6f2301c945
commit f9925887aa
11 changed files with 917 additions and 0 deletions

View File

@@ -25,6 +25,7 @@ from .tools import (
ContextualAIRerankTool,
CouchbaseFTSVectorSearchTool,
CrewaiEnterpriseTools,
CrewaiPlatformTools,
CSVSearchTool,
DallETool,
DatabricksQueryTool,

View File

@@ -19,6 +19,7 @@ from .contextualai_query_tool.contextual_query_tool import ContextualAIQueryTool
from .contextualai_rerank_tool.contextual_rerank_tool import ContextualAIRerankTool
from .couchbase_tool.couchbase_tool import CouchbaseFTSVectorSearchTool
from .crewai_enterprise_tools.crewai_enterprise_tools import CrewaiEnterpriseTools
from .crewai_platform_tools.crewai_platform_tools import CrewaiPlatformTools
from .csv_search_tool.csv_search_tool import CSVSearchTool
from .dalle_tool.dalle_tool import DallETool
from .databricks_query_tool.databricks_query_tool import DatabricksQueryTool

View File

@@ -33,6 +33,13 @@ def CrewaiEnterpriseTools(
A ToolCollection of BaseTool instances for enterprise actions
"""
import warnings
warnings.warn(
"CrewaiEnterpriseTools will be removed in v1.0.0. Considering use `Agent(apps=[...])` instead.",
DeprecationWarning,
stacklevel=2
)
if enterprise_token is None or enterprise_token == "":
enterprise_token = os.environ.get("CREWAI_ENTERPRISE_TOOLS_TOKEN")
if not enterprise_token:

View File

@@ -0,0 +1,16 @@
"""CrewAI Platform Tools
This module provides tools for integrating with various platform applications
through the CrewAI platform API.
"""
from crewai_tools.tools.crewai_platform_tools.crewai_platform_tools import CrewaiPlatformTools
from crewai_tools.tools.crewai_platform_tools.crewai_platform_action_tool import CrewAIPlatformActionTool
from crewai_tools.tools.crewai_platform_tools.crewai_platform_tool_builder import CrewaiPlatformToolBuilder
__all__ = [
"CrewaiPlatformTools",
"CrewAIPlatformActionTool",
"CrewaiPlatformToolBuilder",
]

View File

@@ -0,0 +1,233 @@
"""
Crewai Enterprise Tools
"""
import re
import json
import requests
from typing import Dict, Any, List, Type, Optional, Union, get_origin, cast, Literal
from pydantic import Field, create_model
from crewai.tools import BaseTool
from crewai_tools.tools.crewai_platform_tools.misc import get_platform_api_base_url, get_platform_integration_token
class CrewAIPlatformActionTool(BaseTool):
action_name: str = Field(default="", description="The name of the action")
action_schema: Dict[str, Any] = Field(
default_factory=dict, description="The schema of the action"
)
def __init__(
self,
description: str,
action_name: str,
action_schema: Dict[str, Any],
):
self._model_registry = {}
self._base_name = self._sanitize_name(action_name)
schema_props, required = self._extract_schema_info(action_schema)
field_definitions = {}
for param_name, param_details in schema_props.items():
param_desc = param_details.get("description", "")
is_required = param_name in required
try:
field_type = self._process_schema_type(
param_details, self._sanitize_name(param_name).title()
)
except Exception as e:
field_type = str
field_definitions[param_name] = self._create_field_definition(
field_type, is_required, param_desc
)
if field_definitions:
try:
args_schema = create_model(
f"{self._base_name}Schema", **field_definitions
)
except Exception as e:
print(f"Warning: Could not create main schema model: {e}")
args_schema = create_model(
f"{self._base_name}Schema",
input_text=(str, Field(description="Input for the action")),
)
else:
args_schema = create_model(
f"{self._base_name}Schema",
input_text=(str, Field(description="Input for the action")),
)
super().__init__(name=action_name.lower().replace(" ", "_"), description=description, args_schema=args_schema)
self.action_name = action_name
self.action_schema = action_schema
def _sanitize_name(self, name: str) -> str:
name = name.lower().replace(" ", "_")
sanitized = re.sub(r"[^a-zA-Z0-9_]", "", name)
parts = sanitized.split("_")
return "".join(word.capitalize() for word in parts if word)
def _extract_schema_info(
self, action_schema: Dict[str, Any]
) -> tuple[Dict[str, Any], List[str]]:
schema_props = (
action_schema.get("function", {})
.get("parameters", {})
.get("properties", {})
)
required = (
action_schema.get("function", {}).get("parameters", {}).get("required", [])
)
return schema_props, required
def _process_schema_type(self, schema: Dict[str, Any], type_name: str) -> Type[Any]:
if "anyOf" in schema:
any_of_types = schema["anyOf"]
is_nullable = any(t.get("type") == "null" for t in any_of_types)
non_null_types = [t for t in any_of_types if t.get("type") != "null"]
if non_null_types:
base_type = self._process_schema_type(non_null_types[0], type_name)
return Optional[base_type] if is_nullable else base_type
return cast(Type[Any], Optional[str])
if "oneOf" in schema:
return self._process_schema_type(schema["oneOf"][0], type_name)
if "allOf" in schema:
return self._process_schema_type(schema["allOf"][0], type_name)
json_type = schema.get("type", "string")
if "enum" in schema:
enum_values = schema["enum"]
if not enum_values:
return self._map_json_type_to_python(json_type)
return Literal[tuple(enum_values)]
if json_type == "array":
items_schema = schema.get("items", {"type": "string"})
item_type = self._process_schema_type(items_schema, f"{type_name}Item")
return List[item_type]
if json_type == "object":
return self._create_nested_model(schema, type_name)
return self._map_json_type_to_python(json_type)
def _create_nested_model(self, schema: Dict[str, Any], model_name: str) -> Type[Any]:
full_model_name = f"{self._base_name}{model_name}"
if full_model_name in self._model_registry:
return self._model_registry[full_model_name]
properties = schema.get("properties", {})
required_fields = schema.get("required", [])
if not properties:
return dict
field_definitions = {}
for prop_name, prop_schema in properties.items():
prop_desc = prop_schema.get("description", "")
is_required = prop_name in required_fields
try:
prop_type = self._process_schema_type(
prop_schema, f"{model_name}{self._sanitize_name(prop_name).title()}"
)
except Exception as e:
prop_type = str
field_definitions[prop_name] = self._create_field_definition(
prop_type, is_required, prop_desc
)
try:
nested_model = create_model(full_model_name, **field_definitions)
self._model_registry[full_model_name] = nested_model
return nested_model
except Exception as e:
print(f"Warning: Could not create nested model {full_model_name}: {e}")
return dict
def _create_field_definition(
self, field_type: Type[Any], is_required: bool, description: str
) -> tuple:
if is_required:
return (field_type, Field(description=description))
else:
if get_origin(field_type) is Union:
return (field_type, Field(default=None, description=description))
else:
return (
Optional[field_type],
Field(default=None, description=description),
)
def _map_json_type_to_python(self, json_type: str) -> Type[Any]:
type_mapping = {
"string": str,
"integer": int,
"number": float,
"boolean": bool,
"array": list,
"object": dict,
"null": type(None),
}
return type_mapping.get(json_type, str)
def _get_required_nullable_fields(self) -> List[str]:
schema_props, required = self._extract_schema_info(self.action_schema)
required_nullable_fields = []
for param_name in required:
param_details = schema_props.get(param_name, {})
if self._is_nullable_type(param_details):
required_nullable_fields.append(param_name)
return required_nullable_fields
def _is_nullable_type(self, schema: Dict[str, Any]) -> bool:
if "anyOf" in schema:
return any(t.get("type") == "null" for t in schema["anyOf"])
return schema.get("type") == "null"
def _run(self, **kwargs) -> str:
try:
cleaned_kwargs = {}
for key, value in kwargs.items():
if value is not None:
cleaned_kwargs[key] = value
required_nullable_fields = self._get_required_nullable_fields()
for field_name in required_nullable_fields:
if field_name not in cleaned_kwargs:
cleaned_kwargs[field_name] = None
api_url = f"{get_platform_api_base_url()}/actions/{self.action_name}/execute"
token = get_platform_integration_token()
headers = {
"Authorization": f"Bearer {token}",
"Content-Type": "application/json",
}
payload = cleaned_kwargs
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)}"

View File

@@ -0,0 +1,135 @@
import requests
from typing import List, Any, Dict
from crewai.tools import BaseTool
from crewai_tools.tools.crewai_platform_tools.misc import get_platform_api_base_url, get_platform_integration_token
from crewai_tools.tools.crewai_platform_tools.crewai_platform_action_tool import CrewAIPlatformActionTool
class CrewaiPlatformToolBuilder:
def __init__(
self,
apps: list[str],
):
self._apps = apps
self._actions_schema = {}
self._tools = None
def tools(self) -> list[BaseTool]:
if self._tools is None:
self._fetch_actions()
self._create_tools()
return self._tools if self._tools is not None else []
def _fetch_actions(self):
actions_url = f"{get_platform_api_base_url()}/actions"
headers = {"Authorization": f"Bearer {get_platform_integration_token()}"}
try:
response = requests.get(
actions_url, headers=headers, timeout=30, params={"apps": ",".join(self._apps)}
)
response.raise_for_status()
except Exception as e:
return
raw_data = response.json()
self._actions_schema = {}
action_categories = raw_data.get("actions", {})
for app, action_list in action_categories.items():
if isinstance(action_list, list):
for action in action_list:
if action_name := action.get("name"):
action_schema = {
"function": {
"name": action_name,
"description": action.get("description", f"Execute {action_name}"),
"parameters": action.get("parameters", {}),
"app": app,
}
}
self._actions_schema[action_name] = action_schema
def _generate_detailed_description(
self, schema: Dict[str, Any], indent: int = 0
) -> List[str]:
descriptions = []
indent_str = " " * indent
schema_type = schema.get("type", "string")
if schema_type == "object":
properties = schema.get("properties", {})
required_fields = schema.get("required", [])
if properties:
descriptions.append(f"{indent_str}Object with properties:")
for prop_name, prop_schema in properties.items():
prop_desc = prop_schema.get("description", "")
is_required = prop_name in required_fields
req_str = " (required)" if is_required else " (optional)"
descriptions.append(
f"{indent_str} - {prop_name}: {prop_desc}{req_str}"
)
if prop_schema.get("type") == "object":
descriptions.extend(
self._generate_detailed_description(prop_schema, indent + 2)
)
elif prop_schema.get("type") == "array":
items_schema = prop_schema.get("items", {})
if items_schema.get("type") == "object":
descriptions.append(f"{indent_str} Array of objects:")
descriptions.extend(
self._generate_detailed_description(
items_schema, indent + 3
)
)
elif "enum" in items_schema:
descriptions.append(
f"{indent_str} Array of enum values: {items_schema['enum']}"
)
elif "enum" in prop_schema:
descriptions.append(
f"{indent_str} Enum values: {prop_schema['enum']}"
)
return descriptions
def _create_tools(self):
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}")
parameters = function_details.get("parameters", {})
param_descriptions = []
if parameters.get("properties"):
param_descriptions.append("\nDetailed Parameter Structure:")
param_descriptions.extend(
self._generate_detailed_description(parameters)
)
full_description = description + "\n".join(param_descriptions)
tool = CrewAIPlatformActionTool(
description=full_description,
action_name=action_name,
action_schema=action_schema,
)
tools.append(tool)
self._tools = tools
def __enter__(self):
return self.tools()
def __exit__(self, exc_type, exc_val, exc_tb):
pass

View File

@@ -0,0 +1,28 @@
import re
import os
import typing as t
from typing import Literal
import logging
import json
from crewai.tools import BaseTool
from crewai_tools.tools.crewai_platform_tools.crewai_platform_tool_builder import CrewaiPlatformToolBuilder
from crewai_tools.adapters.tool_collection import ToolCollection
logger = logging.getLogger(__name__)
def CrewaiPlatformTools(
apps: list[str],
) -> ToolCollection[BaseTool]:
"""Factory function that returns crewai platform tools.
Args:
apps: List of platform apps to get tools that are available on the platform.
Returns:
A list of BaseTool instances for platform actions
"""
builder = CrewaiPlatformToolBuilder(apps=apps)
return builder.tools()

View File

@@ -0,0 +1,13 @@
import os
def get_platform_api_base_url() -> str:
"""Get the platform API base URL from environment or use default."""
base_url = os.getenv("CREWAI_PLUS_URL", "https://app.crewai.com")
return f"{base_url}/crewai_plus/api/v1/integrations"
def get_platform_integration_token() -> str:
"""Get the platform API base URL from environment or use default."""
token = os.getenv("CREWAI_PLATFORM_INTEGRATION_TOKEN") or ""
if not token:
raise ValueError("No platform integration token found, please set the CREWAI_PLATFORM_INTEGRATION_TOKEN environment variable")
return token # TODO: Use context manager to get token

View File

@@ -0,0 +1,165 @@
import unittest
from unittest.mock import patch, Mock
import pytest
from crewai_tools.tools.crewai_platform_tools import CrewAIPlatformActionTool
class TestCrewAIPlatformActionTool(unittest.TestCase):
@pytest.fixture
def sample_action_schema(self):
return {
"function": {
"name": "test_action",
"description": "Test action for unit testing",
"parameters": {
"type": "object",
"properties": {
"message": {
"type": "string",
"description": "Message to send"
},
"priority": {
"type": "integer",
"description": "Priority level"
}
},
"required": ["message"]
}
}
}
@pytest.fixture
def platform_action_tool(self, sample_action_schema):
return CrewAIPlatformActionTool(
description="Test Action Tool\nTest description",
action_name="test_action",
action_schema=sample_action_schema
)
@patch.dict("os.environ", {"CREWAI_PLATFORM_INTEGRATION_TOKEN": "test_token"})
@patch("crewai_tools.tools.crewai_platform_tools.crewai_platform_action_tool.requests.post")
def test_run_success(self, mock_post):
schema = {
"function": {
"name": "test_action",
"description": "Test action",
"parameters": {
"type": "object",
"properties": {
"message": {"type": "string", "description": "Message"}
},
"required": ["message"]
}
}
}
tool = CrewAIPlatformActionTool(
description="Test tool",
action_name="test_action",
action_schema=schema
)
mock_response = Mock()
mock_response.ok = True
mock_response.json.return_value = {"result": "success", "data": "test_data"}
mock_post.return_value = mock_response
result = tool._run(message="test message")
mock_post.assert_called_once()
_, kwargs = mock_post.call_args
assert "test_action/execute" in kwargs["url"]
assert kwargs["headers"]["Authorization"] == "Bearer test_token"
assert kwargs["json"]["message"] == "test message"
assert "success" in result
@patch.dict("os.environ", {"CREWAI_PLATFORM_INTEGRATION_TOKEN": "test_token"})
@patch("crewai_tools.tools.crewai_platform_tools.crewai_platform_action_tool.requests.post")
def test_run_api_error(self, mock_post):
schema = {
"function": {
"name": "test_action",
"description": "Test action",
"parameters": {
"type": "object",
"properties": {
"message": {"type": "string", "description": "Message"}
},
"required": ["message"]
}
}
}
tool = CrewAIPlatformActionTool(
description="Test tool",
action_name="test_action",
action_schema=schema
)
mock_response = Mock()
mock_response.ok = False
mock_response.json.return_value = {"error": {"message": "Invalid request"}}
mock_post.return_value = mock_response
result = tool._run(message="test message")
assert "API request failed" in result
assert "Invalid request" in result
@patch.dict("os.environ", {"CREWAI_PLATFORM_INTEGRATION_TOKEN": "test_token"})
@patch("crewai_tools.tools.crewai_platform_tools.crewai_platform_action_tool.requests.post")
def test_run_exception(self, mock_post):
schema = {
"function": {
"name": "test_action",
"description": "Test action",
"parameters": {
"type": "object",
"properties": {
"message": {"type": "string", "description": "Message"}
},
"required": ["message"]
}
}
}
tool = CrewAIPlatformActionTool(
description="Test tool",
action_name="test_action",
action_schema=schema
)
mock_post.side_effect = Exception("Network error")
result = tool._run(message="test message")
assert "Error executing action test_action: Network error" in result
def test_run_without_token(self):
schema = {
"function": {
"name": "test_action",
"description": "Test action",
"parameters": {
"type": "object",
"properties": {
"message": {"type": "string", "description": "Message"}
},
"required": ["message"]
}
}
}
tool = CrewAIPlatformActionTool(
description="Test tool",
action_name="test_action",
action_schema=schema
)
with patch.dict("os.environ", {}, clear=True):
result = tool._run(message="test message")
assert "Error executing action test_action:" in result
assert "No platform integration token found" in result

View File

@@ -0,0 +1,223 @@
import unittest
from unittest.mock import patch, Mock
import pytest
from crewai_tools.tools.crewai_platform_tools import CrewaiPlatformToolBuilder, CrewAIPlatformActionTool
class TestCrewaiPlatformToolBuilder(unittest.TestCase):
@pytest.fixture
def platform_tool_builder(self):
"""Create a CrewaiPlatformToolBuilder instance for testing"""
return CrewaiPlatformToolBuilder(apps=["github", "slack"])
@pytest.fixture
def mock_api_response(self):
return {
"actions": {
"github": [
{
"name": "create_issue",
"description": "Create a GitHub issue",
"parameters": {
"type": "object",
"properties": {
"title": {"type": "string", "description": "Issue title"},
"body": {"type": "string", "description": "Issue body"}
},
"required": ["title"]
}
}
],
"slack": [
{
"name": "send_message",
"description": "Send a Slack message",
"parameters": {
"type": "object",
"properties": {
"channel": {"type": "string", "description": "Channel name"},
"text": {"type": "string", "description": "Message text"}
},
"required": ["channel", "text"]
}
}
]
}
}
@patch.dict("os.environ", {"CREWAI_PLATFORM_INTEGRATION_TOKEN": "test_token"})
@patch("crewai_tools.tools.crewai_platform_tools.crewai_platform_tool_builder.requests.get")
def test_fetch_actions_success(self, mock_get):
mock_api_response = {
"actions": {
"github": [
{
"name": "create_issue",
"description": "Create a GitHub issue",
"parameters": {
"type": "object",
"properties": {
"title": {"type": "string", "description": "Issue title"}
},
"required": ["title"]
}
}
]
}
}
builder = CrewaiPlatformToolBuilder(apps=["github", "slack/send_message"])
mock_response = Mock()
mock_response.raise_for_status.return_value = None
mock_response.json.return_value = mock_api_response
mock_get.return_value = mock_response
builder._fetch_actions()
mock_get.assert_called_once()
args, kwargs = mock_get.call_args
assert "/actions" in args[0]
assert kwargs["headers"]["Authorization"] == "Bearer test_token"
assert kwargs["params"]["apps"] == "github,slack/send_message"
assert "create_issue" in builder._actions_schema
assert builder._actions_schema["create_issue"]["function"]["name"] == "create_issue"
def test_fetch_actions_no_token(self):
builder = CrewaiPlatformToolBuilder(apps=["github"])
with patch.dict("os.environ", {}, clear=True):
with self.assertRaises(ValueError) as context:
builder._fetch_actions()
assert "No platform integration token found" in str(context.exception)
@patch.dict("os.environ", {"CREWAI_PLATFORM_INTEGRATION_TOKEN": "test_token"})
@patch("crewai_tools.tools.crewai_platform_tools.crewai_platform_tool_builder.requests.get")
def test_create_tools(self, mock_get):
mock_api_response = {
"actions": {
"github": [
{
"name": "create_issue",
"description": "Create a GitHub issue",
"parameters": {
"type": "object",
"properties": {
"title": {"type": "string", "description": "Issue title"}
},
"required": ["title"]
}
}
],
"slack": [
{
"name": "send_message",
"description": "Send a Slack message",
"parameters": {
"type": "object",
"properties": {
"channel": {"type": "string", "description": "Channel name"}
},
"required": ["channel"]
}
}
]
}
}
builder = CrewaiPlatformToolBuilder(apps=["github", "slack"])
mock_response = Mock()
mock_response.raise_for_status.return_value = None
mock_response.json.return_value = mock_api_response
mock_get.return_value = mock_response
tools = builder.tools()
assert len(tools) == 2
assert all(isinstance(tool, CrewAIPlatformActionTool) for tool in tools)
tool_names = [tool.action_name for tool in tools]
assert "create_issue" in tool_names
assert "send_message" in tool_names
github_tool = next((t for t in tools if t.action_name == "create_issue"), None)
slack_tool = next((t for t in tools if t.action_name == "send_message"), None)
assert github_tool is not None
assert slack_tool is not None
assert "Create a GitHub issue" in github_tool.description
assert "Send a Slack message" in slack_tool.description
def test_tools_caching(self):
builder = CrewaiPlatformToolBuilder(apps=["github"])
cached_tools = []
def mock_create_tools():
builder._tools = cached_tools
with patch.object(builder, '_fetch_actions') as mock_fetch, \
patch.object(builder, '_create_tools', side_effect=mock_create_tools) as mock_create:
tools1 = builder.tools()
assert mock_fetch.call_count == 1
assert mock_create.call_count == 1
tools2 = builder.tools()
assert mock_fetch.call_count == 1
assert mock_create.call_count == 1
assert tools1 is tools2
@patch.dict("os.environ", {"CREWAI_PLATFORM_INTEGRATION_TOKEN": "test_token"})
def test_empty_apps_list(self):
builder = CrewaiPlatformToolBuilder(apps=[])
with patch("crewai_tools.tools.crewai_platform_tools.crewai_platform_tool_builder.requests.get") as mock_get:
mock_response = Mock()
mock_response.raise_for_status.return_value = None
mock_response.json.return_value = {"actions": {}}
mock_get.return_value = mock_response
tools = builder.tools()
assert isinstance(tools, list)
assert len(tools) == 0
_, kwargs = mock_get.call_args
assert kwargs["params"]["apps"] == ""
def test_detailed_description_generation(self):
builder = CrewaiPlatformToolBuilder(apps=["test"])
complex_schema = {
"type": "object",
"properties": {
"simple_string": {"type": "string", "description": "A simple string"},
"nested_object": {
"type": "object",
"properties": {
"inner_prop": {"type": "integer", "description": "Inner property"}
},
"description": "Nested object"
},
"array_prop": {
"type": "array",
"items": {"type": "string"},
"description": "Array of strings"
}
}
}
descriptions = builder._generate_detailed_description(complex_schema)
assert isinstance(descriptions, list)
assert len(descriptions) > 0
description_text = "\n".join(descriptions)
assert "simple_string" in description_text
assert "nested_object" in description_text
assert "array_prop" in description_text

View File

@@ -0,0 +1,95 @@
import unittest
from unittest.mock import patch, Mock
from crewai_tools.tools.crewai_platform_tools import CrewaiPlatformTools
class TestCrewaiPlatformTools(unittest.TestCase):
@patch.dict("os.environ", {"CREWAI_PLATFORM_INTEGRATION_TOKEN": "test_token"})
@patch("crewai_tools.tools.crewai_platform_tools.crewai_platform_tool_builder.requests.get")
def test_crewai_platform_tools_basic(self, mock_get):
mock_response = Mock()
mock_response.raise_for_status.return_value = None
mock_response.json.return_value = {"actions": {"github": []}}
mock_get.return_value = mock_response
tools = CrewaiPlatformTools(apps=["github"])
assert tools is not None
assert isinstance(tools, list)
@patch.dict("os.environ", {"CREWAI_PLATFORM_INTEGRATION_TOKEN": "test_token"})
@patch("crewai_tools.tools.crewai_platform_tools.crewai_platform_tool_builder.requests.get")
def test_crewai_platform_tools_multiple_apps(self, mock_get):
mock_response = Mock()
mock_response.raise_for_status.return_value = None
mock_response.json.return_value = {
"actions": {
"github": [
{
"name": "create_issue",
"description": "Create a GitHub issue",
"parameters": {
"type": "object",
"properties": {
"title": {"type": "string", "description": "Issue title"},
"body": {"type": "string", "description": "Issue body"}
},
"required": ["title"]
}
}
],
"slack": [
{
"name": "send_message",
"description": "Send a Slack message",
"parameters": {
"type": "object",
"properties": {
"channel": {"type": "string", "description": "Channel to send to"},
"text": {"type": "string", "description": "Message text"}
},
"required": ["channel", "text"]
}
}
]
}
}
mock_get.return_value = mock_response
tools = CrewaiPlatformTools(apps=["github", "slack"])
assert tools is not None
assert isinstance(tools, list)
assert len(tools) == 2
mock_get.assert_called_once()
args, kwargs = mock_get.call_args
assert "apps=github,slack" in args[0] or kwargs.get("params", {}).get("apps") == "github,slack"
@patch.dict("os.environ", {"CREWAI_PLATFORM_INTEGRATION_TOKEN": "test_token"})
def test_crewai_platform_tools_empty_apps(self):
with patch("crewai_tools.tools.crewai_platform_tools.crewai_platform_tool_builder.requests.get") as mock_get:
mock_response = Mock()
mock_response.raise_for_status.return_value = None
mock_response.json.return_value = {"actions": {}}
mock_get.return_value = mock_response
tools = CrewaiPlatformTools(apps=[])
assert tools is not None
assert isinstance(tools, list)
assert len(tools) == 0
@patch.dict("os.environ", {"CREWAI_PLATFORM_INTEGRATION_TOKEN": "test_token"})
@patch("crewai_tools.tools.crewai_platform_tools.crewai_platform_tool_builder.requests.get")
def test_crewai_platform_tools_api_error_handling(self, mock_get):
mock_get.side_effect = Exception("API Error")
tools = CrewaiPlatformTools(apps=["github"])
assert tools is not None
assert isinstance(tools, list)
assert len(tools) == 0
def test_crewai_platform_tools_no_token(self):
with patch.dict("os.environ", {}, clear=True):
with self.assertRaises(ValueError) as context:
CrewaiPlatformTools(apps=["github"])
assert "No platform integration token found" in str(context.exception)