Camel.toolkits.terminal toolkit.terminal toolkit

​

_to_plain

def _to_plain(text: str):

​

TerminalToolkit

class TerminalToolkit(BaseToolkit):

• 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.
• working_dir (str): The base directory for all operations. For the local backend, this acts as a security sandbox.
• session_logs_dir (Optional[str]): The directory to store session logs. Defaults to a ‘terminal_logs’ subfolder in the working_dir.
• timeout (int): The default timeout in seconds for blocking commands. Defaults to 60.

​

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

​

_setup_cloned_environment

def _setup_cloned_environment(self):

​

_setup_initial_environment

def _setup_initial_environment(self):

​

_adapt_command_for_environment

def _adapt_command_for_environment(self, command: str):

​

_write_to_log

def _write_to_log(self, log_file: str, content: str):

• log_file (str): Path to the log file
• content (str): Content to write

​

_sanitize_command

def _sanitize_command(self, command: str):

​

_start_output_reader_thread

def _start_output_reader_thread(self, session_id: str):

​

_collect_output_until_idle

def _collect_output_until_idle(
    self,
    id: str,
    idle_duration: float = 0.5,
    check_interval: float = 0.1,
    max_wait: float = 5.0
):

• id (str): The session ID.
• idle_duration (float): How long the stream must be empty to be considered idle.(default: 0.5)
• check_interval (float): The time to sleep between checks. (default: 0.1)
• max_wait (float): The maximum total time to wait for the process to go idle. (default: 5.0)

​

shell_exec

def shell_exec(
    self,
    id: str,
    command: str,
    block: bool = True
):

• command (str): The command to execute.
• block (bool): If True, the command runs synchronously, waiting for it to complete or time out, and returns its full output. If False, the command runs asynchronously in the background.
• id (Optional[str]): A specific ID for the session. If not provided, a unique ID is generated for non-blocking sessions.

​

shell_write_to_process

def shell_write_to_process(self, id: str, command: str):

• id (str): The unique session ID of the non-blocking process.
• command (str): The text to write to the process’s standard input.

​

shell_view

def shell_view(self, id: str):

• id (str): The unique session ID of the non-blocking process.

​

shell_wait

def shell_wait(self, id: str, wait_seconds: float = 5.0):

• id (str): The unique session ID of the non-blocking process.
• wait_seconds (float): The maximum number of seconds to wait.

​

shell_kill_process

def shell_kill_process(self, id: str):

• id (str): The unique session ID of the process to kill.

​

shell_ask_user_for_help

def shell_ask_user_for_help(self, id: str, prompt: str):

1. If session exists: Shows session output and allows interaction
2. If session doesn’t exist: Creates a temporary session for help

• 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’.”).

​

del

def __del__(self):

​

get_tools

def get_tools(self):