feat(agents): Phase 6 tool system refactoring
Phase 6.1: ToolRegistry infrastructure - Add ToolManifest with ToolCategory, PermissionClass, SideEffectScope - Add ToolRegistry singleton with register/get/unregister/list/search - Add BaseTool abstract class with ReadTool/WriteTool/DBWriteTool/ExternalTool/NetworkTool subclasses - Add migration layer for backward compatibility Phase 6.2: Hook interception system - Add HookType (PRE_TOOL_USE, POST_TOOL_USE, TOOL_ERROR, TOOL_SKIP) - Add HookManager with singleton for hook registration - Add HookExecutor for pre/post/error hook execution Phase 6.3: Streaming execution - Add StreamingToolExecutor with batch execution support Phase 6.4: New builtin tools - Add file_tools: GlobTool, GrepTool, ReadFileTool, WriteFileTool - Add system_tools: BashTool, PowerShellTool - Add dev_tools: LSPTools, GitTool - Add collaboration_tools: TeamAgentTool, TaskBroadcastTool Tests: 29 passed
This commit is contained in:
255
backend/app/agents/tools/builtins/file_tools.py
Normal file
255
backend/app/agents/tools/builtins/file_tools.py
Normal file
@@ -0,0 +1,255 @@
|
||||
"""文件操作工具 - Phase 6.4"""
|
||||
|
||||
import os
|
||||
from typing import Any
|
||||
|
||||
from app.agents.tools.base import ExternalTool, ReadTool, WriteTool
|
||||
from app.agents.tools.manifest import (
|
||||
PermissionClass,
|
||||
SideEffectScope,
|
||||
ToolCategory,
|
||||
)
|
||||
|
||||
|
||||
class GlobTool(ReadTool):
|
||||
"""文件路径匹配工具
|
||||
|
||||
使用 glob 模式查找文件。
|
||||
"""
|
||||
|
||||
def __init__(self, root_dir: str = "."):
|
||||
super().__init__(
|
||||
name="glob",
|
||||
description="使用 glob 模式查找文件路径",
|
||||
permission_class=PermissionClass.READ,
|
||||
side_effect_scope=SideEffectScope.NONE,
|
||||
tags=["file", "search", "glob"],
|
||||
)
|
||||
self.root_dir = root_dir
|
||||
|
||||
def get_parameters(self) -> dict[str, Any]:
|
||||
return {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"pattern": {
|
||||
"type": "string",
|
||||
"description": "Glob 模式，如 **/*.py",
|
||||
},
|
||||
"root_dir": {
|
||||
"type": "string",
|
||||
"description": "搜索根目录（可选）",
|
||||
},
|
||||
},
|
||||
"required": ["pattern"],
|
||||
}
|
||||
|
||||
def get_return_schema(self) -> dict[str, Any]:
|
||||
return {
|
||||
"type": "array",
|
||||
"items": {"type": "string"},
|
||||
}
|
||||
|
||||
async def execute(self, pattern: str, root_dir: str | None = None) -> list[str]:
|
||||
import glob as glob_module
|
||||
|
||||
root = root_dir or self.root_dir
|
||||
return glob_module.glob(pattern, root_dir=root, recursive=True)
|
||||
|
||||
|
||||
class GrepTool(ReadTool):
|
||||
"""文件内容搜索工具
|
||||
|
||||
在文件中搜索匹配的行。
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
super().__init__(
|
||||
name="grep",
|
||||
description="在文件中搜索匹配的文本行",
|
||||
permission_class=PermissionClass.READ,
|
||||
side_effect_scope=SideEffectScope.NONE,
|
||||
tags=["file", "search", "text"],
|
||||
)
|
||||
|
||||
def get_parameters(self) -> dict[str, Any]:
|
||||
return {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"pattern": {
|
||||
"type": "string",
|
||||
"description": "正则表达式模式",
|
||||
},
|
||||
"paths": {
|
||||
"type": "array",
|
||||
"items": {"type": "string"},
|
||||
"description": "要搜索的文件路径列表",
|
||||
},
|
||||
"case_sensitive": {
|
||||
"type": "boolean",
|
||||
"description": "是否区分大小写",
|
||||
},
|
||||
},
|
||||
"required": ["pattern", "paths"],
|
||||
}
|
||||
|
||||
def get_return_schema(self) -> dict[str, Any]:
|
||||
return {
|
||||
"type": "array",
|
||||
"items": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"file": {"type": "string"},
|
||||
"line": {"type": "integer"},
|
||||
"content": {"type": "string"},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
async def execute(
|
||||
self, pattern: str, paths: list[str], case_sensitive: bool = True
|
||||
) -> list[dict[str, Any]]:
|
||||
import re
|
||||
|
||||
flags = 0 if case_sensitive else re.IGNORECASE
|
||||
regex = re.compile(pattern, flags)
|
||||
results = []
|
||||
|
||||
for path in paths:
|
||||
if not os.path.isfile(path):
|
||||
continue
|
||||
|
||||
try:
|
||||
with open(path, "r", encoding="utf-8") as f:
|
||||
for line_num, line in enumerate(f, 1):
|
||||
if regex.search(line):
|
||||
results.append(
|
||||
{
|
||||
"file": path,
|
||||
"line": line_num,
|
||||
"content": line.rstrip(),
|
||||
}
|
||||
)
|
||||
except (UnicodeDecodeError, PermissionError):
|
||||
continue
|
||||
|
||||
return results
|
||||
|
||||
|
||||
class ReadFileTool(ReadTool):
|
||||
"""文件读取工具"""
|
||||
|
||||
def __init__(self):
|
||||
super().__init__(
|
||||
name="read_file",
|
||||
description="读取文件内容",
|
||||
permission_class=PermissionClass.READ,
|
||||
side_effect_scope=SideEffectScope.NONE,
|
||||
tags=["file", "read"],
|
||||
)
|
||||
|
||||
def get_parameters(self) -> dict[str, Any]:
|
||||
return {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"path": {
|
||||
"type": "string",
|
||||
"description": "文件路径",
|
||||
},
|
||||
"limit": {
|
||||
"type": "integer",
|
||||
"description": "最大行数",
|
||||
},
|
||||
"offset": {
|
||||
"type": "integer",
|
||||
"description": "起始行号",
|
||||
},
|
||||
},
|
||||
"required": ["path"],
|
||||
}
|
||||
|
||||
def get_return_schema(self) -> dict[str, Any]:
|
||||
return {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"content": {"type": "string"},
|
||||
"lines": {"type": "integer"},
|
||||
},
|
||||
}
|
||||
|
||||
async def execute(self, path: str, limit: int | None = None, offset: int = 0) -> dict[str, Any]:
|
||||
if not os.path.isfile(path):
|
||||
raise FileNotFoundError(f"File not found: {path}")
|
||||
|
||||
with open(path, "r", encoding="utf-8") as f:
|
||||
lines = f.readlines()
|
||||
|
||||
total_lines = len(lines)
|
||||
start = max(0, offset)
|
||||
end = len(lines) if limit is None else min(start + limit, len(lines))
|
||||
|
||||
content = "".join(lines[start:end])
|
||||
|
||||
return {
|
||||
"content": content,
|
||||
"lines": total_lines,
|
||||
"truncated": limit is not None and end < len(lines),
|
||||
}
|
||||
|
||||
|
||||
class WriteFileTool(WriteTool):
|
||||
"""文件写入工具"""
|
||||
|
||||
def __init__(self):
|
||||
super().__init__(
|
||||
name="write_file",
|
||||
description="写入文件内容",
|
||||
permission_class=PermissionClass.WRITE,
|
||||
side_effect_scope=SideEffectScope.LOCAL_STATE,
|
||||
requires_confirmation=True,
|
||||
tags=["file", "write"],
|
||||
)
|
||||
|
||||
def get_parameters(self) -> dict[str, Any]:
|
||||
return {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"path": {
|
||||
"type": "string",
|
||||
"description": "文件路径",
|
||||
},
|
||||
"content": {
|
||||
"type": "string",
|
||||
"description": "文件内容",
|
||||
},
|
||||
"append": {
|
||||
"type": "boolean",
|
||||
"description": "是否追加模式",
|
||||
},
|
||||
},
|
||||
"required": ["path", "content"],
|
||||
}
|
||||
|
||||
def get_return_schema(self) -> dict[str, Any]:
|
||||
return {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"success": {"type": "boolean"},
|
||||
"bytes_written": {"type": "integer"},
|
||||
},
|
||||
}
|
||||
|
||||
async def execute(self, path: str, content: str, append: bool = False) -> dict[str, Any]:
|
||||
mode = "a" if append else "w"
|
||||
|
||||
# 确保目录存在
|
||||
directory = os.path.dirname(path)
|
||||
if directory and not os.path.exists(directory):
|
||||
os.makedirs(directory, exist_ok=True)
|
||||
|
||||
with open(path, mode, encoding="utf-8") as f:
|
||||
bytes_written = f.write(content)
|
||||
|
||||
return {
|
||||
"success": True,
|
||||
"bytes_written": bytes_written,
|
||||
}
|
||||
Reference in New Issue
Block a user