Skip to main content

_to_plain

def _to_plain(text: str):
Convert ANSI text to plain text using rich if available.

TerminalToolkit

class TerminalToolkit(BaseToolkit):
A toolkit for LLM agents to execute and interact with terminal commands in either a local or a sandboxed Docker environment. Parameters:
  • timeout (Optional[float]): The default timeout in seconds for blocking commands. Defaults to 20.0.
  • working_directory (Optional[str]): The base directory for operations. For the local backend, this acts as a security sandbox. For the Docker backend, this sets the working directory inside the container. If not specified, defaults to ”./workspace” for local and “/workspace” for Docker.
  • use_docker_backend (bool): If True, all commands are executed in a Docker container. Defaults to False.
  • docker_container_name (Optional[str]): The name of the Docker container to use. Required if use_docker_backend is True.
  • session_logs_dir (Optional[str]): The directory to store session logs. Defaults to a ‘terminal_logs’ subfolder in the working directory.
  • safe_mode (bool): Whether to apply security checks to commands. Defaults to True.
  • allowed_commands (Optional[List[str]]): List of allowed commands when safe_mode is True. If None, uses default safety rules.
  • clone_current_env (bool): Whether to clone the current Python environment for local execution. Defaults to False.
  • install_dependencies (List): A list of user specified libraries to install.

init

def __init__(
    self,
    timeout: Optional[float] = 20.0,
    working_directory: Optional[str] = None,
    use_docker_backend: bool = False,
    docker_container_name: Optional[str] = None,
    session_logs_dir: Optional[str] = None,
    safe_mode: bool = True,
    allowed_commands: Optional[List[str]] = None,
    clone_current_env: bool = False,
    install_dependencies: Optional[List[str]] = None
):

_setup_cloned_environment

def _setup_cloned_environment(self):
Set up a cloned Python environment.

_install_dependencies

def _install_dependencies(self):
Install user specified dependencies in the current environment.

_setup_initial_environment

def _setup_initial_environment(self):
Set up an initial environment with Python 3.10.

_get_venv_path

def _get_venv_path(self):
Get the virtual environment path if available.

_write_to_log

def _write_to_log(self, log_file: str, content: str):
Write content to log file with optional ANSI stripping. Parameters:
  • log_file (str): Path to the log file
  • content (str): Content to write

_sanitize_command

def _sanitize_command(self, command: str):
A comprehensive command sanitizer for both local and Docker backends.

_start_output_reader_thread

def _start_output_reader_thread(self, session_id: str):
Starts a thread to read stdout from a non-blocking process.

_collect_output_until_idle

def _collect_output_until_idle(
    self,
    id: str,
    idle_duration: float = 0.5,
    max_wait: float = 5.0
):
Collects output from a session until it’s idle or a max wait time is reached. Parameters:
  • id (str): The session ID.
  • idle_duration (float): How long the stream must be empty to be considered idle.(default: 0.5)
  • max_wait (float): The maximum total time to wait for the process to go idle. (default: 5.0)
Returns: str: The collected output. If max_wait is reached while the process is still outputting, a warning is appended.

shell_exec

def shell_exec(
    self,
    id: str,
    command: str,
    block: bool = True,
    timeout: float = 20.0
):
Executes a shell command in blocking or non-blocking mode. Parameters:
  • id (str): A unique identifier for the command’s session. This ID is used to interact with non-blocking processes.
  • command (str): The shell command to execute.
  • block (bool, optional): Determines the execution mode. Defaults to True. If True (blocking mode), the function waits for the command to complete and returns the full output. Use this for most commands. If False (non-blocking mode), the function starts the command in the background. Use this only for interactive sessions or long-running tasks, or servers.
  • timeout (float, optional): The maximum time in seconds to wait for the command to complete in blocking mode. If the command does not complete within the timeout, it will be restarted automatically as a non-blocking background session. You can then use shell_view(id) to check output, or shell_kill_process(id) to terminate it. This parameter is ignored in non-blocking mode. (default: :obj:20)
Returns: str: The output of the command execution, which varies by mode. In blocking mode, returns the complete standard output and standard error from the command. In non-blocking mode, returns a confirmation message with the session id. To interact with the background process, use other functions: shell_view(id) to see output, shell_write_to_process(id, "input") to send input, and shell_kill_process(id) to terminate.

shell_write_to_process

def shell_write_to_process(self, id: str, command: str):
This function sends command to a running non-blocking process and returns the resulting output after the process becomes idle again. A newline \n is automatically appended to the input command. Parameters:
  • id (str): The unique session ID of the non-blocking process.
  • command (str): The text to write to the process’s standard input.
Returns: str: The output from the process after the command is sent.

shell_view

def shell_view(self, id: str):
Retrieves new output from a non-blocking session. This function returns only NEW output since the last call. It does NOT wait or block - it returns immediately with whatever is available. Parameters:
  • id (str): The unique session ID of the non-blocking process.
Returns: str: New output if available, or a status message.

shell_kill_process

def shell_kill_process(self, id: str):
This function forcibly terminates a running non-blocking process. Parameters:
  • id (str): The unique session ID of the process to kill.
Returns: str: A confirmation message indicating the process was terminated.

shell_ask_user_for_help

def shell_ask_user_for_help(self, id: str, prompt: str):
This function pauses execution and asks a human for help with an interactive session. This method can handle different scenarios:
  1. If session exists: Shows session output and allows interaction
  2. If session doesn’t exist: Creates a temporary session for help
Parameters:
  • id (str): The session ID of the interactive process needing help. Can be empty string for general help without session context.
  • prompt (str): The question or instruction from the LLM to show the human user (e.g., “The program is asking for a filename. Please enter ‘config.json’.”).
Returns: str: The output from the shell session after the user’s command has been executed, or help information for general queries.

enter

def __enter__(self):
Context manager entry.

exit

def __exit__(
    self,
    exc_type,
    exc_val,
    exc_tb
):
Context manager exit - clean up all sessions.

cleanup

def cleanup(self):
Clean up all active sessions.

del

def __del__(self):
Fallback cleanup in destructor.

get_tools

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