first

agent_runtime/write_tools.py

@@ -0,0 +1,43 @@
+import os
+import json
+from pathlib import Path
+from typing import Union
+from qwen_agent.tools.base import BaseTool, register_tool
+
+def _split_items(raw: str) -> list[str]:
+    return [item.strip() for item in raw.split(';') if item.strip()]
+
+def _resolve_write_roots() -> tuple[Path, ...]:
+    roots_value = os.getenv('WRITEABLE_FS_ROOTS', '')
+    return tuple(Path(os.path.expanduser(item)).resolve() for item in _split_items(roots_value))
+
+@register_tool('write_file', allow_overwrite=True)
+class WriteFileTool(BaseTool):
+    description = '文件写入工具。只要路径在白名单内，即可直接创建或覆盖文件。'
+    parameters = {
+        'type': 'object',
+        'properties': {
+            'path': {'type': 'string', 'description': '目标绝对路径'},
+            'content': {'type': 'string', 'description': '要写入的完整内容'}
+        },
+        'required': ['path', 'content'],
+    }
+
+    def call(self, params: Union[str, dict], **kwargs) -> str:
+        params = self._verify_json_format_args(params)
+        target = Path(os.path.expanduser(str(params['path']))).resolve()
+        content = str(params.get('content', ''))
+
+        # 核心防线：检查是否在白名单内
+        roots = _resolve_write_roots()
+        if not any(target.is_relative_to(root) for root in roots):
+            allowed = ", ".join(str(r) for r in roots)
+            return f"拒绝写入：路径不在白名单内。允许范围：{allowed}"
+
+        try:
+            target.parent.mkdir(parents=True, exist_ok=True)
+            with open(target, 'w', encoding='utf-8') as f:
+                f.write(content)
+            return f"✅ 成功：内容已保存至 {target}"
+        except Exception as e:
+            return f"写入失败：{str(e)}"