Bash Tools (Python)

Execute bash commands using either local system execution or secure sandboxed environments with comprehensive command management capabilities.

Overview

The Bash Tools provide a unified interface for executing bash commands across different environments:

Local Execution: Direct execution on the local system
Sandbox Execution: Secure execution in isolated E2B containers
Multi-Command Support: Execute multiple commands in sequence
Real-time Streaming: Live output monitoring with callbacks

Quick Start

Local Execution

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!"'})

Sandbox Execution

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

# Note: Sandbox.create() is synchronous in Python SDK
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'})

Core Features

Single Command Execution

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

Multi-Command Execution

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

Real-time Output Streaming

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

Interactive Commands

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

Command Builders

Pre-built command generators for common operations:

from ag_kit_py.tools.bash import CommandBuilders

# File operations
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)

# Directory operations
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)

# Execute built commands
result = await bash_tool.invoke({'command': list_command})

Security Features

Command Validation

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}")

Argument Escaping

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])
# Result: "ls '-la' 'file with spaces & special chars'"

API Reference

Context Creation

# Local context
create_local_bash_context(
    cwd: Optional[str] = None,
    env: Optional[Dict[str, str]] = None,
    default_timeout: int = 30000
) -> BashExecutionContext

# Sandbox context
create_sandbox_bash_context(
    sandbox: Sandbox,
    cwd: Optional[str] = None,
    env: Optional[Dict[str, str]] = None,
    default_timeout: int = 30000
) -> BashExecutionContext

Tool Creation

# Create bash tool
create_bash_tool(context: BashExecutionContext) -> Tool

# Create multi-command tool
create_multi_command_tool(context: BashExecutionContext) -> Tool

Operator Classes

BaseBashOperator (Abstract)

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

Data Models

CommandOptions

@dataclass
class CommandOptions:
    cwd: Optional[str] = None
    env: Optional[Dict[str, str]] = None
    timeout: Optional[int] = None  # milliseconds
    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  # milliseconds

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

Utility Functions

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

# Validate command
validation = validate_command('echo "safe"')  # CommandValidation

# Escape arguments
escaped = escape_shell_arg("it's a test")  # "'it'\"'\"'s a test'"

# Build command
cmd = build_command('ls', ['-la', '/home'])  # "ls '-la' '/home'"

# Parse output
parsed = parse_command_output(stdout, stderr)  # ParsedOutput

# Format time
time_str = format_execution_time(1500)  # "1.50s"

CommandBuilders Reference

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

Examples

Development Workflow

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

File System Operations

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

Data Processing Pipeline

pipeline = await multi_tool.invoke({
    'commands': [
        'curl -o data.json https://api.example.com/data',
        'jq ".items[]" data.json > processed.json',
        'wc -l processed.json'
    ],
    'continue_on_error': False
})

Error Handling

try:
    result = await bash_tool.invoke({
        'command': 'risky-command',
        'timeout': 5000
    })
    
    if not result.success:
        print(f"Command failed: {result.data['stderr']}")
        print(f"Exit code: {result.data['exit_code']}")
except Exception as error:
    print(f'Execution error: {error}')

Advanced Usage

Using Operator Directly

from ag_kit_py.tools.bash import LocalBashOperator, CommandOptions

operator = LocalBashOperator(cwd='/project')

# Execute with custom options
result = await operator.execute_command(
    'long-running-command',
    CommandOptions(
        timeout=60000,
        env={'DEBUG': '1'},
        on_stdout=lambda data: print(data, end='')
    )
)

# Directory operations
await operator.change_directory('/tmp')
current = await operator.get_current_directory()

# Environment management
await operator.set_environment_variable('API_KEY', 'secret')
env = await operator.get_environment_variables()

Custom Tool Configuration

from ag_kit_py.tools.bash import BashExecutionContext, create_bash_tool

# Create custom context
context = BashExecutionContext(
    bash_operator=LocalBashOperator(),
    working_directory='/custom/path',
    environment_variables={'ENV': 'production'},
    default_timeout=60000
)

# Create tool with custom context
bash_tool = create_bash_tool(context)

Installation

Basic Installation

pip install ag_kit_py

With E2B Support

pip install ag_kit_py e2b-code-interpreter

Environment Variables

When using sandbox execution, configure E2B:

export AG_KIT_SANDBOX_API_KEY="your-e2b-api-key"
export AG_KIT_SANDBOX_DOMAIN="your-e2b-domain"  # optional

File System Tools - File operations
Code Executor - Code execution
Framework Adapters - LangChain & LlamaIndex integration
MCP Integration - Model Context Protocol
TypeScript Bash Tools - TypeScript version

Examples Repository

Find complete working examples at:

bash_tools.py - Comprehensive bash tools examples
langchain_adapter.py - LangChain integration
llamaindex_adapter.py - LlamaIndex integration

Security Considerations

Command Validation: Always validate user-provided commands
Argument Escaping: Use escape_shell_arg() for user input
Sandbox Isolation: Use sandbox execution for untrusted code
Timeout Limits: Set appropriate timeouts for all commands
Environment Variables: Be careful with sensitive data in env vars

Performance Tips

Reuse Operators: Create operator once and reuse for multiple commands
Batch Commands: Use multi-command tool for related operations
Stream Output: Use callbacks for large output to avoid memory issues
Timeout Management: Set realistic timeouts based on expected execution time
Sandbox Lifecycle: Reuse sandbox instances when possible

Documentation Index

​Bash Tools (Python)

​Overview

​Quick Start

​Local Execution

​Sandbox Execution

​Core Features

​Single Command Execution

​Multi-Command Execution

​Real-time Output Streaming

​Interactive Commands

​Command Builders

​Security Features

​Command Validation

​Argument Escaping

​API Reference

​Context Creation

​Tool Creation

​Operator Classes

​BaseBashOperator (Abstract)

​LocalBashOperator

​SandboxBashOperator

​Data Models

​CommandOptions

​CommandResult

​BashToolResponse

​MultiCommandResponse

​Utility Functions

​CommandBuilders Reference

​Examples

​Development Workflow

​File System Operations

​Data Processing Pipeline

​Error Handling

​Advanced Usage

​Using Operator Directly

​Custom Tool Configuration

​Installation

​Basic Installation

​With E2B Support

​Environment Variables

​Related Documentation

​Examples Repository

​Security Considerations

​Performance Tips