fix: validate FileSystemProvider root; use fresh dict on reload

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

src/fastmcp/server/providers/filesystem.py

@@ -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)

src/fastmcp/server/providers/filesystem_discovery.py

@@ -8,6 +8,8 @@ This module provides functions to:
 
 from __future__ import annotations
 
+import contextlib
+import hashlib
 import importlib.util
 import sys
 from dataclasses import dataclass, field
@@ -68,10 +70,16 @@ def _is_package_dir(directory: Path) -> bool:
     return (directory / "__init__.py").exists()
 
 
-def _find_package_root(file_path: Path) -> Path | None:
+def _find_package_root(file_path: Path, stop_at: Path | None = None) -> Path | None:
     """Find the root of the package containing this file.
 
-    Walks up the directory tree until we find a directory without __init__.py.
+    Walks up the directory tree until we find a directory without __init__.py,
+    but never above stop_at (the provider root). This prevents escaping into
+    ancestor packages when the provider is nested inside a larger Python project.
+
+    Args:
+        file_path: Path to the Python file.
+        stop_at: Do not walk above this directory. Typically the provider root.
 
     Returns:
         The package root directory, or None if not in a package.
@@ -80,6 +88,8 @@ def _find_package_root(file_path: Path) -> Path | None:
     package_root = None
 
     while current != current.parent:  # Stop at filesystem root
+        if stop_at is not None and current == stop_at.parent:
+            break  # Don't escape above the provider root
         if _is_package_dir(current):
             package_root = current
             current = current.parent
@@ -106,15 +116,22 @@ def _compute_module_name(file_path: Path, package_root: Path) -> str:
     return ".".join(parts)
 
 
-def import_module_from_file(file_path: Path) -> ModuleType:
+def import_module_from_file(
+    file_path: Path, provider_root: Path | None = None
+) -> ModuleType:
     """Import a Python file as a module.
 
     If the file is part of a package (directory has __init__.py), imports
     it as a proper package member (relative imports work). Otherwise,
     imports directly using spec_from_file_location.
 
+    sys.path is modified only for the duration of the import and restored
+    immediately after, so no permanent pollution occurs.
+
     Args:
         file_path: Path to the Python file.
+        provider_root: The provider's root directory. Prevents package root
+            discovery from walking above this boundary into ancestor packages.
 
     Returns:
         The imported module.
@@ -123,22 +140,24 @@ def import_module_from_file(file_path: Path) -> ModuleType:
         ImportError: If the module cannot be imported.
     """
     file_path = file_path.resolve()
+    if provider_root is not None:
+        provider_root = provider_root.resolve()
 
     # Check if this file is part of a package
-    package_root = _find_package_root(file_path)
+    package_root = _find_package_root(file_path, stop_at=provider_root)
 
     if package_root is not None:
         # Import as part of a package
         module_name = _compute_module_name(file_path, package_root)
 
-        # Ensure package root's parent is in sys.path
+        # Temporarily add package root's parent to sys.path for the import
         package_parent = str(package_root.parent)
-        if package_parent not in sys.path:
+        path_added = package_parent not in sys.path
+        if path_added:
             sys.path.insert(0, package_parent)
 
-        # Import using standard import machinery
-        # If already imported, reload to pick up changes (for reload mode)
         try:
+            # If already imported, reload to pick up changes (for reload mode)
             if module_name in sys.modules:
                 return importlib.reload(sys.modules[module_name])
             return importlib.import_module(module_name)
@@ -146,30 +165,71 @@ def import_module_from_file(file_path: Path) -> ModuleType:
             raise ImportError(
                 f"Failed to import {module_name} from {file_path}: {e}"
             ) from e
+        finally:
+            if path_added:
+                with contextlib.suppress(ValueError):
+                    sys.path.remove(package_parent)
     else:
         # Import directly using spec_from_file_location
-        module_name = file_path.stem
-
-        # Ensure parent directory is in sys.path for imports
+        stem = file_path.stem
         parent_dir = str(file_path.parent)
-        if parent_dir not in sys.path:
+
+        # Determine the sys.modules key. Prefer the bare stem (so that sibling
+        # imports like `import helpers` resolve correctly), but fall back to a
+        # private collision-safe key if the bare stem is already claimed by
+        # something else (stdlib, a third-party package, or another provider file
+        # from a different directory).
+        existing = sys.modules.get(stem)
+        if existing is not None and getattr(existing, "__file__", None) != str(
+            file_path
+        ):
+            module_name = f"_fastmcp_{stem}_{hashlib.sha1(str(file_path).encode()).hexdigest()[:12]}"
+        else:
+            module_name = stem
+
+        # Temporarily add parent to sys.path so module-level sibling imports resolve.
+        # Safe to remove after exec_module: all top-level imports are resolved by then,
+        # and sibling files imported as side effects are already in sys.modules.
+        path_added = parent_dir not in sys.path
+        if path_added:
             sys.path.insert(0, parent_dir)
 
-        spec = importlib.util.spec_from_file_location(module_name, file_path)
-        if spec is None or spec.loader is None:
-            raise ImportError(f"Cannot load spec for {file_path}")
-
-        module = importlib.util.module_from_spec(spec)
-        sys.modules[module_name] = module
-
         try:
-            spec.loader.exec_module(module)
-        except Exception as e:
-            # Clean up sys.modules on failure
-            sys.modules.pop(module_name, None)
-            raise ImportError(f"Failed to execute module {file_path}: {e}") from e
+            spec = importlib.util.spec_from_file_location(module_name, file_path)
+            if spec is None or spec.loader is None:
+                raise ImportError(f"Cannot load spec for {file_path}")
 
-        return module
+            existing = sys.modules.get(module_name)
+            if existing is not None:
+                # Re-exec in place rather than importlib.reload: reload() re-finds
+                # the module by name via sys.path, which fails for private keys
+                # (the file is tool.py, not _fastmcp_tool_xxx.py).
+                existing.__spec__ = spec
+                existing.__loader__ = spec.loader
+                existing.__file__ = str(file_path)
+                try:
+                    spec.loader.exec_module(existing)
+                except Exception as e:
+                    raise ImportError(
+                        f"Failed to reload module {file_path}: {e}"
+                    ) from e
+                return existing
+
+            module = importlib.util.module_from_spec(spec)
+            sys.modules[module_name] = module
+
+            try:
+                spec.loader.exec_module(module)
+            except Exception as e:
+                # Clean up sys.modules on failure
+                sys.modules.pop(module_name, None)
+                raise ImportError(f"Failed to execute module {file_path}: {e}") from e
+
+            return module
+        finally:
+            if path_added:
+                with contextlib.suppress(ValueError):
+                    sys.path.remove(parent_dir)
 
 
 def extract_components(module: ModuleType) -> list[FastMCPComponent]:
@@ -316,10 +376,7 @@ def discover_and_import(root: Path) -> DiscoveryResult:
 
     for file_path in discover_files(root):
         try:
-            module = import_module_from_file(file_path)
-        except ImportError as e:
-            result.failed_files[file_path] = str(e)
-            continue
+            module = import_module_from_file(file_path, provider_root=root)
         except Exception as e:
             result.failed_files[file_path] = str(e)
             continue

tests/fs/test_discovery.py

@@ -1,5 +1,6 @@
 """Tests for filesystem discovery module."""
 
+import sys
 from pathlib import Path
 
 from fastmcp.prompts.base import Prompt
@@ -485,3 +486,101 @@ def my_tool(x: str) -> str:
         assert components[0].version is None
         meta = components[0].get_meta()
         assert "version" not in meta["fastmcp"]
+
+
+class TestImportMachineryFixes:
+    """Tests for import machinery correctness: sys.path cleanup, sys.modules safety, package root boundary."""
+
+    def test_syspath_not_polluted_after_import(self, tmp_path: Path):
+        """sys.path should not contain the file's parent after import_module_from_file returns."""
+        (tmp_path / "mymod.py").write_text("VALUE = 1")
+        path_before = list(sys.path)
+        import_module_from_file(tmp_path / "mymod.py")
+        assert sys.path == path_before
+
+    def test_syspath_not_polluted_after_package_import(self, tmp_path: Path):
+        """sys.path should not contain the package root's parent after a package import."""
+        pkg = tmp_path / "mypkg_syspath"
+        pkg.mkdir()
+        (pkg / "__init__.py").write_text("")
+        (pkg / "mod.py").write_text("VALUE = 2")
+        path_before = list(sys.path)
+        import_module_from_file(pkg / "mod.py", provider_root=tmp_path)
+        assert sys.path == path_before
+
+    def test_stdlib_not_shadowed_by_same_named_file(self, tmp_path: Path):
+        """A provider file named json.py must not overwrite sys.modules['json']."""
+        import json as stdlib_json
+
+        saved = sys.modules["json"]
+        try:
+            (tmp_path / "json.py").write_text(
+                "from fastmcp.tools import tool\n@tool\ndef parse(): return 'provider'"
+            )
+            import_module_from_file(tmp_path / "json.py")
+            assert sys.modules.get("json") is stdlib_json
+        finally:
+            sys.modules["json"] = saved
+
+    def test_same_stem_files_get_independent_modules(self, tmp_path: Path):
+        """Two files with the same stem in different directories must not collide in sys.modules.
+
+        The first-imported file keeps the bare stem key; the second gets a private key.
+        Both modules must be independently accessible with correct content.
+        """
+        dir_a = tmp_path / "a"
+        dir_b = tmp_path / "b"
+        dir_a.mkdir()
+        dir_b.mkdir()
+        (dir_a / "helpers.py").write_text("ORIGIN = 'a'")
+        (dir_b / "helpers.py").write_text("ORIGIN = 'b'")
+
+        mod_a = import_module_from_file(dir_a / "helpers.py")
+        mod_b = import_module_from_file(dir_b / "helpers.py")
+
+        assert mod_a.ORIGIN == "a"
+        assert mod_b.ORIGIN == "b"
+        # The first module retains the bare stem key; the second uses a private key.
+        # They must be distinct objects — the second import must not have clobbered the first.
+        assert mod_a is not mod_b
+        assert sys.modules.get("helpers") is not mod_b
+
+    def test_package_root_bounded_by_provider_root(self, tmp_path: Path):
+        """When the provider root is nested inside a larger package, import_module_from_file
+        with provider_root must not escape into ancestor packages.
+
+        The generated module name should be relative to the provider root (e.g. "myprovider.tools"),
+        not to an ancestor package (e.g. "myproject.myprovider.tools"), and tmp_path (the
+        ancestor's parent) must not be added to sys.path.
+        """
+        # Use a name that won't collide with any installed package
+        project = tmp_path / "myproject"
+        project.mkdir()
+        (project / "__init__.py").write_text("")
+        provider = project / "myprovider"
+        provider.mkdir()
+        (provider / "__init__.py").write_text("")
+        (provider / "tools.py").write_text("VALUE = 42")
+
+        path_before = set(sys.path)
+        mod = import_module_from_file(provider / "tools.py", provider_root=provider)
+        path_after = set(sys.path)
+
+        # Module was correctly imported
+        assert mod.VALUE == 42
+        # sys.path should not contain tmp_path (the ancestor's grandparent);
+        # that would only happen if the package root escaped past the provider boundary
+        assert str(tmp_path) not in (path_after - path_before)
+        # The module name is bounded to the provider root, not "myproject.myprovider.tools"
+        assert mod.__name__ == "myprovider.tools"
+
+    def test_non_package_reload_returns_updated_content(self, tmp_path: Path):
+        """Re-importing a non-package file should reflect file changes (exec_module path)."""
+        f = tmp_path / "reloadable_np.py"
+        f.write_text("VALUE = 'original'")
+        mod = import_module_from_file(f)
+        assert mod.VALUE == "original"
+
+        f.write_text("VALUE = 'updated'")
+        mod2 = import_module_from_file(f)
+        assert mod2.VALUE == "updated"

tests/fs/test_provider.py

@@ -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"