🛠️ 工具集接口

evoagentx.tools

Tool

Tool(**kwargs)

Bases: BaseModule

Base interface for all tools. All tools must implement: - get_tool_schemas: Returns the OpenAI-compatible function schema - get_tools: Returns a list of callable functions for all tools - get_tool_descriptions: Returns a list of descriptions for all tools

Source code in evoagentx/core/module.py

def __init__(self, **kwargs):
    """
    Initializes a BaseModule instance.

    Args:
        **kwargs (Any): Keyword arguments used to initialize the instance

    Raises:
        ValidationError: When parameter validation fails
        Exception: When other errors occur during initialization
    """

    try:
        for field_name, _ in type(self).model_fields.items():
            field_value = kwargs.get(field_name, None)
            if field_value:
                kwargs[field_name] = self._process_data(field_value)
            # if field_value and isinstance(field_value, dict) and "class_name" in field_value:
            #     class_name = field_value.get("class_name")
            #     sub_cls = MODULE_REGISTRY.get_module(cls_name=class_name)
            #     kwargs[field_name] = sub_cls._create_instance(field_value)
        super().__init__(**kwargs) 
        self.init_module()
    except (ValidationError, Exception) as e:
        exception_handler = callback_manager.get_callback("exception_buffer")
        if exception_handler is None:
            error_message = get_base_module_init_error_message(
                cls=self.__class__, 
                data=kwargs, 
                errors=e
            )
            logger.error(error_message)
            raise
        else:
            exception_handler.add(e)

get_tool_schemas

get_tool_schemas() -> List[Dict[str, Any]]

Returns the OpenAI-compatible function schema for this tool. The schema follows the format used by MCP servers and OpenAI function calling.

Returns:

Type	Description
List[Dict[str, Any]]	Dict[str, Any]: The function schema in OpenAI format

Source code in evoagentx/tools/tool.py

def get_tool_schemas(self) -> List[Dict[str, Any]]:
    """
    Returns the OpenAI-compatible function schema for this tool.
    The schema follows the format used by MCP servers and OpenAI function calling.

    Returns:
        Dict[str, Any]: The function schema in OpenAI format
    """
    raise NotImplementedError("All tools must implement get_tool_schemas")

get_tools

get_tools() -> List[Callable]

Returns a list of callable functions for all tools

Returns:

Type	Description
List[Callable]	List[Callable]: A list of callable functions

Source code in evoagentx/tools/tool.py

def get_tools(self) -> List[Callable]:
    """
    Returns a list of callable functions for all tools

    Returns:
        List[Callable]: A list of callable functions
    """
    raise NotImplementedError("All tools must implement get_tools")

BaseInterpreter

BaseInterpreter(name: str = 'BaseInterpreter', **kwargs)

Bases: Tool

Base class for interpreter tools that execute code securely. Implements the standard tool interface with get_tool_schemas and execute methods.

Source code in evoagentx/tools/interpreter_base.py

def __init__(
    self, 
    name: str = 'BaseInterpreter',
    **kwargs
):
    super().__init__(name=name, **kwargs)

DockerInterpreter

DockerInterpreter(name: str = 'DockerInterpreter', image_tag: str = None, dockerfile_path: str = None, require_confirm: bool = False, print_stdout: bool = True, print_stderr: bool = True, host_directory: str = '', container_directory: str = '/home/app/', container_command: str = 'tail -f /dev/null', tmp_directory: str = '/tmp', **data)

Bases: BaseInterpreter

A Docker-based interpreter for executing Python, Bash, and R scripts in an isolated environment.

Initialize a Docker-based interpreter for executing code in an isolated environment.

Parameters:

Name	Type	Description	Default
name	str	The name of the interpreter	'DockerInterpreter'
image_tag	str	The Docker image tag to use. Must be provided if dockerfile_path is not.	None
dockerfile_path	str	Path to the Dockerfile to build. Must be provided if image_tag is not.	None
require_confirm	bool	Whether to require confirmation before executing code	False
print_stdout	bool	Whether to print stdout from code execution	True
print_stderr	bool	Whether to print stderr from code execution	True
host_directory	str	The path to the host directory to mount in the container	''
container_directory	str	The target directory inside the container	'/home/app/'
container_command	str	The command to run in the container	'tail -f /dev/null'
tmp_directory	str	The temporary directory to use for file creation in the container	'/tmp'
**data		Additional data to pass to the parent class	{}

Source code in evoagentx/tools/interpreter_docker.py

def __init__(
    self, 
    name:str = "DockerInterpreter",
    image_tag:str = None,
    dockerfile_path:str = None,
    require_confirm:bool = False,
    print_stdout:bool = True,
    print_stderr:bool = True,
    host_directory:str = "",
    container_directory:str = "/home/app/",
    container_command:str = "tail -f /dev/null",
    tmp_directory:str = "/tmp",
    **data
):
    """
    Initialize a Docker-based interpreter for executing code in an isolated environment.

    Args:
        name (str): The name of the interpreter
        image_tag (str, optional): The Docker image tag to use. Must be provided if dockerfile_path is not.
        dockerfile_path (str, optional): Path to the Dockerfile to build. Must be provided if image_tag is not.
        require_confirm (bool): Whether to require confirmation before executing code
        print_stdout (bool): Whether to print stdout from code execution
        print_stderr (bool): Whether to print stderr from code execution
        host_directory (str): The path to the host directory to mount in the container
        container_directory (str): The target directory inside the container
        container_command (str): The command to run in the container
        tmp_directory (str): The temporary directory to use for file creation in the container
        **data: Additional data to pass to the parent class
    """
    # Extract or generate schemas, descriptions, tools
    schemas = data.pop('schemas', None) or self.get_tool_schemas()
    descriptions = data.pop('descriptions', None) or self.get_tool_descriptions()
    tools = data.pop('tools', None)
    tools = self.get_tools()

    # Pass these to the parent class initialization
    super().__init__(
        name=name,
        schemas=schemas,
        descriptions=descriptions,
        tools=tools,
        **data
    )
    self.require_confirm = require_confirm
    self.print_stdout = print_stdout
    self.print_stderr = print_stderr
    self.host_directory = host_directory
    self.container_directory = container_directory
    self.container_command = container_command
    self.tmp_directory = tmp_directory

    # Initialize Docker client and container
    self.client = docker.from_env()
    self.container = None
    self.image_tag = image_tag
    self.dockerfile_path = dockerfile_path
    self._initialize_if_needed()

    # Upload directory if specified
    if self.host_directory:
        self._upload_directory_to_container(self.host_directory)

execute

execute(code: str, language: str) -> str

Executes code in a Docker container.

Parameters:

Name	Type	Description	Default
code	str	The code to execute	required
language	str	The programming language to use	required

Returns:

Name	Type	Description
str	str	The execution output

Raises:

Type	Description
RuntimeError	If container is not properly initialized or execution fails
ValueError	If code content is invalid or exceeds limits

Source code in evoagentx/tools/interpreter_docker.py

def execute(self, code: str, language: str) -> str:
    """
    Executes code in a Docker container.

    Args:
        code (str): The code to execute
        language (str): The programming language to use

    Returns:
        str: The execution output

    Raises:
        RuntimeError: If container is not properly initialized or execution fails
        ValueError: If code content is invalid or exceeds limits
    """
    if not code or not code.strip():
        raise ValueError("Code content cannot be empty")

    if not self.container:
        raise RuntimeError("Container is not initialized")

    # Check container status
    try:
        container_info = self.client.api.inspect_container(self.container.id)
        if not container_info['State']['Running']:
            raise RuntimeError("Container is not running")
    except Exception as e:
        raise RuntimeError(f"Failed to check container status: {e}")

    if self.host_directory:
        code = f"import sys; sys.path.insert(0, '{self.container_directory}');" + code

    language = self._check_language(language)

    if self.require_confirm:
        confirmation = input(f"Confirm execution of {language} code? [Y/n]: ")
        if confirmation.lower() not in ["y", "yes", ""]:
            raise RuntimeError("Execution aborted by user.")

    try:
        file_path = self._create_file_in_container(code)
        return self._run_file_in_container(file_path, language)
    except Exception as e:
        raise RuntimeError(f"Code execution failed: {e}")
    finally:
        # Clean up temporary files
        try:
            if hasattr(self, 'container') and self.container:
                self.container.exec_run(f"rm -f {file_path}")
        except Exception:
            pass  # Ignore cleanup errors

execute_script

execute_script(file_path: str, language: str = None) -> str

Reads code from a file and executes it in a Docker container.

Parameters:

Name	Type	Description	Default
file_path	str	The path to the script file to execute	required
language	str	The programming language of the code. If None, will be determined from the file extension.	None

Returns:

Name	Type	Description
str	str	The execution output

Raises:

Type	Description
FileNotFoundError	If the script file does not exist
RuntimeError	If container is not properly initialized or execution fails
ValueError	If file content is invalid or exceeds limits

Source code in evoagentx/tools/interpreter_docker.py

def execute_script(self, file_path: str, language: str = None) -> str:
    """
    Reads code from a file and executes it in a Docker container.

    Args:
        file_path (str): The path to the script file to execute
        language (str, optional): The programming language of the code. If None, will be determined from the file extension.

    Returns:
        str: The execution output

    Raises:
        FileNotFoundError: If the script file does not exist
        RuntimeError: If container is not properly initialized or execution fails
        ValueError: If file content is invalid or exceeds limits
    """
    # Check if file exists and is readable
    if not os.path.isfile(file_path):
        raise FileNotFoundError(f"Script file not found: {file_path}")

    if not os.access(file_path, os.R_OK):
        raise PermissionError(f"Cannot read script file: {file_path}")

    # Read the file content
    try:
        with open(file_path, 'r', encoding='utf-8') as f:
            code = f.read()
    except Exception as e:
        raise RuntimeError(f"Failed to read script file: {e}")

    # Execute the code
    return self.execute(code, language)

get_tool_schemas

get_tool_schemas() -> list[Dict[str, Any]]

Returns the OpenAI-compatible function schema for the Docker interpreter.

Returns:

Type	Description
list[Dict[str, Any]]	list[Dict[str, Any]]: Function schema in OpenAI format

Source code in evoagentx/tools/interpreter_docker.py

def get_tool_schemas(self) -> list[Dict[str, Any]]:
    """
    Returns the OpenAI-compatible function schema for the Docker interpreter.

    Returns:
        list[Dict[str, Any]]: Function schema in OpenAI format
    """
    schemas = [
        {
            "type": "function",
            "function": {
                "name": "execute",
                "description": "Execute code in a secure Docker container environment.",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "code": {
                            "type": "string",
                            "description": "The code to execute"
                        },
                        "language": {
                            "type": "string",
                            "description": "The programming language of the code (e.g., python, py, python3)",
                            "enum": list(self.CODE_TYPE_MAPPING.keys())
                        }
                    },
                    "required": ["code", "language"]
                }
            }
        },
        {
            "type": "function",
            "function": {
                "name": "execute_script",
                "description": "Execute code from a script file in a secure Docker container environment.",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "file_path": {
                            "type": "string",
                            "description": "The path to the script file to execute"
                        },
                        "language": {
                            "type": "string",
                            "description": "The programming language of the code. If not provided, will be determined from file extension.",
                            "enum": list(self.CODE_TYPE_MAPPING.keys())
                        }
                    },
                    "required": ["file_path"]
                }
            }
        }
    ]
    return schemas

get_tool_descriptions

get_tool_descriptions() -> List[str]

Returns a brief description of the Docker interpreter tool.

Returns:

Type	Description
List[str]	List[str]: Tool description

Source code in evoagentx/tools/interpreter_docker.py

def get_tool_descriptions(self) -> List[str]:
    """
    Returns a brief description of the Docker interpreter tool.

    Returns:
        List[str]: Tool description
    """
    return [
        "Execute code in a secure Docker container environment.",
        "Execute code from script files in a secure Docker container environment."
    ]

get_tools

get_tools() -> List[callable]

Returns a list of callable methods provided by this tool.

Returns:

Type	Description
List[callable]	List[callable]: List of callable methods

Source code in evoagentx/tools/interpreter_docker.py

def get_tools(self) -> List[callable]:
    """
    Returns a list of callable methods provided by this tool.

    Returns:
        List[callable]: List of callable methods
    """
    return [self.execute, self.execute_script]

PythonInterpreter

PythonInterpreter(name: str = 'PythonInterpreter', project_path: Optional[str] = '.', directory_names: Optional[List[str]] = [], allowed_imports: Optional[Set[str]] = None, **kwargs)

Bases: BaseInterpreter

Initialize a Python interpreter for executing code in a controlled environment.

Parameters:

Name	Type	Description	Default
name	str	The name of the interpreter	'PythonInterpreter'
project_path	Optional[str]	Path to the project directory for module resolution	'.'
directory_names	Optional[List[str]]	List of directory names to check for imports	[]
allowed_imports	Optional[Set[str]]	Set of allowed module imports to enforce security	None
**kwargs		Additional data to pass to the parent class	{}

Source code in evoagentx/tools/interpreter_python.py

def __init__(
    self, 
    name: str = 'PythonInterpreter',
    project_path:Optional[str] = ".",
    directory_names:Optional[List[str]] = [],
    allowed_imports:Optional[Set[str]] = None,
    **kwargs
):
    """
    Initialize a Python interpreter for executing code in a controlled environment.

    Args:
        name (str): The name of the interpreter
        project_path (Optional[str]): Path to the project directory for module resolution
        directory_names (Optional[List[str]]): List of directory names to check for imports
        allowed_imports (Optional[Set[str]]): Set of allowed module imports to enforce security
        **kwargs: Additional data to pass to the parent class
    """
    super().__init__(
        name=name, 
        project_path=project_path,
        directory_names=directory_names,
        allowed_imports=allowed_imports,
        **kwargs
    )
    self.allowed_imports = allowed_imports or set()

execute

execute(code: str, language: str = 'python') -> str

Analyzes and executes the provided Python code in a controlled environment.

NOTE: This method only returns content printed to stdout during execution. It does not return any values from the code itself. To see results, use print statements in your code.

WARNING: This method uses Python's exec() function internally, which executes code with full privileges. While safety checks are performed, there is still a security risk. Do not use with untrusted code.

Parameters:

Name	Type	Description	Default
code	str	The Python code to execute.	required
language	str	The programming language of the code. Defaults to "python".	'python'

Returns:

Name	Type	Description
str	str	The output of the executed code (printed content only), or a list of violations if found.

Source code in evoagentx/tools/interpreter_python.py

def execute(self, code: str, language: str = "python") -> str:
    """
    Analyzes and executes the provided Python code in a controlled environment.

    NOTE: This method only returns content printed to stdout during execution.
    It does not return any values from the code itself. To see results, use
    print statements in your code.

    WARNING: This method uses Python's exec() function internally, which executes
    code with full privileges. While safety checks are performed, there is still
    a security risk. Do not use with untrusted code.

    Args:
        code (str): The Python code to execute.
        language (str, optional): The programming language of the code. Defaults to "python".

    Returns:
        str: The output of the executed code (printed content only), or a list of violations if found.
    """
    # Verify language is python
    if language.lower() != "python":
        return f"Error: This interpreter only supports Python language. Received: {language}"

    self.visited_modules = {}
    self.namespace = {}

    # Change to the project directory and update sys.path for module resolution
    if not self.project_path:
        raise ValueError("Project path (project_path) is not set")

    if not os.path.exists(self.project_path):
        raise ValueError(f"Project path '{self.project_path}' does not exist")

    if not os.path.isdir(self.project_path):
        raise ValueError(f"Project path '{self.project_path}' is not a directory")

    os.chdir(self.project_path)
    sys.path.insert(0, self.project_path)

    if self.allowed_imports:
        violations = self._analyze_code(code)
        if violations:
            return"\n".join(violations)


    # Capture standard output during execution
    stdout_capture = io.StringIO()
    with contextlib.redirect_stdout(stdout_capture):
        try:
            # Execute the code with basic builtins
            exec(code, {"__builtins__": __builtins__})
        except Exception:
            exc_type, exc_value, exc_tb = sys.exc_info()
            error_msg = "".join(traceback.format_exception(exc_type, exc_value, exc_tb))
            return error_msg

    # Retrieve and return the captured output
    return stdout_capture.getvalue().strip()

execute_script

execute_script(file_path: str, language: str = 'python') -> str

Reads Python code from a file and executes it using the execute method.

NOTE: This method only returns content printed to stdout during execution. It does not return any values from the code itself. To see results, use print statements in your code.

WARNING: This method uses Python's exec() function internally, which executes code with full privileges. While safety checks are performed, there is still a security risk. Do not use with untrusted code.

Parameters:

Name	Type	Description	Default
file_path	str	The path to the Python file to be executed.	required
language	str	The programming language of the code. Defaults to "python".	'python'

Returns:

Name	Type	Description
str	str	The output of the executed code (printed content only), or an error message if the execution fails.

Source code in evoagentx/tools/interpreter_python.py

def execute_script(self, file_path: str, language: str = "python") -> str:
    """
    Reads Python code from a file and executes it using the `execute` method.

    NOTE: This method only returns content printed to stdout during execution.
    It does not return any values from the code itself. To see results, use
    print statements in your code.

    WARNING: This method uses Python's exec() function internally, which executes
    code with full privileges. While safety checks are performed, there is still
    a security risk. Do not use with untrusted code.

    Args:
        file_path (str): The path to the Python file to be executed.
        language (str, optional): The programming language of the code. Defaults to "python".

    Returns:
        str: The output of the executed code (printed content only), or an error message if the execution fails.
    """

    if not os.path.isfile(file_path):
        return f"Error: File '{file_path}' does not exist."

    try:
        with open(file_path, 'r', encoding=DEFAULT_ENCODING) as file:
            code = file.read()
    except Exception as e:
        return f"Error reading file: {e}"

    return self.execute(code, language)

get_tool_schemas

get_tool_schemas() -> list[Dict[str, Any]]

Returns the OpenAI-compatible function schema for the Python interpreter.

Returns:

Type	Description
list[Dict[str, Any]]	list[Dict[str, Any]]: Function schema in OpenAI format

Source code in evoagentx/tools/interpreter_python.py

def get_tool_schemas(self) -> list[Dict[str, Any]]:
    """
    Returns the OpenAI-compatible function schema for the Python interpreter.

    Returns:
        list[Dict[str, Any]]: Function schema in OpenAI format
    """
    return [{
        "type": "function",
        "function": {
            "name": "execute",
            "description": "Execute Python code in a secure environment with safety checks.",
            "parameters": {
                "type": "object",
                "properties": {
                    "code": {
                        "type": "string",
                        "description": "The Python code to execute"
                    },
                    "language": {
                        "type": "string",
                        "description": "The programming language of the code (only 'python' is supported)",
                        "enum": ["python"]
                    }
                },
                "required": ["code"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "execute_script",
            "description": "Execute Python code from a file in a secure environment with safety checks.",
            "parameters": {
                "type": "object",
                "properties": {
                    "file_path": {
                        "type": "string",
                        "description": "The path to the Python file to be executed"
                    },
                    "language": {
                        "type": "string",
                        "description": "The programming language of the code (only 'python' is supported)",
                        "enum": ["python"]
                    }
                },
                "required": ["file_path"]
            }
        }
    }]

get_tool_descriptions

get_tool_descriptions() -> list[str]

Returns a brief description of the Python interpreter tool.

Returns:

Type	Description
list[str]	list[str]: Tool descriptions

Source code in evoagentx/tools/interpreter_python.py

def get_tool_descriptions(self) -> list[str]:
    """
    Returns a brief description of the Python interpreter tool.

    Returns:
        list[str]: Tool descriptions
    """
    return [
        "Execute Python code in a secure environment with safety checks.",
        "Execute Python code from a file in a secure environment with safety checks."
    ]

SearchBase

SearchBase(name: str = 'SearchBase', num_search_pages: Optional[int] = 5, max_content_words: Optional[int] = None, **kwargs)

Bases: Tool

Base class for search tools that retrieve information from various sources. Implements the standard tool interface with get_tool_schemas and execute methods.

Initialize the base search tool.

Parameters:

Name	Type	Description	Default
name	str	Name of the tool	'SearchBase'
num_search_pages	int	Number of search results to retrieve	5
max_content_words	int	Maximum number of words to include in content, default None means no limit.	None
**kwargs		Additional keyword arguments for parent class initialization	{}

Source code in evoagentx/tools/search_base.py

def __init__(
    self, 
    name: str = "SearchBase",
    num_search_pages: Optional[int] = 5, 
    max_content_words: Optional[int] = None, 
    **kwargs
):
    """
    Initialize the base search tool.

    Args:
        name (str): Name of the tool
        num_search_pages (int): Number of search results to retrieve
        max_content_words (int): Maximum number of words to include in content, default None means no limit. 
        **kwargs: Additional keyword arguments for parent class initialization
    """ 
    # Pass to parent class initialization
    super().__init__(name=name, num_search_pages=num_search_pages, max_content_words=max_content_words, **kwargs)

SearchGoogleFree

SearchGoogleFree(name: str = 'GoogleFreeSearch', num_search_pages: Optional[int] = 5, max_content_words: Optional[int] = None, **kwargs)

Bases: SearchBase

Free Google Search tool that doesn't require API keys.

Initialize the Free Google Search tool.

Parameters:

Name	Type	Description	Default
name	str	Name of the tool	'GoogleFreeSearch'
num_search_pages	int	Number of search results to retrieve	5
max_content_words	int	Maximum number of words to include in content	None
**kwargs		Additional keyword arguments for parent class initialization	{}

Source code in evoagentx/tools/search_google_f.py

def __init__(
    self, 
    name: str = "GoogleFreeSearch",
    num_search_pages: Optional[int] = 5, 
    max_content_words: Optional[int] = None,
   **kwargs 
):
    """
    Initialize the Free Google Search tool.

    Args:
        name (str): Name of the tool
        num_search_pages (int): Number of search results to retrieve
        max_content_words (int): Maximum number of words to include in content
        **kwargs: Additional keyword arguments for parent class initialization
    """
    super().__init__(name=name, num_search_pages=num_search_pages, max_content_words=max_content_words, **kwargs)

search

search(query: str, num_search_pages: int = None, max_content_words: int = None) -> Dict[str, Any]

Searches Google for the given query and retrieves content from multiple pages.

Parameters:

Name	Type	Description	Default
query	str	The search query.	required
num_search_pages	int	Number of search results to retrieve	None
max_content_words	int	Maximum number of words to include in content, None means no limit	None

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: Contains a list of search results and optional error message.

Source code in evoagentx/tools/search_google_f.py

def search(self, query: str, num_search_pages: int = None, max_content_words: int = None) -> Dict[str, Any]:
    """
    Searches Google for the given query and retrieves content from multiple pages.

    Args:
        query (str): The search query.
        num_search_pages (int): Number of search results to retrieve
        max_content_words (int): Maximum number of words to include in content, None means no limit

    Returns:
        Dict[str, Any]: Contains a list of search results and optional error message.
    """
    # Use class defaults
    num_search_pages = num_search_pages or self.num_search_pages
    max_content_words = max_content_words or self.max_content_words 

    results = []
    try:
        # Step 1: Get top search result links
        logger.info(f"Searching Google (Free) for: {query}, num_results={num_search_pages}, max_content_words={max_content_words}")
        search_results = list(google_f_search(query, num_results=num_search_pages))
        if not search_results:
            return {"results": [], "error": "No search results found."}

        logger.info(f"Found {len(search_results)} search results")

        # Step 2: Fetch content from each page
        for url in search_results:
            try:
                title, content = self._scrape_page(url)
                if content:  # Ensure valid content exists
                    # Use the base class's content truncation method
                    display_content = self._truncate_content(content, max_content_words)

                    results.append({
                        "title": title,
                        "content": display_content,
                        "url": url,
                    })
            except Exception as e:
                logger.warning(f"Error processing URL {url}: {str(e)}")
                continue  # Skip pages that cannot be processed

        return {"results": results, "error": None}

    except Exception as e:
        logger.error(f"Error in free Google search: {str(e)}")
        return {"results": [], "error": str(e)}

get_tool_schemas

get_tool_schemas() -> List[Dict[str, Any]]

Returns the OpenAI-compatible function schema for the free Google search tool.

Returns:

Type	Description
List[Dict[str, Any]]	list[Dict[str, Any]]: Function schema in OpenAI format

Source code in evoagentx/tools/search_google_f.py

def get_tool_schemas(self) -> List[Dict[str, Any]]:
    """
    Returns the OpenAI-compatible function schema for the free Google search tool.

    Returns:
        list[Dict[str, Any]]: Function schema in OpenAI format
    """
    return [{
        "type": "function",
        "function": {
            "name": "search",
            "description": "Search Google without requiring an API key and retrieve content from search results.",
            "parameters": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "The search query to execute on Google"
                },
                "num_search_pages": {
                    "type": "integer",
                    "description": "Number of search results to retrieve. Default: 5"
                },
                "max_content_words": {
                    "type": "integer",
                    "description": "Maximum number of words to include in content per result. None means no limit. Default: None"
                }
            },
            "required": ["query"]
            }
        }
    }]

SearchWiki

SearchWiki(name: str = 'SearchWiki', num_search_pages: Optional[int] = 5, max_content_words: Optional[int] = None, max_summary_sentences: Optional[int] = None, **kwargs)

Bases: SearchBase

Initialize the Wikipedia Search tool.

Parameters:

Name	Type	Description	Default
name	str	The name of the search tool	'SearchWiki'
num_search_pages	int	Number of search results to retrieve	5
max_content_words	int	Maximum number of words to include in content, None means no limit	None
max_summary_sentences	int	Maximum number of sentences in the summary, None means no limit	None
**kwargs		Additional data to pass to the parent class	{}

Source code in evoagentx/tools/search_wiki.py

def __init__(
    self, 
    name: str = 'SearchWiki',
    num_search_pages: Optional[int] = 5, 
    max_content_words: Optional[int] = None,
    max_summary_sentences: Optional[int] = None,
    **kwargs
):
    """
    Initialize the Wikipedia Search tool.

    Args:
        name (str): The name of the search tool
        num_search_pages (int): Number of search results to retrieve
        max_content_words (int, optional): Maximum number of words to include in content, None means no limit
        max_summary_sentences (int, optional): Maximum number of sentences in the summary, None means no limit
        **kwargs: Additional data to pass to the parent class
    """

    super().__init__(
        name=name,
        num_search_pages=num_search_pages,
        max_content_words=max_content_words,
        max_summary_sentences=max_summary_sentences,
        **kwargs
    )

search

search(query: str, num_search_pages: int = None, max_content_words: int = None, max_summary_sentences: int = None) -> Dict[str, Any]

Searches Wikipedia for the given query and returns the summary and truncated full content.

Parameters:

Name	Type	Description	Default
query	str	The search query.	required
num_search_pages	int	Number of search results to retrieve	None
max_content_words	int	Maximum number of words to include in content, None means no limit	None
max_summary_sentences	int	Maximum number of sentences in the summary, None means no limit	None

Returns:

Name	Type	Description
dict	Dict[str, Any]	A dictionary with the title, summary, truncated content, and Wikipedia page link.

Source code in evoagentx/tools/search_wiki.py

def search(self, query: str, num_search_pages: int = None, max_content_words: int = None, max_summary_sentences: int = None) -> Dict[str, Any]:
    """
    Searches Wikipedia for the given query and returns the summary and truncated full content.

    Args:
        query (str): The search query.
        num_search_pages (int): Number of search results to retrieve
        max_content_words (int): Maximum number of words to include in content, None means no limit
        max_summary_sentences (int): Maximum number of sentences in the summary, None means no limit

    Returns:
        dict: A dictionary with the title, summary, truncated content, and Wikipedia page link.
    """
    num_search_pages = num_search_pages or self.num_search_pages
    max_content_words = max_content_words or self.max_content_words
    max_summary_sentences = max_summary_sentences or self.max_summary_sentences

    try:
        logger.info(f"Searching wikipedia: {query}, num_results={num_search_pages}, max_content_words={max_content_words}, max_summary_sentences={max_summary_sentences}")
        # Search for top matching titles
        search_results = wikipedia.search(query, results=num_search_pages)
        logger.info(f"Search results: {search_results}")
        if not search_results:
            return {"results": [], "error": "No search results found."}

        # Try fetching the best available page
        results = []
        for title in search_results:
            try:
                page = wikipedia.page(title, auto_suggest=False)

                # Handle the max_summary_sentences parameter
                if max_summary_sentences is not None and max_summary_sentences > 0:
                    summary = wikipedia.summary(title, sentences=max_summary_sentences)
                else:
                    # Get the full summary without limiting sentences
                    summary = wikipedia.summary(title)

                # Use the base class's content truncation method
                display_content = self._truncate_content(page.content, max_content_words)

                results.append({
                    "title": page.title,
                    "summary": summary,
                    "content": display_content,
                    "url": page.url,
                })
            except wikipedia.exceptions.DisambiguationError:
                # Skip ambiguous results and try the next
                continue
            except wikipedia.exceptions.PageError:
                # Skip non-existing pages and try the next
                continue

        # logger.info(f"get results from wikipedia: {results}")
        return {"results": results, "error": None}

    except Exception as e:
        logger.error(f"Error searching Wikipedia: {str(e)}")
        return {"results": [], "error": str(e)}

get_tool_schemas

get_tool_schemas() -> list[Dict[str, Any]]

Returns the OpenAI-compatible function schema for the Wikipedia search tool.

Returns:

Type	Description
list[Dict[str, Any]]	list[Dict[str, Any]]: Function schema in OpenAI format

Source code in evoagentx/tools/search_wiki.py

def get_tool_schemas(self) -> list[Dict[str, Any]]:
    """
    Returns the OpenAI-compatible function schema for the Wikipedia search tool.

    Returns:
        list[Dict[str, Any]]: Function schema in OpenAI format
    """
    return [{
        "type": "function",
        "function": {
            "name": "search",
            "description": "Search Wikipedia for relevant articles and content.",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "The search query to look up on Wikipedia"
                    },
                    "num_search_pages": {
                        "type": "integer",
                        "description": "Number of search results to retrieve. Default: 5"
                    },
                    "max_content_words": {
                        "type": "integer",
                        "description": "Maximum number of words to include in content per result. None means no limit. Default: None"
                    },
                    "max_summary_sentences": {
                        "type": "integer",
                        "description": "Maximum number of sentences in the summary. None means no limit. Default: None"
                    }
                },
                "required": ["query"]
            }
        }
    }]

get_tool_descriptions

get_tool_descriptions() -> list[str]

Returns a brief description of the Wikipedia search tool.

Returns:

Type	Description
list[str]	list[str]: Tool descriptions

Source code in evoagentx/tools/search_wiki.py

def get_tool_descriptions(self) -> list[str]:
    """
    Returns a brief description of the Wikipedia search tool.

    Returns:
        list[str]: Tool descriptions
    """
    return [
        "Search Wikipedia for relevant articles and content."
    ]

SearchGoogle

SearchGoogle(name: str = 'SearchGoogle', num_search_pages: Optional[int] = 5, max_content_words: Optional[int] = None, **kwargs)

Bases: SearchBase

Initialize the Google Search tool.

Parameters:

Name	Type	Description	Default
name	str	The name of the search tool	'SearchGoogle'
num_search_pages	int	Number of search results to retrieve	5
max_content_words	int	Maximum number of words to include in content, None means no limit	None
**kwargs		Additional data to pass to the parent class	{}

Source code in evoagentx/tools/search_google.py

def __init__(
    self, 
    name: str = 'SearchGoogle',
    num_search_pages: Optional[int] = 5, 
    max_content_words: Optional[int] = None,
    **kwargs
):
    """
    Initialize the Google Search tool.

    Args:
        name (str): The name of the search tool
        num_search_pages (int): Number of search results to retrieve
        max_content_words (int, optional): Maximum number of words to include in content, None means no limit
        **kwargs: Additional data to pass to the parent class
    """
    # Pass these to the parent class initialization
    super().__init__(name=name, num_search_pages=num_search_pages, max_content_words=max_content_words, **kwargs)

search

search(query: str, num_search_pages: int = None, max_content_words: int = None) -> Dict[str, Any]

Search Google using the Custom Search API and retrieve detailed search results with content snippets.

Parameters:

Name	Type	Description	Default
query	str	The search query to execute on Google	required
num_search_pages	int	Number of search results to retrieve	None
max_content_words	int	Maximum number of words to include in content, None means no limit	None

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: Contains search results and optional error message

Source code in evoagentx/tools/search_google.py

def search(self, query: str, num_search_pages: int = None, max_content_words: int = None) -> Dict[str, Any]:
    """
    Search Google using the Custom Search API and retrieve detailed search results with content snippets.

    Args:
        query (str): The search query to execute on Google
        num_search_pages (int): Number of search results to retrieve
        max_content_words (int): Maximum number of words to include in content, None means no limit

    Returns:
        Dict[str, Any]: Contains search results and optional error message
    """
    num_search_pages = num_search_pages or self.num_search_pages
    max_content_words = max_content_words or self.max_content_words
    results = []

    # Get API credentials from environment variables
    api_key = os.getenv('GOOGLE_API_KEY', '')
    search_engine_id = os.getenv('GOOGLE_SEARCH_ENGINE_ID', '')

    # print(f"api_key: {api_key}")
    # print(f"search_engine_id: {search_engine_id}")

    if not api_key or not search_engine_id:
        error_msg = (
            "API key and search engine ID are required. "
            "Please set GOOGLE_API_KEY and GOOGLE_SEARCH_ENGINE_ID environment variables. "
            "You can get these from the Google Cloud Console: https://console.cloud.google.com/apis/"
        )
        logger.error(error_msg)
        return {"results": [], "error": error_msg}

    try:
        # Step 1: Query Google Custom Search API
        logger.info(f"Searching Google for: {query}, num_results={num_search_pages}, max_content_words={max_content_words}")
        search_url = "https://www.googleapis.com/customsearch/v1"
        params = {
            "key": api_key,
            "cx": search_engine_id,
            "q": query,
            "num": num_search_pages,
        }
        response = requests.get(search_url, params=params)
        data = response.json()

        if "items" not in data:
            return {"results": [], "error": "No search results found."}

        search_results = data["items"]
        logger.info(f"Found {len(search_results)} search results")

        # Step 2: Fetch content from each valid search result
        for item in search_results:
            url = item["link"]
            title = item["title"]
            try:
                title, content = self._scrape_page(url)
                if content:  # Ensure valid content exists
                    # Use the base class's content truncation method
                    display_content = self._truncate_content(content, max_content_words)

                    results.append({
                        "title": title,
                        "content": display_content,
                        "url": url,
                    })
            except Exception as e:
                logger.warning(f"Error processing URL {url}: {str(e)}")
                continue  # Skip pages that cannot be processed

        return {"results": results, "error": None}

    except Exception as e:
        logger.error(f"Error searching Google: {str(e)}")
        return {"results": [], "error": str(e)}

get_tool_schemas

get_tool_schemas() -> list[Dict[str, Any]]

Returns the OpenAI-compatible function schema for the Google search tool.

Returns:

Type	Description
list[Dict[str, Any]]	list[Dict[str, Any]]: Function schema in OpenAI format

Source code in evoagentx/tools/search_google.py

def get_tool_schemas(self) -> list[Dict[str, Any]]:
    """
    Returns the OpenAI-compatible function schema for the Google search tool.

    Returns:
        list[Dict[str, Any]]: Function schema in OpenAI format
    """
    return [{
        "type": "function",
        "function": {
            "name": "search",
            "description": "Search Google and retrieve content from search results.",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "The search query to execute on Google"
                    },
                    "num_search_pages": {
                        "type": "integer",
                        "description": "Number of search results to retrieve. Default: 5"
                    },
                    "max_content_words": {
                        "type": "integer",
                        "description": "Maximum number of words to include in content per result. None means no limit. Default: None"
                    }
                },
                "required": ["query"]
            }
        }
    }]

get_tool_descriptions

get_tool_descriptions() -> list[str]

Returns a brief description of the Google search tool.

Returns:

Type	Description
list[str]	list[str]: Tool descriptions

Source code in evoagentx/tools/search_google.py

def get_tool_descriptions(self) -> list[str]:
    """
    Returns a brief description of the Google search tool.

    Returns:
        list[str]: Tool descriptions
    """
    return ["Search Google and retrieve content from search results."]

MCPClient

MCPClient(server_configs: StdioServerParameters | dict[str, Any] | list[StdioServerParameters | dict[str, Any]], connect_timeout: float = 120.0)

Source code in evoagentx/tools/mcp.py

def __init__(
    self, 
    server_configs: StdioServerParameters | dict[str, Any] | list[StdioServerParameters | dict[str, Any]],
    connect_timeout: float = 120.0,
):

    if isinstance(server_configs, list):
        self.server_configs = server_configs
    else:
        self.server_configs = [server_configs]

    self.event_loop = asyncio.new_event_loop()

    self.sessions:list[mcp.ClientSession] = []
    self.mcp_tools:list[list[mcp.types.Tool]] = []

    self.task = None
    self.thread_running = threading.Event()
    ## Testing
    self.working_thread = threading.Thread(target=self._run_event, daemon=True)
    self.connect_timeout = connect_timeout

    self.tools = None
    self.tool_schemas = None
    self.tool_descriptions = None

MCPToolkit

MCPToolkit(servers: Optional[list[MCPClient]] = None, config_path: Optional[str] = None, config: Optional[dict[str, Any]] = None)

Source code in evoagentx/tools/mcp.py

def __init__(self, servers: Optional[list[MCPClient]] = None, config_path: Optional[str] = None, config: Optional[dict[str, Any]] = None):

    parameters = []
    if config_path:
        parameters += self._from_config_file(config_path)
    if config:
        parameters += self._from_config(config)

    self.servers = [MCPClient(parameters)]
    if servers:
        self.servers += servers
    for server in self.servers:
        try:
            server._connect()
            logger.info("Successfully connected to MCP servers")
        except TimeoutError as e:
            logger.warning(f"Timeout connecting to MCP servers: {str(e)}. Some tools may not be available.")
        except Exception as e:
            logger.error(f"Error connecting to MCP servers: {str(e)}")

get_tools

get_tools()

Return a flattened list of all tools across all servers

Source code in evoagentx/tools/mcp.py

def get_tools(self):
    """Return a flattened list of all tools across all servers"""
    all_tools = []
    for server in self.servers:
        try:
            tools = server.get_tools()
            all_tools.extend(tools)
            logger.info(f"Added {len(tools)} tools from MCP server")
        except Exception as e:
            logger.error(f"Error getting tools from MCP server: {str(e)}")
    return all_tools