feat(files-mcp): add Pydantic models and pure fs_ops layer

Adds files_mcp/models.py (FileEntry, OperationResult) and files_mcp/core/fs_ops.py with 9 pure file operations on already-resolved Path objects: create/read/update/delete/list_dir/stat/search/move/copy. All writes go through tempfile.mkstemp + os.replace for atomicity; read enforces utf-8 + 1MB cap; create/update enforce strict presence semantics (create: absent or overwrite; update: present). 39 unit tests in tests/unit/test_fs_ops.py. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-06-30 17:52:00 +08:00
parent d8d541f50d
commit 01f5e54a66
3 changed files with 498 additions and 0 deletions
@@ -0,0 +1,177 @@
+# coding=utf-8
+"""
+@Time   :2026/6/30
+@Author :tao.chen
+
+Pure file-system operations on already-resolved Path objects. Callers
+are responsible for sandbox resolution (see path_guard.resolve_and_check).
+No global state, no logging — kept trivially testable.
+"""
+import os
+import shutil
+import tempfile
+from pathlib import Path
+
+from files_mcp.models import FileEntry, OperationResult
+
+MAX_BYTES = 1_048_576  # 1 MB cap, mirrors read_job_file / update_job_file
+
+
+def _to_entry(p: Path) -> FileEntry:
+    st = p.lstat()  # lstat: do not follow symlink for type detection
+    if p.is_symlink():
+        kind = "symlink"
+    elif p.is_dir():
+        kind = "directory"
+    elif p.is_file():
+        kind = "file"
+    else:
+        kind = "other"
+    return FileEntry(
+        path=str(p),
+        name=p.name,
+        type=kind,
+        size=st.st_size if kind == "file" else 0,
+        mtime=st.st_mtime,
+        mode=st.st_mode,
+    )
+
+
+def _check_size(content: str) -> None:
+    if len(content.encode("utf-8")) > MAX_BYTES:
+        raise ValueError(
+            f"content too large: {len(content)} bytes > {MAX_BYTES}"
+        )
+
+
+def _atomic_write(path: Path, content: str) -> None:
+    fd, tmp = tempfile.mkstemp(dir=path.parent, prefix=".tmp_", suffix=path.suffix)
+    try:
+        with os.fdopen(fd, "w", encoding="utf-8") as f:
+            f.write(content)
+        os.replace(tmp, path)
+    except BaseException:
+        Path(tmp).unlink(missing_ok=True)
+        raise
+
+
+def create(path: Path, content: str, overwrite: bool) -> OperationResult:
+    if path.exists() and not overwrite:
+        raise FileExistsError(f"file already exists: {path}")
+    _check_size(content)
+    _atomic_write(path, content)
+    return OperationResult(path=str(path), status="CREATED", size=path.stat().st_size)
+
+
+def read(path: Path) -> dict:
+    if not path.exists():
+        raise KeyError(f"path not found: {path}")
+    if not path.is_file():
+        raise ValueError(f"path {path!r} is not a regular file")
+    raw = path.read_bytes()
+    if len(raw) > MAX_BYTES:
+        raise ValueError(f"file too large: {len(raw)} bytes > {MAX_BYTES}")
+    try:
+        text = raw.decode("utf-8")
+    except UnicodeDecodeError as e:
+        raise ValueError(f"not valid utf-8: {path}") from e
+    return {"path": str(path), "content": text, "size": len(raw), "encoding": "utf-8"}
+
+
+def update(path: Path, content: str) -> OperationResult:
+    if not path.exists():
+        raise KeyError(f"path not found: {path}")
+    _check_size(content)
+    _atomic_write(path, content)
+    return OperationResult(path=str(path), status="UPDATED", size=path.stat().st_size)
+
+
+def delete(path: Path, recursive: bool) -> OperationResult:
+    if not path.exists():
+        raise KeyError(f"path not found: {path}")
+    if path.is_dir() and not recursive:
+        # refuse to delete non-empty dirs unless recursive
+        if any(path.iterdir()):
+            raise ValueError(
+                f"directory {path!r} is not empty (use recursive=True)"
+            )
+    if path.is_dir():
+        if recursive:
+            shutil.rmtree(path)
+        else:
+            # empty dir (non-empty case raised above)
+            path.rmdir()
+    else:
+        path.unlink()
+    return OperationResult(path=str(path), status="DELETED", size=None)
+
+
+def list_dir(path: Path, recursive: bool, max_depth: int) -> list[FileEntry]:
+    if not path.exists():
+        raise KeyError(f"path not found: {path}")
+    if not path.is_dir():
+        raise ValueError(f"path {path!r} is not a directory")
+    if max_depth < 0:
+        raise ValueError("max_depth must be >= 0")
+    out: list[FileEntry] = []
+    if not recursive:
+        max_depth = 0
+    for cur, dirs, files in os.walk(path):
+        rel = Path(cur).relative_to(path)
+        depth = 0 if str(rel) == "." else len(rel.parts)
+        if depth > max_depth:
+            dirs[:] = []  # don't descend further
+            continue
+        for d in dirs:
+            out.append(_to_entry(Path(cur) / d))
+        for f in files:
+            out.append(_to_entry(Path(cur) / f))
+    return out
+
+
+def stat(path: Path) -> FileEntry:
+    if not path.exists() and not path.is_symlink():
+        raise KeyError(f"path not found: {path}")
+    return _to_entry(path)
+
+
+def search(path: Path, pattern: str, recursive: bool) -> list[str]:
+    if not path.exists():
+        raise KeyError(f"path not found: {path}")
+    if not path.is_dir():
+        raise ValueError(f"path {path!r} is not a directory")
+    glob_fn = path.rglob if recursive else path.glob
+    return [str(p) for p in glob_fn(pattern)]
+
+
+def move(src: Path, dst: Path, overwrite: bool) -> OperationResult:
+    if not src.exists():
+        raise KeyError(f"src not found: {src}")
+    if dst.exists() and not overwrite:
+        raise FileExistsError(f"dst already exists: {dst}")
+    if dst.exists() and overwrite:
+        if dst.is_dir() and not dst.is_symlink():
+            shutil.rmtree(dst)
+        else:
+            dst.unlink()
+    os.replace(src, dst)
+    size = dst.stat().st_size if dst.is_file() else None
+    return OperationResult(path=str(dst), status="MOVED", size=size)
+
+
+def copy(src: Path, dst: Path, overwrite: bool) -> OperationResult:
+    if not src.exists():
+        raise KeyError(f"src not found: {src}")
+    if dst.exists() and not overwrite:
+        raise FileExistsError(f"dst already exists: {dst}")
+    if src.is_dir():
+        if dst.exists() and overwrite:
+            shutil.rmtree(dst)
+        shutil.copytree(src, dst)
+        size = None
+    else:
+        if dst.exists() and overwrite:
+            dst.unlink()
+        shutil.copy2(src, dst)
+        size = dst.stat().st_size
+    return OperationResult(path=str(dst), status="COPIED", size=size)
@@ -0,0 +1,23 @@
+# coding=utf-8
+"""
+@Time   :2026/6/30
+@Author :tao.chen
+"""
+from typing import Literal
+
+from pydantic import BaseModel
+
+
+class FileEntry(BaseModel):
+    path: str            # absolute, resolved, inside sandbox
+    name: str
+    type: Literal["file", "directory", "symlink", "other"]
+    size: int            # bytes; 0 for directories
+    mtime: float         # unix seconds
+    mode: int            # st_mode (octal, e.g. 0o644)
+
+
+class OperationResult(BaseModel):
+    path: str
+    status: str          # "CREATED" | "UPDATED" | "DELETED" | "MOVED" | "COPIED"
+    size: int | None     # None for delete/move