跳转到主要内容

Bash 工具集 (Python)

通过统一接口执行bash命令,支持本地系统执行或安全的沙箱隔离环境,提供全面的命令管理能力。

概述

Bash工具集提供跨环境执行bash命令的统一接口:
  • 本地执行:直接在本地系统运行
  • 沙箱执行:在隔离的E2B容器中安全运行
  • 多命令支持:顺序执行多个命令
  • 实时流输出:通过回调函数实时监控输出

快速开始

本地执行

from ag_kit_py.tools.bash import create_local_bash_context, create_bash_tool

context = create_local_bash_context(
    cwd='/path/to/working/directory',
    env={'CUSTOM_VAR': 'value'}
)

bash_tool = create_bash_tool(context)
result = await bash_tool.invoke({'command': 'echo "Hello World!"'})

沙箱执行

from ag_kit_py.tools.bash import create_sandbox_bash_context, create_bash_tool
from e2b_code_interpreter import Sandbox

# 注意:Python SDK中Sandbox.create()是同步方法
sandbox = Sandbox.create()
context = create_sandbox_bash_context(
    sandbox=sandbox,
    cwd='/home/user'
)

bash_tool = create_bash_tool(context)
result = await bash_tool.invoke({'command': 'ls -la'})

核心功能

单命令执行

result = await bash_tool.invoke({
    'command': 'find . -name "*.py" | head -5',
    'cwd': '/project/src',
    'env': {'NODE_ENV': 'development'},
    'timeout': 10000
})

多命令执行

from ag_kit_py.tools.bash import create_multi_command_tool

multi_tool = create_multi_command_tool(context)
result = await multi_tool.invoke({
    'commands': [
        'mkdir -p /tmp/test',
        'echo "Hello" > /tmp/test/file.txt',
        'cat /tmp/test/file.txt'
    ],
    'continue_on_error': False
})

实时输出流

def on_stdout(data: str):
    print(f"OUT: {data}", end="")

def on_stderr(data: str):
    print(f"ERR: {data}", end="")

result = await bash_tool.invoke({
    'command': 'for i in {1..5}; do echo "Step $i"; sleep 1; done',
    'on_stdout': on_stdout,
    'on_stderr': on_stderr
})

交互式命令

result = await bash_tool.invoke({
    'command': 'python3 -c "name = input(\'Name: \'); print(f\'Hello {name}!\')"',
    'input': 'Alice\n',
    'timeout': 5000
})

命令构建器

预置常见操作的命令生成器:
from ag_kit_py.tools.bash import CommandBuilders

# 文件操作
list_command = CommandBuilders.list_files('/home/user', all=True, long=True, human=True)
find_command = CommandBuilders.find_files('*.py', './src')
grep_command = CommandBuilders.grep('TODO', '*.js', recursive=True, ignore_case=True)

# 目录操作
mkdir_command = CommandBuilders.mkdir('/path/to/dir', parents=True)
copy_command = CommandBuilders.copy('source', 'dest', recursive=True)
remove_command = CommandBuilders.remove('/tmp/file', force=True)

# 执行构建的命令
result = await bash_tool.invoke({'command': list_command})

安全特性

命令验证

from ag_kit_py.tools.bash import validate_command

validation = validate_command('rm -rf /')
if not validation.is_valid:
    print(f"Dangerous command: {validation.reason}")

参数转义

from ag_kit_py.tools.bash import escape_shell_arg, build_command

user_input = "file with spaces & special chars"
command = build_command('ls', ['-la', user_input])
# 结果: "ls '-la' 'file with spaces & special chars'"

应用程序接口参考

上下文创建

# 本地上下文
create_local_bash_context(
    cwd: Optional[str] = None,
    env: Optional[Dict[str, str]] = None,
    default_timeout: int = 30000
) -> BashExecutionContext

# 沙箱上下文
create_sandbox_bash_context(
    sandbox: Sandbox,
    cwd: Optional[str] = None,
    env: Optional[Dict[str, str]] = None,
    default_timeout: int = 30000
) -> BashExecutionContext

工具创建

# 创建bash工具
create_bash_tool(context: BashExecutionContext) -> Tool

# 创建多命令工具
create_multi_command_tool(context: BashExecutionContext) -> Tool

操作器类

BaseBashOperator (抽象类)

class BaseBashOperator:
    async def execute_command(
        self,
        command: str,
        options: Optional[CommandOptions] = None
    ) -> CommandResult
    
    async def get_current_directory(self) -> str
    
    async def change_directory(self, path: str) -> None
    
    async def get_environment_variables(self) -> Dict[str, str]
    
    async def set_environment_variable(self, key: str, value: str) -> None
    
    async def cleanup(self) -> None

LocalBashOperator

operator = LocalBashOperator(
    cwd: Optional[str] = None,
    env: Optional[Dict[str, str]] = None
)

result = await operator.execute_command('echo "Hello"')

SandboxBashOperator

operator = SandboxBashOperator(
    sandbox: Sandbox,
    cwd: Optional[str] = None,
    env: Optional[Dict[str, str]] = None
)

result = await operator.execute_command('ls -la')

数据模型

CommandOptions

@dataclass
class CommandOptions:
    cwd: Optional[str] = None
    env: Optional[Dict[str, str]] = None
    timeout: Optional[int] = None  # 毫秒
    input: Optional[str] = None
    shell: Optional[bool] = True
    on_stdout: Optional[Callable[[str], None]] = None
    on_stderr: Optional[Callable[[str], None]] = None

CommandResult

@dataclass
class CommandResult:
    success: bool
    exit_code: Optional[int]
    stdout: str
    stderr: str
    execution_time: float  # 毫秒

BashToolResponse

class BashToolResponse(BaseModel):
    command: str
    exit_code: Optional[int]
    stdout: str
    stderr: str
    working_directory: str

MultiCommandResponse

class MultiCommandResponse(BaseModel):
    commands: List[str]
    results: List[CommandResultItem]
    successful_commands: int
    failed_commands: int
    working_directory: str

实用函数

from ag_kit_py.tools.bash import (
    validate_command,
    escape_shell_arg,
    build_command,
    parse_command_output,
    format_execution_time,
    CommandBuilders
)

# 验证命令
validation = validate_command('echo "safe"')  # CommandValidation

# 转义参数
escaped = escape_shell_arg("it's a test")  # "'it'\"'\"'s a test'"

# 构建命令
cmd = build_command('ls', ['-la', '/home'])  # "ls '-la' '/home'"

# 解析输出
parsed = parse_command_output(stdout, stderr)  # ParsedOutput

# 格式化时间
time_str = format_execution_time(1500)  # "1.50s"

CommandBuilders 参考

class CommandBuilders:
    @staticmethod
    def list_files(
        path: str = ".",
        all: bool = False,
        long: bool = False,
        human: bool = False
    ) -> str
    
    @staticmethod
    def find_files(pattern: str, path: str = ".") -> str
    
    @staticmethod
    def grep(
        pattern: str,
        files: str = "*",
        recursive: bool = False,
        ignore_case: bool = False,
        line_numbers: bool = False
    ) -> str
    
    @staticmethod
    def exists(path: str) -> str
    
    @staticmethod
    def stat(path: str) -> str
    
    @staticmethod
    def mkdir(path: str, parents: bool = False) -> str
    
    @staticmethod
    def copy(
        source: str,
        destination: str,
        recursive: bool = False,
        preserve: bool = False
    ) -> str
    
    @staticmethod
    def move(source: str, destination: str) -> str
    
    @staticmethod
    def remove(
        path: str,
        recursive: bool = False,
        force: bool = False
    ) -> str

示例

开发工作流

build_result = await multi_tool.invoke({
    'commands': [
        'pip install -r requirements.txt',
        'python setup.py build',
        'pytest tests/'
    ],
    'continue_on_error': False,
    'timeout': 120000
})

文件系统操作

setup_result = await multi_tool.invoke({
    'commands': [
        'mkdir -p project/{src,tests,docs}',
        'touch project/src/__init__.py',
        'echo "# Project" > project/README.md'
    ]
})

数据处理管道

pipeline = await multi_tool.invoke({
    'commands': [
        'curl -o data.json https://api