Method execute

mini_agent/tools/bash_tool.py:309–438

Execute shell command with optional background execution. Args: command: The shell command to execute timeout: Timeout in seconds (default: 120, max: 600) run_in_background: Set true to run command in background Returns: BashExecution

(
        self,
        command: str,
        timeout: int = 120,
        run_in_background: bool = False,
    )

        }

    async def execute(
        self,
        command: str,
        timeout: int = 120,
        run_in_background: bool = False,
    ) -> ToolResult:
        """Execute shell command with optional background execution.

        Args:
            command: The shell command to execute
            timeout: Timeout in seconds (default: 120, max: 600)
            run_in_background: Set true to run command in background

        Returns:
            BashExecutionResult with command output and status
        """

        try:
            # Validate timeout
            if timeout > 600:
                timeout = 600
            elif timeout < 1:
                timeout = 120

            # Prepare shell-specific command execution
            if self.is_windows:
                # Windows: Use PowerShell with appropriate encoding
                shell_cmd = ["powershell.exe", "-NoProfile", "-Command", command]
            else:
                # Unix/Linux/macOS: Use bash
                shell_cmd = command

            if run_in_background:
                # Background execution: Create isolated process
                bash_id = str(uuid.uuid4())[:8]

                # Start background process with combined stdout/stderr
                if self.is_windows:
                    process = await asyncio.create_subprocess_exec(
                        *shell_cmd,
                        stdout=asyncio.subprocess.PIPE,
                        stderr=asyncio.subprocess.STDOUT,
                        cwd=self.workspace_dir,
                    )
                else:
                    process = await asyncio.create_subprocess_shell(
                        shell_cmd,
                        stdout=asyncio.subprocess.PIPE,
                        stderr=asyncio.subprocess.STDOUT,
                        cwd=self.workspace_dir,
                    )

                # Create background shell and add to manager
                bg_shell = BackgroundShell(bash_id=bash_id, command=command, process=process, start_time=time.time())
                BackgroundShellManager.add(bg_shell)

                # Start monitoring task
                await BackgroundShellManager.start_monitor(bash_id)