chore: fix ruff linting issues in tools module

greysonlalonde · web-flow · commit cb0efd05b4f3 · 2025-09-22T13:13:23.000-04:00
linting, args_schema default, and validator check
diff --git a/src/crewai/tools/__init__.py b/src/crewai/tools/__init__.py
@@ -1,7 +1,7 @@
-from .base_tool import BaseTool, tool, EnvVar
+from .base_tool import BaseTool, EnvVar, tool
 
 __all__ = [
     "BaseTool",
-    "tool",
     "EnvVar",
-]
+    "tool",
+]
diff --git a/src/crewai/tools/agent_tools/add_image_tool.py b/src/crewai/tools/agent_tools/add_image_tool.py
@@ -1,5 +1,3 @@
-from typing import Dict, Optional, Union
-
 from pydantic import BaseModel, Field
 
 from crewai.tools.base_tool import BaseTool
@@ -10,22 +8,24 @@
 
 class AddImageToolSchema(BaseModel):
     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"
     )
 
 
 class AddImageTool(BaseTool):
     """Tool for adding images to the content"""
 
-    name: str = Field(default_factory=lambda: i18n.tools("add_image")["name"])  # type: ignore
-    description: str = Field(default_factory=lambda: i18n.tools("add_image")["description"])  # 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[index]
+    )
     args_schema: type[BaseModel] = AddImageToolSchema
 
     def _run(
         self,
         image_url: str,
-        action: Optional[str] = None,
+        action: str | None = None,
         **kwargs,
     ) -> dict:
         action = action or i18n.tools("add_image")["default_action"]  # type: ignore
diff --git a/src/crewai/tools/agent_tools/agent_tools.py b/src/crewai/tools/agent_tools/agent_tools.py
@@ -9,9 +9,9 @@
 class AgentTools:
     """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.i18n = i18n
+        self.i18n = i18n if i18n is not None else I18N()
 
     def tools(self) -> list[BaseTool]:
         """Get all available agent tools"""
diff --git a/src/crewai/tools/agent_tools/ask_question_tool.py b/src/crewai/tools/agent_tools/ask_question_tool.py
@@ -1,5 +1,3 @@
-from typing import Optional
-
 from pydantic import BaseModel, Field
 
 from crewai.tools.agent_tools.base_agent_tools import BaseAgentTool
@@ -21,7 +19,7 @@ def _run(
         self,
         question: str,
         context: str,
-        coworker: Optional[str] = None,
+        coworker: str | None = None,
         **kwargs,
     ) -> str:
         coworker = self._get_coworker(coworker, **kwargs)
diff --git a/src/crewai/tools/agent_tools/base_agent_tools.py b/src/crewai/tools/agent_tools/base_agent_tools.py
@@ -1,5 +1,4 @@
 import logging
-from typing import Optional
 
 from pydantic import Field
 
@@ -38,7 +37,7 @@ def sanitize_agent_name(self, name: str) -> str:
         # Remove quotes and convert to lowercase
         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")
         if coworker:
             is_list = coworker.startswith("[") and coworker.endswith("]")
@@ -47,10 +46,7 @@ def _get_coworker(self, coworker: Optional[str], **kwargs) -> Optional[str]:
         return coworker
 
     def _execute(
-        self,
-        agent_name: Optional[str],
-        task: str,
-        context: Optional[str] = None
+        self, agent_name: str | None, task: str, context: str | None = None
     ) -> str:
         """
         Execute delegation to an agent with case-insensitive and whitespace-tolerant matching.
@@ -77,7 +73,9 @@ def _execute(
             # when it should look like this:
             # {"task": "....", "coworker": "...."}
             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]
             logger.debug(f"Available agents: {available_agents}")
@@ -87,38 +85,47 @@ def _execute(
                 for available_agent in self.agents
                 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:
             # Handle specific exceptions that might occur during role name processing
             return self.i18n.errors("agent_tool_unexisting_coworker").format(
                 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:
             # No matching agent found after sanitization
             return self.i18n.errors("agent_tool_unexisting_coworker").format(
                 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:
             task_with_assigned_agent = Task(
                 description=task,
-                agent=agent,
-                expected_output=agent.i18n.slice("manager_request"),
-                i18n=agent.i18n,
+                agent=selected_agent,
+                expected_output=selected_agent.i18n.slice("manager_request"),
+                i18n=selected_agent.i18n,
+            )
+            logger.debug(
+                f"Created task for agent '{self.sanitize_agent_name(selected_agent.role)}': {task}"
             )
-            logger.debug(f"Created task for agent '{self.sanitize_agent_name(agent.role)}': {task}")
-            return agent.execute_task(task_with_assigned_agent, context)
+            return selected_agent.execute_task(task_with_assigned_agent, context)
         except Exception as e:
             # Handle task creation or execution errors
             return self.i18n.errors("agent_tool_execution_error").format(
-                agent_role=self.sanitize_agent_name(agent.role),
-                error=str(e)
+                agent_role=self.sanitize_agent_name(selected_agent.role), error=str(e)
             )
diff --git a/src/crewai/tools/agent_tools/delegate_work_tool.py b/src/crewai/tools/agent_tools/delegate_work_tool.py
@@ -1,5 +1,3 @@
-from typing import Optional
-
 from pydantic import BaseModel, Field
 
 from crewai.tools.agent_tools.base_agent_tools import BaseAgentTool
@@ -23,7 +21,7 @@ def _run(
         self,
         task: str,
         context: str,
-        coworker: Optional[str] = None,
+        coworker: str | None = None,
         **kwargs,
     ) -> str:
         coworker = self._get_coworker(coworker, **kwargs)
diff --git a/src/crewai/tools/base_tool.py b/src/crewai/tools/base_tool.py
@@ -1,7 +1,8 @@
 import asyncio
 from abc import ABC, abstractmethod
+from collections.abc import Callable
 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 (
     BaseModel,
@@ -19,7 +20,7 @@ class EnvVar(BaseModel):
     name: str
     description: str
     required: bool = True
-    default: Optional[str] = None
+    default: str | None = None
 
 
 class BaseTool(BaseModel, ABC):
@@ -32,10 +33,10 @@ class _ArgsSchemaPlaceholder(PydanticBaseModel):
     """The unique name of the tool that clearly communicates its purpose."""
     description: str
     """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."""
-    args_schema: Type[PydanticBaseModel] = Field(
-        default_factory=_ArgsSchemaPlaceholder, validate_default=True
+    args_schema: type[PydanticBaseModel] = Field(
+        default=_ArgsSchemaPlaceholder, validate_default=True
     )
     """The schema for the arguments that the tool accepts."""
     description_updated: bool = False
@@ -52,9 +53,9 @@ class _ArgsSchemaPlaceholder(PydanticBaseModel):
     @field_validator("args_schema", mode="before")
     @classmethod
     def _default_args_schema(
-        cls, v: Type[PydanticBaseModel]
-    ) -> Type[PydanticBaseModel]:
-        if not isinstance(v, cls._ArgsSchemaPlaceholder):
+        cls, v: type[PydanticBaseModel]
+    ) -> type[PydanticBaseModel]:
+        if v != cls._ArgsSchemaPlaceholder:
             return v
 
         return type(
@@ -139,7 +140,7 @@ def from_langchain(cls, tool: Any) -> "BaseTool":
             # Infer args_schema from the function signature if not provided
             func_signature = signature(tool.func)
             annotations = func_signature.parameters
-            args_fields = {}
+            args_fields: dict[str, Any] = {}
             for name, param in annotations.items():
                 if name != "self":
                     param_annotation = (
@@ -247,7 +248,7 @@ def from_langchain(cls, tool: Any) -> "Tool":
             # Infer args_schema from the function signature if not provided
             func_signature = signature(tool.func)
             annotations = func_signature.parameters
-            args_fields = {}
+            args_fields: dict[str, Any] = {}
             for name, param in annotations.items():
                 if name != "self":
                     param_annotation = (
diff --git a/src/crewai/tools/tool_calling.py b/src/crewai/tools/tool_calling.py
@@ -1,4 +1,4 @@
-from typing import Any, Dict, Optional
+from typing import Any
 
 from pydantic import BaseModel, Field
 from pydantic import BaseModel as PydanticBaseModel
@@ -7,7 +7,7 @@
 
 class ToolCalling(BaseModel):
     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."
     )
 
@@ -16,6 +16,6 @@ class InstructorToolCalling(PydanticBaseModel):
     tool_name: str = PydanticField(
         ..., 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."
     )

Original file line number	Diff line number	Diff line change
`@@ -1,4 +1,4 @@`
`1`		`-from typing import Any, Dict, Optional`
	`1`	`+from typing import Any`
`2`	`2`
`3`	`3`	`from pydantic import BaseModel, Field`
`4`	`4`	`from pydantic import BaseModel as PydanticBaseModel`
`@@ -7,7 +7,7 @@`
`7`	`7`
`8`	`8`	`class ToolCalling(BaseModel):`
`9`	`9`	`tool_name: str = Field(..., description="The name of the tool to be called.")`
`10`		`- arguments: Optional[Dict[str, Any]] = Field(`
	`10`	`+ arguments: dict[str, Any] \| None = Field(`
`11`	`11`	`..., description="A dictionary of arguments to be passed to the tool."`
`12`	`12`	`)`
`13`	`13`
`@@ -16,6 +16,6 @@ class InstructorToolCalling(PydanticBaseModel):`
`16`	`16`	`tool_name: str = PydanticField(`
`17`	`17`	`..., description="The name of the tool to be called."`
`18`	`18`	`)`
`19`		`- arguments: Optional[Dict[str, Any]] = PydanticField(`
	`19`	`+ arguments: dict[str, Any] \| None = PydanticField(`
`20`	`20`	`..., description="A dictionary of arguments to be passed to the tool."`
`21`	`21`	`)`