mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-09 07:09:11 +02:00
fix: validate FileSystemProvider root; use fresh dict on reload
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
parent
4c874609e9
commit
be88708c9f
2 changed files with 26 additions and 4 deletions
|
|
@ -89,6 +89,14 @@ class FileSystemProvider(LocalProvider):
|
|||
) -> None:
|
||||
super().__init__(on_duplicate="replace")
|
||||
self._root = Path(root).resolve()
|
||||
if not self._root.exists():
|
||||
raise FileNotFoundError(
|
||||
f"FileSystemProvider root does not exist: {self._root}"
|
||||
)
|
||||
if not self._root.is_dir():
|
||||
raise NotADirectoryError(
|
||||
f"FileSystemProvider root is not a directory: {self._root}"
|
||||
)
|
||||
self._reload = reload
|
||||
self._loaded = False
|
||||
# Track files we've warned about: path -> mtime when warned
|
||||
|
|
@ -102,10 +110,6 @@ class FileSystemProvider(LocalProvider):
|
|||
|
||||
def _load_components(self) -> None:
|
||||
"""Discover and register all components from the filesystem."""
|
||||
# Clear existing components if reloading
|
||||
if self._loaded:
|
||||
self._components.clear()
|
||||
|
||||
result = discover_and_import(self._root)
|
||||
|
||||
# Log warnings for failed files (only once per file version)
|
||||
|
|
@ -126,6 +130,9 @@ class FileSystemProvider(LocalProvider):
|
|||
for fp in successful_files:
|
||||
self._warned_files.pop(fp, None)
|
||||
|
||||
# Fresh dict (not .clear()) so in-flight iterators over the old dict aren't disrupted.
|
||||
self._components = {}
|
||||
|
||||
for file_path, component in result.components:
|
||||
try:
|
||||
self._register_component(component)
|
||||
|
|
|
|||
|
|
@ -3,6 +3,8 @@
|
|||
import time
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from fastmcp import FastMCP
|
||||
from fastmcp.client import Client
|
||||
from fastmcp.server.providers import FileSystemProvider
|
||||
|
|
@ -16,6 +18,19 @@ class TestFileSystemProvider:
|
|||
provider = FileSystemProvider(tmp_path)
|
||||
assert repr(provider).startswith("FileSystemProvider")
|
||||
|
||||
def test_provider_raises_on_missing_root(self, tmp_path: Path):
|
||||
"""Provider should raise FileNotFoundError for non-existent root."""
|
||||
missing = tmp_path / "does_not_exist"
|
||||
with pytest.raises(FileNotFoundError, match="does not exist"):
|
||||
FileSystemProvider(missing)
|
||||
|
||||
def test_provider_raises_on_file_root(self, tmp_path: Path):
|
||||
"""Provider should raise NotADirectoryError when root is a file."""
|
||||
file_path = tmp_path / "not_a_dir.py"
|
||||
file_path.write_text("x = 1")
|
||||
with pytest.raises(NotADirectoryError, match="not a directory"):
|
||||
FileSystemProvider(file_path)
|
||||
|
||||
def test_provider_discovers_tools(self, tmp_path: Path):
|
||||
"""Provider should discover @tool decorated functions."""
|
||||
tools_dir = tmp_path / "tools"
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue