Skip to main content

Filesystem Tools

Comprehensive filesystem operations across different environments with support for local, in-memory, and sandboxed execution.

Overview

AG-Kit provides filesystem tools that work across three different backends:
  • Local File Operator: Direct filesystem access (maximum performance)
  • In-Memory File Operator: Virtual filesystem in memory (testing/isolation)
  • Sandbox File Operator: Secure operations in E2B containers (production safe)

Available Tools

Individual Tools

  • ReadTool - Read file contents with encoding support
  • WriteTool - Write content to files with directory creation
  • EditTool - Make targeted edits to existing files
  • MultiEditTool - Batch edit multiple files in sequence
  • GlobTool - Find files using glob patterns
  • GrepTool - Search file contents with regex patterns
  • LsTool - List directory contents with detailed information

Comprehensive Tool

  • StrReplaceEditor - Advanced multi-operation file editor for AI coding workflows

Quick Comparison

FeatureLocalIn-MemorySandbox
Performance✅ Maximum✅ High⚠️ Network overhead
Security⚠️ Full access✅ Isolated✅ Sandboxed
Persistence✅ Permanent❌ Temporary✅ Session-based
Testing⚠️ Side effects✅ Perfect⚠️ Complex setup

Setup

Local File Operator

from ag_kit_py.tools.fs import LocalFileOperator, ExecutionContext

file_operator = LocalFileOperator()
context = ExecutionContext(
    working_directory="/path/to/workspace",
    fs_operator=file_operator
)

In-Memory File Operator

from ag_kit_py.tools.fs import InMemoryFileOperator, ExecutionContext
from pyfakefs import fake_filesystem

# Create in-memory filesystem
fs = fake_filesystem.FakeFilesystem()
file_operator = InMemoryFileOperator(filesystem=fs)

context = ExecutionContext(
    working_directory="/virtual",
    fs_operator=file_operator
)

Sandbox File Operator

from ag_kit_py.tools.fs import SandboxFileOperator, ExecutionContext
from ag_kit_py.tools.fs.env_utils import setup_e2b_env
from e2b_code_interpreter import Sandbox

# Setup E2B environment (loads .env file automatically)
api_key = setup_e2b_env()
if not api_key:
    raise ValueError("E2B API key not found. Set E2B_API_KEY in .env file or environment")

# Create sandbox with API key
sandbox = Sandbox.create(api_key=api_key)
file_operator = SandboxFileOperator(sandbox=sandbox)

context = ExecutionContext(
    working_directory="/tmp",
    fs_operator=file_operator
)
Environment Setup: Create a .env file in your project root:
E2B_API_KEY=your_e2b_api_key_here
Get your API key from E2B Dashboard.

Individual Tools Usage

ReadTool

from ag_kit_py.tools.fs import create_read_tool, ExecutionContext

# Create read tool
read_tool = create_read_tool(context)

# Read entire file
result = await read_tool.invoke({
    "file_path": "src/config.json"
})

# Read with line range
partial_result = await read_tool.invoke({
    "file_path": "large-file.txt",
    "offset": 100,
    "end_line": 200
})

# Access result data
if result.success:
    content = result.data['content']
    file_path = result.data['filePath']

WriteTool

from ag_kit_py.tools.fs import create_write_tool, ExecutionContext

# Create write tool
write_tool = create_write_tool(context)

result = await write_tool.invoke({
    "file_path": "output/data.json",
    "content": '{"message": "Hello World"}',
    "create_dirs": True
})

if result.success:
    bytes_written = result.data['bytes_written']
    path = result.data['file_path']

EditTool

from ag_kit_py.tools.fs import create_edit_tool, ExecutionContext

# Create edit tool
edit_tool = create_edit_tool(context)

result = await edit_tool.invoke({
    "file_path": "src/config.py",
    "old_string": 'API_URL = "http://localhost"',
    "new_string": 'API_URL = "https://api.production.com"',
    "expected_replacements": 1  # Expected number of replacements
})

if result.success:
    replacements = result.data['replacements_made']
    is_new_file = result.data['is_new_file']

MultiEditTool

from ag_kit_py.tools.fs import create_multiedit_tool, ExecutionContext

# Create multi-edit tool
multiedit_tool = create_multiedit_tool(context)

# Define multiple edits for a single file
edits = [
    {
        "old_string": "DEBUG = True",
        "new_string": "DEBUG = False",
        "expected_replacements": 1
    },
    {
        "old_string": "ENV = 'dev'",
        "new_string": "ENV = 'prod'",
        "expected_replacements": 1
    }
]

result = await multiedit_tool.invoke({
    "file_path": "src/config.py",
    "edits": edits
})

if result.success:
    successful = result.data['edits_successful']
    total_replacements = result.data['total_replacements']

GlobTool

from ag_kit_py.tools.fs import create_glob_tool, ExecutionContext

# Create glob tool
glob_tool = create_glob_tool(context)

result = await glob_tool.invoke({
    "pattern": "**/*.py",
    "path": ".",
    "exclude": ["venv/**", "__pycache__/**"]
})

if result.success:
    matches = result.data['matches']
    for match in matches:
        print(match['path'])

GrepTool

from ag_kit_py.tools.fs import create_grep_tool, ExecutionContext

# Create grep tool
grep_tool = create_grep_tool(context)

result = await grep_tool.invoke({
    "pattern": "TODO|FIXME",
    "path": "src",
    "case_sensitive": False,
    "include": "**/*.py"
})

if result.success:
    matches = result.data['matches']
    for match in matches:
        print(f"{match['file']}:{match['line_number']}: {match['line']}")

LsTool

from ag_kit_py.tools.fs import create_ls_tool, ExecutionContext

# Create ls tool
ls_tool = create_ls_tool(context)

result = await ls_tool.invoke({
    "path": "src",
    "recursive": False,
    "ignore": ["*.pyc", "__pycache__"]
})

if result.success:
    entries = result.data['entries']
    for entry in entries:
        type_str = "DIR" if entry['type'] == 'directory' else "FILE"
        print(f"[{type_str}] {entry['name']}")

StrReplaceEditor

Advanced file editing tool designed for AI coding workflows:
from ag_kit_py.tools.fs import create_str_replace_editor_tool, ExecutionContext

# Create str-replace-editor tool
editor_tool = create_str_replace_editor_tool(context)

# Create new file
result = await editor_tool.invoke({
    "command": "create",
    "path": "new-file.py",
    "file_text": "def main():\n    pass"
})

# View file contents
result = await editor_tool.invoke({
    "command": "view",
    "path": "file.py",
    "view_range": [1, 50]  # Optional line range
})

# Replace text
result = await editor_tool.invoke({
    "command": "str_replace",
    "path": "src/app.py",
    "old_str": "old implementation",
    "new_str": "new implementation"
})

# Insert text
result = await editor_tool.invoke({
    "command": "insert",
    "path": "src/app.py",
    "insert_line": 10,
    "new_str": "# New comment"
})

API Reference

Common Interfaces

from typing import Optional, List, Dict, Any
from dataclasses import dataclass

@dataclass
class ExecutionContext:
    """Execution context for file operations"""
    working_directory: str
    fs_operator: BaseFileOperator
    message_handler: Optional[Any] = None

@dataclass
class ToolResult:
    """Standardized tool response"""
    success: bool
    data: Optional[Dict[str, Any]] = None
    error: Optional[str] = None
    error_type: Optional[str] = None

Tool Factory Functions

# create_read_tool() - Creates a read tool
def create_read_tool(context: ExecutionContext) -> DynamicTool:
    """Create a tool for reading files
    
    Tool parameters:
    - file_path: str - Path to file
    - offset: Optional[int] - Start line (0-indexed)
    - end_line: Optional[int] - End line (inclusive)
    """
    ...

# create_write_tool() - Creates a write tool
def create_write_tool(context: ExecutionContext) -> DynamicTool:
    """Create a tool for writing files
    
    Tool parameters:
    - file_path: str - Path to file
    - content: str - Content to write
    - create_dirs: bool - Create parent directories (default: True)
    """
    ...

# create_edit_tool() - Creates an edit tool
def create_edit_tool(context: ExecutionContext) -> DynamicTool:
    """Create a tool for editing files
    
    Tool parameters:
    - file_path: str - Path to file
    - old_string: str - Text to replace
    - new_string: str - Replacement text
    - expected_replacements: int - Expected number of replacements
    """
    ...

# create_multiedit_tool() - Creates a multi-edit tool
def create_multiedit_tool(context: ExecutionContext) -> DynamicTool:
    """Create a tool for multiple edits on a single file
    
    Tool parameters:
    - file_path: str - Path to file
    - edits: List[Dict] - List of edit operations
    """
    ...

# create_glob_tool() - Creates a glob tool
def create_glob_tool(context: ExecutionContext) -> DynamicTool:
    """Create a tool for finding files by pattern
    
    Tool parameters:
    - pattern: str - Glob pattern
    - path: str - Base path (default: ".")
    - exclude: Optional[List[str]] - Exclude patterns
    """
    ...

# create_grep_tool() - Creates a grep tool
def create_grep_tool(context: ExecutionContext) -> DynamicTool:
    """Create a tool for searching file contents
    
    Tool parameters:
    - pattern: str - Regex pattern
    - path: str - Base path (default: ".")
    - case_sensitive: bool - Case sensitive search (default: True)
    - include: Optional[str] - Include pattern
    - exclude: Optional[str] - Exclude pattern
    """
    ...

# create_ls_tool() - Creates an ls tool
def create_ls_tool(context: ExecutionContext) -> DynamicTool:
    """Create a tool for listing directory contents
    
    Tool parameters:
    - path: str - Directory path (default: ".")
    - recursive: bool - Recursive listing (default: False)
    - ignore: Optional[List[str]] - Ignore patterns
    """
    ...

# create_str_replace_editor_tool() - Creates a str-replace-editor tool
def create_str_replace_editor_tool(context: ExecutionContext) -> DynamicTool:
    """Create an advanced file editor tool
    
    Tool parameters:
    - command: str - Command ("view", "create", "str_replace", "insert")
    - path: str - File path
    - old_str: Optional[str] - Text to replace
    - new_str: Optional[str] - Replacement text
    - insert_line: Optional[int] - Line number for insert
    - view_range: Optional[List[int]] - Line range for view
    - file_text: Optional[str] - Content for create
    """
    ...

Advanced Usage

Multi-Backend Workflow

import os
from ag_kit_py.tools.fs import (
    LocalFileOperator,
    InMemoryFileOperator,
    ExecutionContext
)
from pyfakefs import fake_filesystem

def create_file_operator():
    env = os.environ.get("ENV", "development")
    
    if env == "test":
        # Create in-memory filesystem
        fs = fake_filesystem.FakeFilesystem()
        return InMemoryFileOperator(filesystem=fs)
    else:
        # Use local filesystem
        return LocalFileOperator()

file_operator = create_file_operator()
context = ExecutionContext(
    working_directory="/workspace",
    fs_operator=file_operator
)

Batch Operations

import asyncio
from ag_kit_py.tools.fs import create_read_tool, ExecutionContext

# Create read tool
read_tool = create_read_tool(context)

files = ["config.json", "package.json", "pyproject.toml"]
results = await asyncio.gather(*[
    read_tool.invoke({"file_path": file})
    for file in files
])

for result in results:
    if result.success:
        print(f"Read {result.data['filePath']}")

Error Handling

from ag_kit_py.tools.fs import create_read_tool, ExecutionContext

# Create read tool
read_tool = create_read_tool(context)

result = await read_tool.invoke({"file_path": "nonexistent.txt"})

if not result.success:
    error_type = result.error_type
    if error_type == "file_not_found":
        print("File does not exist")
    elif error_type == "permission":
        print("Permission denied")
    elif error_type == "security":
        print("Security violation")
    else:
        print(f"Unknown error: {result.error}")

Security Features

Path Validation

All tools automatically validate paths to prevent directory traversal attacks:
# ✅ Safe - relative paths within workspace
await read(context, file_path="src/config.py")

# ❌ Blocked - directory traversal
await read(context, file_path="../../../etc/passwd")

Production Recommendations

  • Use Sandbox File Operator for untrusted code
  • Use In-Memory File Operator for testing
  • Use Local File Operator only for trusted environments

Examples

Development Workflow

import json
from ag_kit_py.tools.fs import (
    create_read_tool,
    create_edit_tool,
    create_grep_tool,
    create_glob_tool,
    ExecutionContext
)

# Create tools
read_tool = create_read_tool(context)
edit_tool = create_edit_tool(context)
grep_tool = create_grep_tool(context)
glob_tool = create_glob_tool(context)

# Read configuration
config_result = await read_tool.invoke({"file_path": "config.json"})
config = json.loads(config_result.data['content'])

# Update configuration
await edit_tool.invoke({
    "file_path": "config.json",
    "old_string": '"debug": false',
    "new_string": '"debug": true',
    "expected_replacements": 1
})

# Find all Python files
py_files = await glob_tool.invoke({
    "pattern": "**/*.py",
    "path": ".",
    "exclude": ["venv/**"]
})

# Search for TODOs
todos = await grep_tool.invoke({
    "pattern": "TODO|FIXME",
    "path": ".",
    "include": "**/*.py"
})

for match in todos.data['matches']:
    print(f"{match['file']}:{match['line_number']}: {match['line']}")

File Refactoring Workflow

from ag_kit_py.tools.fs import (
    create_grep_tool,
    create_edit_tool,
    ExecutionContext
)

# Create tools
grep_tool = create_grep_tool(context)
edit_tool = create_edit_tool(context)

# Find all files with old API
result = await grep_tool.invoke({
    "pattern": "old_api_function",
    "path": ".",
    "include": "**/*.py"
})

# Apply edits to each file
for match in result.data['matches']:
    file_path = match['file']
    edit_result = await edit_tool.invoke({
        "file_path": file_path,
        "old_string": "old_api_function",
        "new_string": "new_api_function",
        "expected_replacements": 1
    })
    if edit_result.success:
        print(f"Updated {file_path}")

File Processing Pipeline

from ag_kit_py.tools.fs import (
    create_glob_tool,
    create_read_tool,
    create_write_tool,
    ExecutionContext
)

# Create tools
glob_tool = create_glob_tool(context)
read_tool = create_read_tool(context)
write_tool = create_write_tool(context)

# Find all text files
files_result = await glob_tool.invoke({
    "pattern": "**/*.txt",
    "path": "input"
})

# Process each file
for match in files_result.data['matches']:
    file_path = match['path']
    
    # Read file
    read_result = await read_tool.invoke({"file_path": file_path})
    if read_result.success:
        content = read_result.data['content']
        
        # Process content
        processed = content.upper()
        
        # Write to output
        output_path = file_path.replace("input", "output")
        await write_tool.invoke({
            "file_path": output_path,
            "content": processed,
            "create_dirs": True
        })

Performance Optimization

Parallel Operations

import asyncio
from ag_kit_py.tools.fs import create_read_tool, ExecutionContext

# Create read tool
read_tool = create_read_tool(context)

# Read multiple files in parallel
files = ["file1.txt", "file2.txt", "file3.txt"]
results = await asyncio.gather(*[
    read_tool.invoke({"file_path": f})
    for f in files
])

for result in results:
    if result.success:
        print(f"Read {result.data['filePath']}")

Best Practices

  1. Validate paths: Always use relative paths within workspace
  2. Handle errors: Check success and error_type fields
  3. Use appropriate backend: Match environment needs
  4. Batch operations: Use multiedit for multiple changes
  5. Clean up: Remove temporary files
  6. Log operations: Track file changes

Troubleshooting

File Not Found

read_tool = create_read_tool(context)
result = await read_tool.invoke({"file_path": "file.txt"})
if not result.success and result.error_type == "file_not_found":
    print("File does not exist")

Path Security Violation

read_tool = create_read_tool(context)

# Use relative paths within workspace
# ✅ Good
await read_tool.invoke({"file_path": "src/config.py"})

# ❌ Bad
await read_tool.invoke({"file_path": "/etc/passwd"})

Examples Repository

Find complete working examples at: