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