tao.chen/mcp-server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request 'Feat/files mcp' (#2) from feat/files-mcp into main

Browse Source 
Reviewed-on: https://gitea-production-a772.up.railway.app/taochen/mcp-server/pulls/2
This commit is contained in:
tao.chen
2026-06-30 11:13:02 +00:00
parent 9305ac750b c17a33152f
commit 4d593ae13c
17 changed files with 3485 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/superpowers/plans/2026-06-30-files-mcp.md
+1799
View File
File diff suppressed because it is too large Load Diff
docs/superpowers/specs/2026-06-30-files-mcp-design.md
+384
View File
@@ -0,0 +1,384 @@
# Files MCP — Design Spec
**Date:** 2026-06-30
**Branch:** `feat/files-mcp`
**Mount path:** `/files-mcp`
**New package:** `files_mcp/`
**Status:** Design approved; awaiting implementation plan
## Goal
Add a sibling MCP service that exposes file-system CRUD (plus a handful of
introspection and rearrangement operations) over the same `fastapi-mcp` transport
the existing `spark-executor-mcp` already uses. The service is sandboxed to a
configurable root directory and only handles utf-8 text content.
## Background & Motivation
`spark-executor-mcp` ships 12 tools for the full Spark-on-YARN lifecycle
(submit, status, logs, kill, connections, plus Stage-2 PySpark file helpers).
The `MCP_SERVICES` registry in `main.py` was designed for additional sibling
services — the comment block above the list literally shows the intended
pattern: `McpService(name="metrics", app=metrics_app, mount_path="/metrics-mcp")`.
This spec fills in one of those slots: a generic, sandboxed file-CRUD service.
Use cases the LLM can now express through the same MCP transport:
- Read/write configuration snippets (YAML, JSON, `.env`) on the host.
- Stage input datasets in a known working directory before submitting a Spark job.
- Inspect prior job logs and outputs.
- Move/rename artifacts after a run completes.
## Non-Goals
- **Binary files.** MCP is JSON; binary would force base64 round-trips and
inflates the LLM context. We only do utf-8 text. (Mirrors the existing
`job_writer` "reject code that is not valid UTF-8" rule.)
- **Streaming / partial reads.** Read returns the whole file, capped at 1 MB.
YARN log tails are a separate tool (`get_job_logs`) on the other service.
- **Multi-tenant isolation / per-user roots.** Stage 1 ships a single shared
root from one env var. Per-user roots are a Stage 3 concern (per the existing
plan `docs/superpowers/plans/2026-06-24-spark-executor-mcp.md`).
- **File locking / concurrent writer arbitration.** Each write is atomic
(`tempfile + os.replace`) but we do not serialize writers. This is fine for
the actual usage (one LLM agent at a time per process); revisit if needed.
## Architecture
```
main.py # append one McpService entry
└── FastApiMCP(files_app, mount_path="/files-mcp")
files_mcp/ # new sibling package
__init__.py # re-exports app; triggers root initialization
server.py # FastAPI("Files MCP") + 9 routes + exception handlers
models.py # FileEntry, OperationResult
core/
path_guard.py # resolve_and_check(path) -> Path; sandbox check
fs_ops.py # pure file operations on already-resolved Path
tools/
files.py # business wrappers: log + delegate to fs_ops
requests.py # Pydantic body models (fastapi-mcp requires)
tests/
unit/test_path_guard.py
unit/test_fs_ops.py
unit/test_files_tool.py
integration/test_files_mcp_routes.py
```
### Why a sibling package (not a sub-module of `spark_executor`)?
The existing `MCP_SERVICES` registry is the extension point the codebase was
designed around. Adding a sibling:
- Uses the reserved extension point without bypassing it.
- Keeps the 12 Spark tools and 9 file tools on separate mount paths, so
`tools/list` for `/spark-executor-mcp` is not bloated.
- Aligns with the project's stated principle of additive change: 0 existing
files get modified beyond the one-line append in `main.py`.
## Components
### `core/path_guard.py` — the sandbox
```python
_ROOT: Path # set once at import time from FILES_MCP_ROOT
def _init_root() -> Path:
root = os.environ.get("FILES_MCP_ROOT")
if not root:
raise RuntimeError("FILES_MCP_ROOT env var is required")
p = Path(root).expanduser().resolve()
if not p.is_dir():
raise RuntimeError(f"FILES_MCP_ROOT={root!r} is not an existing directory")
return p
def resolve_and_check(path: str) -> Path:
"""Resolve an arbitrary input path and verify it lies inside the sandbox.
Raises:
ValueError: empty, contains NUL byte, or escapes the root.
"""
if not path:
raise ValueError("path is empty")
if "\x00" in path:
raise ValueError("path contains NUL byte")
p = Path(path).expanduser().resolve()
try:
p.relative_to(_ROOT)
except ValueError:
raise ValueError(f"path escapes sandbox root {_ROOT}: got {p}") from None
return p
def set_root_for_testing(p: Path) -> None:
"""Test-only escape hatch; never call from production code."""
global _ROOT
_ROOT = p
```
Key properties:
- One `.resolve()` call handles `~`, `.`, `..`, symlinks, double slashes.
- `relative_to` is the only sandbox check — no whitelist to keep in sync.
- A symlink inside the root pointing outside is rejected because
`resolve()` follows the link before we check.
- `set_root_for_testing` is the only mutation point, used by the
`conftest.py` autouse fixture so each test gets a fresh `tmp_path` root.
### `core/fs_ops.py` — pure file operations
All functions take an **already-resolved** `Path` (the caller is responsible for
running `path_guard.resolve_and_check`). This keeps `fs_ops` trivially
testable — no global state, no side effects beyond the operation it performs.
```python
MAX_BYTES = 1_048_576 # 1 MB cap, mirrors read_job_file / update_job_file
def create(path: Path, content: str, overwrite: bool) -> OperationResult: ...
def read(path: Path) -> dict: ...
def update(path: Path, content: str) -> OperationResult: ...
def delete(path: Path, recursive: bool) -> OperationResult: ...
def list_dir(path: Path, recursive: bool, max_depth: int) -> list[FileEntry]: ...
def stat(path: Path) -> FileEntry: ...
def search(path: Path, pattern: str, recursive: bool) -> list[str]: ...
def move(src: Path, dst: Path, overwrite: bool) -> OperationResult: ...
def copy(src: Path, dst: Path, overwrite: bool) -> OperationResult: ...
```
Atomic write pattern (used by `create` and `update`):
```python
fd, tmp_path = 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, path) # atomic on same fs
except BaseException:
Path(tmp_path).unlink(missing_ok=True)
raise
```
### `tools/files.py` — business wrappers
One function per tool, each:
1. Logs DEBUG on entry.
2. Calls `path_guard.resolve_and_check` on every user-supplied path.
3. Delegates to `fs_ops`.
4. Returns a Pydantic-serialized dict.
No business logic of its own — same shape as
`spark_executor/tools/connections.py`.
### `tools/requests.py` — Pydantic body models
One model per route. fastapi-mcp passes tool args as a JSON body, so dict-typed
params (none here, but the pattern matters) need a model, not query params.
Mirrors the pattern in `spark_executor/tools/requests.py`.
### `server.py` — FastAPI app + routes
9 routes, all `POST`, each with a `summary` and a `description`. The same
exception handlers as `spark_executor/server.py`:
- `KeyError` → 404
- `ValueError` → 400
- `FileExistsError` → 409 (new in this service — needed for create / move / copy)
The `/health` route is intentionally **not** added: the root app already has
`/health`. (Linus: don't add a duplicate.)
## Data Model
```python
# files_mcp/models.py
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
```
## API Surface
| Tool | Inputs | Returns | Errors |
|---|---|---|---|
| `create_file` | `path`, `content`, `overwrite=false` | `OperationResult` | 409 if exists & !overwrite; 400 if >1MB or outside root |
| `read_file` | `path` | `{path, content, size, encoding:"utf-8"}` | 404 if not found; 400 if dir / >1MB / non-utf-8 / outside root |
| `update_file` | `path`, `content` | `OperationResult` | 404 if not found; 400 if >1MB / outside root |
| `delete_file` | `path`, `recursive=false` | `OperationResult` | 404 if not found; 400 if dir & !recursive |
| `list_dir` | `path`, `recursive=false`, `max_depth=1` | `{path, entries: [FileEntry]}` | 404 if not found; 400 if not dir |
| `stat` | `path` | `FileEntry` | 404 if not found; 400 if outside root |
| `search` | `path`, `pattern`, `recursive=true` | `{pattern, matches: [str]}` | 404 if root not found; 400 if outside root |
| `move` | `src`, `dst`, `overwrite=false` | `OperationResult` | 404 if src missing; 409 if dst exists & !overwrite |
| `copy` | `src`, `dst`, `overwrite=false` | `OperationResult` | Same as move |
## Error Handling
| Exception | HTTP | When |
|---|---|---|
| `KeyError("path 'X' not found")` | 404 | read/update/delete/stat on missing path |
| `ValueError("path 'X' is not a directory")` | 400 | list_dir with a file path |
| `ValueError("path escapes sandbox")` | 400 | path_guard rejects |
| `ValueError("file 'X' already exists")` | 409 | create / move / copy with dst existing and `overwrite=False` |
| `ValueError("content too large")` | 400 | write > 1 MB |
| `ValueError("not valid utf-8")` | 400 | read encounters non-utf-8 bytes |
| `FileExistsError` | 409 | (extra handler) atomic rename target exists |
| `IsADirectoryError` / `NotADirectoryError` | 400 | operation type mismatches path type |
| Other `OSError` | 500 | real I/O failure (EACCES, ENOSPC, etc.) |
The `FileExistsError` handler is **new** (spark_executor has only the first
two). It is registered on `files_mcp.server.app`, not on the root app, so
existing `KeyError` / `ValueError` handlers in `spark_executor.server` are
unaffected.
## Configuration
- **`FILES_MCP_ROOT`** (required env var): absolute path to the sandbox root.
Must exist and be a directory at process start, otherwise `RuntimeError`
halts startup with a clear message. (Mirrors the way `SPARK_EXECUTOR_DATA_DIR`
is documented in the project plan.)
- **No new dependencies.** All operations use stdlib (`pathlib`, `shutil`,
`tempfile`, `os`, `os.replace`).
- **No config file.** A single env var is enough; multi-tenant per-user roots
is explicitly out of scope.
## Integration with `main.py`
One import, one new entry:
```python
# main.py
from files_mcp import app as files_app
MCP_SERVICES: list[McpService] = [
McpService(name="spark_executor", app=spark_executor_app, mount_path="/spark-executor-mcp"),
McpService(name="files", app=files_app, mount_path="/files-mcp"),
]
```
The existing `lifespan` already iterates `MCP_SERVICES` and calls
`init_mcp_server(svc.app).mount_http(...)` for each. No edit to the lifespan
function.
## Backward Compatibility
- **0 existing tools changed.**
- **0 existing tests changed** (only `conftest.py` gains a new autouse fixture
for the files_mcp tests; spark_executor tests are unaffected).
- **0 new dependencies.**
- **API change is purely additive**: the new `/files-mcp` endpoint exists;
`/spark-executor-mcp` is byte-for-byte identical.
- The existing `GUNICORN_WORKERS=1` warning now covers both mounts, because
both share the root app process. No edit to `gunicorn.conf.py` is needed —
the comment block there already says the warning is process-wide.
## Testing
### `tests/unit/test_path_guard.py` — 7 cases
- `tmp/a.txt` → accept, resolved equals expected.
- `../etc/passwd` → reject.
- `/etc/passwd` (root not `/`) → reject.
- `tmp/../etc/passwd` → reject.
- `~/outside` (root not under home) → reject.
- symlink `tmp/link → /etc/passwd` → reject (resolve follows link).
- empty / NUL → reject.
### `tests/unit/test_fs_ops.py` — ~10 cases
- create: success / overwrite=True / overwrite=False + existing → `FileExistsError`.
- read: success / directory path → `ValueError` / >1MB → `ValueError` / non-utf-8 bytes → `ValueError`.
- update: success / missing → `KeyError`.
- delete: success / missing → `KeyError` / dir without recursive → `ValueError`.
- list_dir: single / recursive / depth cap.
- search: simple glob / recursive / zero matches.
- move/copy: same-fs rename / cross-fs (non-atomic but allowed) / dst exists → `FileExistsError`.
### `tests/unit/test_files_tool.py` — ~6 cases
- Business function calls `path_guard`; escape attempt → `ValueError`.
- DEBUG log line emitted.
- Returns Pydantic-serialized dict.
### `tests/integration/test_files_mcp_routes.py` — ~12 cases
`TestClient(files_app)`, one happy-path POST per tool (9), plus error
coverage: 404 (read missing), 400 (escape), 400 (>1MB), 409 (create hit
existing). Mirrors `tests/integration/test_mcp_routes.py` style for
`spark_executor`.
### `tests/conftest.py` addition
```python
@pytest.fixture(autouse=True)
def _isolate_files_root(monkeypatch, tmp_path):
"""Point files_mcp at a fresh tmp_path for every test."""
monkeypatch.setenv("FILES_MCP_ROOT", str(tmp_path))
from files_mcp.core import path_guard
path_guard.set_root_for_testing(tmp_path.resolve())
yield
```
Total expected: existing 242 tests + ~36 new, full suite under 5 s.
## Risks
1. **Symlink-based escape.** Mitigated by `.resolve()` following the link
before `relative_to` runs. Verified by the symlink test case.
2. **Write race between agents.** Two writers calling `create_file` for the
same path in the same process can still both succeed (each holds a different
`tempfile`); the loser wins the `os.replace`. This is acceptable for the
stated usage (single agent per process) and explicitly called out in
Non-Goals.
3. **`FILES_MCP_ROOT` misconfiguration.** If the env var points at `/`,
any path is "inside" the sandbox. The startup-time check that it is a
directory is the only guard. Document in the README that operators must
choose a narrow root.
4. **Pydantic model drift.** Mirrors existing patterns, low risk.
5. **MCP session affinity.** New endpoint shares the existing constraint
(`GUNICORN_WORKERS=1`). The existing `gunicorn.conf.py` warning already
fires for any workers > 1 regardless of which mount the request targets.
## Open Questions
None — all five clarifying questions resolved before this spec was written
(scope=sandbox; ops=CRUD+list+stat+search+move+copy; content=utf-8 only;
write=atomic+strict create/update; naming=`feat/files-mcp` + `/files-mcp` +
`files_mcp/`).
## Implementation Order (preview)
This is a preview only — the full plan comes from the `writing-plans` skill.
1. `files_mcp/__init__.py` + `core/path_guard.py` + `tests/unit/test_path_guard.py`
2. `files_mcp/core/fs_ops.py` + `tests/unit/test_fs_ops.py`
3. `files_mcp/models.py` + `tools/requests.py` + `tools/files.py`
+ `tests/unit/test_files_tool.py`
4. `files_mcp/server.py` + `tests/integration/test_files_mcp_routes.py`
5. `main.py` one-line append
6. `tests/conftest.py` autouse fixture
7. Run full suite; commit.
## Acceptance Criteria
- [ ] `uv run pytest` passes (existing 242 + new ~36).
- [ ] `uv run main.py` starts cleanly with `FILES_MCP_ROOT` set; fails with a
clear `RuntimeError` if it is unset or points at a non-directory.
- [ ] `tools/list` against `/files-mcp` returns 9 tools with their descriptions.
- [ ] End-to-end smoke: a `TestClient` POST to `create_file` followed by
`read_file` returns the same content.
- [ ] `git diff main` shows changes only in: new `files_mcp/` package, new
tests, the one-line append in `main.py`, and the conftest fixture addition.
- [ ] No new top-level dependencies in `pyproject.toml`.
files_mcp/__init__.py
+28
View File
@@ -0,0 +1,28 @@
# coding=utf-8
"""
@Time :2026/6/30
@Author :tao.chen
"""
from pathlib import Path
from files_mcp.core import path_guard
# Fail fast at process start if FILES_MCP_ROOT is unset / bad. During tests
# the root is injected via set_root_for_testing(), so tolerate the env var
# being unset here rather than crashing at import/collection time.
_ROOT: Path | None
try:
_ROOT = path_guard._init_root()
except RuntimeError:
_ROOT = None
# Re-export the FastAPI app. server.py now exists, so this import is
# unconditional — a missing/corrupt server.py fails loud at import time
# (production-correct behavior). The sandbox root is only needed by the
# tool routes at request time, not at import, so this stays safe to import
# even when FILES_MCP_ROOT is unset (e.g. during pytest collection).
from files_mcp.server import app # noqa: E402
ROOT: Path | None = _ROOT
__all__ = ["app", "ROOT"]
files_mcp/core/__init__.py
+5
View File
@@ -0,0 +1,5 @@
# coding=utf-8
"""
@Time :2026/6/30
@Author :tao.chen
"""
files_mcp/core/fs_ops.py
+177
View File
@@ -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)
files_mcp/core/path_guard.py
+83
View File
@@ -0,0 +1,83 @@
# coding=utf-8
"""
@Time :2026/6/30
@Author :tao.chen
Sandbox: every user-supplied path is resolve()d and checked against a
single frozen root directory. resolve() follows symlinks and normalizes
.., ~, and duplicate slashes, so a single relative_to() check is enough.
Relative inputs are anchored at the root: a bare "inside.txt" means
"<root>/inside.txt", not "<cwd>/inside.txt". Absolute paths and paths
that already contain a separator (e.g. "~/x") are left intact so ~ and
leading-slash inputs behave as expected.
"""
import os
from pathlib import Path
from common.logging import logger
_ROOT: Path | None = None
def _init_root() -> Path:
"""Resolve and freeze the sandbox root from FILES_MCP_ROOT.
Called once at package import time. Raises RuntimeError if the env
var is unset, points at a missing path, or points at a non-directory.
Idempotent: subsequent calls return the cached root.
"""
global _ROOT
if _ROOT is not None:
return _ROOT
raw = os.environ.get("FILES_MCP_ROOT")
if not raw:
raise RuntimeError(
"FILES_MCP_ROOT env var is required (e.g. export "
"FILES_MCP_ROOT=/srv/files-mcp-sandbox)"
)
p = Path(raw).expanduser().resolve()
if not p.is_dir():
raise RuntimeError(
f"FILES_MCP_ROOT={raw!r} resolves to {p} which is not an existing directory"
)
_ROOT = p
logger.info(f"Files MCP sandbox root: {_ROOT}")
return _ROOT
def resolve_and_check(path: str) -> Path:
"""Resolve an arbitrary input path and verify it lies inside the sandbox.
Raises ValueError if the path is empty, contains NUL, or escapes the
root after symlink/.. resolution.
"""
if _ROOT is None:
_init_root()
assert _ROOT is not None
if not path:
raise ValueError("path is empty")
if "\x00" in path:
raise ValueError("path contains NUL byte")
# Anchor bare relative names at the sandbox root so "inside.txt" means
# "<root>/inside.txt", not "<cwd>/inside.txt". Keep user home ("~...") and
# absolute paths untouched so they still resolve / reject as expected.
if not os.path.isabs(path) and not path.startswith("~"):
candidate = _ROOT / path
else:
candidate = Path(path)
p = candidate.expanduser().resolve()
try:
p.relative_to(_ROOT)
except ValueError:
raise ValueError(
f"path escapes sandbox root {_ROOT}: got {p}"
) from None
return p
def set_root_for_testing(p: Path) -> None:
"""Test-only escape hatch. Never call from production code."""
global _ROOT
_ROOT = p.resolve()
os.environ["FILES_MCP_ROOT"] = str(_ROOT)
files_mcp/models.py
+23
View File
@@ -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
files_mcp/server.py
+185
View File
@@ -0,0 +1,185 @@
# coding=utf-8
"""
@Time :2026/6/30
@Author :tao.chen
"""
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from files_mcp.tools.files import (
copy as copy_file,
create_file,
delete_file,
list_dir,
move as move_file,
read_file,
search,
stat,
update_file,
)
from files_mcp.tools.requests import (
CopyRequest,
CreateFileRequest,
DeleteFileRequest,
ListDirRequest,
MoveRequest,
ReadFileRequest,
SearchRequest,
StatRequest,
UpdateFileRequest,
)
app = FastAPI(
title="Files MCP",
version="0.0.1",
description="Files MCP Server — sandboxed file-system CRUD over fastapi-mcp",
)
# --- Exception handlers ---
# KeyError -> 404 (path not found)
# ValueError -> 400 (bad input: escape, non-utf-8, too large, wrong type)
# FileExistsError -> 409 (create/move/copy hit existing dst without overwrite)
#
# Only the third handler is new here; the first two mirror the
# spark_executor.server convention. They live on this app only, so the
# spark_executor handlers are unaffected.
@app.exception_handler(KeyError)
async def _keyerror_handler(_request: Request, exc: KeyError) -> JSONResponse:
return JSONResponse(status_code=404, content={"detail": str(exc)})
@app.exception_handler(ValueError)
async def _valueerror_handler(_request: Request, exc: ValueError) -> JSONResponse:
return JSONResponse(status_code=400, content={"detail": str(exc)})
@app.exception_handler(FileExistsError)
async def _fileexists_handler(_request: Request, exc: FileExistsError) -> JSONResponse:
return JSONResponse(status_code=409, content={"detail": str(exc)})
# --- Routes ---
@app.post(
"/create_file",
operation_id="create_file",
summary="Create a new text file inside the sandbox",
description=(
"Writes `content` to `path` (resolved against FILES_MCP_ROOT). "
"Atomic write via tempfile + os.replace. Default fails with 409 "
"if the file already exists; pass overwrite=true to replace. "
"Content is capped at 1 MB. utf-8 only."
),
)
def _create_file(req: CreateFileRequest):
return create_file(path=req.path, content=req.content, overwrite=req.overwrite)
@app.post(
"/read_file",
operation_id="read_file",
summary="Read a text file inside the sandbox",
description=(
"Returns the full utf-8 content. Files larger than 1 MB and "
"non-utf-8 files are rejected with 400. Directories are rejected "
"with 400; use list_dir instead."
),
)
def _read_file(req: ReadFileRequest):
return read_file(path=req.path)
@app.post(
"/update_file",
operation_id="update_file",
summary="Overwrite an existing text file inside the sandbox",
description=(
"Replaces the file in full. 404 if the path does not exist. "
"Atomic write, 1 MB cap, utf-8 only."
),
)
def _update_file(req: UpdateFileRequest):
return update_file(path=req.path, content=req.content)
@app.post(
"/delete_file",
operation_id="delete_file",
summary="Delete a file or empty directory",
description=(
"404 if the path does not exist. To delete a non-empty directory, "
"pass recursive=true."
),
)
def _delete_file(req: DeleteFileRequest):
return delete_file(path=req.path, recursive=req.recursive)
@app.post(
"/list_dir",
operation_id="list_dir",
summary="List entries in a directory",
description=(
"Returns a list of FileEntry records. recursive=false lists only "
"the directory's direct children; recursive=true descends up to "
"max_depth levels (0 = the directory itself, 1 = one level deep)."
),
)
def _list_dir(req: ListDirRequest):
return list_dir(path=req.path, recursive=req.recursive, max_depth=req.max_depth)
@app.post(
"/stat",
operation_id="stat",
summary="Get a single FileEntry (path, type, size, mtime, mode)",
description="404 if the path does not exist.",
)
def _stat(req: StatRequest):
return stat(path=req.path)
@app.post(
"/search",
operation_id="search",
summary="Find paths matching a glob pattern",
description=(
"Returns the absolute paths of entries inside `path` whose name "
"matches the glob `pattern`. recursive=true descends into "
"subdirectories; recursive=false matches the directory's direct "
"children only."
),
)
def _search(req: SearchRequest):
return search(path=req.path, pattern=req.pattern, recursive=req.recursive)
@app.post(
"/move",
operation_id="move_file",
summary="Rename or move a file or directory",
description=(
"404 if src does not exist. 409 if dst exists and overwrite=false. "
"Cross-filesystem moves are not atomic."
),
)
def _move(req: MoveRequest):
return move_file(src=req.src, dst=req.dst, overwrite=req.overwrite)
@app.post(
"/copy",
operation_id="copy_file",
summary="Copy a file or directory",
description=(
"404 if src does not exist. 409 if dst exists and overwrite=false. "
"For files, metadata is preserved (shutil.copy2). For directories, "
"the entire tree is copied (shutil.copytree)."
),
)
def _copy(req: CopyRequest):
return copy_file(src=req.src, dst=req.dst, overwrite=req.overwrite)
files_mcp/tools/__init__.py
+5
View File
@@ -0,0 +1,5 @@
# coding=utf-8
"""
@Time :2026/6/30
@Author :tao.chen
"""
files_mcp/tools/files.py
+76
View File
@@ -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()
files_mcp/tools/requests.py
+58
View File
@@ -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
main.py
+6
View File
@@ -10,6 +10,7 @@ from fastapi import FastAPI
from common.factory import init_mcp_server
from common.logging import logger
from files_mcp import app as files_app
from spark_executor import app as spark_executor_app
@@ -39,6 +40,11 @@ MCP_SERVICES: list[McpService] = [
app=spark_executor_app,
mount_path="/spark-executor-mcp",
),
McpService(
name="files",
app=files_app,
mount_path="/files-mcp",
),
]
tests/conftest.py
+17
View File
@@ -6,3 +6,20 @@ from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
import pytest
@pytest.fixture(autouse=True)
def _isolate_files_mcp_root(tmp_path, monkeypatch):
"""Point files_mcp at a fresh tmp_path for every test.
Sets FILES_MCP_ROOT and rebinds files_mcp.core.path_guard's module-level
_ROOT, so each test starts with an empty sandbox. Spark_executor tests
are unaffected by this fixture — they don't touch the files_mcp sandbox.
"""
monkeypatch.setenv("FILES_MCP_ROOT", str(tmp_path))
from files_mcp.core import path_guard
path_guard.set_root_for_testing(tmp_path.resolve())
yield
tests/integration/test_files_mcp_routes.py
+141
View File
@@ -0,0 +1,141 @@
# coding=utf-8
"""
@Time :2026/6/30
@Author :tao.chen
End-to-end HTTP tests for files_mcp.server.app. Uses FastAPI TestClient
directly (no MCP transport) — same pattern as
tests/integration/test_mcp_routes.py.
"""
from fastapi.testclient import TestClient
from files_mcp.core import path_guard
from files_mcp.server import app
def _client(tmp_path) -> TestClient:
import os
os.environ["FILES_MCP_ROOT"] = str(tmp_path)
path_guard.set_root_for_testing(tmp_path.resolve())
return TestClient(app)
class TestCrud:
def test_create_then_read(self, tmp_path):
with _client(tmp_path) as c:
r = c.post("/create_file", json={"path": "a.txt", "content": "hello"})
assert r.status_code == 200
assert r.json() == {"path": str(tmp_path / "a.txt"), "status": "CREATED", "size": 5}
r = c.post("/read_file", json={"path": "a.txt"})
assert r.status_code == 200
assert r.json()["content"] == "hello"
def test_update_existing(self, tmp_path):
with _client(tmp_path) as c:
c.post("/create_file", json={"path": "a.txt", "content": "old"})
r = c.post("/update_file", json={"path": "a.txt", "content": "new"})
assert r.status_code == 200
assert r.json()["status"] == "UPDATED"
r = c.post("/read_file", json={"path": "a.txt"})
assert r.json()["content"] == "new"
def test_delete(self, tmp_path):
with _client(tmp_path) as c:
c.post("/create_file", json={"path": "a.txt", "content": "x"})
r = c.post("/delete_file", json={"path": "a.txt"})
assert r.status_code == 200
assert r.json()["status"] == "DELETED"
r = c.post("/read_file", json={"path": "a.txt"})
assert r.status_code == 404
class TestErrors:
def test_create_existing_returns_409(self, tmp_path):
with _client(tmp_path) as c:
c.post("/create_file", json={"path": "a.txt", "content": "x"})
r = c.post("/create_file", json={"path": "a.txt", "content": "y"})
assert r.status_code == 409
def test_update_missing_returns_404(self, tmp_path):
with _client(tmp_path) as c:
r = c.post("/update_file", json={"path": "nope.txt", "content": "x"})
assert r.status_code == 404
def test_escape_returns_400(self, tmp_path):
with _client(tmp_path) as c:
r = c.post("/read_file", json={"path": "/etc/passwd"})
assert r.status_code == 400
assert "escapes sandbox" in r.json()["detail"]
def test_too_large_returns_400(self, tmp_path):
with _client(tmp_path) as c:
r = c.post(
"/create_file",
json={"path": "big.txt", "content": "x" * (1024 * 1024 + 1)},
)
assert r.status_code == 400
assert "too large" in r.json()["detail"]
def test_non_utf8_returns_400(self, tmp_path):
with _client(tmp_path) as c:
(tmp_path / "bin").write_bytes(b"\xff\xfe\x00")
r = c.post("/read_file", json={"path": "bin"})
assert r.status_code == 400
assert "utf-8" in r.json()["detail"]
class TestIntrospection:
def test_list_dir(self, tmp_path):
with _client(tmp_path) as c:
(tmp_path / "a.txt").write_text("a", encoding="utf-8")
(tmp_path / "b").mkdir()
r = c.post("/list_dir", json={"path": ".", "recursive": False, "max_depth": 1})
assert r.status_code == 200
names = sorted(e["name"] for e in r.json()["entries"])
assert names == ["a.txt", "b"]
def test_stat(self, tmp_path):
with _client(tmp_path) as c:
(tmp_path / "f.txt").write_text("hi", encoding="utf-8")
r = c.post("/stat", json={"path": "f.txt"})
assert r.status_code == 200
assert r.json()["type"] == "file"
assert r.json()["size"] == 2
def test_search(self, tmp_path):
with _client(tmp_path) as c:
(tmp_path / "a.py").write_text("a", encoding="utf-8")
(tmp_path / "b.txt").write_text("b", encoding="utf-8")
r = c.post("/search", json={"path": ".", "pattern": "*.py", "recursive": False})
assert r.status_code == 200
assert len(r.json()["matches"]) == 1
class TestMoveCopy:
def test_move(self, tmp_path):
with _client(tmp_path) as c:
c.post("/create_file", json={"path": "a.txt", "content": "x"})
r = c.post("/move", json={"src": "a.txt", "dst": "b.txt"})
assert r.status_code == 200
assert r.json()["status"] == "MOVED"
r = c.post("/read_file", json={"path": "b.txt"})
assert r.json()["content"] == "x"
def test_move_existing_dst_409(self, tmp_path):
with _client(tmp_path) as c:
c.post("/create_file", json={"path": "a.txt", "content": "x"})
c.post("/create_file", json={"path": "b.txt", "content": "y"})
r = c.post("/move", json={"src": "a.txt", "dst": "b.txt"})
assert r.status_code == 409
def test_copy(self, tmp_path):
with _client(tmp_path) as c:
c.post("/create_file", json={"path": "a.txt", "content": "x"})
r = c.post("/copy", json={"src": "a.txt", "dst": "b.txt"})
assert r.status_code == 200
assert r.json()["status"] == "COPIED"
# source unchanged
r = c.post("/read_file", json={"path": "a.txt"})
assert r.json()["content"] == "x"
r = c.post("/read_file", json={"path": "b.txt"})
assert r.json()["content"] == "x"
tests/unit/test_files_tool.py
+116
View File
@@ -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"
tests/unit/test_fs_ops.py
+298
View File
@@ -0,0 +1,298 @@
# coding=utf-8
"""
@Time :2026/6/30
@Author :tao.chen
"""
from pathlib import Path
import pytest
from files_mcp.core import fs_ops
from files_mcp.models import FileEntry, OperationResult
# --- create ------------------------------------------------------------
class TestCreate:
def test_creates_new_file(self, tmp_path: Path):
target = tmp_path / "new.txt"
result = fs_ops.create(target, "hello", overwrite=False)
assert isinstance(result, OperationResult)
assert result.status == "CREATED"
assert result.size == 5
assert target.read_text(encoding="utf-8") == "hello"
def test_overwrite_true_replaces(self, tmp_path: Path):
target = tmp_path / "f.txt"
target.write_text("old", encoding="utf-8")
result = fs_ops.create(target, "new", overwrite=True)
assert result.status == "CREATED"
assert target.read_text(encoding="utf-8") == "new"
def test_overwrite_false_raises(self, tmp_path: Path):
target = tmp_path / "f.txt"
target.write_text("old", encoding="utf-8")
with pytest.raises(FileExistsError, match="already exists"):
fs_ops.create(target, "new", overwrite=False)
def test_too_large_raises(self, tmp_path: Path):
target = tmp_path / "big.txt"
with pytest.raises(ValueError, match="content too large"):
fs_ops.create(target, "x" * (fs_ops.MAX_BYTES + 1), overwrite=False)
def test_atomic_write_leaves_no_tmp_on_success(self, tmp_path: Path):
target = tmp_path / "f.txt"
fs_ops.create(target, "hi", overwrite=False)
# No leftover .tmp_* in the directory
leftovers = [p for p in tmp_path.iterdir() if p.name.startswith(".tmp_")]
assert leftovers == []
# --- read --------------------------------------------------------------
class TestRead:
def test_reads_existing(self, tmp_path: Path):
f = tmp_path / "f.txt"
f.write_text("hello", encoding="utf-8")
out = fs_ops.read(f)
assert out == {
"path": str(f),
"content": "hello",
"size": 5,
"encoding": "utf-8",
}
def test_missing_raises_keyerror(self, tmp_path: Path):
with pytest.raises(KeyError, match="not found"):
fs_ops.read(tmp_path / "nope.txt")
def test_directory_raises(self, tmp_path: Path):
with pytest.raises(ValueError, match="not a regular file"):
fs_ops.read(tmp_path)
def test_too_large_raises(self, tmp_path: Path):
f = tmp_path / "big.bin"
f.write_bytes(b"x" * (fs_ops.MAX_BYTES + 1))
with pytest.raises(ValueError, match="file too large"):
fs_ops.read(f)
def test_non_utf8_raises(self, tmp_path: Path):
f = tmp_path / "binary.bin"
f.write_bytes(b"\xff\xfe\x00\x01")
with pytest.raises(ValueError, match="not valid utf-8"):
fs_ops.read(f)
# --- update ------------------------------------------------------------
class TestUpdate:
def test_updates_existing(self, tmp_path: Path):
f = tmp_path / "f.txt"
f.write_text("old", encoding="utf-8")
result = fs_ops.update(f, "new")
assert result.status == "UPDATED"
assert result.size == 3
assert f.read_text(encoding="utf-8") == "new"
def test_missing_raises_keyerror(self, tmp_path: Path):
with pytest.raises(KeyError, match="not found"):
fs_ops.update(tmp_path / "nope.txt", "x")
def test_too_large_raises(self, tmp_path: Path):
f = tmp_path / "f.txt"
f.write_text("old", encoding="utf-8")
with pytest.raises(ValueError, match="content too large"):
fs_ops.update(f, "x" * (fs_ops.MAX_BYTES + 1))
# --- delete ------------------------------------------------------------
class TestDelete:
def test_deletes_file(self, tmp_path: Path):
f = tmp_path / "f.txt"
f.write_text("x", encoding="utf-8")
result = fs_ops.delete(f, recursive=False)
assert result.status == "DELETED"
assert result.size is None
assert not f.exists()
def test_missing_raises_keyerror(self, tmp_path: Path):
with pytest.raises(KeyError, match="not found"):
fs_ops.delete(tmp_path / "nope.txt", recursive=False)
def test_non_empty_dir_raises_without_recursive(self, tmp_path: Path):
d = tmp_path / "d"
d.mkdir()
(d / "child").write_text("x", encoding="utf-8")
with pytest.raises(ValueError, match="not empty"):
fs_ops.delete(d, recursive=False)
def test_recursive_deletes_tree(self, tmp_path: Path):
d = tmp_path / "d"
d.mkdir()
(d / "child").write_text("x", encoding="utf-8")
(d / "sub").mkdir()
(d / "sub" / "leaf").write_text("y", encoding="utf-8")
fs_ops.delete(d, recursive=True)
assert not d.exists()
def test_empty_dir_ok_without_recursive(self, tmp_path: Path):
d = tmp_path / "empty"
d.mkdir()
fs_ops.delete(d, recursive=False)
assert not d.exists()
# --- list_dir ----------------------------------------------------------
class TestListDir:
def test_single_level(self, tmp_path: Path):
(tmp_path / "a.txt").write_text("a", encoding="utf-8")
(tmp_path / "b").mkdir()
entries = fs_ops.list_dir(tmp_path, recursive=False, max_depth=1)
names = sorted(e.name for e in entries)
assert names == ["a.txt", "b"]
assert all(isinstance(e, FileEntry) for e in entries)
def test_recursive_respects_max_depth(self, tmp_path: Path):
(tmp_path / "a.txt").write_text("a", encoding="utf-8")
(tmp_path / "sub").mkdir()
(tmp_path / "sub" / "b.txt").write_text("b", encoding="utf-8")
(tmp_path / "sub" / "deep").mkdir()
(tmp_path / "sub" / "deep" / "c.txt").write_text("c", encoding="utf-8")
depth0 = fs_ops.list_dir(tmp_path, recursive=True, max_depth=0)
assert sorted(e.name for e in depth0) == ["a.txt", "sub"]
depth1 = fs_ops.list_dir(tmp_path, recursive=True, max_depth=1)
assert sorted(e.name for e in depth1) == ["a.txt", "b.txt", "deep", "sub"]
def test_missing_raises(self, tmp_path: Path):
with pytest.raises(KeyError, match="not found"):
fs_ops.list_dir(tmp_path / "nope", recursive=False, max_depth=1)
def test_file_raises(self, tmp_path: Path):
f = tmp_path / "f.txt"
f.write_text("x", encoding="utf-8")
with pytest.raises(ValueError, match="not a directory"):
fs_ops.list_dir(f, recursive=False, max_depth=1)
def test_negative_depth_raises(self, tmp_path: Path):
with pytest.raises(ValueError, match="max_depth"):
fs_ops.list_dir(tmp_path, recursive=True, max_depth=-1)
# --- stat --------------------------------------------------------------
class TestStat:
def test_file(self, tmp_path: Path):
f = tmp_path / "f.txt"
f.write_text("hello", encoding="utf-8")
e = fs_ops.stat(f)
assert e.name == "f.txt"
assert e.type == "file"
assert e.size == 5
def test_directory(self, tmp_path: Path):
e = fs_ops.stat(tmp_path)
assert e.type == "directory"
assert e.size == 0
def test_missing_raises(self, tmp_path: Path):
with pytest.raises(KeyError, match="not found"):
fs_ops.stat(tmp_path / "nope")
# --- search ------------------------------------------------------------
class TestSearch:
def test_simple_glob(self, tmp_path: Path):
(tmp_path / "a.py").write_text("a", encoding="utf-8")
(tmp_path / "b.txt").write_text("b", encoding="utf-8")
(tmp_path / "c.py").write_text("c", encoding="utf-8")
matches = fs_ops.search(tmp_path, "*.py", recursive=False)
assert sorted(Path(m).name for m in matches) == ["a.py", "c.py"]
def test_recursive_glob(self, tmp_path: Path):
(tmp_path / "sub").mkdir()
(tmp_path / "sub" / "x.py").write_text("x", encoding="utf-8")
(tmp_path / "a.py").write_text("a", encoding="utf-8")
matches = fs_ops.search(tmp_path, "*.py", recursive=True)
assert sorted(Path(m).name for m in matches) == ["a.py", "x.py"]
def test_no_match_returns_empty(self, tmp_path: Path):
(tmp_path / "a.txt").write_text("a", encoding="utf-8")
assert fs_ops.search(tmp_path, "*.py", recursive=False) == []
def test_missing_raises(self, tmp_path: Path):
with pytest.raises(KeyError, match="not found"):
fs_ops.search(tmp_path / "nope", "*", recursive=False)
def test_non_directory_raises(self, tmp_path: Path):
f = tmp_path / "f.txt"
f.write_text("x", encoding="utf-8")
with pytest.raises(ValueError, match="not a directory"):
fs_ops.search(f, "*", recursive=False)
# --- move / copy -------------------------------------------------------
class TestMove:
def test_rename(self, tmp_path: Path):
src = tmp_path / "a.txt"
src.write_text("hi", encoding="utf-8")
result = fs_ops.move(src, tmp_path / "b.txt", overwrite=False)
assert result.status == "MOVED"
assert result.path == str(tmp_path / "b.txt")
assert not src.exists()
assert (tmp_path / "b.txt").read_text(encoding="utf-8") == "hi"
def test_missing_src_raises(self, tmp_path: Path):
with pytest.raises(KeyError, match="src not found"):
fs_ops.move(tmp_path / "nope", tmp_path / "dst", overwrite=False)
def test_existing_dst_raises_without_overwrite(self, tmp_path: Path):
src = tmp_path / "a.txt"
src.write_text("a", encoding="utf-8")
dst = tmp_path / "b.txt"
dst.write_text("b", encoding="utf-8")
with pytest.raises(FileExistsError, match="dst already exists"):
fs_ops.move(src, dst, overwrite=False)
def test_overwrite_replaces(self, tmp_path: Path):
src = tmp_path / "a.txt"
src.write_text("new", encoding="utf-8")
dst = tmp_path / "b.txt"
dst.write_text("old", encoding="utf-8")
fs_ops.move(src, dst, overwrite=True)
assert dst.read_text(encoding="utf-8") == "new"
assert not src.exists()
class TestCopy:
def test_copies_file(self, tmp_path: Path):
src = tmp_path / "a.txt"
src.write_text("hi", encoding="utf-8")
result = fs_ops.copy(src, tmp_path / "b.txt", overwrite=False)
assert result.status == "COPIED"
assert result.size == 2
assert src.read_text(encoding="utf-8") == "hi" # source unchanged
assert (tmp_path / "b.txt").read_text(encoding="utf-8") == "hi"
def test_missing_src_raises(self, tmp_path: Path):
with pytest.raises(KeyError, match="src not found"):
fs_ops.copy(tmp_path / "nope", tmp_path / "dst", overwrite=False)
def test_existing_dst_raises_without_overwrite(self, tmp_path: Path):
src = tmp_path / "a.txt"
src.write_text("a", encoding="utf-8")
dst = tmp_path / "b.txt"
dst.write_text("b", encoding="utf-8")
with pytest.raises(FileExistsError, match="dst already exists"):
fs_ops.copy(src, dst, overwrite=False)
def test_copies_directory(self, tmp_path: Path):
src = tmp_path / "d"
src.mkdir()
(src / "child").write_text("x", encoding="utf-8")
result = fs_ops.copy(src, tmp_path / "d2", overwrite=False)
assert result.status == "COPIED"
assert (tmp_path / "d2" / "child").read_text(encoding="utf-8") == "x"
tests/unit/test_path_guard.py
+84
View File
@@ -0,0 +1,84 @@
# coding=utf-8
"""
@Time :2026/6/30
@Author :tao.chen
"""
from pathlib import Path
import pytest
from files_mcp.core import path_guard
@pytest.fixture
def sandbox_root(tmp_path, monkeypatch):
"""Fresh tmp_path-based sandbox root for each test.
Pre-populates a regular file, a nested file inside a subdir, and a
symlink that points outside the sandbox (for the symlink-escape test).
"""
monkeypatch.setenv("FILES_MCP_ROOT", str(tmp_path))
path_guard.set_root_for_testing(tmp_path.resolve())
(tmp_path / "inside.txt").write_text("hi", encoding="utf-8")
(tmp_path / "subdir").mkdir()
(tmp_path / "subdir" / "nested.txt").write_text("nested", encoding="utf-8")
(tmp_path / "escape_link").symlink_to("/etc/passwd")
return tmp_path.resolve()
class TestResolveAndCheck:
def test_accepts_inside_path(self, sandbox_root):
p = path_guard.resolve_and_check("inside.txt")
assert p == sandbox_root / "inside.txt"
def test_accepts_nested_path(self, sandbox_root):
p = path_guard.resolve_and_check("subdir/nested.txt")
assert p == sandbox_root / "subdir" / "nested.txt"
def test_accepts_dot_segments(self, sandbox_root):
p = path_guard.resolve_and_check("subdir/../inside.txt")
assert p == sandbox_root / "inside.txt"
def test_rejects_double_dot_escape(self, sandbox_root):
with pytest.raises(ValueError, match="escapes sandbox"):
path_guard.resolve_and_check("../etc/passwd")
def test_rejects_absolute_outside_path(self, sandbox_root):
with pytest.raises(ValueError, match="escapes sandbox"):
path_guard.resolve_and_check("/etc/passwd")
def test_rejects_mixed_escape(self, sandbox_root):
with pytest.raises(ValueError, match="escapes sandbox"):
path_guard.resolve_and_check("subdir/../../etc/passwd")
def test_rejects_symlink_escape(self, sandbox_root):
# escape_link -> /etc/passwd; resolve() follows the link first.
with pytest.raises(ValueError, match="escapes sandbox"):
path_guard.resolve_and_check("escape_link")
def test_rejects_empty(self, sandbox_root):
with pytest.raises(ValueError, match="path is empty"):
path_guard.resolve_and_check("")
def test_rejects_nul_byte(self, sandbox_root):
with pytest.raises(ValueError, match="NUL byte"):
path_guard.resolve_and_check("foo\x00bar")
class TestInitRoot:
def test_init_root_raises_if_unset(self, monkeypatch):
monkeypatch.delenv("FILES_MCP_ROOT", raising=False)
path_guard._ROOT = None
with pytest.raises(RuntimeError, match="FILES_MCP_ROOT"):
path_guard._init_root()
def test_init_root_raises_if_not_directory(self, tmp_path, monkeypatch):
monkeypatch.setenv("FILES_MCP_ROOT", str(tmp_path / "missing"))
path_guard._ROOT = None
with pytest.raises(RuntimeError, match="not an existing directory"):
path_guard._init_root()
def test_init_root_is_idempotent(self, sandbox_root):
first = path_guard._ROOT
path_guard._init_root()
assert path_guard._ROOT == first