Add more service functions

src/agentscope/service/__init__.py

@@ -3,13 +3,16 @@
 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,
     move_file,
     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",

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),
+        )

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,
+        )

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()

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)