diff --git a/src/agentscope/service/__init__.py b/src/agentscope/service/__init__.py index 22714a707..b52023514 100644 --- a/src/agentscope/service/__init__.py +++ b/src/agentscope/service/__init__.py @@ -3,6 +3,7 @@ from loguru import logger from .execute_code.exec_python import execute_python_code +from .execute_code.exec_shell import execute_shell_command from .file.common import ( create_file, delete_file, @@ -10,6 +11,8 @@ create_directory, delete_directory, move_directory, + list_directory_content, + get_current_directory, ) from .file.text import read_text_file, write_text_file from .file.json import read_json_file, write_json_file @@ -40,12 +43,15 @@ def get_help() -> None: "ServiceFactory", "get_help", "execute_python_code", + "execute_shell_command", "create_file", "delete_file", "move_file", "create_directory", "delete_directory", "move_directory", + "list_directory_content", + "get_current_directory", "read_text_file", "write_text_file", "read_json_file", diff --git a/src/agentscope/service/execute_code/exec_shell.py b/src/agentscope/service/execute_code/exec_shell.py new file mode 100644 index 000000000..ffde21b9d --- /dev/null +++ b/src/agentscope/service/execute_code/exec_shell.py @@ -0,0 +1,57 @@ +# -*- coding: utf-8 -*- +"""Service to execute shell commands.""" +import subprocess +from agentscope.service.service_status import ServiceExecStatus +from agentscope.service.service_response import ServiceResponse + + +def execute_shell_command(command: str) -> ServiceResponse: + """ + Executes a given shell command. + + Args: + command (str): The shell command to execute. + + Returns: + ServiceResponse: Contains either the output from the shell command as a + string if sucessful, or an error message include the error type. + + Note: + Use any bash/shell commands you want (e.g. find, grep, cat, ls), + but note that : + 1. interactive session commands (e.g. python, vim) or commands that + change current state (e.g. cd that change the current directory) + are NOT supported yet, so please do not invoke them. + 2. be VERY CAREFUL when using commands that will + change/edit the files current directory (e.g. rm, sed). + ... + """ + try: + result = subprocess.run( + command, + shell=True, + check=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + return ServiceResponse( + status=ServiceExecStatus.SUCCESS, + content=result.stdout.strip() if result.stdout else "Success.", + ) + except subprocess.CalledProcessError as e: + error_message = ( + e.stderr.strip() + if e.stderr + else "An error occurred \ + while executing the command." + ) + return ServiceResponse( + status=ServiceExecStatus.ERROR, + content=error_message, + ) + except Exception as e: + return ServiceResponse( + status=ServiceExecStatus.ERROR, + content=str(e), + ) diff --git a/src/agentscope/service/file/common.py b/src/agentscope/service/file/common.py index 82ec9eed9..adeb5a0ad 100644 --- a/src/agentscope/service/file/common.py +++ b/src/agentscope/service/file/common.py @@ -2,6 +2,7 @@ """ Common operators for file and directory. """ import os import shutil +from typing import List from agentscope.utils.common import write_file from agentscope.service.service_response import ServiceResponse @@ -201,3 +202,61 @@ def move_directory( status=ServiceExecStatus.ERROR, content=error_message, ) + + +def list_directory_content(directory_path: str) -> ServiceResponse: + """ + List the contents of a directory. i.e. ls -a + + Args: + directory_path (`str`): + The path of the directory to show. + + Returns: + `ServiceResponse`: The results contain a list of direcotry contents, + or an error message if any, including the error type. + """ + if not os.path.exists(directory_path): + return ServiceResponse( + status=ServiceExecStatus.ERROR, + content="FileNotFoundError: The directory does not exist.", + ) + if not os.path.isdir(directory_path): + return ServiceResponse( + status=ServiceExecStatus.ERROR, + content="FileNotFoundError: The path is not a directory", + ) + try: + ls_result: List[str] = os.listdir(directory_path) + return ServiceResponse( + status=ServiceExecStatus.SUCCESS, + content=ls_result, + ) + except Exception as e: + error_message = f"{e.__class__.__name__}: {e}" + return ServiceResponse( + status=ServiceExecStatus.ERROR, + content=error_message, + ) + + +def get_current_directory() -> ServiceResponse: + """ + Get the current working directory path. + + Returns: + `ServiceResponse`: The current working directory path, or an error + message if any, including the error type. + """ + try: + cwd = os.getcwd() + return ServiceResponse( + status=ServiceExecStatus.SUCCESS, + content=cwd, + ) + except Exception as e: + error_message = f"{e.__class__.__name__}: {e}" + return ServiceResponse( + status=ServiceExecStatus.ERROR, + content=error_message, + ) diff --git a/tests/execute_shell_command_test.py b/tests/execute_shell_command_test.py new file mode 100644 index 000000000..206e3bcba --- /dev/null +++ b/tests/execute_shell_command_test.py @@ -0,0 +1,62 @@ +# -*- coding: utf-8 -*- +""" Python code execution test.""" +import unittest +import platform + +from agentscope.service import execute_shell_command +from agentscope.service import ServiceExecStatus + + +class ExecuteShellCommandTest(unittest.TestCase): + """ + Python code execution test. + """ + + def setUp(self) -> None: + """Init for ExecuteShellCommandTest.""" + + # Basic expression + self.arg0 = "touch tmp_a.text" + + self.arg1 = "echo 'Helloworld' >> tmp_a.txt" + + self.arg2 = "cat tmp_a.txt" + + self.arg3 = "rm tmp_a.txt" + + def test(self) -> None: + """test command, skip on windows""" + if platform.system() == "Windows": + return + result = execute_shell_command( + command=self.arg0, + ) + assert result.status == ServiceExecStatus.SUCCESS + assert result.content == "Success." + + result = execute_shell_command( + command=self.arg1, + ) + assert result.status == ServiceExecStatus.SUCCESS + assert result.content == "Success." + + result = execute_shell_command( + command=self.arg2, + ) + assert result.status == ServiceExecStatus.SUCCESS + assert result.content == "Helloworld" + + result = execute_shell_command( + command=self.arg3, + ) + assert result.status == ServiceExecStatus.SUCCESS + assert result.content == "Success." + + result = execute_shell_command( + command=self.arg3, + ) + assert result.status == ServiceExecStatus.ERROR + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/operate_file_test.py b/tests/operate_file_test.py index 7c1b5a2e9..c0c663867 100644 --- a/tests/operate_file_test.py +++ b/tests/operate_file_test.py @@ -15,6 +15,8 @@ write_text_file, read_json_file, write_json_file, + list_directory_content, + get_current_directory, ) from agentscope.service.service_status import ServiceExecStatus @@ -62,12 +64,19 @@ def test_file(self) -> None: def test_dir(self) -> None: """Execute dir test.""" + + is_success = get_current_directory().status + self.assertEqual(is_success, ServiceExecStatus.SUCCESS) + is_success = create_directory(self.dir_name).status self.assertEqual(is_success, ServiceExecStatus.SUCCESS) is_success = move_directory(self.dir_name, self.moved_dir_name).status self.assertEqual(is_success, ServiceExecStatus.SUCCESS) + is_success = list_directory_content(self.moved_dir_name).status + self.assertEqual(is_success, ServiceExecStatus.SUCCESS) + is_success = delete_directory(self.moved_dir_name).status self.assertEqual(is_success, ServiceExecStatus.SUCCESS)