41 lines
1.3 KiB
Python
41 lines
1.3 KiB
Python
from typing import Any
|
|
|
|
from duck_core.tools.base import ToolResult
|
|
from duck_core.tools.paths import WorkspacePathError, resolve_workspace_path
|
|
|
|
|
|
class FileWriteTool:
|
|
name = "file_write"
|
|
risk_level = "medium"
|
|
|
|
def __init__(self, workspace: str):
|
|
self.workspace = workspace
|
|
|
|
async def run(self, args: dict[str, Any]) -> ToolResult:
|
|
raw_path = str(args.get("path", ""))
|
|
content = str(args.get("content", ""))
|
|
overwrite = bool(args.get("overwrite", False))
|
|
try:
|
|
path = resolve_workspace_path(self.workspace, raw_path)
|
|
except WorkspacePathError as exc:
|
|
return ToolResult(ok=False, error=str(exc))
|
|
if path.exists() and not overwrite:
|
|
return ToolResult(
|
|
ok=False,
|
|
error="Refusing to overwrite existing file without overwrite=true or approval",
|
|
metadata={"path": str(path)},
|
|
)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
existed = path.exists()
|
|
path.write_text(content)
|
|
return ToolResult(
|
|
ok=True,
|
|
output=f"Wrote {raw_path}",
|
|
metadata={
|
|
"path": str(path),
|
|
"bytes_written": len(content.encode()),
|
|
"created": not existed,
|
|
"updated": existed,
|
|
},
|
|
)
|