🛠️ Tools

evoagentx.tools

Tool

Tool(**kwargs)

Bases: BaseModule

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)

required class-attribute instance-attribute

required: Optional[List[str]] = None

inputs: {"input_name": {"type": "string", "description": "input description"}, ...}

DockerInterpreterToolkit

DockerInterpreterToolkit(name: str = 'DockerInterpreterToolkit', image_tag: Optional[str] = None, dockerfile_path: Optional[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', storage_handler: FileStorageHandler = None, auto_cleanup: bool = True, auto_destroy: bool = True, **kwargs)

Bases: Toolkit

Source code in evoagentx/tools/interpreter_docker.py

def __init__(
    self,
    name: str = "DockerInterpreterToolkit",
    image_tag: Optional[str] = None,
    dockerfile_path: Optional[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",
    storage_handler: FileStorageHandler = None,
    auto_cleanup: bool = True,
    auto_destroy: bool = True,
    **kwargs
):
    # Initialize storage handler if not provided
    if storage_handler is None:
        from .storage_file import LocalStorageHandler
        storage_handler = LocalStorageHandler(base_path="./workplace/docker")

    # Create the shared Docker interpreter instance with storage handler
    docker_interpreter = DockerInterpreter(
        name="DockerInterpreter",
        image_tag=image_tag,
        dockerfile_path=dockerfile_path,
        require_confirm=require_confirm,
        print_stdout=print_stdout,
        print_stderr=print_stderr,
        host_directory=host_directory,
        container_directory=container_directory,
        container_command=container_command,
        tmp_directory=tmp_directory,
        storage_handler=storage_handler,
        auto_cleanup=auto_cleanup,
        auto_destroy=auto_destroy,
        **kwargs
    )

    # Create tools with the shared interpreter
    tools = [
        DockerExecuteTool(docker_interpreter=docker_interpreter),
        DockerExecuteScriptTool(docker_interpreter=docker_interpreter)
    ]

    # Initialize parent with tools
    super().__init__(name=name, tools=tools)

    # Store docker_interpreter as instance variable
    self.docker_interpreter = docker_interpreter
    self.storage_handler = storage_handler
    self.auto_cleanup = auto_cleanup
    self.auto_destroy = auto_destroy

cleanup

cleanup()

Clean up the Docker interpreter and storage handler.

Source code in evoagentx/tools/interpreter_docker.py

def cleanup(self):
    """Clean up the Docker interpreter and storage handler."""
    try:
        if hasattr(self, 'auto_cleanup') and self.auto_cleanup:
            if hasattr(self, 'docker_interpreter') and self.docker_interpreter:
                self.docker_interpreter.cleanup()
            if hasattr(self, 'storage_handler') and self.storage_handler:
                try:
                    self.storage_handler.cleanup()
                except Exception:
                    pass
    except Exception:
        pass  # Silently ignore cleanup errors

__del__

__del__()

Cleanup when toolkit is destroyed.

Source code in evoagentx/tools/interpreter_docker.py

def __del__(self):
    """Cleanup when toolkit is destroyed."""
    try:
        if hasattr(self, 'auto_destroy') and self.auto_destroy:
            self.cleanup()
    except Exception:
        pass  # Silently ignore errors during destruction

BrowserToolkit

BrowserToolkit(name: str = 'BrowserToolkit', browser_type: str = 'chrome', headless: bool = False, timeout: int = 10, **kwargs)

Bases: Toolkit

Browser toolkit with auto-initialization and cleanup.

The browser is automatically initialized when any tool is first used, and automatically closed when the toolkit instance is destroyed. No explicit initialization or cleanup is required.

Source code in evoagentx/tools/browser_tool.py

def __init__(
    self,
    name: str = "BrowserToolkit",
    browser_type: str = "chrome",
    headless: bool = False,
    timeout: int = 10,
    **kwargs

):
    # Create the shared browser tool instance
    browser_tool = BrowserBase(
        name="BrowserBase",
        browser_type=browser_type,
        headless=headless,
        timeout=timeout,
        **kwargs
    )

    # Create tools with the shared browser tool instance
    # Note: Browser auto-initializes when first used and auto-closes when destroyed
    tools = [
        NavigateToUrlTool(browser_tool=browser_tool),
        InputTextTool(browser_tool=browser_tool),
        BrowserClickTool(browser_tool=browser_tool),
        BrowserSnapshotTool(browser_tool=browser_tool),
        BrowserConsoleMessagesTool(browser_tool=browser_tool)
    ]

    # Initialize parent with tools
    super().__init__(name=name, tools=tools)

    # Store browser_tool as instance variable
    self.browser_tool = browser_tool

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)

    # Initialize servers list
    self.servers = []

    # Only create MCPClient if we have parameters
    if parameters:
        self.servers.append(MCPClient(parameters))

    # Add additional servers if provided
    if servers:
        self.servers.extend(servers)

    # Connect to all servers
    failed_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.")
            failed_servers.append(server)
        except Exception as e:
            logger.error(f"Error connecting to MCP servers: {str(e)}")
            failed_servers.append(server)

    # Remove failed servers
    for failed_server in failed_servers:
        if failed_server in self.servers:
            self.servers.remove(failed_server)

get_toolkits

get_toolkits() -> List[Toolkit]

Return a flattened list of all tools across all servers

Source code in evoagentx/tools/mcp.py

def get_toolkits(self) -> List[Toolkit]:
    """Return a flattened list of all tools across all servers"""
    all_tools = []
    if not self.servers:
        logger.info("No MCP servers configured, returning empty toolkit list")
        return all_tools

    for server in self.servers:
        try:
            # Add timeout to prevent hanging
            import threading
            import queue

            result_queue = queue.Queue()
            exception_queue = queue.Queue()

            def get_tools_with_timeout():
                try:
                    tools = server.get_toolkits()
                    result_queue.put(tools)
                except Exception as e:
                    exception_queue.put(e)

            # Start tool retrieval in a separate thread with timeout
            thread = threading.Thread(target=get_tools_with_timeout)
            thread.daemon = True
            thread.start()
            thread.join(timeout=30)  # 30 second timeout

            if thread.is_alive():
                logger.warning("Timeout getting tools from MCP server after 30 seconds")
                continue

            if not exception_queue.empty():
                raise exception_queue.get()

            tools = result_queue.get()
            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

BrowserUseToolkit

BrowserUseToolkit(name: str = 'BrowserUseToolkit', model: str = 'gpt-4o-mini', api_key: str = None, browser_type: str = 'chromium', headless: bool = True)

Bases: Toolkit

Toolkit for browser automation using Browser Use.

Initialize the BrowserUse toolkit.

Parameters:

Name	Type	Description	Default
name	str	Toolkit name	'BrowserUseToolkit'
model	str	LLM model to use	'gpt-4o-mini'
api_key	str	API key for the LLM	None
browser_type	str	Browser type (chromium, firefox, webkit)	'chromium'
headless	bool	Whether to run browser in headless mode	True

Source code in evoagentx/tools/browser_use.py

def __init__(self, name: str = "BrowserUseToolkit", model: str = "gpt-4o-mini", 
             api_key: str = None, browser_type: str = "chromium", 
             headless: bool = True):
    """
    Initialize the BrowserUse toolkit.

    Args:
        name: Toolkit name
        model: LLM model to use
        api_key: API key for the LLM
        browser_type: Browser type (chromium, firefox, webkit)
        headless: Whether to run browser in headless mode
    """
    # Create the shared browser base instance
    browser_base = BrowserUseBase(
        model=model,
        api_key=api_key,
        browser_type=browser_type,
        headless=headless
    )

    # Create tools with the shared base
    tools = [
        BrowserUseTool(browser_base=browser_base)
    ]

    # Initialize parent with tools
    super().__init__(name=name, tools=tools)

    # Store browser_base as instance variable
    self.browser_base = browser_base

GoogleMapsToolkit

GoogleMapsToolkit(api_key: str = None, timeout: int = 10, name: str = 'GoogleMapsToolkit')

Bases: Toolkit

Complete Google Maps Platform toolkit containing all available tools.

Initialize the Google Maps toolkit.

Parameters:

Name	Type	Description	Default
api_key	str	Google Maps Platform API key. If not provided, will try to get from GOOGLE_MAPS_API_KEY environment variable.	None
timeout	int	Request timeout in seconds	10
name	str	Toolkit name	'GoogleMapsToolkit'

Source code in evoagentx/tools/google_maps_tool.py

def __init__(self, api_key: str = None, timeout: int = 10, name: str = "GoogleMapsToolkit"):
    """
    Initialize the Google Maps toolkit.

    Args:
        api_key (str, optional): Google Maps Platform API key. If not provided, will try to get from GOOGLE_MAPS_API_KEY environment variable.
        timeout (int): Request timeout in seconds
        name (str): Toolkit name
    """
    # Create shared Google Maps base instance
    google_maps_base = GoogleMapsBase(api_key=api_key, timeout=timeout)

    # Create all tools with shared base
    tools = [
        GeocodeAddressTool(google_maps_base=google_maps_base),
        ReverseGeocodeTool(google_maps_base=google_maps_base),
        PlacesSearchTool(google_maps_base=google_maps_base),
        PlaceDetailsTool(google_maps_base=google_maps_base),
        DirectionsTool(google_maps_base=google_maps_base),
        DistanceMatrixTool(google_maps_base=google_maps_base),
        TimeZoneTool(google_maps_base=google_maps_base)
    ]

    # Initialize parent with tools
    super().__init__(name=name, tools=tools)

    # Store base instance for access
    self.google_maps_base = google_maps_base

MongoDBToolkit

MongoDBToolkit(name: str = 'MongoDBToolkit', connection_string: str = None, database_name: str = None, local_path: str = None, auto_save: bool = True, read_only: bool = False, **kwargs)

Bases: Toolkit

MongoDB-specific toolkit with simplified design. Automatically handles remote, local file-based, or new database creation.

Initialize the MongoDB toolkit.

Parameters:

Name	Type	Description	Default
name	str	Name of the toolkit	'MongoDBToolkit'
connection_string	str	MongoDB connection string (for remote/existing)	None
database_name	str	Name of the database to use	None
local_path	str	Path for local file-based database	None
auto_save	bool	Automatically save changes to local files	True
read_only	bool	If True, only read operations are allowed (no insert, update, delete)	False
**kwargs		Additional connection parameters	{}

Source code in evoagentx/tools/database_mongodb.py

def __init__(self, 
             name: str = "MongoDBToolkit",
             connection_string: str = None,
             database_name: str = None,
             local_path: str = None,
             auto_save: bool = True,
             read_only: bool = False,
             **kwargs):
    """
    Initialize the MongoDB toolkit.

    Args:
        name: Name of the toolkit
        connection_string: MongoDB connection string (for remote/existing)
        database_name: Name of the database to use
        local_path: Path for local file-based database
        auto_save: Automatically save changes to local files
        read_only: If True, only read operations are allowed (no insert, update, delete)
        **kwargs: Additional connection parameters
    """
    # Initialize database with automatic detection
    database = MongoDBDatabase(
        connection_string=connection_string,
        database_name=database_name,
        local_path=local_path,
        auto_save=auto_save,
        read_only=read_only,
        **kwargs
    )

    # Initialize tools based on read-only mode
    if read_only:
        # Only include read-only tools
        tools = [
            MongoDBExecuteQueryTool(database=database),
            MongoDBFindTool(database=database),
            MongoDBInfoTool(database=database)
        ]
    else:
        # Include all tools for write mode
        tools = [
            MongoDBExecuteQueryTool(database=database),
            MongoDBFindTool(database=database),
            MongoDBUpdateTool(database=database),
            MongoDBDeleteTool(database=database),
            MongoDBInfoTool(database=database)
        ]

    # Initialize parent with tools
    super().__init__(name=name, tools=tools)

    # Store configuration after parent initialization
    self.database = database
    self.connection_string = connection_string
    self.database_name = database_name
    self.local_path = local_path
    self.auto_save = auto_save

    # Set up automatic cleanup
    import atexit
    atexit.register(self._cleanup)

get_capabilities

get_capabilities() -> Dict[str, Any]

Get MongoDB-specific capabilities

Source code in evoagentx/tools/database_mongodb.py

def get_capabilities(self) -> Dict[str, Any]:
    """Get MongoDB-specific capabilities"""
    if self.database:
        capabilities = self.database.get_capabilities()
        capabilities.update({
            "is_local_database": self.database.is_local_database,
            "local_path": str(self.database.local_path) if self.database.local_path else None,
            "auto_save": self.database.auto_save,
            "read_only": self.database.read_only
        })
        return capabilities
    return {"error": "MongoDB database not initialized"}

connect

connect() -> bool

Connect to MongoDB

Source code in evoagentx/tools/database_mongodb.py

def connect(self) -> bool:
    """Connect to MongoDB"""
    return self.database.connect() if self.database else False

disconnect

disconnect() -> bool

Disconnect from MongoDB

Source code in evoagentx/tools/database_mongodb.py

def disconnect(self) -> bool:
    """Disconnect from MongoDB"""
    return self.database.disconnect() if self.database else False

test_connection

test_connection() -> bool

Test MongoDB connection

Source code in evoagentx/tools/database_mongodb.py

def test_connection(self) -> bool:
    """Test MongoDB connection"""
    return self.database.test_connection() if self.database else False

get_database

get_database() -> MongoDBDatabase

Get the underlying MongoDB database instance

Source code in evoagentx/tools/database_mongodb.py

def get_database(self) -> MongoDBDatabase:
    """Get the underlying MongoDB database instance"""
    return self.database

get_local_info

get_local_info() -> Dict[str, Any]

Get information about local database setup

Source code in evoagentx/tools/database_mongodb.py

def get_local_info(self) -> Dict[str, Any]:
    """Get information about local database setup"""
    return {
        "is_local_database": self.database.is_local_database,
        "local_path": str(self.database.local_path) if self.database.local_path else None,
        "auto_save": self.database.auto_save,
        "read_only": self.database.read_only,
        "database_name": self.database_name,
        "connection_string": self.connection_string
    } if self.database else {"error": "Database not initialized"}

FileStorageHandler

FileStorageHandler(base_path: str = '.', **kwargs)

Bases: StorageBase

Unified storage handler that provides a consistent interface for all storage operations. This class serves as the main entry point for storage operations, inheriting from StorageBase and providing the core CRUD operations (Create, Read, Update, Delete).

Initialize the storage handler.

Parameters:

Name	Type	Description	Default
base_path	str	Base directory for storage operations (default: current directory)	'.'
**kwargs		Additional keyword arguments for parent class initialization	{}

Source code in evoagentx/tools/storage_handler.py

def __init__(self, base_path: str = ".", **kwargs):
    """
    Initialize the storage handler.

    Args:
        base_path (str): Base directory for storage operations (default: current directory)
        **kwargs: Additional keyword arguments for parent class initialization
    """
    super().__init__(base_path=base_path, **kwargs)

translate_in

translate_in(file_path: str) -> str

Translate input file path by combining it with base_path. This method takes a user-provided path and converts it to the full system path.

Parameters:

Name	Type	Description	Default
file_path	str	User-provided file path (can be relative or absolute)	required

Returns:

Name	Type	Description
str	str	Full system path combining base_path and file_path

Source code in evoagentx/tools/storage_handler.py

def translate_in(self, file_path: str) -> str:
    """
    Translate input file path by combining it with base_path.
    This method takes a user-provided path and converts it to the full system path.

    Args:
        file_path (str): User-provided file path (can be relative or absolute)

    Returns:
        str: Full system path combining base_path and file_path
    """
    # If the path is already absolute, return as is
    if os.path.isabs(file_path):
        return file_path

    # Always combine base_path with file_path to ensure working directory is respected
    # Check if this is a remote storage handler (like Supabase)
    if hasattr(self, 'bucket_name') and hasattr(self, 'supabase'):
        # For remote storage, treat base_path as a prefix within the bucket
        # Don't use os.path.join as it's designed for local filesystems
        if self.base_path.startswith('/'):
            # Remove leading slash and combine
            clean_base = self.base_path.lstrip('/')
            if clean_base:
                return f"{clean_base}/{file_path}"
            else:
                return file_path
        else:
            # Combine base_path and file_path with forward slash
            return f"{self.base_path}/{file_path}"
    else:
        # For local storage, use os.path.join for proper filesystem handling
        combined_path = os.path.join(self.base_path, file_path)
        normalized_path = os.path.normpath(combined_path)
        return normalized_path

translate_out

translate_out(full_path: str) -> str

Translate output full path by removing the base_path prefix. This method takes a full system path and converts it back to the user-relative path.

Parameters:

Name	Type	Description	Default
full_path	str	Full system path	required

Returns:

Name	Type	Description
str	str	User-relative path with base_path removed

Source code in evoagentx/tools/storage_handler.py

def translate_out(self, full_path: str) -> str:
    """
    Translate output full path by removing the base_path prefix.
    This method takes a full system path and converts it back to the user-relative path.

    Args:
        full_path (str): Full system path

    Returns:
        str: User-relative path with base_path removed
    """
    # If base_path is just "." or empty, return the full_path as is
    if self.base_path in [".", "", None]:
        return full_path

    # Check if this is a remote storage handler (like Supabase)
    if hasattr(self, 'bucket_name') and hasattr(self, 'supabase'):
        # For remote storage, handle path prefix removal
        if self.base_path.startswith('/'):
            clean_base = self.base_path.lstrip('/')
        else:
            clean_base = self.base_path

        if clean_base and full_path.startswith(f"{clean_base}/"):
            # Remove the base_path prefix
            relative_path = full_path[len(f"{clean_base}/"):]
            return relative_path
        elif clean_base and full_path == clean_base:
            # If the full_path is exactly the base_path, return empty string
            return ""
        else:
            # If the path doesn't start with base_path, return as is
            return full_path
    else:
        # For local storage, use os.path operations for proper filesystem handling
        # Convert both paths to absolute paths for comparison
        base_abs = os.path.abspath(self.base_path)
        full_abs = os.path.abspath(full_path)

        # Check if the full_path starts with base_path
        if full_abs.startswith(base_abs):
            # Remove the base_path prefix
            relative_path = full_abs[len(base_abs):]
            # Remove leading separator if present
            if relative_path.startswith(os.sep):
                relative_path = relative_path[1:]
            return relative_path

        # If the path doesn't start with base_path, return as is
        return full_path

create_file

create_file(file_path: str, content: Any, **kwargs) -> Dict[str, Any]

Create a new file with the specified content.

Parameters:

Name	Type	Description	Default
file_path	str	Path where the file should be created	required
content	Any	Content to write to the file	required
**kwargs		Additional arguments for file creation (encoding, format, etc.)	{}

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: Result of the operation with success status and details

Source code in evoagentx/tools/storage_handler.py

def create_file(self, file_path: str, content: Any, **kwargs) -> Dict[str, Any]:
    """
    Create a new file with the specified content.

    Args:
        file_path (str): Path where the file should be created
        content (Any): Content to write to the file
        **kwargs: Additional arguments for file creation (encoding, format, etc.)

    Returns:
        Dict[str, Any]: Result of the operation with success status and details
    """
    try:
        # Get file type to determine the appropriate save method
        file_extension = self.get_file_type(file_path)
        target_file_path = self.translate_in(file_path)

        # Route to specialized save methods based on file type
        if file_extension == '.json':
            return self._save_json(target_file_path, content, **kwargs)
        elif file_extension in ['.txt', '.md', '.log']:
            return self._save_text(target_file_path, content, **kwargs)
        elif file_extension == '.csv':
            return self._save_csv(target_file_path, content, **kwargs)
        elif file_extension in ['.yaml', '.yml']:
            return self._save_yaml(target_file_path, content, **kwargs)
        elif file_extension == '.xml':
            return self._save_xml(target_file_path, content, **kwargs)
        elif file_extension == '.xlsx':
            return self._save_excel(target_file_path, content, **kwargs)
        elif file_extension == '.pickle':
            return self._save_pickle(target_file_path, content, **kwargs)
        elif file_extension == '.pdf':
            return self._save_pdf(target_file_path, content, **kwargs)
        elif file_extension in ['.png', '.jpg', '.jpeg', '.gif', '.bmp', '.tiff']:
            return self._save_image(target_file_path, content, **kwargs)
        else:
            # For other file types, use the generic approach
            # Convert content to bytes if it's not already
            if isinstance(content, str):
                content_bytes = content.encode(kwargs.get('encoding', 'utf-8'))
            elif isinstance(content, bytes):
                content_bytes = content
            else:
                content_bytes = str(content).encode(kwargs.get('encoding', 'utf-8'))

            # Write the file using the raw method
            success = self._write_raw(target_file_path, content_bytes, **kwargs)

            if success:
                return {
                    "success": True,
                    "message": f"File '{file_path}' created successfully",
                    "file_path": file_path,
                    "full_path": target_file_path,
                    "size": len(content_bytes)
                }
            else:
                return {
                    "success": False,
                    "message": f"Failed to create file '{file_path}'",
                    "file_path": file_path,
                    "full_path": target_file_path
                }
    except Exception as e:
        logger.error(f"Error creating file {file_path}: {str(e)}")
        return {
            "success": False,
            "message": f"Error creating file: {str(e)}",
            "file_path": file_path
        }

read_file

read_file(file_path: str, **kwargs) -> Dict[str, Any]

Read content from an existing file.

Parameters:

Name	Type	Description	Default
file_path	str	Path of the file to read	required
**kwargs		Additional arguments for file reading (encoding, format, etc.)	{}

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: Result of the operation with file content and success status

Source code in evoagentx/tools/storage_handler.py

def read_file(self, file_path: str, **kwargs) -> Dict[str, Any]:
    """
    Read content from an existing file.

    Args:
        file_path (str): Path of the file to read
        **kwargs: Additional arguments for file reading (encoding, format, etc.)

    Returns:
        Dict[str, Any]: Result of the operation with file content and success status
    """
    try:
        # Use translate_in to get the full path
        full_path = self.translate_in(file_path)

        # Read the file using the raw method
        content_bytes = self._read_raw(full_path, **kwargs)

        # Convert to string if encoding is specified
        if 'encoding' in kwargs:
            content = content_bytes.decode(kwargs['encoding'])
        else:
            content = content_bytes

        return {
            "success": True,
            "message": f"File '{file_path}' read successfully",
            "file_path": file_path,
            "full_path": full_path,
            "content": content,
            "size": len(content_bytes)
        }
    except Exception as e:
        logger.error(f"Error reading file {file_path}: {str(e)}")
        return {
            "success": False,
            "message": f"Error reading file: {str(e)}",
            "file_path": file_path
        }

update_file

update_file(file_path: str, content: Any, **kwargs) -> Dict[str, Any]

Update an existing file with new content.

Parameters:

Name	Type	Description	Default
file_path	str	Path of the file to update	required
content	Any	New content to write to the file	required
**kwargs		Additional arguments for file update (encoding, format, etc.)	{}

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: Result of the operation with success status and details

Source code in evoagentx/tools/storage_handler.py

def update_file(self, file_path: str, content: Any, **kwargs) -> Dict[str, Any]:
    """
    Update an existing file with new content.

    Args:
        file_path (str): Path of the file to update
        content (Any): New content to write to the file
        **kwargs: Additional arguments for file update (encoding, format, etc.)

    Returns:
        Dict[str, Any]: Result of the operation with success status and details
    """
    # For update, we use the same create_file method as it handles both create and update
    return self.create_file(file_path, content, **kwargs)

delete_file

delete_file(file_path: str) -> Dict[str, Any]

Delete a file or directory.

Parameters:

Name	Type	Description	Default
file_path	str	Path of the file or directory to delete	required

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: Result of the operation with success status and details

Source code in evoagentx/tools/storage_handler.py

def delete_file(self, file_path: str) -> Dict[str, Any]:
    """
    Delete a file or directory.

    Args:
        file_path (str): Path of the file or directory to delete

    Returns:
        Dict[str, Any]: Result of the operation with success status and details
    """
    try:
        # Use translate_in to get the full path
        full_path = self.translate_in(file_path)

        # Delete the file using the raw method
        success = self._delete_raw(full_path)

        if success:
            return {
                "success": True,
                "message": f"File '{file_path}' deleted successfully",
                "file_path": file_path,
                "full_path": full_path
            }
        else:
            return {
                "success": False,
                "message": f"Failed to delete file '{file_path}'",
                "file_path": file_path,
                "full_path": full_path
            }
    except Exception as e:
        logger.error(f"Error deleting file {file_path}: {str(e)}")
        return {
            "success": False,
            "message": f"Error deleting file: {str(e)}",
            "file_path": file_path
        }

list_files

list_files(path: str = None, max_depth: int = 3, include_hidden: bool = False) -> Dict[str, Any]

List files and directories in the specified path.

Parameters:

Name	Type	Description	Default
path	str	Path to list (default: base_path)	None
max_depth	int	Maximum depth for recursive listing	3
include_hidden	bool	Whether to include hidden files	False

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: Result of the operation with file list and success status

Source code in evoagentx/tools/storage_handler.py

def list_files(self, path: str = None, max_depth: int = 3, include_hidden: bool = False) -> Dict[str, Any]:
    """
    List files and directories in the specified path.

    Args:
        path (str): Path to list (default: base_path)
        max_depth (int): Maximum depth for recursive listing
        include_hidden (bool): Whether to include hidden files

    Returns:
        Dict[str, Any]: Result of the operation with file list and success status
    """
    try:
        # Use translate_in to get the full path if provided
        full_path = self.translate_in(path) if path else self.base_path

        # List files using the raw method
        items = self._list_raw(full_path, max_depth, include_hidden)

        return {
            "success": True,
            "message": f"Listed {len(items)} items from '{path or 'base directory'}'",
            "path": path,
            "full_path": full_path,
            "items": items,
            "total_count": len(items)
        }
    except Exception as e:
        logger.error(f"Error listing files in {path}: {str(e)}")
        return {
            "success": False,
            "message": f"Error listing files: {str(e)}",
            "path": path
        }

file_exists

file_exists(file_path: str) -> bool

Check if a file exists.

Parameters:

Name	Type	Description	Default
file_path	str	Path of the file to check	required

Returns:

Name	Type	Description
bool	bool	True if file exists, False otherwise

Source code in evoagentx/tools/storage_handler.py

def file_exists(self, file_path: str) -> bool:
    """
    Check if a file exists.

    Args:
        file_path (str): Path of the file to check

    Returns:
        bool: True if file exists, False otherwise
    """
    try:
        # Use translate_in to get the full path
        full_path = self.translate_in(file_path)
        return self._exists_raw(full_path)
    except Exception as e:
        logger.error(f"Error checking if file exists {file_path}: {str(e)}")
        return False

get_file_information

get_file_information(file_path: str) -> Dict[str, Any]

Get comprehensive information about a file.

Parameters:

Name	Type	Description	Default
file_path	str	Path of the file to get information for	required

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: File information including size, type, timestamps, etc.

Source code in evoagentx/tools/storage_handler.py

def get_file_information(self, file_path: str) -> Dict[str, Any]:
    """
    Get comprehensive information about a file.

    Args:
        file_path (str): Path of the file to get information for

    Returns:
        Dict[str, Any]: File information including size, type, timestamps, etc.
    """
    try:
        # Use translate_in to get the full path
        full_path = self.translate_in(file_path)

        if hasattr(self, '_get_file_info_raw'):
            file_info = self._get_file_info_raw(full_path)
            # Add the translated paths to the result
            file_info['file_path'] = file_path
            file_info['full_path'] = full_path
            return file_info
        else:
            # Fallback to basic info
            return {
                'file_path': file_path,
                'full_path': full_path,
                'exists': self._exists_raw(full_path),
                'type': 'file'  # Default assumption
            }
    except Exception as e:
        logger.error(f"Error getting file info for {file_path}: {str(e)}")
        return {}

create_directory

create_directory(path: str) -> Dict[str, Any]

Create a new directory.

Parameters:

Name	Type	Description	Default
path	str	Path where the directory should be created	required

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: Result of the operation with success status and details

Source code in evoagentx/tools/storage_handler.py

def create_directory(self, path: str) -> Dict[str, Any]:
    """
    Create a new directory.

    Args:
        path (str): Path where the directory should be created

    Returns:
        Dict[str, Any]: Result of the operation with success status and details
    """
    try:
        # Use translate_in to get the full path
        full_path = self.translate_in(path)

        # Create directory using the raw method
        success = self._create_directory_raw(full_path)

        if success:
            return {
                "success": True,
                "message": f"Directory '{path}' created successfully",
                "path": path,
                "full_path": full_path
            }
        else:
            return {
                "success": False,
                "message": f"Failed to create directory '{path}'",
                "path": path,
                "full_path": full_path
            }
    except Exception as e:
        logger.error(f"Error creating directory {path}: {str(e)}")
        return {
            "success": False,
            "message": f"Error creating directory: {str(e)}",
            "path": path
        }

copy_file

copy_file(source: str, destination: str) -> Dict[str, Any]

Copy a file from source to destination.

Parameters:

Name	Type	Description	Default
source	str	Source file path	required
destination	str	Destination file path	required

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: Result of the operation with success status and details

Source code in evoagentx/tools/storage_handler.py

def copy_file(self, source: str, destination: str) -> Dict[str, Any]:
    """
    Copy a file from source to destination.

    Args:
        source (str): Source file path
        destination (str): Destination file path

    Returns:
        Dict[str, Any]: Result of the operation with success status and details
    """
    try:
        # Use translate_in to get the full paths
        full_source = self.translate_in(source)
        full_destination = self.translate_in(destination)

        # Read source file
        source_content = self._read_raw(full_source)

        # Write to destination
        success = self._write_raw(full_destination, source_content)

        if success:
            return {
                "success": True,
                "message": f"File copied from '{source}' to '{destination}'",
                "source": source,
                "destination": destination,
                "full_source": full_source,
                "full_destination": full_destination
            }
        else:
            return {
                "success": False,
                "message": f"Failed to copy file from '{source}' to '{destination}'",
                "source": source,
                "destination": destination
            }
    except Exception as e:
        logger.error(f"Error copying file from {source} to {destination}: {str(e)}")
        return {
            "success": False,
            "message": f"Error copying file: {str(e)}",
            "source": source,
            "destination": destination
        }

move_file

move_file(source: str, destination: str) -> Dict[str, Any]

Move/rename a file from source to destination.

Parameters:

Name	Type	Description	Default
source	str	Source file path	required
destination	str	Destination file path	required

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: Result of the operation with success status and details

Source code in evoagentx/tools/storage_handler.py

def move_file(self, source: str, destination: str) -> Dict[str, Any]:
    """
    Move/rename a file from source to destination.

    Args:
        source (str): Source file path
        destination (str): Destination file path

    Returns:
        Dict[str, Any]: Result of the operation with success status and details
    """
    try:
        # Use translate_in to get the full paths
        full_source = self.translate_in(source)
        full_destination = self.translate_in(destination)

        # Copy file to destination
        copy_success = self._write_raw(full_destination, self._read_raw(full_source))

        if copy_success:
            # Delete source file
            delete_success = self._delete_raw(full_source)

            if delete_success:
                return {
                    "success": True,
                    "message": f"File moved from '{source}' to '{destination}'",
                    "source": source,
                    "destination": destination,
                    "full_source": full_source,
                    "full_destination": full_destination
                }
            else:
                return {
                    "success": False,
                    "message": f"File copied but failed to delete source '{source}'",
                    "source": source,
                    "destination": destination
                }
        else:
            return {
                "success": False,
                "message": f"Failed to copy file from '{source}' to '{destination}'",
                "source": source,
                "destination": destination
            }
    except Exception as e:
        logger.error(f"Error moving file from {source} to {destination}: {str(e)}")
        return {
            "success": False,
            "message": f"Error moving file: {str(e)}",
            "source": source,
            "destination": destination
        }

append_to_file

append_to_file(file_path: str, content: Any, **kwargs) -> Dict[str, Any]

Append content to an existing file.

Parameters:

Name	Type	Description	Default
file_path	str	Path of the file to append to	required
content	Any	Content to append	required
**kwargs		Additional arguments for appending (encoding, format, etc.)	{}

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: Result of the operation with success status and details

Source code in evoagentx/tools/storage_handler.py

def append_to_file(self, file_path: str, content: Any, **kwargs) -> Dict[str, Any]:
    """
    Append content to an existing file.

    Args:
        file_path (str): Path of the file to append to
        content (Any): Content to append
        **kwargs: Additional arguments for appending (encoding, format, etc.)

    Returns:
        Dict[str, Any]: Result of the operation with success status and details
    """
    try:
        # Use translate_in to get the full path
        full_path = self.translate_in(file_path)

        # Read existing content
        existing_content = self._read_raw(full_path)

        # Convert new content to bytes
        if isinstance(content, str):
            new_content_bytes = content.encode(kwargs.get('encoding', 'utf-8'))
            # For text files, decode existing content and append
            existing_content_str = existing_content.decode(kwargs.get('encoding', 'utf-8'))
            combined_content = existing_content_str + content
            combined_bytes = combined_content.encode(kwargs.get('encoding', 'utf-8'))
        else:
            new_content_bytes = str(content).encode(kwargs.get('encoding', 'utf-8'))
            # For binary files, just concatenate bytes
            combined_bytes = existing_content + new_content_bytes

        # Write back to file
        success = self._write_raw(full_path, combined_bytes, **kwargs)

        if success:
            return {
                "success": True,
                "message": f"Content appended to '{file_path}'",
                "file_path": file_path,
                "full_path": full_path,
                "content_length": len(new_content_bytes)
            }
        else:
            return {
                "success": False,
                "message": f"Failed to append content to '{file_path}'",
                "file_path": file_path,
                "full_path": full_path
            }
    except Exception as e:
        logger.error(f"Error appending to file {file_path}: {str(e)}")
        return {
            "success": False,
            "message": f"Error appending to file: {str(e)}",
            "file_path": file_path
        }

save

save(file_path: str, content: Any, **kwargs) -> Dict[str, Any]

Save content to a file (alias for create_file). This method is expected by some tools like flux image generation.

Parameters:

Name	Type	Description	Default
file_path	str	Path where the file should be saved	required
content	Any	Content to save to the file	required
**kwargs		Additional arguments for file creation	{}

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: Result of the operation

Source code in evoagentx/tools/storage_handler.py

def save(self, file_path: str, content: Any, **kwargs) -> Dict[str, Any]:
    """
    Save content to a file (alias for create_file).
    This method is expected by some tools like flux image generation.

    Args:
        file_path (str): Path where the file should be saved
        content (Any): Content to save to the file
        **kwargs: Additional arguments for file creation

    Returns:
        Dict[str, Any]: Result of the operation
    """
    return self.create_file(file_path, content, **kwargs)

LocalStorageHandler

LocalStorageHandler(base_path: str = '.', **kwargs)

Bases: FileStorageHandler

Local filesystem storage implementation. Provides all file operations for local storage with default working directory.

Initialize local storage handler.

Parameters:

Name	Type	Description	Default
base_path	str	Base directory for storage operations (default: current directory)	'.'
**kwargs		Additional keyword arguments for parent class initialization	{}

Source code in evoagentx/tools/storage_handler.py

def __init__(self, base_path: str = ".", **kwargs):
    """
    Initialize local storage handler.

    Args:
        base_path (str): Base directory for storage operations (default: current directory)
        **kwargs: Additional keyword arguments for parent class initialization
    """
    super().__init__(base_path=base_path, **kwargs)

get_file_info

get_file_info(file_path: str) -> Dict[str, Any]

Get comprehensive information about a file

Source code in evoagentx/tools/storage_handler.py

def get_file_info(self, file_path: str) -> Dict[str, Any]:
    """Get comprehensive information about a file"""
    try:
        resolved_path = self._resolve_path(file_path)
        path_obj = Path(resolved_path)

        if not path_obj.exists():
            return {"success": False, "error": f"File {file_path} does not exist"}

        stat = path_obj.stat()
        return {
            "success": True,
            "file_path": resolved_path,
            "file_name": path_obj.name,
            "file_extension": path_obj.suffix.lower(),
            "mime_type": self.get_mime_type(resolved_path),
            "size_bytes": stat.st_size,
            "size_mb": round(stat.st_size / (1024 * 1024), 2),
            "created_time": datetime.fromtimestamp(stat.st_ctime).isoformat(),
            "modified_time": datetime.fromtimestamp(stat.st_mtime).isoformat(),
            "is_file": path_obj.is_file(),
            "is_directory": path_obj.is_dir(),
            "is_readable": os.access(path_obj, os.R_OK),
            "is_writable": os.access(path_obj, os.W_OK),
            "exists": True
        }
    except Exception as e:
        logger.error(f"Error getting file info for {file_path}: {str(e)}")
        return {"success": False, "error": str(e), "file_path": file_path}

SupabaseStorageHandler

SupabaseStorageHandler(bucket_name: str = None, base_path: str = '/', **kwargs)

Bases: FileStorageHandler

Supabase remote storage implementation. Provides file operations via Supabase Storage API with environment-based configuration.

Initialize Supabase storage handler.

Parameters:

Name	Type	Description	Default
bucket_name	str	Supabase storage bucket name (default: from environment or "default")	None
base_path	str	Base path for storage operations (default: "/")	'/'
**kwargs		Additional keyword arguments for parent class initialization	{}

Source code in evoagentx/tools/storage_handler.py

def __init__(self, bucket_name: str = None, base_path: str = "/", **kwargs):
    """
    Initialize Supabase storage handler.

    Args:
        bucket_name: Supabase storage bucket name (default: from environment or "default")
        base_path: Base path for storage operations (default: "/")
        **kwargs: Additional keyword arguments for parent class initialization
    """
    # Call parent constructor first
    super().__init__(base_path=base_path, **kwargs)

    # Get bucket name from environment or use default
    self.bucket_name = bucket_name or os.getenv("SUPABASE_BUCKET_STORAGE") or "default"
    self.supabase_url = os.getenv("SUPABASE_URL_STORAGE")
    self.supabase_key = os.getenv("SUPABASE_KEY_STORAGE")

    if not self.supabase_url or not self.supabase_key:
        raise ValueError(
            "Supabase configuration not found in environment variables. "
            "Please set SUPABASE_URL/SUPABASE_KEY environment variables."
        )

    # Initialize Supabase client
    try:
        from supabase import create_client, Client
        logger.info(f"Creating Supabase client with URL: {self.supabase_url[:30]}...")
        self.supabase: Client = create_client(self.supabase_url, self.supabase_key)
        logger.info(f"Successfully initialized Supabase client for bucket: {bucket_name}")
    except ImportError:
        raise ImportError(
            "Supabase Python client not installed. "
            "Please install it with: pip install supabase"
        )
    except Exception as e:
        logger.error(f"Failed to initialize Supabase client: {str(e)}")
        raise Exception(f"Failed to initialize Supabase client: {str(e)}")

    # Initialize storage after all attributes are set
    self._initialize_storage()

get_file_info

get_file_info(file_path: str) -> Dict[str, Any]

Get comprehensive information about a file in Supabase Storage

Source code in evoagentx/tools/storage_handler.py

def get_file_info(self, file_path: str) -> Dict[str, Any]:
    """Get comprehensive information about a file in Supabase Storage"""
    try:
        resolved_path = self._resolve_path(file_path)

        if not self._exists_raw(resolved_path):
            return {"success": False, "error": f"File {file_path} does not exist"}

        # Get file content to determine size
        content = self._read_raw(resolved_path)

        # Try to get additional metadata from list
        list_path = os.path.dirname(resolved_path.lstrip('/'))
        file_name = os.path.basename(resolved_path)

        metadata = {}
        try:
            files = self.supabase.storage.from_(self.bucket_name).list(list_path)
            for file_info in files:
                if file_info.get('name') == file_name:
                    metadata = file_info.get('metadata', {})
                    break
        except Exception:
            pass

        return {
            "success": True,
            "file_path": resolved_path,
            "file_name": Path(resolved_path).name,
            "file_extension": Path(resolved_path).suffix.lower(),
            "mime_type": metadata.get('mimetype', self.get_mime_type(resolved_path)),
            "size_bytes": len(content),
            "size_mb": round(len(content) / (1024 * 1024), 2),
            "modified_time": metadata.get('updated_at', datetime.now().isoformat()),
            "is_file": True,
            "is_directory": False,
            "is_readable": True,
            "is_writable": True,
            "exists": True
        }
    except Exception as e:
        logger.error(f"Error getting file info for {file_path}: {str(e)}")
        return {"success": False, "error": str(e), "file_path": file_path}

StorageToolkit

StorageToolkit(name: str = 'StorageToolkit', base_path: str = './workplace/storage', storage_handler: FileStorageHandler = None)

Bases: Toolkit

Comprehensive storage toolkit with local filesystem operations. Provides tools for reading, writing, appending, deleting, moving, copying files, creating directories, and listing files with support for various file formats.

Initialize the storage toolkit.

Parameters:

Name	Type	Description	Default
name	str	Name of the toolkit	'StorageToolkit'
base_path	str	Base directory for storage operations (default: ./workplace/storage)	'./workplace/storage'
storage_handler	FileStorageHandler	Storage handler instance (defaults to LocalStorageHandler)	None

Source code in evoagentx/tools/storage_file.py

def __init__(self, name: str = "StorageToolkit", base_path: str = "./workplace/storage", storage_handler: FileStorageHandler = None):
    """
    Initialize the storage toolkit.

    Args:
        name: Name of the toolkit
        base_path: Base directory for storage operations (default: ./workplace/storage)
        storage_handler: Storage handler instance (defaults to LocalStorageHandler)
    """
    # Create the shared storage handler instance
    if storage_handler is None:
        storage_handler = LocalStorageHandler(base_path=base_path)

    # Create tools with the storage handler
    tools = [
        SaveTool(storage_handler=storage_handler),
        ReadTool(storage_handler=storage_handler),
        AppendTool(storage_handler=storage_handler),
        DeleteTool(storage_handler=storage_handler),
        MoveTool(storage_handler=storage_handler),
        CopyTool(storage_handler=storage_handler),
        CreateDirectoryTool(storage_handler=storage_handler),
        ListFileTool(storage_handler=storage_handler),
        ExistsTool(storage_handler=storage_handler),
        ListSupportedFormatsTool(storage_handler=storage_handler)
    ]

    super().__init__(name=name, tools=tools)
    self.storage_handler = storage_handler 

FluxImageGenerationToolkit

FluxImageGenerationToolkit(name: str = 'FluxImageGenerationToolkit', api_key: str = None, save_path: str = './imgs', storage_handler: FileStorageHandler = None)

Bases: Toolkit

Toolkit for Flux image generation with storage handler integration.

Initialize the Flux image generation toolkit.

Parameters:

Name	Type	Description	Default
name	str	Name of the toolkit	'FluxImageGenerationToolkit'
api_key	str	API key for Flux image generation	None
save_path	str	Default save path for images	'./imgs'
storage_handler	FileStorageHandler	Storage handler for file operations	None

Source code in evoagentx/tools/images_flux_generation.py

def __init__(self, name: str = "FluxImageGenerationToolkit", api_key: str = None, save_path: str = "./imgs", storage_handler: FileStorageHandler = None):
    """
    Initialize the Flux image generation toolkit.

    Args:
        name: Name of the toolkit
        api_key: API key for Flux image generation
        save_path: Default save path for images
        storage_handler: Storage handler for file operations
    """
    # Initialize storage handler if not provided
    if storage_handler is None:
        from .storage_file import LocalStorageHandler
        storage_handler = LocalStorageHandler(base_path=save_path)

    # Create the image generation tool
    tool = FluxImageGenerationTool(
        api_key=api_key,
        save_path=save_path,
        storage_handler=storage_handler
    )

    # Create tools list
    tools = [tool]

    # Initialize parent with tools
    super().__init__(name=name, tools=tools)

    # Store instance variables
    self.api_key = api_key
    self.save_path = save_path
    self.storage_handler = storage_handler

OpenAIImageGenerationToolkit

OpenAIImageGenerationToolkit(name: str = 'OpenAIImageGenerationToolkit', api_key: str = None, organization_id: str = None, model: str = 'gpt-4o', save_path: str = './', storage_handler: FileStorageHandler = None)

Bases: Toolkit

Toolkit for OpenAI image generation with storage handler integration.

Initialize the OpenAI image generation toolkit.

Parameters:

Name	Type	Description	Default
name	str	Name of the toolkit	'OpenAIImageGenerationToolkit'
api_key	str	API key for OpenAI	None
organization_id	str	Organization ID for OpenAI	None
model	str	Model to use for image generation	'gpt-4o'
save_path	str	Default save path for images	'./'
storage_handler	FileStorageHandler	Storage handler for file operations	None

Source code in evoagentx/tools/images_openai_generation.py

def __init__(self, name: str = "OpenAIImageGenerationToolkit", api_key: str = None, organization_id: str = None, model: str = "gpt-4o", save_path: str = "./", storage_handler: FileStorageHandler = None):
    """
    Initialize the OpenAI image generation toolkit.

    Args:
        name: Name of the toolkit
        api_key: API key for OpenAI
        organization_id: Organization ID for OpenAI
        model: Model to use for image generation
        save_path: Default save path for images
        storage_handler: Storage handler for file operations
    """
    # Initialize storage handler if not provided
    if storage_handler is None:
        from .storage_file import LocalStorageHandler
        storage_handler = LocalStorageHandler(base_path="./workplace/images")

    # Create the image generation tool
    tool = OpenAI_ImageGenerationTool(
        api_key=api_key,
        organization_id=organization_id,
        model=model,
        save_path=save_path,
        storage_handler=storage_handler
    )

    # Create tools list
    tools = [tool]

    # Initialize parent with tools
    super().__init__(name=name, tools=tools)

    # Store instance variables
    self.api_key = api_key
    self.organization_id = organization_id
    self.model = model
    self.save_path = save_path
    self.storage_handler = storage_handler

ImageAnalysisToolkit

ImageAnalysisToolkit(name: str = 'ImageAnalysisToolkit', api_key: str = None, model: str = 'openai/gpt-4o', storage_handler: FileStorageHandler = None)

Bases: Toolkit

Toolkit for image analysis with storage handler integration.

Initialize the image analysis toolkit.

Parameters:

Name	Type	Description	Default
name	str	Name of the toolkit	'ImageAnalysisToolkit'
api_key	str	API key for OpenRouter	None
model	str	Model to use for image analysis	'openai/gpt-4o'
storage_handler	FileStorageHandler	Storage handler for file operations	None

Source code in evoagentx/tools/image_analysis.py

def __init__(self, name: str = "ImageAnalysisToolkit", api_key: str = None, model: str = "openai/gpt-4o", storage_handler: FileStorageHandler = None):
    """
    Initialize the image analysis toolkit.

    Args:
        name: Name of the toolkit
        api_key: API key for OpenRouter
        model: Model to use for image analysis
        storage_handler: Storage handler for file operations
    """
    # Initialize storage handler if not provided
    if storage_handler is None:
        from .storage_file import LocalStorageHandler
        storage_handler = LocalStorageHandler(base_path="./workplace/analysis")

    # Create the image analysis tool
    tool = ImageAnalysisTool(
        api_key=api_key,
        model=model,
        storage_handler=storage_handler
    )

    # Create tools list
    tools = [tool]

    # Initialize parent with tools
    super().__init__(name=name, tools=tools)

    # Store instance variables
    self.api_key = api_key
    self.model = model
    self.storage_handler = storage_handler 

CMDToolkit

CMDToolkit(name: str = 'CMDToolkit', default_shell: str = None, storage_handler: FileStorageHandler = None)

Bases: Toolkit

Command line toolkit that provides safe command execution with permission checking and cross-platform support. Supports Linux, macOS, and Windows.

Initialize the CMDToolkit with a shared command base instance.

Parameters:

Name	Type	Description	Default
name	str	Name of the toolkit	'CMDToolkit'
default_shell	str	Override default shell detection	None
storage_handler	FileStorageHandler	Storage handler for file operations	None

Source code in evoagentx/tools/cmd_toolkit.py

def __init__(self, name: str = "CMDToolkit", default_shell: str = None, storage_handler: FileStorageHandler = None):
    """
    Initialize the CMDToolkit with a shared command base instance.

    Args:
        name: Name of the toolkit
        default_shell: Override default shell detection
        storage_handler: Storage handler for file operations
    """
    # Initialize storage handler if not provided
    if storage_handler is None:
        from .storage_file import LocalStorageHandler
        storage_handler = LocalStorageHandler(base_path="./workplace/cmd")

    # Create the shared command base instance with storage handler
    cmd_base = CMDBase(default_shell=default_shell, storage_handler=storage_handler)

    # Initialize tools with the shared command base
    tools = [
        ExecuteCommandTool(cmd_base=cmd_base)
    ]

    # Initialize parent with tools
    super().__init__(name=name, tools=tools)
    self.cmd_base = cmd_base
    self.storage_handler = storage_handler 

RSSToolkit

RSSToolkit(name: str = 'RSSToolkit')

Bases: Toolkit

Toolkit for RSS feed operations.

Source code in evoagentx/tools/rss_feed.py

def __init__(self, name: str = "RSSToolkit"):
    # Create the shared RSS base instance
    rss_base = RSSBase()

    # Create tools with the shared RSS base
    tools = [
        RSSFetchTool(rss_base=rss_base),
        RSSValidateTool(rss_base=rss_base)
    ]

    super().__init__(name=name, tools=tools)

SerperAPIToolkit

SerperAPIToolkit(name: str = 'SerperAPIToolkit', api_key: Optional[str] = None, num_search_pages: Optional[int] = 10, max_content_words: Optional[int] = None, default_location: Optional[str] = None, default_language: Optional[str] = 'en', default_country: Optional[str] = 'us', enable_content_scraping: Optional[bool] = True, **kwargs)

Bases: Toolkit

Initialize SerperAPI Toolkit.

Parameters:

Name	Type	Description	Default
name	str	Name of the toolkit	'SerperAPIToolkit'
api_key	str	SerperAPI authentication key	None
num_search_pages	int	Default number of search results to retrieve	10
max_content_words	int	Default maximum words per result content	None
default_location	str	Default geographic location	None
default_language	str	Default interface language	'en'
default_country	str	Default country code	'us'
enable_content_scraping	bool	Whether to enable content scraping	True
**kwargs		Additional keyword arguments	{}

Source code in evoagentx/tools/search_serperapi.py

def __init__(
    self,
    name: str = "SerperAPIToolkit",
    api_key: Optional[str] = None,
    num_search_pages: Optional[int] = 10,
    max_content_words: Optional[int] = None,
    default_location: Optional[str] = None,
    default_language: Optional[str] = "en",
    default_country: Optional[str] = "us",
    enable_content_scraping: Optional[bool] = True,
    **kwargs
):
    """
    Initialize SerperAPI Toolkit.

    Args:
        name (str): Name of the toolkit
        api_key (str): SerperAPI authentication key
        num_search_pages (int): Default number of search results to retrieve
        max_content_words (int): Default maximum words per result content
        default_location (str): Default geographic location
        default_language (str): Default interface language
        default_country (str): Default country code
        enable_content_scraping (bool): Whether to enable content scraping
        **kwargs: Additional keyword arguments
    """
    # Create the shared SerperAPI search instance
    search_serperapi = SearchSerperAPI(
        name="SearchSerperAPI",
        api_key=api_key,
        num_search_pages=num_search_pages,
        max_content_words=max_content_words,
        default_location=default_location,
        default_language=default_language,
        default_country=default_country,
        enable_content_scraping=enable_content_scraping,
        **kwargs
    )

    # Create tools with the shared search instance
    tools = [
        SerperAPITool(search_serperapi=search_serperapi)
    ]

    # Initialize parent with tools
    super().__init__(name=name, tools=tools)

    # Store search_serperapi as instance variable
    self.search_serperapi = search_serperapi

SerpAPIToolkit

SerpAPIToolkit(name: str = 'SerpAPIToolkit', api_key: Optional[str] = None, num_search_pages: Optional[int] = 5, max_content_words: Optional[int] = None, default_engine: Optional[str] = 'google', default_location: Optional[str] = None, default_language: Optional[str] = 'en', default_country: Optional[str] = 'us', enable_content_scraping: Optional[bool] = True, **kwargs)

Bases: Toolkit

Initialize SerpAPI Toolkit.

Parameters:

Name	Type	Description	Default
name	str	Name of the toolkit	'SerpAPIToolkit'
api_key	str	SerpAPI authentication key	None
num_search_pages	int	Default number of search results to retrieve	5
max_content_words	int	Default maximum words per result content	None
default_engine	str	Default search engine	'google'
default_location	str	Default geographic location	None
default_language	str	Default interface language	'en'
default_country	str	Default country code	'us'
enable_content_scraping	bool	Whether to enable content scraping	True
**kwargs		Additional keyword arguments	{}

Source code in evoagentx/tools/search_serpapi.py

def __init__(
    self,
    name: str = "SerpAPIToolkit",
    api_key: Optional[str] = None,
    num_search_pages: Optional[int] = 5,
    max_content_words: Optional[int] = None,
    default_engine: Optional[str] = "google",
    default_location: Optional[str] = None,
    default_language: Optional[str] = "en",
    default_country: Optional[str] = "us",
    enable_content_scraping: Optional[bool] = True,
    **kwargs
):
    """
    Initialize SerpAPI Toolkit.

    Args:
        name (str): Name of the toolkit
        api_key (str): SerpAPI authentication key
        num_search_pages (int): Default number of search results to retrieve
        max_content_words (int): Default maximum words per result content
        default_engine (str): Default search engine
        default_location (str): Default geographic location
        default_language (str): Default interface language
        default_country (str): Default country code
        enable_content_scraping (bool): Whether to enable content scraping
        **kwargs: Additional keyword arguments
    """
    # Create the shared SerpAPI search instance
    search_serpapi = SearchSerpAPI(
        name="SearchSerpAPI",
        api_key=api_key,
        num_search_pages=num_search_pages,
        max_content_words=max_content_words,
        default_engine=default_engine,
        default_location=default_location,
        default_language=default_language,
        default_country=default_country,
        enable_content_scraping=enable_content_scraping,
        **kwargs
    )

    # Create tools with the shared search instance
    tools = [
        SerpAPITool(search_serpapi=search_serpapi)
    ]

    # Initialize parent with tools
    super().__init__(name=name, tools=tools)

    # Store search_serpapi as instance variable
    self.search_serpapi = search_serpapi