Anton
052591281d
- Complete agent system with planning and execution - Bridge API with FastAPI server - WhatsApp-style mobile dashboard - Enhanced web search with query improvement - Security guards and policy checks - Comprehensive documentation
70 lines
2.4 KiB
Python
70 lines
2.4 KiB
Python
"""File operation tools — write_file, read_file, finish."""
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
from core.state import PlanStep, StepResult, StepStatus
|
|
|
|
# All file ops are confined to agent/workspace/
|
|
# parents: [0]=tools, [1]=runtime, [2]=agent → agent/workspace
|
|
WORKSPACE_ROOT: Path = Path(__file__).resolve().parents[2] / "workspace"
|
|
|
|
|
|
def write_file(step: PlanStep) -> StepResult:
|
|
"""Write content to a file inside agent/workspace/. Path traversal is blocked."""
|
|
rel_path: str = step.payload.get("path", "").strip()
|
|
content: str = step.payload.get("content", "")
|
|
|
|
if not rel_path:
|
|
return StepResult(step_id=step.step_id, status=StepStatus.FAILED, error="payload missing 'path'")
|
|
|
|
target = (WORKSPACE_ROOT / rel_path).resolve()
|
|
|
|
if not str(target).startswith(str(WORKSPACE_ROOT)):
|
|
return StepResult(
|
|
step_id=step.step_id,
|
|
status=StepStatus.FAILED,
|
|
error=f"Path traversal blocked: '{rel_path}' resolves outside workspace",
|
|
)
|
|
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
target.write_text(content, encoding="utf-8")
|
|
|
|
return StepResult(
|
|
step_id=step.step_id,
|
|
status=StepStatus.DONE,
|
|
output_text=f"Written {len(content)} chars to {rel_path}",
|
|
)
|
|
|
|
|
|
def read_file(step: PlanStep) -> StepResult:
|
|
"""Read a file from inside agent/workspace/. Path traversal is blocked."""
|
|
rel_path: str = step.payload.get("path", "").strip()
|
|
|
|
if not rel_path:
|
|
return StepResult(step_id=step.step_id, status=StepStatus.FAILED, error="payload missing 'path'")
|
|
|
|
target = (WORKSPACE_ROOT / rel_path).resolve()
|
|
|
|
if not str(target).startswith(str(WORKSPACE_ROOT)):
|
|
return StepResult(
|
|
step_id=step.step_id,
|
|
status=StepStatus.FAILED,
|
|
error=f"Path traversal blocked: '{rel_path}' resolves outside workspace",
|
|
)
|
|
|
|
if not target.exists():
|
|
return StepResult(
|
|
step_id=step.step_id,
|
|
status=StepStatus.FAILED,
|
|
error=f"File not found: {rel_path}",
|
|
)
|
|
|
|
content = target.read_text(encoding="utf-8")
|
|
return StepResult(step_id=step.step_id, status=StepStatus.DONE, output_text=content)
|
|
|
|
|
|
def finish_step(step: PlanStep) -> StepResult:
|
|
"""Terminal step — signals successful plan completion."""
|
|
return StepResult(step_id=step.step_id, status=StepStatus.DONE, output_text="Agent task complete.")
|