File system editor for apply_patch operations.
currently supports local filesystem through allowed_paths patterns.
Initialize workspace editor.
| PARAMETER | DESCRIPTION |
workspace_dir | Root directory for file operations (local filesystem path). TYPE: str | Path |
allowed_paths | List of allowed path patterns (for security). Supports glob-style patterns with ** for recursive matching. currently works for local filesystem. Examples: - [""] - Allow all paths (default) - ["src/"] - Allow all files in src/ and subdirectories - [".py"] - Allow Python files in root directory - ["src/", "tests/*"] - Allow paths in multiple directories TYPE: list[str] | None DEFAULT: None |
Source code in autogen/tools/experimental/apply_patch/apply_patch_tool.py
| def __init__(
self,
workspace_dir: str | Path,
allowed_paths: list[str] | None = None,
):
"""Initialize workspace editor.
Args:
workspace_dir: Root directory for file operations (local filesystem path).
allowed_paths: List of allowed path patterns (for security).
Supports glob-style patterns with ** for recursive matching.
currently works for local filesystem.
Examples:
- ["**"] - Allow all paths (default)
- ["src/**"] - Allow all files in src/ and subdirectories
- ["*.py"] - Allow Python files in root directory
- ["src/**", "tests/**"] - Allow paths in multiple directories
"""
workspace_dir = workspace_dir if workspace_dir is not None else os.getcwd()
self.workspace_dir = Path(workspace_dir).resolve()
# Use "**" to match all files and directories recursively (including root)
self.allowed_paths = allowed_paths if allowed_paths is not None else ["**"]
|
workspace_dir = resolve()
allowed_paths = allowed_paths if allowed_paths is not None else ['**']
Create a new file.
Source code in autogen/tools/experimental/apply_patch/apply_patch_tool.py
| def create_file(self, operation: dict[str, Any]) -> dict[str, Any]:
"""Create a new file."""
try:
path = operation.get("path")
if not path:
return {"status": "failed", "output": "Missing 'path' in operation"}
diff = operation.get("diff", "")
full_path = self._validate_path(path)
# Ensure parent directory exists
full_path.parent.mkdir(parents=True, exist_ok=True)
# Apply diff to get file content
content = apply_diff("", diff, create=True)
# Write file
full_path.write_text(content, encoding="utf-8")
return {"status": "completed", "output": f"Created {path}"}
except Exception as e:
return {"status": "failed", "output": f"Error creating {path}: {str(e)}"}
|
Create a new file.
Source code in autogen/tools/experimental/apply_patch/apply_patch_tool.py
| async def a_create_file(self, operation: dict[str, Any]) -> dict[str, Any]:
"""Create a new file."""
if aiofiles is None:
raise RuntimeError("aiofiles is required for async file operations. Install it with: pip install aiofiles")
try:
path = operation.get("path")
if not path:
return {"status": "failed", "output": "Missing 'path' in operation"}
diff = operation.get("diff", "")
full_path = self._validate_path(path)
# Ensure parent directory exists (use asyncio.to_thread for blocking mkdir)
await asyncio.to_thread(full_path.parent.mkdir, parents=True, exist_ok=True)
# Apply diff to get file content
content = apply_diff("", diff, create=True)
# Write file using async I/O
async with aiofiles.open(str(full_path), "w", encoding="utf-8") as f:
await f.write(content)
return {"status": "completed", "output": f"Created {path}"}
except Exception as e:
return {"status": "failed", "output": f"Error creating {path}: {str(e)}"}
|
Update an existing file.
Source code in autogen/tools/experimental/apply_patch/apply_patch_tool.py
| def update_file(self, operation: dict[str, Any]) -> dict[str, Any]:
"""Update an existing file."""
try:
path = operation.get("path")
if not path:
return {"status": "failed", "output": "Missing 'path' in operation"}
diff = operation.get("diff", "")
full_path = self._validate_path(path)
if not full_path.exists():
return {"status": "failed", "output": f"Error: File not found at path '{path}'"}
# Read current content
current_content = full_path.read_text(encoding="utf-8")
# Apply diff
new_content = apply_diff(current_content, diff)
# Write updated content
full_path.write_text(new_content, encoding="utf-8")
return {"status": "completed", "output": f"Updated {path}"}
except Exception as e:
return {"status": "failed", "output": f"Error updating {path}: {str(e)}"}
|
Update an existing file.
Source code in autogen/tools/experimental/apply_patch/apply_patch_tool.py
| async def a_update_file(self, operation: dict[str, Any]) -> dict[str, Any]:
"""Update an existing file."""
if aiofiles is None:
raise RuntimeError("aiofiles is required for async file operations. Install it with: pip install aiofiles")
try:
path = operation.get("path")
if not path:
return {"status": "failed", "output": "Missing 'path' in operation"}
diff = operation.get("diff", "")
full_path = self._validate_path(path)
# Check if file exists using asyncio.to_thread
exists = await asyncio.to_thread(full_path.exists)
if not exists:
return {"status": "failed", "output": f"Error: File not found at path '{path}'"}
# Read current content using async I/O
async with aiofiles.open(str(full_path), "r", encoding="utf-8") as f:
current_content = await f.read()
# Apply diff
new_content = apply_diff(current_content, diff)
# Write updated content using async I/O
async with aiofiles.open(str(full_path), "w", encoding="utf-8") as f:
await f.write(new_content)
return {"status": "completed", "output": f"Updated {path}"}
except Exception as e:
return {"status": "failed", "output": f"Error updating {path}: {str(e)}"}
|
Delete a file.
Source code in autogen/tools/experimental/apply_patch/apply_patch_tool.py
| def delete_file(self, operation: dict[str, Any]) -> dict[str, Any]:
"""Delete a file."""
try:
path = operation.get("path")
if not path:
return {"status": "failed", "output": "Missing 'path' in operation"}
full_path = self._validate_path(path)
if not full_path.exists():
return {"status": "failed", "output": f"Error: File not found at path '{path}'"}
full_path.unlink()
return {"status": "completed", "output": f"Deleted {path}"}
except Exception as e:
return {"status": "failed", "output": f"Error deleting {path}: {str(e)}"}
|
Delete a file.
Source code in autogen/tools/experimental/apply_patch/apply_patch_tool.py
| async def a_delete_file(self, operation: dict[str, Any]) -> dict[str, Any]:
"""Delete a file."""
if aiofiles is None:
raise RuntimeError("aiofiles is required for async file operations. Install it with: pip install aiofiles")
try:
path = operation.get("path")
if not path:
return {"status": "failed", "output": "Missing 'path' in operation"}
full_path = self._validate_path(path)
# Check if file exists using asyncio.to_thread
exists = await asyncio.to_thread(full_path.exists)
if not exists:
return {"status": "failed", "output": f"Error: File not found at path '{path}'"}
# Delete file using async I/O
await aiofiles.os.remove(str(full_path))
return {"status": "completed", "output": f"Deleted {path}"}
except Exception as e:
return {"status": "failed", "output": f"Error deleting {path}: {str(e)}"}
|