ceos/fastmcp
1
0
Fork 0
mirror of https://github.com/PrefectHQ/fastmcp.git synced 2026-08-09 07:09:11 +02:00
Code Issues Projects Releases Packages Wiki Activity

fix: filesystem provider import machinery (#3626)

Browse source 
* fix: filesystem provider import machinery

- Temporary sys.path entries (both package and non-package mode) now removed
  immediately after exec_module via try/finally, eliminating permanent process-wide pollution
- Non-package files use bare stem as sys.modules key only if unclaimed; falls back
  to private hash-based key to prevent stdlib shadowing (e.g. json.py clobbering json)
- Reload of private-key modules uses spec.loader.exec_module directly instead of
  importlib.reload, which cannot find files by their private synthetic name
- _find_package_root gains stop_at parameter; discover_and_import passes provider_root
  to prevent package root discovery from escaping above the provider boundary

Closes #3625 (issues 2, 3, 6)

🤖 Generated with Claude Code

* fix: use contextlib.suppress for SIM105 linting

🤖 Generated with Claude Code

* test: add import machinery regression tests

🤖 Generated with Claude Code

* fix: resolve provider_root before path comparison; improve tests

- Resolve provider_root in import_module_from_file so the stop_at boundary
  in _find_package_root works correctly when provider_root is a relative path
  (e.g. FileSystemProvider(Path("./mcp"))) — previously the resolved file_path
  and unresolved stop_at.parent would never compare equal
- Fix test_stdlib_not_shadowed: use unconditional finally to restore sys.modules["json"]
- Strengthen test_same_stem_files: assert mod_a is not mod_b and that sys.modules["helpers"]
  was not clobbered by the second import
- Replace direct _find_package_root unit test with an integration test through
  import_module_from_file(provider_root=...) that also verifies the module name
  and that tmp_path is not added to sys.path

🤖 Generated with Claude Code
This commit is contained in:
Bill Easton 2026-03-26 16:53:12 -05:00 • committed by GitHub
commit 65ea57963a
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 185 additions and 29 deletions
Download patch file Download diff file Expand all files Collapse all files

115
src/fastmcp/server/providers/filesystem_discovery.py
View file

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

99
tests/fs/test_discovery.py
View file

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