From 36da3f7024b14998b1aeb3af859951d564f30418 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 30 Jun 2026 17:57:02 +0800 Subject: [PATCH] feat(files-mcp): business wrappers and Pydantic request models Adds files_mcp/tools/requests.py with 9 Pydantic body models (one per tool, since fastapi-mcp passes args as JSON body) and files_mcp/tools/files.py with 9 thin wrappers that call path_guard then fs_ops and return model_dump() form. Adds tests/unit/test_files_tool.py covering happy path, sandbox escape, and KeyError/FileExistsError propagation for each wrapper. Co-Authored-By: Claude Fable 5 --- files_mcp/tools/__init__.py | 5 ++ files_mcp/tools/files.py | 76 ++++++++++++++++++++++ files_mcp/tools/requests.py | 58 +++++++++++++++++ tests/unit/test_files_tool.py | 116 ++++++++++++++++++++++++++++++++++ 4 files changed, 255 insertions(+) create mode 100644 files_mcp/tools/__init__.py create mode 100644 files_mcp/tools/files.py create mode 100644 files_mcp/tools/requests.py create mode 100644 tests/unit/test_files_tool.py diff --git a/files_mcp/tools/__init__.py b/files_mcp/tools/__init__.py new file mode 100644 index 0000000..b246a57 --- /dev/null +++ b/files_mcp/tools/__init__.py @@ -0,0 +1,5 @@ +# coding=utf-8 +""" +@Time :2026/6/30 +@Author :tao.chen +""" diff --git a/files_mcp/tools/files.py b/files_mcp/tools/files.py new file mode 100644 index 0000000..86828f2 --- /dev/null +++ b/files_mcp/tools/files.py @@ -0,0 +1,76 @@ +# coding=utf-8 +""" +@Time :2026/6/30 +@Author :tao.chen + +Business wrappers: each tool function logs on entry, calls path_guard +on every user-supplied path, delegates to fs_ops, and returns the +Pydantic-serialized result. No business logic of its own — same shape +as spark_executor/tools/connections.py. +""" +from common.logging import logger +from files_mcp.core import fs_ops, path_guard + + +def create_file(*, path: str, content: str, overwrite: bool = False) -> dict: + logger.debug( + f"create_file enter path={path} content_len={len(content)} overwrite={overwrite}" + ) + p = path_guard.resolve_and_check(path) + return fs_ops.create(p, content, overwrite).model_dump() + + +def read_file(*, path: str) -> dict: + logger.debug(f"read_file enter path={path}") + p = path_guard.resolve_and_check(path) + return fs_ops.read(p) + + +def update_file(*, path: str, content: str) -> dict: + logger.debug(f"update_file enter path={path} content_len={len(content)}") + p = path_guard.resolve_and_check(path) + return fs_ops.update(p, content).model_dump() + + +def delete_file(*, path: str, recursive: bool = False) -> dict: + logger.debug(f"delete_file enter path={path} recursive={recursive}") + p = path_guard.resolve_and_check(path) + return fs_ops.delete(p, recursive).model_dump() + + +def list_dir( + *, path: str, recursive: bool = False, max_depth: int = 1 +) -> dict: + logger.debug( + f"list_dir enter path={path} recursive={recursive} max_depth={max_depth}" + ) + p = path_guard.resolve_and_check(path) + return {"path": str(p), "entries": [e.model_dump() for e in fs_ops.list_dir(p, recursive, max_depth)]} + + +def stat(*, path: str) -> dict: + logger.debug(f"stat enter path={path}") + p = path_guard.resolve_and_check(path) + return fs_ops.stat(p).model_dump() + + +def search(*, path: str, pattern: str, recursive: bool = True) -> dict: + logger.debug( + f"search enter path={path} pattern={pattern!r} recursive={recursive}" + ) + p = path_guard.resolve_and_check(path) + return {"pattern": pattern, "matches": fs_ops.search(p, pattern, recursive)} + + +def move(*, src: str, dst: str, overwrite: bool = False) -> dict: + logger.debug(f"move enter src={src} dst={dst} overwrite={overwrite}") + s = path_guard.resolve_and_check(src) + d = path_guard.resolve_and_check(dst) + return fs_ops.move(s, d, overwrite).model_dump() + + +def copy(*, src: str, dst: str, overwrite: bool = False) -> dict: + logger.debug(f"copy enter src={src} dst={dst} overwrite={overwrite}") + s = path_guard.resolve_and_check(src) + d = path_guard.resolve_and_check(dst) + return fs_ops.copy(s, d, overwrite).model_dump() diff --git a/files_mcp/tools/requests.py b/files_mcp/tools/requests.py new file mode 100644 index 0000000..cba577c --- /dev/null +++ b/files_mcp/tools/requests.py @@ -0,0 +1,58 @@ +# coding=utf-8 +""" +@Time :2026/6/30 +@Author :tao.chen + +Pydantic body models for the FastAPI route layer. fastapi-mcp passes +tool args as a JSON body, so every tool needs a model (even single-arg +ones) — dict-typed query params would 422. +""" +from pydantic import BaseModel, Field + + +class CreateFileRequest(BaseModel): + path: str + content: str + overwrite: bool = False + + +class ReadFileRequest(BaseModel): + path: str + + +class UpdateFileRequest(BaseModel): + path: str + content: str + + +class DeleteFileRequest(BaseModel): + path: str + recursive: bool = False + + +class ListDirRequest(BaseModel): + path: str + recursive: bool = False + max_depth: int = Field(default=1, ge=0) + + +class StatRequest(BaseModel): + path: str + + +class SearchRequest(BaseModel): + path: str + pattern: str + recursive: bool = True + + +class MoveRequest(BaseModel): + src: str + dst: str + overwrite: bool = False + + +class CopyRequest(BaseModel): + src: str + dst: str + overwrite: bool = False diff --git a/tests/unit/test_files_tool.py b/tests/unit/test_files_tool.py new file mode 100644 index 0000000..1290ae9 --- /dev/null +++ b/tests/unit/test_files_tool.py @@ -0,0 +1,116 @@ +# coding=utf-8 +""" +@Time :2026/6/30 +@Author :tao.chen +""" +from pathlib import Path + +import pytest + +from files_mcp.core import path_guard +from files_mcp.tools import files + + +@pytest.fixture +def sandbox_root(tmp_path, monkeypatch): + monkeypatch.setenv("FILES_MCP_ROOT", str(tmp_path)) + path_guard.set_root_for_testing(tmp_path.resolve()) + return tmp_path.resolve() + + +class TestCreateFile: + def test_creates_file(self, sandbox_root: Path): + result = files.create_file(path="new.txt", content="hello") + assert result["status"] == "CREATED" + assert (sandbox_root / "new.txt").read_text(encoding="utf-8") == "hello" + + def test_escape_raises_value_error(self, sandbox_root: Path): + with pytest.raises(ValueError, match="escapes sandbox"): + files.create_file(path="../escape.txt", content="x") + + def test_existing_raises_file_exists(self, sandbox_root: Path): + (sandbox_root / "f.txt").write_text("old", encoding="utf-8") + with pytest.raises(FileExistsError): + files.create_file(path="f.txt", content="new", overwrite=False) + + +class TestReadFile: + def test_reads_existing(self, sandbox_root: Path): + (sandbox_root / "f.txt").write_text("hi", encoding="utf-8") + out = files.read_file(path="f.txt") + assert out["content"] == "hi" + assert out["encoding"] == "utf-8" + + def test_escape_raises_value_error(self, sandbox_root: Path): + with pytest.raises(ValueError, match="escapes sandbox"): + files.read_file(path="/etc/passwd") + + +class TestUpdateFile: + def test_updates_existing(self, sandbox_root: Path): + (sandbox_root / "f.txt").write_text("old", encoding="utf-8") + result = files.update_file(path="f.txt", content="new") + assert result["status"] == "UPDATED" + assert (sandbox_root / "f.txt").read_text(encoding="utf-8") == "new" + + def test_missing_raises_keyerror(self, sandbox_root: Path): + with pytest.raises(KeyError, match="not found"): + files.update_file(path="nope.txt", content="x") + + +class TestDeleteFile: + def test_deletes_file(self, sandbox_root: Path): + (sandbox_root / "f.txt").write_text("x", encoding="utf-8") + result = files.delete_file(path="f.txt") + assert result["status"] == "DELETED" + + +class TestListDir: + def test_returns_entries(self, sandbox_root: Path): + (sandbox_root / "a.txt").write_text("a", encoding="utf-8") + (sandbox_root / "b").mkdir() + out = files.list_dir(path=".", recursive=False, max_depth=1) + assert out["path"] == str(sandbox_root) + names = sorted(e["name"] for e in out["entries"]) + assert names == ["a.txt", "b"] + + +class TestStat: + def test_returns_file_entry(self, sandbox_root: Path): + (sandbox_root / "f.txt").write_text("hi", encoding="utf-8") + out = files.stat(path="f.txt") + assert out["name"] == "f.txt" + assert out["type"] == "file" + + +class TestSearch: + def test_finds_matches(self, sandbox_root: Path): + (sandbox_root / "a.py").write_text("a", encoding="utf-8") + (sandbox_root / "b.txt").write_text("b", encoding="utf-8") + out = files.search(path=".", pattern="*.py", recursive=False) + assert out["pattern"] == "*.py" + assert len(out["matches"]) == 1 + assert out["matches"][0].endswith("a.py") + + +class TestMove: + def test_moves_file(self, sandbox_root: Path): + (sandbox_root / "a.txt").write_text("x", encoding="utf-8") + result = files.move(src="a.txt", dst="b.txt") + assert result["status"] == "MOVED" + assert not (sandbox_root / "a.txt").exists() + assert (sandbox_root / "b.txt").exists() + + def test_escape_on_dst_raises(self, sandbox_root: Path): + (sandbox_root / "a.txt").write_text("x", encoding="utf-8") + with pytest.raises(ValueError, match="escapes sandbox"): + files.move(src="a.txt", dst="../escape.txt") + + +class TestCopy: + def test_copies_file(self, sandbox_root: Path): + (sandbox_root / "a.txt").write_text("x", encoding="utf-8") + result = files.copy(src="a.txt", dst="b.txt") + assert result["status"] == "COPIED" + assert (sandbox_root / "a.txt").exists() # source unchanged + assert (sandbox_root / "b.txt").read_text(encoding="utf-8") == "x"