andre/crewAI
1
0
Fork 0
mirror of https://github.com/crewAIInc/crewAI.git synced 2026-01-11 09:08:31 +00:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

feat: add apps & actions attributes to Agent (#3504)

Browse Source 
* feat: add app attributes to Agent

* feat: add actions attribute to Agent

* chore: resolve linter issues

* refactor: merge the apps and actions parameters into a single one

* fix: remove unnecessary print

* feat: logging error when CrewaiPlatformTools fails

* chore: export CrewaiPlatformTools directly from crewai_tools

* style: resolver linter issues

* test: fix broken tests

* style: solve linter issues

* fix: fix broken test
This commit is contained in:
Lucas Gomide
2025-09-25 17:46:51 -03:00
committed by GitHub
parent e070c1400c
commit 13e5ec711d
7 changed files with 315 additions and 104 deletions
Download Patch File Download Diff File Expand all files Collapse all files

160
src/crewai/agent.py
View File

@@ -1,24 +1,36 @@
import shutil
import subprocess
import time
from collections.abc import Callable, Sequence
from typing import (
Any,
Callable,
Dict,
List,
Literal,
Optional,
Sequence,
Tuple,
Type,
Union,
cast,
)
from pydantic import Field, InstanceOf, PrivateAttr, model_validator
from crewai.agents import CacheHandler
from crewai.agents.agent_builder.base_agent import BaseAgent
from crewai.agents.agent_builder.base_agent import BaseAgent, PlatformAppOrAction
from crewai.agents.crew_agent_executor import CrewAgentExecutor
from crewai.events.event_bus import crewai_event_bus
from crewai.events.types.agent_events import (
AgentExecutionCompletedEvent,
AgentExecutionErrorEvent,
AgentExecutionStartedEvent,
)
from crewai.events.types.knowledge_events import (
KnowledgeQueryCompletedEvent,
KnowledgeQueryFailedEvent,
KnowledgeQueryStartedEvent,
KnowledgeRetrievalCompletedEvent,
KnowledgeRetrievalStartedEvent,
KnowledgeSearchQueryFailedEvent,
)
from crewai.events.types.memory_events import (
MemoryRetrievalCompletedEvent,
MemoryRetrievalStartedEvent,
)
from crewai.knowledge.knowledge import Knowledge
from crewai.knowledge.source.base_knowledge_source import BaseKnowledgeSource
from crewai.knowledge.utils.knowledge_utils import extract_knowledge_context
@@ -38,24 +50,6 @@ from crewai.utilities.agent_utils import (
)
from crewai.utilities.constants import TRAINED_AGENTS_DATA_FILE, TRAINING_DATA_FILE
from crewai.utilities.converter import generate_model_description
from crewai.events.types.agent_events import (
AgentExecutionCompletedEvent,
AgentExecutionErrorEvent,
AgentExecutionStartedEvent,
)
from crewai.events.event_bus import crewai_event_bus
from crewai.events.types.memory_events import (
MemoryRetrievalStartedEvent,
MemoryRetrievalCompletedEvent,
)
from crewai.events.types.knowledge_events import (
KnowledgeQueryCompletedEvent,
KnowledgeQueryFailedEvent,
KnowledgeQueryStartedEvent,
KnowledgeRetrievalCompletedEvent,
KnowledgeRetrievalStartedEvent,
KnowledgeSearchQueryFailedEvent,
)
from crewai.utilities.llm_utils import create_llm
from crewai.utilities.token_counter_callback import TokenCalcHandler
from crewai.utilities.training_handler import CrewTrainingHandler
@@ -84,39 +78,40 @@ class Agent(BaseAgent):
step_callback: Callback to be executed after each step of the agent execution.
knowledge_sources: Knowledge sources for the agent.
embedder: Embedder configuration for the agent.
apps: List of applications that the agent can access through CrewAI Platform.
"""
_times_executed: int = PrivateAttr(default=0)
max_execution_time: Optional[int] = Field(
max_execution_time: int | None = Field(
default=None,
description="Maximum execution time for an agent to execute a task",
)
agent_ops_agent_name: str = None # type: ignore # Incompatible types in assignment (expression has type "None", variable has type "str")
agent_ops_agent_id: str = None # type: ignore # Incompatible types in assignment (expression has type "None", variable has type "str")
step_callback: Optional[Any] = Field(
step_callback: Any | None = Field(
default=None,
description="Callback to be executed after each step of the agent execution.",
)
use_system_prompt: Optional[bool] = Field(
use_system_prompt: bool | None = Field(
default=True,
description="Use system prompt for the agent.",
)
llm: Union[str, InstanceOf[BaseLLM], Any] = Field(
llm: str | InstanceOf[BaseLLM] | Any = Field(
description="Language model that will run the agent.", default=None
)
function_calling_llm: Optional[Union[str, InstanceOf[BaseLLM], Any]] = Field(
function_calling_llm: str | InstanceOf[BaseLLM] | Any | None = Field(
description="Language model that will run the agent.", default=None
)
system_template: Optional[str] = Field(
system_template: str | None = Field(
default=None, description="System format for the agent."
)
prompt_template: Optional[str] = Field(
prompt_template: str | None = Field(
default=None, description="Prompt format for the agent."
)
response_template: Optional[str] = Field(
response_template: str | None = Field(
default=None, description="Response format for the agent."
)
allow_code_execution: Optional[bool] = Field(
allow_code_execution: bool | None = Field(
default=False, description="Enable code execution for the agent."
)
respect_context_window: bool = Field(
@@ -147,31 +142,31 @@ class Agent(BaseAgent):
default=False,
description="Whether the agent should reflect and create a plan before executing a task.",
)
max_reasoning_attempts: Optional[int] = Field(
max_reasoning_attempts: int | None = Field(
default=None,
description="Maximum number of reasoning attempts before executing the task. If None, will try until ready.",
)
embedder: Optional[Dict[str, Any]] = Field(
embedder: dict[str, Any] | None = Field(
default=None,
description="Embedder configuration for the agent.",
)
agent_knowledge_context: Optional[str] = Field(
agent_knowledge_context: str | None = Field(
default=None,
description="Knowledge context for the agent.",
)
crew_knowledge_context: Optional[str] = Field(
crew_knowledge_context: str | None = Field(
default=None,
description="Knowledge context for the crew.",
)
knowledge_search_query: Optional[str] = Field(
knowledge_search_query: str | None = Field(
default=None,
description="Knowledge search query for the agent dynamically generated by the agent.",
)
from_repository: Optional[str] = Field(
from_repository: str | None = Field(
default=None,
description="The Agent's role to be used from your repository.",
)
guardrail: Optional[Union[Callable[[Any], Tuple[bool, Any]], str]] = Field(
guardrail: Callable[[Any], tuple[bool, Any]] | str | None = Field(
default=None,
description="Function or string description of a guardrail to validate agent output",
)
@@ -180,6 +175,7 @@ class Agent(BaseAgent):
)
@model_validator(mode="before")
@classmethod
def validate_from_repository(cls, v):
if v is not None and (from_repository := v.get("from_repository")):
return load_agent_from_repository(from_repository) | v
@@ -208,7 +204,7 @@ class Agent(BaseAgent):
self.cache_handler = CacheHandler()
self.set_cache_handler(self.cache_handler)
def set_knowledge(self, crew_embedder: Optional[Dict[str, Any]] = None):
def set_knowledge(self, crew_embedder: dict[str, Any] | None = None):
try:
if self.embedder is None and crew_embedder:
self.embedder = crew_embedder
@@ -224,7 +220,7 @@ class Agent(BaseAgent):
)
self.knowledge.add_sources()
except (TypeError, ValueError) as e:
raise ValueError(f"Invalid Knowledge Configuration: {str(e)}")
raise ValueError(f"Invalid Knowledge Configuration: {e!s}") from e
def _is_any_available_memory(self) -> bool:
"""Check if any memory is available."""
@@ -244,8 +240,8 @@ class Agent(BaseAgent):
def execute_task(
self,
task: Task,
context: Optional[str] = None,
tools: Optional[List[BaseTool]] = None,
context: str | None = None,
tools: list[BaseTool] | None = None,
) -> str:
"""Execute a task with the agent.
@@ -277,13 +273,9 @@ class Agent(BaseAgent):
# Add the reasoning plan to the task description
task.description += f"\n\nReasoning Plan:\n{reasoning_output.plan.plan}"
except Exception as e:
if hasattr(self, "_logger"):
self._logger.log(
"error", f"Error during reasoning process: {str(e)}"
)
else:
print(f"Error during reasoning process: {str(e)}")
self._logger.log(
"error", f"Error during reasoning process: {e!s}"
)
self._inject_date_to_task(task)
if self.tools_handler:
@@ -335,7 +327,7 @@ class Agent(BaseAgent):
agent=self,
task=task,
)
memory = contextual_memory.build_context_for_task(task, context)
memory = contextual_memory.build_context_for_task(task, context or "")
if memory.strip() != "":
task_prompt += self.i18n.slice("memory").format(memory=memory)
@@ -525,14 +517,14 @@ class Agent(BaseAgent):
try:
return future.result(timeout=timeout)
except concurrent.futures.TimeoutError:
except concurrent.futures.TimeoutError as e:
future.cancel()
raise TimeoutError(
f"Task '{task.description}' execution timed out after {timeout} seconds. Consider increasing max_execution_time or optimizing the task."
)
) from e
except Exception as e:
future.cancel()
raise RuntimeError(f"Task execution failed: {str(e)}")
raise RuntimeError(f"Task execution failed: {e!s}") from e
def _execute_without_timeout(self, task_prompt: str, task: Task) -> str:
"""Execute a task without a timeout.
@@ -554,14 +546,14 @@ class Agent(BaseAgent):
)["output"]
def create_agent_executor(
self, tools: Optional[List[BaseTool]] = None, task=None
self, tools: list[BaseTool] | None = None, task=None
) -> None:
"""Create an agent executor for the agent.
Returns:
An instance of the CrewAgentExecutor class.
"""
raw_tools: List[BaseTool] = tools or self.tools or []
raw_tools: list[BaseTool] = tools or self.tools or []
parsed_tools = parse_tools(raw_tools)
prompt = Prompts(
@@ -587,7 +579,7 @@ class Agent(BaseAgent):
agent=self,
crew=self.crew,
tools=parsed_tools,
prompt=prompt,
prompt=cast(dict[str, str], prompt),
original_tools=raw_tools,
stop_words=stop_words,
max_iter=self.max_iter,
@@ -603,10 +595,18 @@ class Agent(BaseAgent):
callbacks=[TokenCalcHandler(self._token_process)],
)
def get_delegation_tools(self, agents: List[BaseAgent]):
def get_delegation_tools(self, agents: list[BaseAgent]):
agent_tools = AgentTools(agents=agents)
tools = agent_tools.tools()
return tools
return agent_tools.tools()
def get_platform_tools(self, apps: list[PlatformAppOrAction]) -> list[BaseTool]:
try:
from crewai_tools import CrewaiPlatformTools # type: ignore[import-untyped]
return CrewaiPlatformTools(apps=apps)
except Exception as e:
self._logger.log("error", f"Error getting platform tools: {e!s}")
return []
def get_multimodal_tools(self) -> Sequence[BaseTool]:
from crewai.tools.agent_tools.add_image_tool import AddImageTool
@@ -654,7 +654,7 @@ class Agent(BaseAgent):
)
return task_prompt
def _render_text_description(self, tools: List[Any]) -> str:
def _render_text_description(self, tools: list[Any]) -> str:
"""Render the tool name and description in plain text.
Output will be in the format of:
@@ -664,14 +664,13 @@ class Agent(BaseAgent):
search: This tool is used for search
calculator: This tool is used for math
"""
description = "\n".join(
return "\n".join(
[
f"Tool name: {tool.name}\nTool description:\n{tool.description}"
for tool in tools
]
)
return description
def _inject_date_to_task(self, task):
"""Inject the current date into the task description if inject_date is enabled."""
@@ -700,28 +699,33 @@ class Agent(BaseAgent):
task.description += f"\n\nCurrent Date: {current_date}"
except Exception as e:
if hasattr(self, "_logger"):
self._logger.log("warning", f"Failed to inject date: {str(e)}")
self._logger.log("warning", f"Failed to inject date: {e!s}")
else:
print(f"Warning: Failed to inject date: {str(e)}")
print(f"Warning: Failed to inject date: {e!s}")
def _validate_docker_installation(self) -> None:
"""Check if Docker is installed and running."""
if not shutil.which("docker"):
docker_path = shutil.which("docker")
if not docker_path:
raise RuntimeError(
f"Docker is not installed. Please install Docker to use code execution with agent: {self.role}"
)
try:
subprocess.run(
["docker", "info"],
subprocess.run( # noqa: S603
[docker_path, "info"],
check=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
except subprocess.CalledProcessError:
except subprocess.CalledProcessError as e:
raise RuntimeError(
f"Docker is not running. Please start Docker to use code execution with agent: {self.role}"
)
) from e
except subprocess.TimeoutExpired as e:
raise RuntimeError(
f"Docker command timed out. Please check your Docker installation for agent: {self.role}"
) from e
def __repr__(self):
return f"Agent(role={self.role}, goal={self.goal}, backstory={self.backstory})"
@@ -796,8 +800,8 @@ class Agent(BaseAgent):
def kickoff(
self,
messages: Union[str, List[Dict[str, str]]],
response_format: Optional[Type[Any]] = None,
messages: str | list[dict[str, str]],
response_format: type[Any] | None = None,
) -> LiteAgentOutput:
"""
Execute the agent with the given messages using a LiteAgent instance.
@@ -836,8 +840,8 @@ class Agent(BaseAgent):
async def kickoff_async(
self,
messages: Union[str, List[Dict[str, str]]],
response_format: Optional[Type[Any]] = None,
messages: str | list[dict[str, str]],
response_format: type[Any] | None = None,
) -> LiteAgentOutput:
"""
Execute the agent asynchronously with the given messages using a LiteAgent instance.

63
src/crewai/agents/agent_builder/base_agent.py
View File

@@ -3,7 +3,7 @@ from abc import ABC, abstractmethod
from collections.abc import Callable
from copy import copy as shallow_copy
from hashlib import md5
from typing import Any, TypeVar
from typing import Any, Literal, TypeVar
from pydantic import (
UUID4,
@@ -30,6 +30,27 @@ from crewai.utilities.string_utils import interpolate_only
T = TypeVar("T", bound="BaseAgent")
PlatformApp = Literal[
"asana",
"box",
"clickup",
"github",
"gmail",
"google_calendar",
"google_sheets",
"hubspot",
"jira",
"linear",
"notion",
"salesforce",
"shopify",
"slack",
"stripe",
"zendesk",
]
PlatformAppOrAction = PlatformApp | str
class BaseAgent(ABC, BaseModel):
"""Abstract Base Class for all third party agents compatible with CrewAI.
@@ -40,11 +61,11 @@ class BaseAgent(ABC, BaseModel):
goal (str): Objective of the agent.
backstory (str): Backstory of the agent.
cache (bool): Whether the agent should use a cache for tool usage.
config (Optional[Dict[str, Any]]): Configuration for the agent.
config (dict[str, Any] | None): Configuration for the agent.
verbose (bool): Verbose mode for the Agent Execution.
max_rpm (Optional[int]): Maximum number of requests per minute for the agent execution.
max_rpm (int | None): Maximum number of requests per minute for the agent execution.
allow_delegation (bool): Allow delegation of tasks to agents.
tools (Optional[List[Any]]): Tools at the agent's disposal.
tools (list[Any] | None): Tools at the agent's disposal.
max_iter (int): Maximum iterations for an agent to execute a task.
agent_executor (InstanceOf): An instance of the CrewAgentExecutor class.
llm (Any): Language model that will run the agent.
@@ -56,18 +77,22 @@ class BaseAgent(ABC, BaseModel):
knowledge_sources: Knowledge sources for the agent.
knowledge_storage: Custom knowledge storage for the agent.
security_config: Security configuration for the agent, including fingerprinting.
apps: List of enterprise applications that the agent can access through CrewAI Enterprise Tools.
actions: List of actions that the agent can access through CrewAI Enterprise Tools.
Methods:
execute_task(task: Any, context: Optional[str] = None, tools: Optional[List[BaseTool]] = None) -> str:
execute_task(task: Any, context: str | None = None, tools: list[BaseTool] | None = None) -> str:
Abstract method to execute a task.
create_agent_executor(tools=None) -> None:
Abstract method to create an agent executor.
get_delegation_tools(agents: List["BaseAgent"]):
get_delegation_tools(agents: list["BaseAgent"]):
Abstract method to set the agents task tools for handling delegation and question asking to other agents in crew.
get_platform_tools(apps: list[PlatformAppOrAction]):
Abstract method to get platform tools for the specified list of applications and/or application/action combinations.
get_output_converter(llm, model, instructions):
Abstract method to get the converter class for the agent to create json/pydantic outputs.
interpolate_inputs(inputs: Dict[str, Any]) -> None:
interpolate_inputs(inputs: dict[str, Any]) -> None:
Interpolate inputs into the agent description and backstory.
set_cache_handler(cache_handler: CacheHandler) -> None:
Set the cache handler for the agent.
@@ -160,6 +185,10 @@ class BaseAgent(ABC, BaseModel):
default=None,
description="Knowledge configuration for the agent such as limits and threshold",
)
apps: list[PlatformAppOrAction] | None = Field(
default=None,
description="List of applications or application/action combinations that the agent can access through CrewAI Platform. Can contain app names (e.g., 'gmail') or specific actions (e.g., 'gmail/send_email')",
)
@model_validator(mode="before")
@classmethod
@@ -195,6 +224,20 @@ class BaseAgent(ABC, BaseModel):
)
return processed_tools
@field_validator("apps")
@classmethod
def validate_apps(cls, apps: list[PlatformAppOrAction] | None) -> list[PlatformAppOrAction] | None:
if not apps:
return apps
validated_apps = []
for app in apps:
if app.count("/") > 1:
raise ValueError(f"Invalid app format '{app}'. Apps can only have one '/' for app/action format (e.g., 'gmail/send_email')")
validated_apps.append(app)
return list(set(validated_apps))
@model_validator(mode="after")
def validate_and_set_attributes(self):
# Validate required fields
@@ -265,6 +308,10 @@ class BaseAgent(ABC, BaseModel):
def get_delegation_tools(self, agents: list["BaseAgent"]) -> list[BaseTool]:
"""Set the task tools that init BaseAgenTools class."""
@abstractmethod
def get_platform_tools(self, apps: list[PlatformAppOrAction]) -> list[BaseTool]:
"""Get platform tools for the specified list of applications and/or application/action combinations."""
def copy(self: T) -> T: # type: ignore # Signature of "copy" incompatible with supertype "BaseModel"
"""Create a deep copy of the Agent."""
exclude = {
@@ -281,6 +328,8 @@ class BaseAgent(ABC, BaseModel):
"knowledge_sources",
"knowledge_storage",
"knowledge",
"apps",
"actions",
}
# Copy llm

29
src/crewai/crew.py
View File

@@ -984,7 +984,10 @@ class Crew(FlowTrackable, BaseModel):
):
tools = self._add_multimodal_tools(agent, tools)
# Return a List[BaseTool] compatible with Task.execute_sync and execute_async
if agent and (hasattr(agent, "apps") and getattr(agent, "apps", None)):
tools = self._add_platform_tools(task, tools)
# Return a list[BaseTool] compatible with Task.execute_sync and execute_async
return cast(list[BaseTool], tools)
def _get_agent_to_use(self, task: Task) -> BaseAgent | None:
@@ -1024,6 +1027,18 @@ class Crew(FlowTrackable, BaseModel):
return self._merge_tools(tools, cast(list[BaseTool], delegation_tools))
return cast(list[BaseTool], tools)
def _inject_platform_tools(
self,
tools: list[Tool] | list[BaseTool],
task_agent: BaseAgent,
) -> list[BaseTool]:
apps = getattr(task_agent, "apps", None) or []
if hasattr(task_agent, "get_platform_tools") and apps:
platform_tools = task_agent.get_platform_tools(apps=apps)
return self._merge_tools(tools, cast(list[BaseTool], platform_tools))
return cast(list[BaseTool], tools)
def _add_multimodal_tools(
self, agent: BaseAgent, tools: list[Tool] | list[BaseTool]
) -> list[BaseTool]:
@@ -1054,10 +1069,18 @@ class Crew(FlowTrackable, BaseModel):
)
return cast(list[BaseTool], tools)
def _add_platform_tools(
self, task: Task, tools: list[Tool] | list[BaseTool]
) -> list[BaseTool]:
if task.agent:
tools = self._inject_platform_tools(tools, task.agent)
return cast(list[BaseTool], tools or [])
def _log_task_start(self, task: Task, role: str = "None"):
if self.output_log_file:
self._file_handler.log(
task_name=task.name, task=task.description, agent=role, status="started"
task_name=task.name or "unnamed_task", task=task.description, agent=role, status="started"
)
def _update_manager_tools(
@@ -1086,7 +1109,7 @@ class Crew(FlowTrackable, BaseModel):
role = task.agent.role if task.agent is not None else "None"
if self.output_log_file:
self._file_handler.log(
task_name=task.name,
task_name=task.name or "unnamed_task",
task=task.description,
agent=role,
status="completed",

2
src/crewai/utilities/training_handler.py
View File

@@ -5,7 +5,7 @@ from crewai.utilities.file_handler import PickleHandler
class CrewTrainingHandler(PickleHandler):
def save_trained_data(self, agent_id: str, trained_data: dict[int, Any]) -> None:
def save_trained_data(self, agent_id: str, trained_data: dict[str, Any]) -> None:
"""Save the trained data for a specific agent.
Args:

21
tests/agents/agent_adapters/test_base_agent_adapter.py
View File

@@ -1,4 +1,4 @@
from typing import Any, Dict, List, Optional
from typing import Any
import pytest
from pydantic import BaseModel
@@ -12,7 +12,7 @@ from crewai.utilities.token_counter_callback import TokenProcess
# Concrete implementation for testing
class ConcreteAgentAdapter(BaseAgentAdapter):
def configure_tools(
self, tools: Optional[List[BaseTool]] = None, **kwargs: Any
self, tools: list[BaseTool] | None = None, **kwargs: Any
) -> None:
# Simple implementation for testing
self.tools = tools or []
@@ -20,19 +20,19 @@ class ConcreteAgentAdapter(BaseAgentAdapter):
def execute_task(
self,
task: Any,
context: Optional[str] = None,
tools: Optional[List[Any]] = None,
context: str | None = None,
tools: list[Any] | None = None,
) -> str:
# Dummy implementation needed due to BaseAgent inheritance
return "Task executed"
def create_agent_executor(self, tools: Optional[List[BaseTool]] = None) -> Any:
def create_agent_executor(self, tools: list[BaseTool] | None = None) -> Any:
# Dummy implementation
return None
def get_delegation_tools(
self, tools: List[BaseTool], tool_map: Optional[Dict[str, BaseTool]]
) -> List[BaseTool]:
self, tools: list[BaseTool], tool_map: dict[str, BaseTool] | None
) -> list[BaseTool]:
# Dummy implementation
return []
@@ -40,10 +40,14 @@ class ConcreteAgentAdapter(BaseAgentAdapter):
# Dummy implementation
pass
def get_output_converter(self, tools: Optional[List[BaseTool]] = None) -> Any:
def get_output_converter(self, tools: list[BaseTool] | None = None) -> Any:
# Dummy implementation
return None
def get_platform_tools(self, apps: Any) -> list[BaseTool]:
# Dummy implementation
return []
def test_base_agent_adapter_initialization():
"""Test initialization of the concrete agent adapter."""
@@ -95,7 +99,6 @@ def test_configure_structured_output_method_exists():
adapter.configure_structured_output(structured_output)
# Add assertions here if configure_structured_output modifies state
# For now, just ensuring it runs without error is sufficient
pass
def test_base_agent_adapter_inherits_base_agent():

12
tests/agents/agent_builder/test_base_agent.py
View File

@@ -1,5 +1,5 @@
import hashlib
from typing import Any, List, Optional
from typing import Any
from pydantic import BaseModel
@@ -11,14 +11,16 @@ class MockAgent(BaseAgent):
def execute_task(
self,
task: Any,
context: Optional[str] = None,
tools: Optional[List[BaseTool]] = None,
context: str | None = None,
tools: list[BaseTool] | None = None,
) -> str:
return ""
def create_agent_executor(self, tools=None) -> None: ...
def get_delegation_tools(self, agents: List["BaseAgent"]): ...
def get_delegation_tools(self, agents: list["BaseAgent"]): ...
def get_platform_tools(self, apps: list[Any]): ...
def get_output_converter(
self, llm: Any, text: str, model: type[BaseModel] | None, instructions: str
@@ -31,5 +33,5 @@ def test_key():
goal="test goal",
backstory="test backstory",
)
hash = hashlib.md5("test role|test goal|test backstory".encode()).hexdigest()
hash = hashlib.md5("test role|test goal|test backstory".encode(), usedforsecurity=False).hexdigest()
assert agent.key == hash

132
tests/agents/test_agent.py
View File

@@ -1,5 +1,6 @@
"""Test Agent creation and execution basic functionality."""
# ruff: noqa: S106
import os
from unittest import mock
from unittest.mock import MagicMock, patch
@@ -2368,7 +2369,7 @@ def test_agent_from_repository(mock_get_agent, mock_get_auth_token):
tool_action = EnterpriseActionTool(
name="test_name",
description="test_description",
enterprise_action_token="test_token", # noqa: S106
enterprise_action_token="test_token",
action_name="test_action_name",
action_schema={"test": "test"},
)
@@ -2522,3 +2523,132 @@ def test_agent_from_repository_without_org_set(
"No organization currently set. We recommend setting one before using: `crewai org switch <org_id>` command.",
style="yellow",
)
def test_agent_apps_consolidated_functionality():
agent = Agent(
role="Platform Agent",
goal="Use platform tools",
backstory="Platform specialist",
apps=["gmail/create_task", "slack/update_status", "hubspot"]
)
expected = {"gmail/create_task", "slack/update_status", "hubspot"}
assert set(agent.apps) == expected
agent_apps_only = Agent(
role="App Agent",
goal="Use apps",
backstory="App specialist",
apps=["gmail", "slack"]
)
assert set(agent_apps_only.apps) == {"gmail", "slack"}
agent_default = Agent(
role="Regular Agent",
goal="Regular tasks",
backstory="Regular agent"
)
assert agent_default.apps is None
def test_agent_apps_validation():
agent = Agent(
role="Custom Agent",
goal="Test validation",
backstory="Test agent",
apps=["custom_app", "another_app/action"]
)
assert set(agent.apps) == {"custom_app", "another_app/action"}
with pytest.raises(ValueError, match=r"Invalid app format.*Apps can only have one '/' for app/action format"):
Agent(
role="Invalid Agent",
goal="Test validation",
backstory="Test agent",
apps=["app/action/invalid"]
)
@patch.object(Agent, 'get_platform_tools')
def test_app_actions_propagated_to_platform_tools(mock_get_platform_tools):
from crewai.tools import tool
@tool
def action_tool() -> str:
"""Mock action platform tool."""
return "action tool result"
mock_get_platform_tools.return_value = [action_tool]
agent = Agent(
role="Action Agent",
goal="Execute actions",
backstory="Action specialist",
apps=["gmail/send_email", "slack/update_status"]
)
task = Task(
description="Test task",
expected_output="Test output",
agent=agent
)
crew = Crew(agents=[agent], tasks=[task])
tools = crew._prepare_tools(agent, task, [])
mock_get_platform_tools.assert_called_once()
call_args = mock_get_platform_tools.call_args[1]
assert set(call_args["apps"]) == {"gmail/send_email", "slack/update_status"}
assert len(tools) >= 1
@patch.object(Agent, 'get_platform_tools')
def test_mixed_apps_and_actions_propagated(mock_get_platform_tools):
from crewai.tools import tool
@tool
def combined_tool() -> str:
"""Mock combined platform tool."""
return "combined tool result"
mock_get_platform_tools.return_value = [combined_tool]
agent = Agent(
role="Combined Agent",
goal="Use apps and actions",
backstory="Platform specialist",
apps=["gmail", "slack", "gmail/create_task", "slack/update_status"]
)
task = Task(
description="Test task",
expected_output="Test output",
agent=agent
)
crew = Crew(agents=[agent], tasks=[task])
tools = crew._prepare_tools(agent, task, [])
mock_get_platform_tools.assert_called_once()
call_args = mock_get_platform_tools.call_args[1]
expected_apps = {"gmail", "slack", "gmail/create_task", "slack/update_status"}
assert set(call_args["apps"]) == expected_apps
assert len(tools) >= 1
def test_agent_without_apps_no_platform_tools():
"""Test that agents without apps don't trigger platform tools integration."""
agent = Agent(
role="Regular Agent",
goal="Regular tasks",
backstory="Regular agent"
)
task = Task(
description="Test task",
expected_output="Test output",
agent=agent
)
crew = Crew(agents=[agent], tasks=[task])
tools = crew._prepare_tools(agent, task, [])
assert tools == []