Spaces:
Running
Running
File size: 5,369 Bytes
9d0fd45 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 | """read_text tool — plain-text / line-range reader for allowed local dirs or E2B sandbox.
This is the cheap host-local reader (formerly named ``read_file``): it reads a
file's text directly off the host (``plugins/skills/``, ``data/``,
``/tmp/agent-outputs/`` + an explicit workspace root), with an E2B sandbox
fallback for absolute paths, applies an optional 1-indexed line range and
truncates at 15 KB. No subprocess, no structured parsing — for office/PDF/xlsx
documents use the ``read_file`` tool (the sandbox-backed structured reader).
The path-authorization gate lives in :mod:`plugins.tools._path_auth`; this
module only does the reading. :func:`read_text_file` is the importable async
util; :data:`read_text` is the same function exposed as a registered tool.
References:
- DeerFlow: sandbox/tools.py read_file_tool()
"""
from __future__ import annotations
import logging
import os
from pathlib import Path
from frontier_agent.core.tool import tool
from plugins.tools._path_auth import (
_ALLOWED_ABSOLUTE_PREFIXES,
_authorized_local_path,
)
from plugins.tools._paths import virtual_to_physical
logger = logging.getLogger(__name__)
_MAX_CONTENT_BYTES = 15_000
# Document files written by write_file always land here (host-local); see
# the matching prefix in :mod:`plugins.tools._path_auth`.
_LOCAL_OUTPUT_DIR = "/tmp/agent-outputs"
async def read_text_file(path: str, start_line: int = 0, end_line: int = 0) -> str:
"""Read a file's plain text content from allowed directories or E2B sandbox.
Local mode: reads from plugins/skills/, data/, /tmp/agent-outputs/.
Sandbox mode: reads any file from the E2B sandbox.
For skills, use paths like: plugins/skills/deep-research/SKILL.md
For office/PDF/spreadsheet documents, use `read_file` instead (structured).
Args:
path: File path (relative for local, absolute for sandbox).
start_line: Optional 1-indexed start line (0 = from beginning).
end_line: Optional 1-indexed end line (0 = to end of file).
Returns:
File content as text, or error message.
"""
if not path or not path.strip():
return "Error: file path is required."
path = virtual_to_physical(path.strip())
def _finish(content: str) -> str:
"""Apply the line range, then cap the result at 15 KB."""
if start_line > 0 or end_line > 0:
lines = content.splitlines(keepends=True)
s = max(0, start_line - 1) if start_line > 0 else 0
e = end_line if end_line > 0 else len(lines)
sliced = "".join(lines[s:e])
if s > 0 or e < len(lines):
content = f"[lines {s + 1}-{min(e, len(lines))} of {len(lines)}]\n" + sliced
else:
content = sliced
if len(content) > _MAX_CONTENT_BYTES:
content = content[:_MAX_CONTENT_BYTES] + "\n\n... (truncated at 15KB)"
return content
# ── Try host /tmp/agent-outputs/ first ──────────────────────
# Check here BEFORE the sandbox to find persistent write_file outputs.
if os.path.isabs(path) and not path.startswith(_LOCAL_OUTPUT_DIR):
alt_resolved = Path(os.path.join(_LOCAL_OUTPUT_DIR, os.path.basename(path))).resolve()
if alt_resolved.is_file():
try:
return _finish(alt_resolved.read_text(encoding="utf-8"))
except Exception:
pass # Fall through to other strategies
# ── Try E2B sandbox for absolute paths ────────────────────────────
# Skip sandbox for paths that are known to live on the host (e.g. skills).
_is_host_path = any(
os.path.normpath(path).startswith(os.path.normpath(p))
for p in _ALLOWED_ABSOLUTE_PREFIXES
)
if os.path.isabs(path) and not _is_host_path:
try:
from plugins.tools._sandbox import (
aget_sandbox,
arun_sandbox_cmd,
sandbox_available,
shell_quote,
)
if sandbox_available():
sandbox = await aget_sandbox()
result = await arun_sandbox_cmd(
sandbox,
f"cat -- {shell_quote(path)}",
timeout=15,
)
if result.exit_code != 0:
raise FileNotFoundError(result.stderr)
return _finish(result.stdout)
except Exception as e:
logger.debug("Sandbox read failed for '%s': %s, trying local", path, e)
# ── Local file read with security checks ──────────────────────────
resolved, reason = _authorized_local_path(path)
if resolved is None:
return f"Access denied: {reason}"
if not resolved.is_file():
return f"File not found: {path}"
try:
return _finish(resolved.read_text(encoding="utf-8"))
except UnicodeDecodeError:
return f"Cannot read '{path}': not a text file"
except Exception as e:
logger.warning("read_text error for '%s': %s", path, e)
return f"Error reading file: {e}"
# The util IS the tool — decorate it directly so there is one body + docstring.
read_text = tool(name="read_text")(read_text_file)
|