跳转到主要内容

​
文件系统工具

支持跨不同环境执行全面的文件系统操作，涵盖本地、内存和沙箱执行模式。

​
概述

AG-Kit 提供可在三种不同后端运行的文件系统工具：
  • 本地文件操作器：直接访问文件系统（最高性能）
  • 内存文件操作器：内存中的虚拟文件系统（测试/隔离场景）
  • 沙箱文件操作器：在 E2B 容器中执行安全操作（生产环境安全）

​
可用工具

​
独立工具

  • ReadTool - 支持编码读取文件内容
  • WriteTool - 写入文件内容并自动创建目录
  • EditTool - 对现有文件进行定向编辑
  • MultiEditTool - 批量顺序编辑多个文件
  • GlobTool - 使用通配符模式查找文件
  • GrepTool - 通过正则表达式搜索文件内容
  • LsTool - 列出目录内容并显示详细信息

​
综合工具

  • StrReplaceEditor - 专为AI编码工作流设计的高级多操作文件编辑器

​
快速对比

特性本地内存沙箱
性能✅ 最高✅ 高⚠️ 网络开销
安全性⚠️ 完全访问✅ 隔离✅ 沙箱环境
持久性✅ 永久❌ 临时✅ 会话级
测试便利性⚠️ 副作用✅ 理想环境⚠️ 复杂配置

​
配置方法

​
本地文件操作器

from ag_kit_py.tools.fs import LocalFileOperator, ExecutionContext

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

​
内存文件操作器

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

# 创建内存文件系统
fs = fake_filesystem.FakeFilesystem()
file_operator = InMemoryFileOperator(filesystem=fs)

context = ExecutionContext(
    working_directory="/virtual",
    fs_operator=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

# 配置E2B环境（自动加载.env文件）
api_key = setup_e2b_env()
if not api_key:
    raise ValueError("未找到E2B API密钥。请在.env文件或环境变量中设置E2B_API_KEY")

# 使用API密钥创建沙箱
sandbox = Sandbox.create(api_key=api_key)
file_operator = SandboxFileOperator(sandbox=sandbox)

context = ExecutionContext(
    working_directory="/tmp",
    fs_operator=file_operator
)
环境配置：在项目根目录创建.env文件： 
E2B_API_KEY=your_e2b_api_key_here
从E2B控制台获取API密钥。

​
独立工具用法

​
ReadTool

from ag_kit_py.tools.fs import create_read_tool, ExecutionContext

# 创建读取工具
read_tool = create_read_tool(context)

# 读取整个文件
result = await read_tool.invoke({
    "file_path": "src/config.json"
})

# 按行范围读取
partial_result = await read_tool.invoke({
    "file_path": "large-file.txt",
    "offset": 100,
    "end_line": 200
})

# 访问结果数据
if result.success:
    content = result.data['content']
    file_path = result.data['filePath']

​
WriteTool

from ag_kit_py.tools.fs import create_write_tool, ExecutionContext

# 创建写入工具
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

# 创建编辑工具
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  # 预期替换次数
})

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

# 创建多编辑工具
multiedit_tool = create_multiedit_tool(context)

# 定义单个文件的多个编辑操作
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

# 创建glob工具
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

# 创建grep工具
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

# 创建ls工具
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 = "
**关键要求：术语表 - 必须遵守：**
以下特定术语必须使用指定的翻译。在内容中遇到这些术语时，必须严格按照以下方式翻译：

- "API" 必须翻译为 "应用程序接口"
- "SDK" 必须翻译为 "AG-Kit"
- "CLI" 必须翻译为 "命令行界面"
- "GitHub" 必须翻译为 "GitHub"
- "Markdown" 必须翻译为 "Markdown"
- "README" 必须翻译为 "README"
- "Agent" 必须翻译为 "Agent"
- "human-in-loop" 必须翻译为 "Human-in-the-loop"
- "JavaScript" 必须翻译为 "JavaScript"
- "TypeScript" 必须翻译为 "TypeScript"
- "Node.js" 必须翻译为 "Node.js"
- "npm" 必须翻译为 "npm"
- "package.json" 必须翻译为 "package.json"
- "git" 必须翻译为 "Git"
- "repository" 必须翻译为 "仓库"
- "commit" 必须翻译为 "提交"
- "branch" 必须翻译为 "分支"
- "merge" 必须翻译为 "合并"
- "pull request" 必须翻译为 "拉取请求"
- "Build" 必须翻译为 "构建"
- "Integrate" 必须翻译为 "集成"
- "Doc" 必须翻译为 "文档"

**重要提示**：本术语表为强制性要求。必须严格遵循这些翻译并在整个文档中保持一致使用。不得对这些术语使用其他翻译。

**待翻译内容：**

    - command: str - 命令（"view", "create", "str_replace", "insert"）
    - path: str - 文件路径
    - old_str: Optional[str] - 待替换文本
    - new_str: Optional[str] - 替换文本
    - insert_line: Optional[int] - 插入行号
    - view_range: Optional[List[int]] - 查看行范围
    - file_text: Optional[str] - 创建文件内容
    """
    ...

​
高级用法

​
多后端工作流

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":
        # 创建内存文件系统
        fs = fake_filesystem.FakeFilesystem()
        return InMemoryFileOperator(filesystem=fs)
    else:
        # 使用本地文件系统
        return LocalFileOperator()

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

​
批量操作

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

# 创建读取工具
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']}")

​
错误处理

from ag_kit_py.tools.fs import create_read_tool, ExecutionContext

# 创建读取工具
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("文件不存在")
    elif error_type == "permission":
        print("权限不足")
    elif error_type == "security":
        print("安全违规")
    else:
        print(f"未知错误: {result.error}")

​
安全特性

​
路径验证

所有工具自动验证路径以防止目录遍历攻击： 
# ✅ 安全 - 工作区内的相对路径
await read(context, file_path="src/config.py")

# ❌ 被阻止 - 目录遍历
await read(context, file_path="../../../etc/passwd")

​
生产环境建议

  • 使用沙盒文件操作器处理不可信代码
  • 使用内存文件操作器进行测试
  • 仅限可信环境使用本地文件操作器

​
示例

​
开发工作流

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

# 创建工具
read_tool = create_read_tool(context)
edit_tool = create_edit_tool(context)
grep_tool = create_grep_tool(context)
glob_tool = create_glob_tool(context)

# 读取配置
config_result = await read_tool.invoke({"file_path": "config.json"})
config = json.loads(config_result.data['content'])

# 更新配置
await edit_tool.invoke({
    "file_path": "config.json",
    "old_string": '"debug": false',
    "new_string": '"debug": true',
    "expected_replacements": 1
})

# 查找所有Python文件
py_files = await glob_tool.invoke({
    "pattern": "**/*.py",
    "path": ".",
    "exclude": ["venv/**"]
})

# 搜索TODO标记
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']}")

​
文件重构工作流

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

# 创建工具
grep_tool = create_grep_tool(context)
edit_tool = create_edit_tool(context)

# 查找所有使用旧API的文件
result = await grep_tool.invoke({
    "pattern": "old_api_function",
    "path": ".",
    "include": "**/*.py"
})

# 对每个文件应用修改
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"已更新 {file_path}")

​
文件处理流水线

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

# 创建工具
glob_tool = create_glob_tool(context)
read_tool = create_read_tool(context)
write_tool = create_write_tool(context)

# 查找所有文本文件
files_result = await glob_tool.invoke({
    "pattern": "**/*.txt",
    "path": "input"
})

# 处理每个文件
for match in files_result.data['matches']:
    file_path = match['path']
    
    # 读取文件
    read_result = await read_tool.invoke({"file_path": file_path})
    if read_result.success:
        content = read_result.data['content']
        
        # 处理内容
        processed = content.upper()
        
        # 写入输出
        output_path = file_path.replace("input", "output")
        await write_tool.invoke({
            "file_path": output_path,
            "content": processed,
            "create_dirs": True
        })

​
性能优化

​
并行操作

import asyncio
from ag_