""" define a generic tool class, any tool can be used by the agent. A tool can be used by a llm like so: ``` ``` we call these "blocks". For example: ```python print("Hello world") ``` This is then executed by the tool with its own class implementation of execute(). A tool is not just for code tool but also API, internet search, MCP, etc.. """ import sys import os import configparser from abc import abstractmethod from sources.logger import Logger class Tools(): """ Abstract class for all tools. """ def __init__(self): self.tag = "undefined" self.client = None self.messages = [] self.logger = Logger("tools.log") self.config = configparser.ConfigParser() self.work_dir = self.create_work_dir() self.excutable_blocks_found = False self.safe_mode = True self.allow_language_exec_bash = False def get_work_dir(self): return self.work_dir def set_allow_language_exec_bash(value: bool) -> None: self.allow_language_exec_bash = value def check_config_dir_validity(self): """Check if the config directory is valid.""" path = self.config['MAIN']['work_dir'] if path == "": print("WARNING: Work directory not set in config.ini") return False if path.lower() == "none": print("WARNING: Work directory set to none in config.ini") return False if not os.path.exists(path): print(f"WARNING: Work directory {path} does not exist") return False return True def config_exists(self): """Check if the config file exists.""" return os.path.exists('./config.ini') def create_work_dir(self): """Create the work directory if it does not exist.""" default_path = os.path.dirname(os.getcwd()) if self.config_exists(): self.config.read('./config.ini') config_path = self.config['MAIN']['work_dir'] dir_path = default_path if not self.check_config_dir_validity() else config_path else: dir_path = default_path return dir_path @abstractmethod def execute(self, blocks:[str], safety:bool) -> str: """ Abstract method that must be implemented by child classes to execute the tool's functionality. Args: blocks (List[str]): The codes or queries blocks to execute safety (bool): Whenever human intervention is required Returns: str: The output/result from executing the tool """ pass @abstractmethod def execution_failure_check(self, output:str) -> bool: """ Abstract method that must be implemented by child classes to check if tool execution failed. Args: output (str): The output string from the tool execution to analyze Returns: bool: True if execution failed, False if successful """ pass @abstractmethod def interpreter_feedback(self, output:str) -> str: """ Abstract method that must be implemented by child classes to provide feedback to the AI from the tool. Args: output (str): The output string from the tool execution to analyze Returns: str: The feedback message to the AI """ pass def save_block(self, blocks:[str], save_path:str) -> None: """ Save code or query blocks to a file at the specified path. Creates the directory path if it doesn't exist. Args: blocks (List[str]): List of code/query blocks to save save_path (str): File path where blocks should be saved """ if save_path is None: return self.logger.info(f"Saving blocks to {save_path}") save_path_dir = os.path.dirname(save_path) save_path_file = os.path.basename(save_path) directory = os.path.join(self.work_dir, save_path_dir) if directory and not os.path.exists(directory): self.logger.info(f"Creating directory {directory}") os.makedirs(directory) for block in blocks: with open(os.path.join(directory, save_path_file), 'w') as f: f.write(block) def get_parameter_value(self, block: str, parameter_name: str) -> str: """ Get a parameter name. Args: block (str): The block of text to search for the parameter parameter_name (str): The name of the parameter to retrieve Returns: str: The value of the parameter """ for param_line in block.split('\n'): if parameter_name in param_line: param_value = param_line.split('=')[1].strip() return param_value return None def found_executable_blocks(self): """ Check if executable blocks were found. """ tmp = self.excutable_blocks_found self.excutable_blocks_found = False return tmp def load_exec_block(self, llm_text: str) -> tuple[list[str], str | None]: """ Extract code/query blocks from LLM-generated text and process them for execution. This method parses the text looking for code blocks marked with the tool's tag (e.g. ```python). Args: llm_text (str): The raw text containing code blocks from the LLM Returns: tuple[list[str], str | None]: A tuple containing: - List of extracted and processed code blocks - The path the code blocks was saved to """ assert self.tag != "undefined", "Tag not defined" start_tag = f'```{self.tag}' end_tag = '```' code_blocks = [] start_index = 0 save_path = None if start_tag not in llm_text: return None, None while True: start_pos = llm_text.find(start_tag, start_index) if start_pos == -1: break line_start = llm_text.rfind('\n', 0, start_pos)+1 leading_whitespace = llm_text[line_start:start_pos] end_pos = llm_text.find(end_tag, start_pos + len(start_tag)) if end_pos == -1: break content = llm_text[start_pos + len(start_tag):end_pos] lines = content.split('\n') if leading_whitespace: processed_lines = [] for line in lines: if line.startswith(leading_whitespace): processed_lines.append(line[len(leading_whitespace):]) else: processed_lines.append(line) content = '\n'.join(processed_lines) if ':' in content.split('\n')[0]: save_path = content.split('\n')[0].split(':')[1] content = content[content.find('\n')+1:] self.excutable_blocks_found = True code_blocks.append(content) start_index = end_pos + len(end_tag) self.logger.info(f"Found {len(code_blocks)} blocks to execute") return code_blocks, save_path if __name__ == "__main__": tool = Tools() tool.tag = "python" rt = tool.load_exec_block("""```python import os for file in os.listdir(): if file.endswith('.py'): print(file) ``` goodbye! """) print(rt)