From 9a7e7ec6b93f4d32c233e55a9675201383779fed Mon Sep 17 00:00:00 2001 From: mirivlad Date: Tue, 7 Apr 2026 17:25:53 +0800 Subject: [PATCH] Expand coding toolset --- README.md | 2 +- serv/tools.py | 203 ++++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 204 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index 44d54aa..f96eeca 100644 --- a/README.md +++ b/README.md @@ -28,7 +28,7 @@ Qwen OAuth + OpenAI-compatible endpoint - хранение токенов в `~/.qwen/oauth_creds.json` - HTTP API сервера - агентный цикл с tool calling -- инструменты: `list_files`, `glob_search`, `grep_text`, `stat_path`, `read_file`, `apply_unified_diff`, `replace_in_file`, `write_file`, `make_directory`, `exec_command` +- инструменты: `list_files`, `glob_search`, `grep_text`, `stat_path`, `read_file`, `append_file`, `apply_unified_diff`, `replace_in_file`, `write_file`, `make_directory`, `delete_path`, `move_path`, `copy_path`, `git_status`, `git_diff`, `exec_command` - Telegram polling без внешних библиотек - JSON-хранилище сессий - API списка и просмотра сессий diff --git a/serv/tools.py b/serv/tools.py index 1eb2f48..8f3c3c2 100644 --- a/serv/tools.py +++ b/serv/tools.py @@ -1,6 +1,7 @@ from __future__ import annotations import fnmatch +import shutil import json import os import re @@ -26,10 +27,16 @@ class ToolRegistry: "grep_text": self.grep_text, "stat_path": self.stat_path, "read_file": self.read_file, + "append_file": self.append_file, "apply_unified_diff": self.apply_unified_diff, "replace_in_file": self.replace_in_file, "write_file": self.write_file, "make_directory": self.make_directory, + "delete_path": self.delete_path, + "move_path": self.move_path, + "copy_path": self.copy_path, + "git_status": self.git_status, + "git_diff": self.git_diff, "exec_command": self.exec_command, } @@ -109,6 +116,21 @@ class ToolRegistry: }, }, }, + { + "type": "function", + "function": { + "name": "append_file", + "description": "Append UTF-8 text to a file inside the workspace.", + "parameters": { + "type": "object", + "properties": { + "path": {"type": "string"}, + "content": {"type": "string"}, + }, + "required": ["path", "content"], + }, + }, + }, { "type": "function", "function": { @@ -170,6 +192,79 @@ class ToolRegistry: }, }, }, + { + "type": "function", + "function": { + "name": "delete_path", + "description": "Delete a file or directory inside the workspace.", + "parameters": { + "type": "object", + "properties": { + "path": {"type": "string"}, + "recursive": {"type": "boolean"}, + }, + "required": ["path"], + }, + }, + }, + { + "type": "function", + "function": { + "name": "move_path", + "description": "Move or rename a file or directory inside the workspace.", + "parameters": { + "type": "object", + "properties": { + "source_path": {"type": "string"}, + "destination_path": {"type": "string"}, + }, + "required": ["source_path", "destination_path"], + }, + }, + }, + { + "type": "function", + "function": { + "name": "copy_path", + "description": "Copy a file or directory inside the workspace.", + "parameters": { + "type": "object", + "properties": { + "source_path": {"type": "string"}, + "destination_path": {"type": "string"}, + "recursive": {"type": "boolean"}, + }, + "required": ["source_path", "destination_path"], + }, + }, + }, + { + "type": "function", + "function": { + "name": "git_status", + "description": "Return a compact git status for the workspace or a subdirectory.", + "parameters": { + "type": "object", + "properties": { + "path": {"type": "string"}, + }, + }, + }, + }, + { + "type": "function", + "function": { + "name": "git_diff", + "description": "Return a unified git diff for the workspace or a specific path.", + "parameters": { + "type": "object", + "properties": { + "path": {"type": "string"}, + "cached": {"type": "boolean"}, + }, + }, + }, + }, { "type": "function", "function": { @@ -219,10 +314,14 @@ class ToolRegistry: def requires_approval(self, tool_name: str) -> bool: policy = self.config.tool_policy write_tools = { + "append_file", "apply_unified_diff", "replace_in_file", "write_file", "make_directory", + "delete_path", + "move_path", + "copy_path", } shell_tools = {"exec_command"} if policy == "ask-all": @@ -349,6 +448,16 @@ class ToolRegistry: "bytes_written": len(arguments["content"].encode("utf-8")), } + def append_file(self, arguments: dict[str, Any]) -> dict[str, Any]: + target = self._resolve(arguments["path"]) + target.parent.mkdir(parents=True, exist_ok=True) + with target.open("a", encoding="utf-8") as handle: + handle.write(arguments["content"]) + return { + "path": target.relative_to(self.workspace_root).as_posix(), + "bytes_appended": len(arguments["content"].encode("utf-8")), + } + def replace_in_file(self, arguments: dict[str, Any]) -> dict[str, Any]: target = self._resolve(arguments["path"]) if not target.exists(): @@ -424,6 +533,100 @@ class ToolRegistry: target.mkdir(parents=True, exist_ok=True) return {"path": target.relative_to(self.workspace_root).as_posix(), "created": True} + def delete_path(self, arguments: dict[str, Any]) -> dict[str, Any]: + target = self._resolve(arguments["path"]) + recursive = bool(arguments.get("recursive", False)) + if not target.exists(): + raise ToolError("Path does not exist") + rel_path = target.relative_to(self.workspace_root).as_posix() + if target.is_dir(): + if not recursive: + raise ToolError("Directory deletion requires recursive=true") + shutil.rmtree(target) + else: + target.unlink() + return {"path": rel_path, "deleted": True} + + def move_path(self, arguments: dict[str, Any]) -> dict[str, Any]: + source = self._resolve(arguments["source_path"]) + destination = self._resolve(arguments["destination_path"]) + if not source.exists(): + raise ToolError("Source path does not exist") + destination.parent.mkdir(parents=True, exist_ok=True) + shutil.move(str(source), str(destination)) + return { + "source_path": source.relative_to(self.workspace_root).as_posix(), + "destination_path": destination.relative_to(self.workspace_root).as_posix(), + "moved": True, + } + + def copy_path(self, arguments: dict[str, Any]) -> dict[str, Any]: + source = self._resolve(arguments["source_path"]) + destination = self._resolve(arguments["destination_path"]) + recursive = bool(arguments.get("recursive", False)) + if not source.exists(): + raise ToolError("Source path does not exist") + destination.parent.mkdir(parents=True, exist_ok=True) + if source.is_dir(): + if not recursive: + raise ToolError("Directory copy requires recursive=true") + shutil.copytree(source, destination, dirs_exist_ok=True) + else: + shutil.copy2(source, destination) + return { + "source_path": source.relative_to(self.workspace_root).as_posix(), + "destination_path": destination.relative_to(self.workspace_root).as_posix(), + "copied": True, + } + + def git_status(self, arguments: dict[str, Any]) -> dict[str, Any]: + target = self._resolve(arguments.get("path", ".")) + completed = subprocess.run( + ["git", "status", "--short"], + cwd=str(target if target.is_dir() else target.parent), + capture_output=True, + text=True, + timeout=60, + ) + if completed.returncode != 0: + raise ToolError(completed.stderr.strip() or completed.stdout.strip() or "git status failed") + lines = completed.stdout.splitlines() + return {"status": lines[:200], "truncated": len(lines) > 200} + + def git_diff(self, arguments: dict[str, Any]) -> dict[str, Any]: + target = arguments.get("path") + cached = bool(arguments.get("cached", False)) + cmd = ["git", "diff"] + if cached: + cmd.append("--cached") + if target: + resolved_target = self._resolve(target) + cwd = str(resolved_target if resolved_target.is_dir() else resolved_target.parent) + if resolved_target.exists(): + cmd.extend(["--", str(resolved_target)]) + else: + cmd.extend(["--", target]) + else: + cwd = str(self.workspace_root) + completed = subprocess.run( + cmd, + cwd=cwd, + capture_output=True, + text=True, + timeout=60, + ) + if completed.returncode != 0: + raise ToolError(completed.stderr.strip() or completed.stdout.strip() or "git diff failed") + diff_text = completed.stdout + truncated = False + if len(diff_text.encode("utf-8")) > self.config.max_file_read_bytes: + diff_text = diff_text.encode("utf-8")[: self.config.max_file_read_bytes].decode( + "utf-8", + errors="ignore", + ) + truncated = True + return {"diff": diff_text, "truncated": truncated, "cached": cached} + def exec_command(self, arguments: dict[str, Any]) -> dict[str, Any]: cwd = self._resolve(arguments.get("cwd", ".")) command = arguments["command"]