from typing import Any from duck_core.tools.base import ToolResult from duck_core.tools.paths import WorkspacePathError, candidate_path, 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)) approved = bool(args.get("_approved")) try: path = resolve_workspace_path(self.workspace, raw_path, allow_outside=approved) except WorkspacePathError as exc: path = candidate_path(self.workspace, raw_path) return ToolResult( ok=False, error=f"{exc}. Writing outside workspace requires approval.", metadata={ "path": str(path), "requires_approval": True, "risk_level": self.risk_level, "reason": "Path is outside workspace", }, ) if path.exists() and not overwrite and not approved: return ToolResult( ok=False, error="Refusing to overwrite existing file without overwrite=true or approval", metadata={ "path": str(path), "requires_approval": True, "risk_level": self.risk_level, "reason": "File exists and overwrite was not requested", }, ) 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, }, )