feat: implement file read tool with support for text and image files, including validation for large files

astrbot/core/astr_main_agent.py

@@ -1350,7 +1350,15 @@ async def build_main_agent(
         )
 
         if config.computer_use_runtime == "local":
-            tool_prompt += f"\nCurrent workspace you can use: `{os.path.join(get_astrbot_workspaces_path(), event.unified_msg_origin)}`\n"
+            from astrbot.core.computer.tools.fs import _normalize_umo_for_workspace
+
+            normalized_umo = _normalize_umo_for_workspace(event.unified_msg_origin)
+            tool_prompt += (
+                f"\nCurrent workspace you can use: "
+                f"`{os.path.join(get_astrbot_workspaces_path(), normalized_umo)}`\n"
+                "Unless the user explicitly specifies a different directory, "
+                "perform all file-related operations in this workspace.\n"
+            )
 
         req.system_prompt += f"\n{tool_prompt}\n"
 

astrbot/core/computer/booters/local.py

@@ -16,6 +16,7 @@ from astrbot.core.utils.astrbot_path import get_astrbot_root
 
 from ..olayer import FileSystemComponent, PythonComponent, ShellComponent
 from .base import ComputerBooter
+from .shipyard_search_file_util import _truncate_long_lines
 
 _BLOCKED_COMMAND_PATTERNS = [
     " rm -rf ",
@@ -236,7 +237,7 @@ class LocalFileSystemComponent(FileSystemComponent):
                 before_context=before_context,
                 line_number=True,
             )
-            return {"success": True, "content": "".join(results)}
+            return {"success": True, "content": _truncate_long_lines("".join(results))}
 
         return await asyncio.to_thread(_run)
 

astrbot/core/computer/booters/shipyard_search_file_util.py

@@ -5,6 +5,27 @@ from typing import Any
 
 from ..olayer import ShellComponent
 
+_MAX_SEARCH_LINE_COLUMNS = 1000
+
+
+def _truncate_long_lines(text: str) -> str:
+    output_lines: list[str] = []
+    for line in text.splitlines(keepends=True):
+        line_ending = ""
+        line_body = line
+        if line.endswith("\r\n"):
+            line_body = line[:-2]
+            line_ending = "\r\n"
+        elif line.endswith("\n") or line.endswith("\r"):
+            line_body = line[:-1]
+            line_ending = line[-1]
+
+        if len(line_body) > _MAX_SEARCH_LINE_COLUMNS:
+            line_body = line_body[:_MAX_SEARCH_LINE_COLUMNS]
+
+        output_lines.append(f"{line_body}{line_ending}")
+    return "".join(output_lines)
+
 
 def _build_rg_command(
     *,
@@ -14,7 +35,15 @@ def _build_rg_command(
     after_context: int | None,
     before_context: int | None,
 ) -> list[str]:
-    command = ["rg", "--color=never", "-n", "-e", pattern]
+    command = [
+        "rg",
+        "--color=never",
+        "-n",
+        "--max-columns",
+        str(_MAX_SEARCH_LINE_COLUMNS),
+        "-e",
+        pattern,
+    ]
     if glob:
         command.extend(["-g", glob])
     if after_context is not None:
@@ -104,7 +133,7 @@ async def search_files_via_shell(
         before_context=before_context,
     )
     result = await shell.exec(command, timeout=timeout)
-    stdout = str(result.get("stdout", "") or "")
+    stdout = _truncate_long_lines(str(result.get("stdout", "") or ""))
     stderr = str(result.get("stderr", "") or "")
     exit_code = result.get("exit_code")
     if exit_code in (0, None):

astrbot/core/computer/tools/fs.py

@@ -51,6 +51,7 @@ from astrbot.core.utils.astrbot_path import (
 
 from ..computer_client import get_booter
 from .permissions import check_admin_permission
+from .tool_utils.file_read import read_file_tool_result
 
 
 def _normalize_umo_for_workspace(umo: str) -> str:
@@ -175,12 +176,12 @@ class FileReadTool(FunctionTool):
                 },
                 "offset": {
                     "type": "integer",
-                    "description": "Line offset to start reading from. Defaults to 0.",
+                    "description": "Optional line offset to start reading from. 0-based index.",
                     "minimum": 0,
                 },
                 "limit": {
                     "type": "integer",
-                    "description": "Maximum number of lines to read. Defaults to 4000.",
+                    "description": "Optional maximum number of lines to read.",
                     "minimum": 1,
                 },
             },
@@ -226,23 +227,12 @@ class FileReadTool(FunctionTool):
                 context.context.context,
                 context.context.event.unified_msg_origin,
             )
-            result = await sb.fs.read_file(
+            return await read_file_tool_result(
+                sb,
                 path=normalized_path,
-                encoding="utf-8",
                 offset=offset,
                 limit=limit,
             )
-            if not result.get("success", False):
-                error_detail = str(result.get("error", "") or "").strip()
-                return (
-                    "Error reading file: "
-                    f"{error_detail or 'unknown filesystem read error'}"
-                )
-
-            content = str(result.get("content", "") or "")
-            if not content:
-                return "No content found at the requested line offset."
-            return content or ""
         except PermissionError as exc:
             return f"Error: {exc}"
         except Exception as exc:
@@ -438,7 +428,7 @@ class GrepTool(FunctionTool):
                 },
                 "result_limit": {
                     "type": "integer",
-                    "description": "Maximum number of result groups returned by the tool. Defaults to 50.",
+                    "description": "Maximum number of result groups returned by the tool. Defaults to 100.",
                     "minimum": 1,
                 },
             },
@@ -544,7 +534,7 @@ class GrepTool(FunctionTool):
         pattern: str,
         path: str | None = None,
         glob: str | None = None,
-        result_limit: int = 50,
+        result_limit: int = 100,
         **kwargs,
     ) -> ToolExecResult:
         normalized_pattern = pattern.strip()

astrbot/core/computer/tools/tool_utils/file_read.py

@@ -0,0 +1,458 @@
+from __future__ import annotations
+
+import base64
+import json
+from asyncio import to_thread
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Literal
+
+import mcp
+
+from astrbot.core.agent.context.token_counter import EstimateTokenCounter
+from astrbot.core.agent.message import Message
+from astrbot.core.agent.tool import ToolExecResult
+from astrbot.core.computer.booters.base import ComputerBooter
+from astrbot.core.computer.booters.local import LocalBooter
+from astrbot.core.utils.astrbot_path import get_astrbot_temp_path
+from astrbot.core.utils.media_utils import (
+    IMAGE_COMPRESS_DEFAULT_MAX_SIZE,
+    IMAGE_COMPRESS_DEFAULT_OPTIMIZE,
+    IMAGE_COMPRESS_DEFAULT_QUALITY,
+    _compress_image_sync,
+)
+
+_MAX_FILE_READ_BYTES = 128 * 1024
+_MAX_FILE_READ_TOKENS = 25_000
+_MAX_TEXT_FILE_FULL_READ_BYTES = 256 * 1024
+_FILE_SNIFF_BYTES = 512
+_TOKEN_COUNTER = EstimateTokenCounter()
+_TEXT_ENCODINGS = (
+    "utf-8-sig",
+    "utf-8",
+    "gb18030",
+    "utf-16",
+    "utf-16-le",
+    "utf-16-be",
+    "utf-32",
+    "utf-32-le",
+    "utf-32-be",
+)
+_UTF_BOMS = (
+    b"\xef\xbb\xbf",
+    b"\xff\xfe",
+    b"\xfe\xff",
+    b"\xff\xfe\x00\x00",
+    b"\x00\x00\xfe\xff",
+)
+_BINARY_MAGIC_PREFIXES = (
+    b"%PDF-",
+    b"PK\x03\x04",
+    b"PK\x05\x06",
+    b"PK\x07\x08",
+    b"\x1f\x8b",
+    b"7z\xbc\xaf\x27\x1c",
+    b"Rar!\x1a\x07",
+    b"\x7fELF",
+    b"MZ",
+)
+
+
+@dataclass(frozen=True)
+class FileProbe:
+    kind: Literal["text", "image", "binary"]
+    encoding: str | None
+    mime_type: str | None
+    size_bytes: int
+
+
+def _build_probe_script(path: str) -> str:
+    return f"""
+import base64
+import json
+from pathlib import Path
+
+path = Path({path!r})
+with path.open("rb") as file_obj:
+    sample = file_obj.read({_FILE_SNIFF_BYTES})
+print(
+    json.dumps(
+        {{
+            "size_bytes": path.stat().st_size,
+            "sample_b64": base64.b64encode(sample).decode("ascii"),
+        }}
+    )
+)
+""".strip()
+
+
+def _build_text_read_script(
+    path: str,
+    *,
+    encoding: str,
+    offset: int | None,
+    limit: int | None,
+) -> str:
+    start_expr = "0" if offset is None else str(offset)
+    limit_expr = "None" if limit is None else str(limit)
+    return f"""
+import json
+from pathlib import Path
+
+path = Path({path!r})
+start = {start_expr}
+limit = {limit_expr}
+end = None if limit is None else start + limit
+lines = []
+with path.open("r", encoding={encoding!r}, newline="") as file_obj:
+    for index, line in enumerate(file_obj):
+        if index < start:
+            continue
+        if end is not None and index >= end:
+            break
+        lines.append(line)
+content = "".join(lines)
+print(json.dumps({{"content": content}}, ensure_ascii=False))
+""".strip()
+
+
+def _build_image_read_script(path: str) -> str:
+    return f"""
+import base64
+import json
+from pathlib import Path
+
+path = Path({path!r})
+data = path.read_bytes()
+print(
+    json.dumps(
+        {{
+            "size_bytes": len(data),
+            "base64": base64.b64encode(data).decode("ascii"),
+        }}
+    )
+)
+""".strip()
+
+
+async def _exec_python_json(
+    booter: ComputerBooter,
+    script: str,
+    *,
+    action: str,
+) -> dict:
+    result = await booter.python.exec(script)
+    data = result.get("data") if isinstance(result.get("data"), dict) else {}
+    if not isinstance(data, dict):
+        raise RuntimeError(f"{action} failed: invalid result format")
+    output = data.get("output") if isinstance(data.get("output"), dict) else {}
+    if not isinstance(output, dict):
+        raise RuntimeError(f"{action} failed: invalid output format")
+    error_text = str(data.get("error", "") or result.get("error", "") or "").strip()
+    if error_text:
+        raise RuntimeError(f"{action} failed: {error_text}")
+
+    text = str(output.get("text", "") or "").strip()
+    if not text:
+        raise RuntimeError(f"{action} failed: empty output")
+
+    try:
+        payload = json.loads(text)
+    except json.JSONDecodeError as exc:
+        raise RuntimeError(f"{action} failed: invalid JSON output") from exc
+
+    if not isinstance(payload, dict):
+        raise RuntimeError(f"{action} failed: invalid JSON payload")
+    return payload
+
+
+async def _probe_local_file(path: str) -> dict[str, str | int]:
+    def _run() -> dict[str, str | int]:
+        file_path = Path(path)
+        with file_path.open("rb") as file_obj:
+            sample = file_obj.read(_FILE_SNIFF_BYTES)
+        return {
+            "size_bytes": file_path.stat().st_size,
+            "sample_b64": base64.b64encode(sample).decode("ascii"),
+        }
+
+    return await to_thread(_run)
+
+
+async def _read_local_text_range(
+    path: str,
+    *,
+    encoding: str,
+    offset: int | None,
+    limit: int | None,
+) -> str:
+    def _run() -> str:
+        lines: list[str] = []
+        start = 0 if offset is None else offset
+        end = None if limit is None else start + limit
+        with Path(path).open("r", encoding=encoding, newline="") as file_obj:
+            for index, line in enumerate(file_obj):
+                if index < start:
+                    continue
+                if end is not None and index >= end:
+                    break
+                lines.append(line)
+        return "".join(lines)
+
+    return await to_thread(_run)
+
+
+async def _read_local_image_base64(path: str) -> dict[str, str | int]:
+    def _run() -> dict[str, str | int]:
+        data = Path(path).read_bytes()
+        return {
+            "size_bytes": len(data),
+            "base64": base64.b64encode(data).decode("ascii"),
+        }
+
+    return await to_thread(_run)
+
+
+async def _compress_image_bytes_to_base64(data: bytes) -> dict[str, str | int]:
+    def _run() -> dict[str, str | int]:
+        temp_dir = Path(get_astrbot_temp_path())
+        temp_dir.mkdir(parents=True, exist_ok=True)
+        compressed_path = Path(
+            _compress_image_sync(
+                data,
+                temp_dir,
+                IMAGE_COMPRESS_DEFAULT_MAX_SIZE,
+                IMAGE_COMPRESS_DEFAULT_QUALITY,
+                IMAGE_COMPRESS_DEFAULT_OPTIMIZE,
+            )
+        )
+        try:
+            compressed_bytes = compressed_path.read_bytes()
+        finally:
+            compressed_path.unlink(missing_ok=True)
+
+        return {
+            "size_bytes": len(compressed_bytes),
+            "base64": base64.b64encode(compressed_bytes).decode("ascii"),
+            "mime_type": "image/jpeg",
+        }
+
+    return await to_thread(_run)
+
+
+def _detect_image_mime(sample: bytes) -> str | None:
+    if sample.startswith(b"\x89PNG\r\n\x1a\n"):
+        return "image/png"
+    if sample.startswith(b"\xff\xd8\xff"):
+        return "image/jpeg"
+    if sample.startswith((b"GIF87a", b"GIF89a")):
+        return "image/gif"
+    if sample.startswith(b"BM"):
+        return "image/bmp"
+    if sample.startswith((b"II*\x00", b"MM\x00*")):
+        return "image/tiff"
+    if sample.startswith(b"\x00\x00\x01\x00"):
+        return "image/x-icon"
+    if len(sample) >= 12 and sample[:4] == b"RIFF" and sample[8:12] == b"WEBP":
+        return "image/webp"
+    if len(sample) >= 12 and sample[4:12] in (b"ftypavif", b"ftypavis"):
+        return "image/avif"
+    return None
+
+
+def _looks_like_known_binary(sample: bytes) -> bool:
+    return any(sample.startswith(prefix) for prefix in _BINARY_MAGIC_PREFIXES)
+
+
+def _looks_like_text(decoded: str) -> bool:
+    if not decoded:
+        return True
+
+    disallowed = 0
+    printable = 0
+    for char in decoded:
+        if char in "\n\r\t\f\b":
+            printable += 1
+            continue
+        if char.isprintable():
+            printable += 1
+        code = ord(char)
+        if (0 <= code < 32) or (127 <= code < 160):
+            disallowed += 1
+
+    total = max(len(decoded), 1)
+    return disallowed / total <= 0.02 and printable / total >= 0.85
+
+
+def _detect_text_encoding(sample: bytes) -> str | None:
+    if not sample:
+        return "utf-8"
+
+    if b"\x00" in sample and not sample.startswith(_UTF_BOMS):
+        odd_bytes = sample[1::2]
+        even_bytes = sample[0::2]
+        odd_zero_ratio = odd_bytes.count(0) / max(len(odd_bytes), 1)
+        even_zero_ratio = even_bytes.count(0) / max(len(even_bytes), 1)
+        if odd_zero_ratio < 0.8 and even_zero_ratio < 0.8:
+            return None
+
+    for encoding in _TEXT_ENCODINGS:
+        try:
+            decoded = sample.decode(encoding)
+        except UnicodeDecodeError:
+            continue
+        if _looks_like_text(decoded):
+            return encoding
+
+    return None
+
+
+def _probe_file(sample: bytes, *, size_bytes: int) -> FileProbe:
+    if image_mime := _detect_image_mime(sample):
+        return FileProbe(
+            kind="image",
+            encoding=None,
+            mime_type=image_mime,
+            size_bytes=size_bytes,
+        )
+
+    if _looks_like_known_binary(sample):
+        return FileProbe(
+            kind="binary",
+            encoding=None,
+            mime_type=None,
+            size_bytes=size_bytes,
+        )
+
+    if encoding := _detect_text_encoding(sample):
+        return FileProbe(
+            kind="text",
+            encoding=encoding,
+            mime_type="text/plain",
+            size_bytes=size_bytes,
+        )
+
+    return FileProbe(
+        kind="binary",
+        encoding=None,
+        mime_type=None,
+        size_bytes=size_bytes,
+    )
+
+
+def _validate_text_output(content: str) -> str | None:
+    content_bytes = len(content.encode("utf-8"))
+    if content_bytes > _MAX_FILE_READ_BYTES:
+        return (
+            "Error reading file: "
+            f"output exceeds {_MAX_FILE_READ_BYTES} bytes "
+            f"({content_bytes} bytes). Use `offset`, `limit` to narrow the read window."
+        )
+
+    content_tokens = _TOKEN_COUNTER.count_tokens(
+        [Message(role="user", content=content)]
+    )
+    if content_tokens > _MAX_FILE_READ_TOKENS:
+        return (
+            "Error reading file: "
+            f"output exceeds {_MAX_FILE_READ_TOKENS} tokens "
+            f"({content_tokens} tokens). Use `offset`, `limit` to narrow the read window."
+        )
+
+    return None
+
+
+def _validate_full_text_read_request(probe: FileProbe) -> str | None:
+    if probe.size_bytes > _MAX_TEXT_FILE_FULL_READ_BYTES:
+        return (
+            "Error reading file: "
+            f"text file exceeds {_MAX_TEXT_FILE_FULL_READ_BYTES} bytes "
+            f"({probe.size_bytes} bytes). Use `offset` and `limit` to narrow the read window."
+        )
+    return None
+
+
+async def read_file_tool_result(
+    booter: ComputerBooter,
+    *,
+    path: str,
+    offset: int | None,
+    limit: int | None,
+) -> ToolExecResult:
+    if isinstance(booter, LocalBooter):
+        probe_payload = await _probe_local_file(path)
+    else:
+        probe_payload = await _exec_python_json(
+            booter,
+            _build_probe_script(path),
+            action="file probe",
+        )
+    sample_b64 = str(probe_payload.get("sample_b64", "") or "")
+    sample = base64.b64decode(sample_b64) if sample_b64 else b""
+    size_bytes = int(probe_payload.get("size_bytes", 0) or 0)
+    probe = _probe_file(sample, size_bytes=size_bytes)
+
+    if probe.kind == "binary":
+        return "Error reading file: binary files are not supported by this tool."
+
+    if probe.kind == "image":
+        if isinstance(booter, LocalBooter):
+            image_payload = await _read_local_image_base64(path)
+        else:
+            image_payload = await _exec_python_json(
+                booter,
+                _build_image_read_script(path),
+                action="image read",
+            )
+        raw_base64_data = str(image_payload.get("base64", "") or "")
+        if not raw_base64_data:
+            return "Error reading file: image payload is empty."
+        raw_bytes = base64.b64decode(raw_base64_data)
+        compressed_payload = await _compress_image_bytes_to_base64(raw_bytes)
+        base64_data = str(compressed_payload.get("base64", "") or "")
+        if not base64_data:
+            return "Error reading file: compressed image payload is empty."
+        return mcp.types.CallToolResult(
+            content=[
+                mcp.types.ImageContent(
+                    type="image",
+                    data=base64_data,
+                    mimeType=str(
+                        compressed_payload.get("mime_type", "") or "image/jpeg"
+                    ),
+                )
+            ]
+        )
+
+    if isinstance(booter, LocalBooter):
+        if offset is None and limit is None:
+            if validation_error := _validate_full_text_read_request(probe):
+                return validation_error
+        content = await _read_local_text_range(
+            path,
+            encoding=probe.encoding or "utf-8",
+            offset=offset,
+            limit=limit,
+        )
+    else:
+        if offset is None and limit is None:
+            if validation_error := _validate_full_text_read_request(probe):
+                return validation_error
+        text_payload = await _exec_python_json(
+            booter,
+            _build_text_read_script(
+                path,
+                encoding=probe.encoding or "utf-8",
+                offset=offset,
+                limit=limit,
+            ),
+            action="text read",
+        )
+        content = str(text_payload.get("content", "") or "")
+    if not content:
+        return "No content found at the requested line offset."
+
+    if validation_error := _validate_text_output(content):
+        return validation_error
+
+    return content

tests/test_computer_fs_tools.py

@@ -0,0 +1,215 @@
+from __future__ import annotations
+
+import base64
+from types import SimpleNamespace
+from typing import Any
+
+import pytest
+from mcp.types import CallToolResult, ImageContent
+from PIL import Image
+
+from astrbot.core.agent.run_context import ContextWrapper
+from astrbot.core.computer.booters.local import LocalBooter
+from astrbot.core.computer.tools import fs as fs_tools
+from astrbot.core.computer.tools.tool_utils import file_read as file_read_utils
+
+
+def _make_context(
+    *,
+    require_admin: bool = True,
+    role: str = "admin",
+    runtime: str = "local",
+    umo: str = "qq:friend:user-1",
+) -> ContextWrapper:
+    config_holder = SimpleNamespace(
+        get_config=lambda umo=None: {
+            "provider_settings": {
+                "computer_use_require_admin": require_admin,
+                "computer_use_runtime": runtime,
+            }
+        }
+    )
+    event = SimpleNamespace(
+        role=role,
+        unified_msg_origin=umo,
+        get_sender_id=lambda: "user-1",
+    )
+    astr_ctx = SimpleNamespace(context=config_holder, event=event)
+    return ContextWrapper(context=astr_ctx)
+
+
+def _setup_local_fs_tools(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path,
+    *,
+    umo: str = "qq:friend:user-1",
+) -> Any:
+    workspaces_root = tmp_path / "workspaces"
+    skills_root = tmp_path / "skills"
+    temp_root = tmp_path / "temp"
+    workspaces_root.mkdir()
+    skills_root.mkdir()
+    temp_root.mkdir()
+
+    monkeypatch.setattr(
+        fs_tools,
+        "get_astrbot_workspaces_path",
+        lambda: str(workspaces_root),
+    )
+    monkeypatch.setattr(
+        fs_tools,
+        "get_astrbot_skills_path",
+        lambda: str(skills_root),
+    )
+    monkeypatch.setattr(
+        fs_tools,
+        "get_astrbot_temp_path",
+        lambda: str(temp_root),
+    )
+    monkeypatch.setattr(
+        file_read_utils,
+        "get_astrbot_temp_path",
+        lambda: str(temp_root),
+    )
+
+    booter = LocalBooter()
+
+    async def _fake_get_booter(_ctx, _umo):
+        return booter
+
+    monkeypatch.setattr(fs_tools, "get_booter", _fake_get_booter)
+
+    normalized_umo = fs_tools._normalize_umo_for_workspace(umo)
+    workspace = workspaces_root / normalized_umo
+    workspace.mkdir(parents=True, exist_ok=True)
+    return workspace
+
+
+def _make_large_text() -> str:
+    return "".join(f"line-{index:05d}-{'x' * 48}\n" for index in range(6000))
+
+
+@pytest.mark.asyncio
+async def test_file_read_tool_rejects_large_full_text_read_before_local_stream_read(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path,
+):
+    workspace = _setup_local_fs_tools(monkeypatch, tmp_path)
+    large_file = workspace / "large.txt"
+    large_file.write_text(_make_large_text(), encoding="utf-8")
+
+    async def _unexpected_read(*args, **kwargs):
+        raise AssertionError("full file read should be rejected before streaming")
+
+    monkeypatch.setattr(file_read_utils, "_read_local_text_range", _unexpected_read)
+
+    result = await fs_tools.FileReadTool().call(
+        _make_context(),
+        path="large.txt",
+    )
+
+    assert "text file exceeds 262144 bytes" in result
+    assert "Use `offset` and `limit`" in result
+
+
+@pytest.mark.asyncio
+async def test_file_read_tool_allows_partial_read_for_large_text_file(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path,
+):
+    workspace = _setup_local_fs_tools(monkeypatch, tmp_path)
+    large_file = workspace / "large.txt"
+    lines = [f"line-{index:05d}\n" for index in range(50000)]
+    large_file.write_text("".join(lines), encoding="utf-8")
+
+    result = await fs_tools.FileReadTool().call(
+        _make_context(),
+        path="large.txt",
+        offset=1000,
+        limit=3,
+    )
+
+    assert result == "".join(lines[1000:1003])
+
+
+@pytest.mark.asyncio
+async def test_file_read_tool_returns_image_call_tool_result_for_images(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path,
+):
+    workspace = _setup_local_fs_tools(monkeypatch, tmp_path)
+    image_path = workspace / "sample.png"
+    Image.new("RGB", (32, 16), color=(255, 0, 0)).save(image_path, format="PNG")
+
+    result = await fs_tools.FileReadTool().call(
+        _make_context(),
+        path="sample.png",
+    )
+
+    assert isinstance(result, CallToolResult)
+    assert len(result.content) == 1
+    assert isinstance(result.content[0], ImageContent)
+    assert result.content[0].mimeType == "image/jpeg"
+    assert base64.b64decode(result.content[0].data).startswith(b"\xff\xd8\xff")
+
+
+@pytest.mark.asyncio
+async def test_file_read_tool_treats_svg_as_text(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path,
+):
+    workspace = _setup_local_fs_tools(monkeypatch, tmp_path)
+    svg_path = workspace / "shape.svg"
+    svg_text = (
+        "<svg xmlns='http://www.w3.org/2000/svg'><rect width='10' height='10'/></svg>"
+    )
+    svg_path.write_text(svg_text, encoding="utf-8")
+
+    result = await fs_tools.FileReadTool().call(
+        _make_context(),
+        path="shape.svg",
+    )
+
+    assert result == svg_text
+
+
+@pytest.mark.asyncio
+async def test_file_read_tool_rejects_pdf_as_binary(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path,
+):
+    workspace = _setup_local_fs_tools(monkeypatch, tmp_path)
+    pdf_path = workspace / "doc.pdf"
+    pdf_path.write_bytes(b"%PDF-1.7\n%\xe2\xe3\xcf\xd3\n1 0 obj\n<<>>\nendobj\n")
+
+    result = await fs_tools.FileReadTool().call(
+        _make_context(),
+        path="doc.pdf",
+    )
+
+    assert result == "Error reading file: binary files are not supported by this tool."
+
+
+@pytest.mark.asyncio
+async def test_grep_tool_applies_result_limit(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path,
+):
+    workspace = _setup_local_fs_tools(monkeypatch, tmp_path)
+    text_path = workspace / "grep.txt"
+    text_path.write_text(
+        "match-1\nmatch-2\nmatch-3\nmatch-4\n",
+        encoding="utf-8",
+    )
+
+    result = await fs_tools.GrepTool().call(
+        _make_context(),
+        pattern="match",
+        path="grep.txt",
+        result_limit=2,
+    )
+
+    assert "match-1" in result
+    assert "match-2" in result
+    assert "match-3" not in result
+    assert "[Truncated to first 2 result groups.]" in result