TerminalToolkit

class TerminalToolkit(BaseToolkit):
A toolkit for terminal operations across multiple operating systems. This toolkit provides a set of functions for terminal operations such as searching for files by name or content, executing shell commands, and managing terminal sessions. Parameters:
  • timeout (Optional[float]): The timeout for terminal operations. (default: :obj:20.0)
  • shell_sessions (Optional[Dict[str, Any]]): A dictionary to store shell session information. If :obj:None, an empty dictionary will be used. (default: :obj:None)
  • working_directory (Optional[str]): The working directory for operations. If not provided, it will be determined by the CAMEL_WORKDIR environment variable (if set). If the environment variable is not set, it defaults to ./workspace. All execution and write operations will be restricted to this directory. Read operations can access paths outside this directory. (default: :obj:None)
  • need_terminal (bool): Whether to create a terminal interface. (default: :obj:True)
  • use_shell_mode (bool): Whether to use shell mode for command execution. (default: :obj:True)
  • clone_current_env (bool): Whether to clone the current Python environment. (default: :obj:False)
  • safe_mode (bool): Whether to enable safe mode to restrict operations. (default: :obj:True)
Note: Most functions are compatible with Unix-based systems (macOS, Linux). For Windows compatibility, additional implementation details are needed.

init

def __init__(
    self,
    timeout: Optional[float] = 20.0,
    shell_sessions: Optional[Dict[str, Any]] = None,
    working_directory: Optional[str] = None,
    need_terminal: bool = True,
    use_shell_mode: bool = True,
    clone_current_env: bool = False,
    safe_mode: bool = True
):

_setup_file_output

def _setup_file_output(self):
Set up file output to replace GUI, using a fixed file to simulate terminal.

_clone_current_environment

def _clone_current_environment(self):
Create a new Python virtual environment.

_is_uv_environment

def _is_uv_environment(self):
Detect whether the current Python runtime is managed by uv.

_ensure_uv_available

def _ensure_uv_available(self):
Returns: bool: True if uv is available (either already installed or successfully installed), False otherwise.

_prepare_initial_environment

def _prepare_initial_environment(self):
Prepare initial environment with Python 3.10, pip, and other essential tools.

_setup_initial_env_with_uv

def _setup_initial_env_with_uv(self):
Set up initial environment using uv.

_setup_initial_env_with_venv

def _setup_initial_env_with_venv(self):
Set up initial environment using standard venv.

_check_nodejs_availability

def _check_nodejs_availability(self):
Check if Node.js is available without modifying the system.

_create_terminal

def _create_terminal(self):
Create a terminal GUI. If GUI creation fails, fallback to file output.

_update_terminal_output

def _update_terminal_output(self, output: str):
Update terminal output and send to agent. Parameters:
  • output (str): The output to be sent to the agent

_is_path_within_working_dir

def _is_path_within_working_dir(self, path: str):
Check if the path is within the working directory. Parameters:
  • path (str): The path to check
Returns: bool: Returns True if the path is within the working directory, otherwise returns False

_enforce_working_dir_for_execution

def _enforce_working_dir_for_execution(self, path: str):
Enforce working directory restrictions, return error message if execution path is not within the working directory. Parameters:
  • path (str): The path to be used for executing operations
Returns: Optional[str]: Returns error message if the path is not within the working directory, otherwise returns None

_copy_external_file_to_workdir

def _copy_external_file_to_workdir(self, external_file: str):
Copy external file to working directory. Parameters:
  • external_file (str): The path of the external file
Returns: Optional[str]: New path after copying to the working directory, returns None on failure

_sanitize_command

def _sanitize_command(self, command: str, exec_dir: str):
Check and modify command to ensure safety. Parameters:
  • command (str): The command to check
  • exec_dir (str): The directory to execute the command in
Returns: Tuple: (is safe, modified command or error message)

shell_exec

def shell_exec(
    self,
    id: str,
    command: str,
    interactive: bool = False
):
Executes a shell command in a specified session. This function creates and manages shell sessions to execute commands, simulating a real terminal. It can run commands in both non-interactive (capturing output) and interactive modes. Each session is identified by a unique ID. If a session with the given ID does not exist, it will be created. Parameters:
  • id (str): A unique identifier for the shell session. This is used to manage multiple concurrent shell processes.
  • command (str): The shell command to be executed.
  • interactive (bool, optional): If True, the command runs in interactive mode, connecting it to the terminal’s standard input. This is useful for commands that require user input, like ssh. Defaults to False. Interactive mode is only supported on macOS and Linux. (default: :obj:False)
Returns: str: The standard output and standard error from the command. If an error occurs during execution, a descriptive error message is returned. Note: When interactive is set to True, this function may block if the command requires input. In safe mode, some commands that are considered dangerous are restricted.

shell_view

def shell_view(self, id: str):
View the full output history of a specified shell session. Retrieves the accumulated output (both stdout and stderr) generated by commands in the specified session since its creation. This is useful for checking the complete history of a session, especially after a command has finished execution. Parameters:
  • id (str): The unique identifier of the shell session to view.
Returns: str: The complete output history of the shell session. Returns an error message if the session is not found.

shell_wait

def shell_wait(self, id: str, seconds: Optional[int] = None):
Wait for a command to finish in a specified shell session. Blocks execution and waits for the running process in a shell session to complete. This is useful for ensuring a long-running command has finished before proceeding. Parameters:
  • id (str): The unique identifier of the target shell session.
  • seconds (Optional[int], optional): The maximum time to wait, in seconds. If None, it waits indefinitely. (default: :obj:None)
Returns: str: A message indicating that the process has completed, including the final output. If the process times out, it returns a timeout message.

shell_write_to_process

def shell_write_to_process(
    self,
    id: str,
    input: str,
    press_enter: bool
):
Write input to a running process in a specified shell session. Sends a string of text to the standard input of a running process. This is useful for interacting with commands that require input. This function cannot be used with a command that was started in interactive mode. Parameters:
  • id (str): The unique identifier of the target shell session.
  • input (str): The text to write to the process’s stdin.
  • press_enter (bool): If True, a newline character (\n) is appended to the input, simulating pressing the Enter key.
Returns: str: A status message indicating whether the input was sent, or an error message if the operation fails.

shell_kill_process

def shell_kill_process(self, id: str):
Terminate a running process in a specified shell session. Forcibly stops a command that is currently running in a shell session. This is useful for ending processes that are stuck, running too long, or need to be cancelled. Parameters:
  • id (str): The unique identifier of the shell session containing the process to be terminated.
Returns: str: A status message indicating that the process has been terminated, or an error message if the operation fails.

ask_user_for_help

def ask_user_for_help(self, id: str):
Pause the agent and ask a human for help with a command. This function should be used when the agent is stuck and requires manual intervention, such as solving a CAPTCHA or debugging a complex issue. It pauses the agent’s execution and allows a human to take control of a specified shell session. The human can execute one command to resolve the issue, and then control is returned to the agent. Parameters:
  • id (str): The identifier of the shell session for the human to interact with. If the session does not exist, it will be created.
Returns: str: A status message indicating that the human has finished, including the number of commands executed. If the takeover times out or fails, an error message is returned.

del

def __del__(self):
Clean up resources when the object is being destroyed. Terminates all running processes and closes any open file handles.

get_tools

def get_tools(self):
Returns: List[FunctionTool]: A list of FunctionTool objects representing the functions in the toolkit.