chore: fix ruff linting issues in tools module
Some checks failed
CodeQL Advanced / Analyze (actions) (push) Has been cancelled
CodeQL Advanced / Analyze (python) (push) Has been cancelled
Notify Downstream / notify-downstream (push) Has been cancelled
Update Test Durations / update-durations (3.13) (push) Has been cancelled
Update Test Durations / update-durations (3.10) (push) Has been cancelled
Update Test Durations / update-durations (3.11) (push) Has been cancelled
Update Test Durations / update-durations (3.12) (push) Has been cancelled

linting, args_schema default, and validator check
This commit is contained in:
Greyson LaLonde
2025-09-22 13:13:23 -04:00
committed by GitHub
parent db5f565dea
commit cb0efd05b4
8 changed files with 54 additions and 50 deletions

View File

@@ -1,7 +1,7 @@
from .base_tool import BaseTool, tool, EnvVar from .base_tool import BaseTool, EnvVar, tool
__all__ = [ __all__ = [
"BaseTool", "BaseTool",
"tool",
"EnvVar", "EnvVar",
] "tool",
]

View File

@@ -1,5 +1,3 @@
from typing import Dict, Optional, Union
from pydantic import BaseModel, Field from pydantic import BaseModel, Field
from crewai.tools.base_tool import BaseTool from crewai.tools.base_tool import BaseTool
@@ -10,7 +8,7 @@ i18n = I18N()
class AddImageToolSchema(BaseModel): class AddImageToolSchema(BaseModel):
image_url: str = Field(..., description="The URL or path of the image to add") image_url: str = Field(..., description="The URL or path of the image to add")
action: Optional[str] = Field( action: str | None = Field(
default=None, description="Optional context or question about the image" default=None, description="Optional context or question about the image"
) )
@@ -18,14 +16,16 @@ class AddImageToolSchema(BaseModel):
class AddImageTool(BaseTool): class AddImageTool(BaseTool):
"""Tool for adding images to the content""" """Tool for adding images to the content"""
name: str = Field(default_factory=lambda: i18n.tools("add_image")["name"]) # type: ignore name: str = Field(default_factory=lambda: i18n.tools("add_image")["name"]) # type: ignore[index]
description: str = Field(default_factory=lambda: i18n.tools("add_image")["description"]) # type: ignore description: str = Field(
default_factory=lambda: i18n.tools("add_image")["description"] # type: ignore[index]
)
args_schema: type[BaseModel] = AddImageToolSchema args_schema: type[BaseModel] = AddImageToolSchema
def _run( def _run(
self, self,
image_url: str, image_url: str,
action: Optional[str] = None, action: str | None = None,
**kwargs, **kwargs,
) -> dict: ) -> dict:
action = action or i18n.tools("add_image")["default_action"] # type: ignore action = action or i18n.tools("add_image")["default_action"] # type: ignore

View File

@@ -9,9 +9,9 @@ from .delegate_work_tool import DelegateWorkTool
class AgentTools: class AgentTools:
"""Manager class for agent-related tools""" """Manager class for agent-related tools"""
def __init__(self, agents: list[BaseAgent], i18n: I18N = I18N()): def __init__(self, agents: list[BaseAgent], i18n: I18N | None = None):
self.agents = agents self.agents = agents
self.i18n = i18n self.i18n = i18n if i18n is not None else I18N()
def tools(self) -> list[BaseTool]: def tools(self) -> list[BaseTool]:
"""Get all available agent tools""" """Get all available agent tools"""

View File

@@ -1,5 +1,3 @@
from typing import Optional
from pydantic import BaseModel, Field from pydantic import BaseModel, Field
from crewai.tools.agent_tools.base_agent_tools import BaseAgentTool from crewai.tools.agent_tools.base_agent_tools import BaseAgentTool
@@ -21,7 +19,7 @@ class AskQuestionTool(BaseAgentTool):
self, self,
question: str, question: str,
context: str, context: str,
coworker: Optional[str] = None, coworker: str | None = None,
**kwargs, **kwargs,
) -> str: ) -> str:
coworker = self._get_coworker(coworker, **kwargs) coworker = self._get_coworker(coworker, **kwargs)

View File

@@ -1,5 +1,4 @@
import logging import logging
from typing import Optional
from pydantic import Field from pydantic import Field
@@ -38,7 +37,7 @@ class BaseAgentTool(BaseTool):
# Remove quotes and convert to lowercase # Remove quotes and convert to lowercase
return normalized.replace('"', "").casefold() return normalized.replace('"', "").casefold()
def _get_coworker(self, coworker: Optional[str], **kwargs) -> Optional[str]: def _get_coworker(self, coworker: str | None, **kwargs) -> str | None:
coworker = coworker or kwargs.get("co_worker") or kwargs.get("coworker") coworker = coworker or kwargs.get("co_worker") or kwargs.get("coworker")
if coworker: if coworker:
is_list = coworker.startswith("[") and coworker.endswith("]") is_list = coworker.startswith("[") and coworker.endswith("]")
@@ -47,10 +46,7 @@ class BaseAgentTool(BaseTool):
return coworker return coworker
def _execute( def _execute(
self, self, agent_name: str | None, task: str, context: str | None = None
agent_name: Optional[str],
task: str,
context: Optional[str] = None
) -> str: ) -> str:
""" """
Execute delegation to an agent with case-insensitive and whitespace-tolerant matching. Execute delegation to an agent with case-insensitive and whitespace-tolerant matching.
@@ -77,7 +73,9 @@ class BaseAgentTool(BaseTool):
# when it should look like this: # when it should look like this:
# {"task": "....", "coworker": "...."} # {"task": "....", "coworker": "...."}
sanitized_name = self.sanitize_agent_name(agent_name) sanitized_name = self.sanitize_agent_name(agent_name)
logger.debug(f"Sanitized agent name from '{agent_name}' to '{sanitized_name}'") logger.debug(
f"Sanitized agent name from '{agent_name}' to '{sanitized_name}'"
)
available_agents = [agent.role for agent in self.agents] available_agents = [agent.role for agent in self.agents]
logger.debug(f"Available agents: {available_agents}") logger.debug(f"Available agents: {available_agents}")
@@ -87,38 +85,47 @@ class BaseAgentTool(BaseTool):
for available_agent in self.agents for available_agent in self.agents
if self.sanitize_agent_name(available_agent.role) == sanitized_name if self.sanitize_agent_name(available_agent.role) == sanitized_name
] ]
logger.debug(f"Found {len(agent)} matching agents for role '{sanitized_name}'") logger.debug(
f"Found {len(agent)} matching agents for role '{sanitized_name}'"
)
except (AttributeError, ValueError) as e: except (AttributeError, ValueError) as e:
# Handle specific exceptions that might occur during role name processing # Handle specific exceptions that might occur during role name processing
return self.i18n.errors("agent_tool_unexisting_coworker").format( return self.i18n.errors("agent_tool_unexisting_coworker").format(
coworkers="\n".join( coworkers="\n".join(
[f"- {self.sanitize_agent_name(agent.role)}" for agent in self.agents] [
f"- {self.sanitize_agent_name(agent.role)}"
for agent in self.agents
]
), ),
error=str(e) error=str(e),
) )
if not agent: if not agent:
# No matching agent found after sanitization # No matching agent found after sanitization
return self.i18n.errors("agent_tool_unexisting_coworker").format( return self.i18n.errors("agent_tool_unexisting_coworker").format(
coworkers="\n".join( coworkers="\n".join(
[f"- {self.sanitize_agent_name(agent.role)}" for agent in self.agents] [
f"- {self.sanitize_agent_name(agent.role)}"
for agent in self.agents
]
), ),
error=f"No agent found with role '{sanitized_name}'" error=f"No agent found with role '{sanitized_name}'",
) )
agent = agent[0] selected_agent = agent[0]
try: try:
task_with_assigned_agent = Task( task_with_assigned_agent = Task(
description=task, description=task,
agent=agent, agent=selected_agent,
expected_output=agent.i18n.slice("manager_request"), expected_output=selected_agent.i18n.slice("manager_request"),
i18n=agent.i18n, i18n=selected_agent.i18n,
) )
logger.debug(f"Created task for agent '{self.sanitize_agent_name(agent.role)}': {task}") logger.debug(
return agent.execute_task(task_with_assigned_agent, context) f"Created task for agent '{self.sanitize_agent_name(selected_agent.role)}': {task}"
)
return selected_agent.execute_task(task_with_assigned_agent, context)
except Exception as e: except Exception as e:
# Handle task creation or execution errors # Handle task creation or execution errors
return self.i18n.errors("agent_tool_execution_error").format( return self.i18n.errors("agent_tool_execution_error").format(
agent_role=self.sanitize_agent_name(agent.role), agent_role=self.sanitize_agent_name(selected_agent.role), error=str(e)
error=str(e)
) )

View File

@@ -1,5 +1,3 @@
from typing import Optional
from pydantic import BaseModel, Field from pydantic import BaseModel, Field
from crewai.tools.agent_tools.base_agent_tools import BaseAgentTool from crewai.tools.agent_tools.base_agent_tools import BaseAgentTool
@@ -23,7 +21,7 @@ class DelegateWorkTool(BaseAgentTool):
self, self,
task: str, task: str,
context: str, context: str,
coworker: Optional[str] = None, coworker: str | None = None,
**kwargs, **kwargs,
) -> str: ) -> str:
coworker = self._get_coworker(coworker, **kwargs) coworker = self._get_coworker(coworker, **kwargs)

View File

@@ -1,7 +1,8 @@
import asyncio import asyncio
from abc import ABC, abstractmethod from abc import ABC, abstractmethod
from collections.abc import Callable
from inspect import signature from inspect import signature
from typing import Any, Callable, Type, get_args, get_origin, Optional, List from typing import Any, get_args, get_origin
from pydantic import ( from pydantic import (
BaseModel, BaseModel,
@@ -19,7 +20,7 @@ class EnvVar(BaseModel):
name: str name: str
description: str description: str
required: bool = True required: bool = True
default: Optional[str] = None default: str | None = None
class BaseTool(BaseModel, ABC): class BaseTool(BaseModel, ABC):
@@ -32,10 +33,10 @@ class BaseTool(BaseModel, ABC):
"""The unique name of the tool that clearly communicates its purpose.""" """The unique name of the tool that clearly communicates its purpose."""
description: str description: str
"""Used to tell the model how/when/why to use the tool.""" """Used to tell the model how/when/why to use the tool."""
env_vars: List[EnvVar] = [] env_vars: list[EnvVar] = []
"""List of environment variables used by the tool.""" """List of environment variables used by the tool."""
args_schema: Type[PydanticBaseModel] = Field( args_schema: type[PydanticBaseModel] = Field(
default_factory=_ArgsSchemaPlaceholder, validate_default=True default=_ArgsSchemaPlaceholder, validate_default=True
) )
"""The schema for the arguments that the tool accepts.""" """The schema for the arguments that the tool accepts."""
description_updated: bool = False description_updated: bool = False
@@ -52,9 +53,9 @@ class BaseTool(BaseModel, ABC):
@field_validator("args_schema", mode="before") @field_validator("args_schema", mode="before")
@classmethod @classmethod
def _default_args_schema( def _default_args_schema(
cls, v: Type[PydanticBaseModel] cls, v: type[PydanticBaseModel]
) -> Type[PydanticBaseModel]: ) -> type[PydanticBaseModel]:
if not isinstance(v, cls._ArgsSchemaPlaceholder): if v != cls._ArgsSchemaPlaceholder:
return v return v
return type( return type(
@@ -139,7 +140,7 @@ class BaseTool(BaseModel, ABC):
# Infer args_schema from the function signature if not provided # Infer args_schema from the function signature if not provided
func_signature = signature(tool.func) func_signature = signature(tool.func)
annotations = func_signature.parameters annotations = func_signature.parameters
args_fields = {} args_fields: dict[str, Any] = {}
for name, param in annotations.items(): for name, param in annotations.items():
if name != "self": if name != "self":
param_annotation = ( param_annotation = (
@@ -247,7 +248,7 @@ class Tool(BaseTool):
# Infer args_schema from the function signature if not provided # Infer args_schema from the function signature if not provided
func_signature = signature(tool.func) func_signature = signature(tool.func)
annotations = func_signature.parameters annotations = func_signature.parameters
args_fields = {} args_fields: dict[str, Any] = {}
for name, param in annotations.items(): for name, param in annotations.items():
if name != "self": if name != "self":
param_annotation = ( param_annotation = (

View File

@@ -1,4 +1,4 @@
from typing import Any, Dict, Optional from typing import Any
from pydantic import BaseModel, Field from pydantic import BaseModel, Field
from pydantic import BaseModel as PydanticBaseModel from pydantic import BaseModel as PydanticBaseModel
@@ -7,7 +7,7 @@ from pydantic import Field as PydanticField
class ToolCalling(BaseModel): class ToolCalling(BaseModel):
tool_name: str = Field(..., description="The name of the tool to be called.") tool_name: str = Field(..., description="The name of the tool to be called.")
arguments: Optional[Dict[str, Any]] = Field( arguments: dict[str, Any] | None = Field(
..., description="A dictionary of arguments to be passed to the tool." ..., description="A dictionary of arguments to be passed to the tool."
) )
@@ -16,6 +16,6 @@ class InstructorToolCalling(PydanticBaseModel):
tool_name: str = PydanticField( tool_name: str = PydanticField(
..., description="The name of the tool to be called." ..., description="The name of the tool to be called."
) )
arguments: Optional[Dict[str, Any]] = PydanticField( arguments: dict[str, Any] | None = PydanticField(
..., description="A dictionary of arguments to be passed to the tool." ..., description="A dictionary of arguments to be passed to the tool."
) )