import json import logging import warnings from abc import ABC, abstractmethod from inspect import signature from typing import Any, Callable, Dict, Optional, Type, Union, get_args, get_origin logger = logging.getLogger(__name__) from pydantic import ( BaseModel, ConfigDict, Field, PydanticDeprecatedSince20, create_model, validator, ) from pydantic import BaseModel as PydanticBaseModel from crewai.tools.structured_tool import CrewStructuredTool # Ignore all "PydanticDeprecatedSince20" warnings globally warnings.filterwarnings("ignore", category=PydanticDeprecatedSince20) class BaseTool(BaseModel, ABC): class _ArgsSchemaPlaceholder(PydanticBaseModel): pass model_config = ConfigDict() name: str """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.""" args_schema: Type[PydanticBaseModel] = Field(default_factory=_ArgsSchemaPlaceholder) """The schema for the arguments that the tool accepts.""" description_updated: bool = False """Flag to check if the description has been updated.""" cache_function: Callable = lambda _args=None, _result=None: True """Function that will be used to determine if the tool should be cached, should return a boolean. If None, the tool will be cached.""" result_as_answer: bool = False """Flag to check if the tool should be the final agent answer.""" @validator("args_schema", always=True, pre=True) def _default_args_schema( cls, v: Type[PydanticBaseModel] ) -> Type[PydanticBaseModel]: if not isinstance(v, cls._ArgsSchemaPlaceholder): return v return type( f"{cls.__name__}Schema", (PydanticBaseModel,), { "__annotations__": { k: v for k, v in cls._run.__annotations__.items() if k != "return" }, }, ) def model_post_init(self, __context: Any) -> None: self._generate_description() super().model_post_init(__context) def run( self, *args: Any, **kwargs: Any, ) -> Any: print(f"Using Tool: {self.name}") return self._run(*args, **kwargs) @abstractmethod def _run( self, *args: Any, **kwargs: Any, ) -> Any: """Here goes the actual implementation of the tool.""" def invoke( self, input: Union[str, dict], config: Optional[dict] = None, **kwargs: Any ) -> Any: """Main method for tool execution. This method provides a fallback implementation for models that don't support function calling natively (like QwQ-32B-Preview and deepseek-chat). It parses the input and calls the _run method with the appropriate arguments. Args: input: Either a string (raw or JSON) or a dictionary of arguments config: Optional configuration dictionary **kwargs: Additional keyword arguments to pass to _run Returns: The result of calling the tool's _run method Raises: ValueError: If input is neither a string nor a dictionary """ if not isinstance(input, (str, dict)): raise ValueError(f"Input must be string or dict, got {type(input)}") if isinstance(input, str): # Try to parse as JSON if it's a string try: input = json.loads(input) logger.debug(f"Successfully parsed JSON input: {input}") except json.JSONDecodeError as e: # If not valid JSON, pass as a single argument logger.debug(f"Input string is not JSON format: {e}") return self._run(input) if not isinstance(input, dict): # If input is not a dict after parsing, pass it directly logger.debug(f"Using non-dict input directly: {input}") return self._run(input) # Get the expected arguments from the schema if hasattr(self, 'args_schema') and self.args_schema is not None: try: # Extract argument names from the schema arg_names = list(self.args_schema.model_json_schema()["properties"].keys()) # Filter the input to only include valid arguments filtered_args = {} for k in input.keys(): if k in arg_names: filtered_args[k] = input[k] else: logger.warning(f"Ignoring unexpected argument: {k}") logger.debug(f"Calling _run with filtered arguments: {filtered_args}") # Call _run with the filtered arguments return self._run(**filtered_args) except Exception as e: # Fallback to passing the entire input dict if schema parsing fails logger.warning(f"Schema parsing failed, using raw input: {e}") # If we couldn't parse the schema or there was an error, just pass the input dict logger.debug(f"Calling _run with unfiltered arguments: {input}") return self._run(**input) def to_structured_tool(self) -> CrewStructuredTool: """Convert this tool to a CrewStructuredTool instance.""" self._set_args_schema() return CrewStructuredTool( name=self.name, description=self.description, args_schema=self.args_schema, func=self._run, result_as_answer=self.result_as_answer, ) @classmethod def from_langchain(cls, tool: Any) -> "BaseTool": """Create a Tool instance from a CrewStructuredTool. This method takes a CrewStructuredTool object and converts it into a Tool instance. It ensures that the provided tool has a callable 'func' attribute and infers the argument schema if not explicitly provided. """ if not hasattr(tool, "func") or not callable(tool.func): raise ValueError("The provided tool must have a callable 'func' attribute.") args_schema = getattr(tool, "args_schema", None) if args_schema is None: # Infer args_schema from the function signature if not provided func_signature = signature(tool.func) annotations = func_signature.parameters args_fields = {} for name, param in annotations.items(): if name != "self": param_annotation = ( param.annotation if param.annotation != param.empty else Any ) field_info = Field( default=..., description="", ) args_fields[name] = (param_annotation, field_info) if args_fields: args_schema = create_model(f"{tool.name}Input", **args_fields) else: # Create a default schema with no fields if no parameters are found args_schema = create_model( f"{tool.name}Input", __base__=PydanticBaseModel ) return cls( name=getattr(tool, "name", "Unnamed Tool"), description=getattr(tool, "description", ""), func=tool.func, args_schema=args_schema, ) def _set_args_schema(self): if self.args_schema is None: class_name = f"{self.__class__.__name__}Schema" self.args_schema = type( class_name, (PydanticBaseModel,), { "__annotations__": { k: v for k, v in self._run.__annotations__.items() if k != "return" }, }, ) def _generate_description(self): args_schema = { name: { "description": field.description, "type": BaseTool._get_arg_annotations(field.annotation), } for name, field in self.args_schema.model_fields.items() } self.description = f"Tool Name: {self.name}\nTool Arguments: {args_schema}\nTool Description: {self.description}" @staticmethod def _get_arg_annotations(annotation: type[Any] | None) -> str: if annotation is None: return "None" origin = get_origin(annotation) args = get_args(annotation) if origin is None: return ( annotation.__name__ if hasattr(annotation, "__name__") else str(annotation) ) if args: args_str = ", ".join(BaseTool._get_arg_annotations(arg) for arg in args) return f"{origin.__name__}[{args_str}]" return origin.__name__ class Tool(BaseTool): """The function that will be executed when the tool is called.""" func: Callable def _run(self, *args: Any, **kwargs: Any) -> Any: return self.func(*args, **kwargs) @classmethod def from_langchain(cls, tool: Any) -> "Tool": """Create a Tool instance from a CrewStructuredTool. This method takes a CrewStructuredTool object and converts it into a Tool instance. It ensures that the provided tool has a callable 'func' attribute and infers the argument schema if not explicitly provided. Args: tool (Any): The CrewStructuredTool object to be converted. Returns: Tool: A new Tool instance created from the provided CrewStructuredTool. Raises: ValueError: If the provided tool does not have a callable 'func' attribute. """ if not hasattr(tool, "func") or not callable(tool.func): raise ValueError("The provided tool must have a callable 'func' attribute.") args_schema = getattr(tool, "args_schema", None) if args_schema is None: # Infer args_schema from the function signature if not provided func_signature = signature(tool.func) annotations = func_signature.parameters args_fields = {} for name, param in annotations.items(): if name != "self": param_annotation = ( param.annotation if param.annotation != param.empty else Any ) field_info = Field( default=..., description="", ) args_fields[name] = (param_annotation, field_info) if args_fields: args_schema = create_model(f"{tool.name}Input", **args_fields) else: # Create a default schema with no fields if no parameters are found args_schema = create_model( f"{tool.name}Input", __base__=PydanticBaseModel ) return cls( name=getattr(tool, "name", "Unnamed Tool"), description=getattr(tool, "description", ""), func=tool.func, args_schema=args_schema, ) def to_langchain( tools: list[BaseTool | CrewStructuredTool], ) -> list[CrewStructuredTool]: return [t.to_structured_tool() if isinstance(t, BaseTool) else t for t in tools] def tool(*args): """ Decorator to create a tool from a function. """ def _make_with_name(tool_name: str) -> Callable: def _make_tool(f: Callable) -> BaseTool: if f.__doc__ is None: raise ValueError("Function must have a docstring") if f.__annotations__ is None: raise ValueError("Function must have type annotations") class_name = "".join(tool_name.split()).title() args_schema = type( class_name, (PydanticBaseModel,), { "__annotations__": { k: v for k, v in f.__annotations__.items() if k != "return" }, }, ) return Tool( name=tool_name, description=f.__doc__, func=f, args_schema=args_schema, ) return _make_tool if len(args) == 1 and callable(args[0]): return _make_with_name(args[0].__name__)(args[0]) if len(args) == 1 and isinstance(args[0], str): return _make_with_name(args[0]) raise ValueError("Invalid arguments")