feat: 添加 fs_write 和 py_build 工具

agent/executor.py

@@ -124,9 +124,19 @@ def decide_next_action(task: Task, current_index: int, last_result: Optional[Dic
         "   适用意图: 确保仓库已可用/若不存在则克隆（避免重复克隆）\n"
         "   重要规则: 不要把 '.' 作为 dest 传入；如无特定目标目录，省略 dest 即可\n\n"
 
-        "四、通用执行工具（兜底，仅在专用工具不适用时使用）：\n\n"
+        "四、文件系统工具（execute 模式优先）：\n\n" 
 
-        "8. run_instruction - 执行任意 shell 命令\n"
+        "11. files_write - 写入文件内容（覆盖或追加）\n"
+        "   参数: {\"path\": \"文件路径\", \"content\": \"要写入的文本\", \"append\": false, \"create_dirs\": true}\n"
+        "   适用意图: 创建配置文件、修改代码、写入测试数据\n\n"
+
+        "12. files_delete - 删除文件或目录（递归）\n" 
+        "   参数: {\"path\": \"文件路径\", \"recursive\": false}\n"
+        "   适用意图: 清理临时文件、删除旧目录\n\n"
+
+        "五、通用执行工具（兜底...）：\n\n" #
+
+        "13. run_instruction - 执行任意 shell 命令\n"
         "   用于: git clone、pip install、运行脚本等所有命令行操作\n"
         "   适用意图: 所有非探测性的执行操作\n\n"
 

agent/workflow.py

@@ -22,6 +22,9 @@
     GIT_REPO_STATUS_TOOL,
     GIT_ENSURE_CLONED_TOOL,
     MD_READ_INFO_TOOL,
+    FILES_WRITE_TOOL,
+    FILES_DELETE_TOOL,
+    PYBUILD_CHECK_INSTALLED_TOOL,
 )
 from agent.observer import observe, observe_v2
 from agent.discover_react import run_discover_react
@@ -628,6 +631,9 @@ def create_task_graph():
         PYENV_SELECT_INSTALLER_TOOL,
         GIT_REPO_STATUS_TOOL,
         GIT_ENSURE_CLONED_TOOL,
+        FILES_WRITE_TOOL,
+        FILES_DELETE_TOOL,
+        PYBUILD_CHECK_INSTALLED_TOOL,
     ]))
     workflow.add_node("observe", observe_node)
 

tools/__init__.py

@@ -32,6 +32,15 @@
     MD_READ_INFO_TOOL,
 )
 
+from .fs_write import ( # <--- 新增导入
+    FILES_WRITE_TOOL,
+    FILES_DELETE_TOOL,
+)
+
+from .py_build import ( # <--- 新增导入
+    PYBUILD_CHECK_INSTALLED_TOOL,
+)
+
 __all__ = [
     "FILES_EXISTS_TOOL",
     "FILES_STAT_TOOL",
@@ -49,7 +58,10 @@
     "GIT_REPO_STATUS_TOOL",
     "GIT_ENSURE_CLONED_TOOL",
     "RUN_INSTRUCTION_TOOL",
-    "MD_READ_INFO_TOOL"
+    "MD_READ_INFO_TOOL",
+    "FILES_WRITE_TOOL",
+    "FILES_DELETE_TOOL",
+    "PYBUILD_CHECK_INSTALLED_TOOL",
 ]
 
 

tools/fs_write.py

@@ -0,0 +1,155 @@
+#write tool
+#输入：
+#参数名,类型,是否可选,描述,默认值
+#path,str,否,要写入的文件路径，相对于工作区根目录。,无
+#content,str,否,要写入文件的文本内容。,无
+#append,bool,是,如果为 True，则将内容追加到文件末尾；如果为 False，则覆盖原有内容。,False
+#create_dirs,bool,是,如果文件的父目录不存在，是否自动创建它们。,True
+#输出：
+#{
+#    "ok": true,
+#    "tool": "files_write",
+#    "data": {
+#        "path": "写入的绝对路径",
+#        "size_written": 写入的内容长度,
+#        "append": true | false
+#    }
+#}
+
+
+
+#delete tool
+#输入：
+#path	str	要删除的文件或目录的路径，相对于工作区根目录。	无
+#recursive	bool	如果为 True，则递归删除非空目录；如果为 False，则只能删除文件或空目录。
+#输出：
+#{
+#    "ok": true,
+#    "tool": "files_delete",
+#    "data": {
+#        "path": "实际删除的绝对路径",
+#        "deleted": true | false,
+#        "type": "file" | "dir" | "dir_recursive" | "unknown",
+#
+#     }
+#}
+
+
+from __future__ import annotations
+
+import os
+import json
+import shutil
+from pathlib import Path
+from typing import Any, Dict, List, Optional, Tuple
+
+from langchain_core.tools import tool
+
+from config import get_config
+from agent.debug import dispInfo, debug
+from tools.base import tool_response
+from tools.fs import _resolve_and_guard 
+
+
+@tool("files_write")
+@dispInfo("fs_write")
+def FILES_WRITE_TOOL(path: str, content: str, append: bool = False, create_dirs: bool = True) -> str:
+    """写入文件内容。如果文件不存在，则创建；如果存在，则覆盖或追加（append=True）。"""
+
+    ok, violation, p = _resolve_and_guard(path)
+    if not ok or p is None:
+        return tool_response(
+            tool="files_write",
+            ok=False,
+            data={"path": str(path)},
+            error=violation or "invalid_path"
+        )
+
+    try:
+        if create_dirs and not p.parent.exists():
+            p.parent.mkdir(parents=True, exist_ok=True)
+            try:
+                debug.note("created_parent_dirs", str(p.parent))
+            except Exception:
+                pass
+
+        mode = 'a' if append else 'w'
+        content_bytes = content.encode("utf-8", errors="replace")
+
+        with open(p, mode + 'b') as f:
+            f.write(content_bytes)
+
+        return tool_response(
+            tool="files_write",
+            ok=True,
+            data={"path": str(p), "size_written": len(content_bytes), "append": append}
+        )
+    except Exception as e:
+        return tool_response(
+            tool="files_write",
+            ok=False,
+            data={"path": str(p)},
+            error=f"{type(e).__name__}: {e}"
+        )
+
+
+@tool("files_delete")
+@dispInfo("fs_delete")
+def FILES_DELETE_TOOL(path: str, recursive: bool = False) -> str:
+
+
+    ok, violation, p = _resolve_and_guard(path)
+    if not ok or p is None:
+        return tool_response(
+            tool="files_delete",
+            ok=False,
+            data={"path": str(path)},
+            error=violation or "invalid_path"
+        )
+
+    if not p.exists():
+        return tool_response(
+            tool="files_delete",
+            ok=True,
+            data={"path": str(p), "deleted": False, "reason": "not_found"}
+        )
+
+    try:
+        if p.is_file() or p.is_symlink():
+            os.remove(p)
+            return tool_response(
+                tool="files_delete",
+                ok=True,
+                data={"path": str(p), "deleted": True, "type": "file"}
+            )
+
+        if p.is_dir():
+            if recursive:
+                shutil.rmtree(p)
+                return tool_response(
+                    tool="files_delete",
+                    ok=True,
+                    data={"path": str(p), "deleted": True, "type": "dir_recursive"}
+                )
+            else:
+                os.rmdir(p)
+                return tool_response(
+                    tool="files_delete",
+                    ok=True,
+                    data={"path": str(p), "deleted": True, "type": "dir"}
+                )
+
+        return tool_response(
+            tool="files_delete",
+            ok=False,
+            data={"path": str(p)},
+            error="unsupported_file_type"
+        )
+
+    except Exception as e:
+        return tool_response(
+            tool="files_delete",
+            ok=False,
+            data={"path": str(p)},
+            error=f"delete_failed: {type(e).__name__}: {e}"
+        )

tools/py_build.py

@@ -0,0 +1,66 @@
+#输入：package_name	str	要检查的 Python 包的名称。
+#输出：
+#installed	bool	包是否已安装 (True 或 False)。
+#package_name	str	被检查的包名称。
+#version	str	如果已安装，返回包的版本号。
+#location	str	如果已安装，返回包的安装路径。
+#summary	str	如果已安装，返回包的简短摘要。
+#pip_error	str	如果未安装，返回pip show 命令返回的错误信息。
+
+
+from __future__ import annotations
+
+import json
+import subprocess
+from typing import Any, Dict, List, Optional, Tuple
+
+from langchain_core.tools import tool
+
+from config import get_config
+from agent.debug import dispInfo, debug
+from tools.base import tool_response
+from tools.pyenv import _run_cmd # 沿用 pyenv.py 的命令执行
+
+
+@tool("pybuild_check_installed")
+@dispInfo("pybuild_check")
+def PYBUILD_CHECK_INSTALLED_TOOL(package_name: str) -> str:
+
+    if not package_name:
+        return tool_response(
+            tool="pybuild_check_installed",
+            ok=False,
+            data={"package_name": ""},
+            error="package_name_required"
+        )
+
+    code, stdout, stderr = _run_cmd(["pip", "show", package_name])
+
+    if code == 0:
+        info = {}
+        for line in stdout.splitlines():
+            if ":" in line:
+                key, value = line.split(":", 1)
+                info[key.strip().lower()] = value.strip()
+
+        return tool_response(
+            tool="pybuild_check_installed",
+            ok=True,
+            data={
+                "installed": True,
+                "package_name": package_name,
+                "version": info.get("version"),
+                "location": info.get("location"),
+                "summary": info.get("summary")
+            }
+        )
+    else:
+        return tool_response(
+            tool="pybuild_check_installed",
+            ok=True,
+            data={
+                "installed": False,
+                "package_name": package_name,
+                "pip_error": stderr
+            }
+        )