mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-09 15:19:10 +02:00
523 lines
16 KiB
Python
523 lines
16 KiB
Python
"""Tests for FileSystemProvider."""
|
|
|
|
import time
|
|
from pathlib import Path
|
|
|
|
from fastmcp import FastMCP
|
|
from fastmcp.client import Client
|
|
from fastmcp.server.providers import FileSystemProvider
|
|
|
|
|
|
class TestFileSystemProvider:
|
|
"""Tests for FileSystemProvider."""
|
|
|
|
def test_provider_empty_directory(self, tmp_path: Path):
|
|
"""Provider should work with empty directory."""
|
|
provider = FileSystemProvider(tmp_path)
|
|
assert repr(provider).startswith("FileSystemProvider")
|
|
|
|
def test_provider_discovers_tools(self, tmp_path: Path):
|
|
"""Provider should discover @tool decorated functions."""
|
|
tools_dir = tmp_path / "tools"
|
|
tools_dir.mkdir()
|
|
(tools_dir / "greet.py").write_text(
|
|
"""\
|
|
from fastmcp.tools import tool
|
|
|
|
@tool
|
|
def greet(name: str) -> str:
|
|
'''Greet someone by name.'''
|
|
return f"Hello, {name}!"
|
|
"""
|
|
)
|
|
|
|
provider = FileSystemProvider(tmp_path)
|
|
|
|
# Check tool was registered
|
|
assert len(provider._components) == 1
|
|
|
|
def test_provider_discovers_resources(self, tmp_path: Path):
|
|
"""Provider should discover @resource decorated functions."""
|
|
(tmp_path / "config.py").write_text(
|
|
"""\
|
|
from fastmcp.resources import resource
|
|
|
|
@resource("config://app")
|
|
def get_config() -> dict:
|
|
'''Get app config.'''
|
|
return {"setting": "value"}
|
|
"""
|
|
)
|
|
|
|
provider = FileSystemProvider(tmp_path)
|
|
assert len(provider._components) == 1
|
|
|
|
def test_provider_discovers_resource_templates(self, tmp_path: Path):
|
|
"""Provider should discover resource templates."""
|
|
(tmp_path / "users.py").write_text(
|
|
"""\
|
|
from fastmcp.resources import resource
|
|
|
|
@resource("users://{user_id}/profile")
|
|
def get_profile(user_id: str) -> dict:
|
|
'''Get user profile.'''
|
|
return {"id": user_id}
|
|
"""
|
|
)
|
|
|
|
provider = FileSystemProvider(tmp_path)
|
|
assert len(provider._components) == 1
|
|
|
|
def test_provider_discovers_prompts(self, tmp_path: Path):
|
|
"""Provider should discover @prompt decorated functions."""
|
|
(tmp_path / "analyze.py").write_text(
|
|
"""\
|
|
from fastmcp.prompts import prompt
|
|
|
|
@prompt
|
|
def analyze(topic: str) -> list:
|
|
'''Analyze a topic.'''
|
|
return [{"role": "user", "content": f"Analyze: {topic}"}]
|
|
"""
|
|
)
|
|
|
|
provider = FileSystemProvider(tmp_path)
|
|
assert len(provider._components) == 1
|
|
|
|
def test_provider_discovers_multiple_in_one_file(self, tmp_path: Path):
|
|
"""Provider should discover multiple components in one file."""
|
|
(tmp_path / "multi.py").write_text(
|
|
"""\
|
|
from fastmcp.tools import tool
|
|
from fastmcp.resources import resource
|
|
|
|
@tool
|
|
def tool1() -> str:
|
|
return "tool1"
|
|
|
|
@tool
|
|
def tool2() -> str:
|
|
return "tool2"
|
|
|
|
@resource("config://app")
|
|
def get_config() -> dict:
|
|
return {}
|
|
"""
|
|
)
|
|
|
|
provider = FileSystemProvider(tmp_path)
|
|
assert len(provider._components) == 3
|
|
|
|
def test_provider_skips_undecorated_files(self, tmp_path: Path):
|
|
"""Provider should skip files with no decorated functions."""
|
|
(tmp_path / "utils.py").write_text(
|
|
"""\
|
|
def helper_function():
|
|
return "helper"
|
|
|
|
SOME_CONSTANT = 42
|
|
"""
|
|
)
|
|
(tmp_path / "tool.py").write_text(
|
|
"""\
|
|
from fastmcp.tools import tool
|
|
|
|
@tool
|
|
def my_tool() -> str:
|
|
return "tool"
|
|
"""
|
|
)
|
|
|
|
provider = FileSystemProvider(tmp_path)
|
|
# Only the tool should be registered
|
|
assert len(provider._components) == 1
|
|
|
|
|
|
class TestFileSystemProviderReloadMode:
|
|
"""Tests for FileSystemProvider reload mode."""
|
|
|
|
def test_reload_false_caches_at_init(self, tmp_path: Path):
|
|
"""With reload=False, components are cached at init."""
|
|
(tmp_path / "tool.py").write_text(
|
|
"""\
|
|
from fastmcp.tools import tool
|
|
|
|
@tool
|
|
def original() -> str:
|
|
return "original"
|
|
"""
|
|
)
|
|
|
|
provider = FileSystemProvider(tmp_path, reload=False)
|
|
assert len(provider._components) == 1
|
|
|
|
# Add another file - should NOT be picked up
|
|
(tmp_path / "tool2.py").write_text(
|
|
"""\
|
|
from fastmcp.tools import tool
|
|
|
|
@tool
|
|
def added() -> str:
|
|
return "added"
|
|
"""
|
|
)
|
|
|
|
# Still only one component
|
|
assert len(provider._components) == 1
|
|
|
|
async def test_reload_true_rescans(self, tmp_path: Path):
|
|
"""With reload=True, components are rescanned on each request."""
|
|
(tmp_path / "tool.py").write_text(
|
|
"""\
|
|
from fastmcp.tools import tool
|
|
|
|
@tool
|
|
def original() -> str:
|
|
return "original"
|
|
"""
|
|
)
|
|
|
|
provider = FileSystemProvider(tmp_path, reload=True)
|
|
|
|
# Always loaded once at init (to catch errors early)
|
|
assert provider._loaded
|
|
assert len(provider._components) == 1
|
|
|
|
# Add another file - should be picked up on next _ensure_loaded
|
|
(tmp_path / "tool2.py").write_text(
|
|
"""\
|
|
from fastmcp.tools import tool
|
|
|
|
@tool
|
|
def added() -> str:
|
|
return "added"
|
|
"""
|
|
)
|
|
|
|
# With reload=True, _ensure_loaded re-scans
|
|
await provider._ensure_loaded()
|
|
assert len(provider._components) == 2
|
|
|
|
async def test_warning_deduplication_same_file(self, tmp_path: Path, capsys):
|
|
"""Warnings for the same broken file should not repeat."""
|
|
bad_file = tmp_path / "bad.py"
|
|
bad_file.write_text("1/0 # division by zero")
|
|
|
|
provider = FileSystemProvider(tmp_path, reload=True)
|
|
|
|
# First load - should warn
|
|
captured = capsys.readouterr()
|
|
# Check for warning indicator (rich may truncate long paths)
|
|
assert "WARNING" in captured.err and "Failed to import" in captured.err
|
|
|
|
# Second load (same file, unchanged) - should NOT warn again
|
|
await provider._ensure_loaded()
|
|
captured = capsys.readouterr()
|
|
assert "Failed to import" not in captured.err
|
|
|
|
async def test_warning_on_file_change(self, tmp_path: Path, capsys):
|
|
"""Warnings should reappear when a broken file changes."""
|
|
bad_file = tmp_path / "bad.py"
|
|
bad_file.write_text("1/0 # division by zero")
|
|
|
|
provider = FileSystemProvider(tmp_path, reload=True)
|
|
|
|
# First load - should warn
|
|
captured = capsys.readouterr()
|
|
# Check for warning indicator (rich may truncate long paths)
|
|
assert "WARNING" in captured.err and "Failed to import" in captured.err
|
|
|
|
# Modify the file (different error) - need to ensure mtime changes
|
|
time.sleep(0.01) # Ensure mtime differs
|
|
bad_file.write_text("syntax error here !!!")
|
|
|
|
# Next load - should warn again (file changed)
|
|
await provider._ensure_loaded()
|
|
captured = capsys.readouterr()
|
|
# Check for warning indicator (rich may truncate long paths)
|
|
assert "WARNING" in captured.err and "Failed to import" in captured.err
|
|
|
|
async def test_warning_cleared_when_fixed(self, tmp_path: Path, capsys):
|
|
"""Warnings should clear when a file is fixed, and reappear if broken again."""
|
|
bad_file = tmp_path / "tool.py"
|
|
bad_file.write_text("1/0 # broken")
|
|
|
|
provider = FileSystemProvider(tmp_path, reload=True)
|
|
|
|
# First load - should warn
|
|
captured = capsys.readouterr()
|
|
# Check for warning indicator (rich may truncate long paths)
|
|
assert "WARNING" in captured.err and "Failed to import" in captured.err
|
|
|
|
# Fix the file
|
|
time.sleep(0.01)
|
|
bad_file.write_text(
|
|
"""\
|
|
from fastmcp.tools import tool
|
|
|
|
@tool
|
|
def my_tool() -> str:
|
|
return "fixed"
|
|
"""
|
|
)
|
|
|
|
# Load again - should NOT warn, file is fixed
|
|
await provider._ensure_loaded()
|
|
captured = capsys.readouterr()
|
|
assert "Failed to import" not in captured.err
|
|
assert len(provider._components) == 1
|
|
|
|
# Break it again
|
|
time.sleep(0.01)
|
|
bad_file.write_text("1/0 # broken again")
|
|
|
|
# Should warn again
|
|
await provider._ensure_loaded()
|
|
captured = capsys.readouterr()
|
|
# Check for warning indicator (rich may truncate long paths)
|
|
assert "WARNING" in captured.err and "Failed to import" in captured.err
|
|
|
|
|
|
class TestFileSystemProviderIntegration:
|
|
"""Integration tests with FastMCP server."""
|
|
|
|
async def test_provider_with_fastmcp_server(self, tmp_path: Path):
|
|
"""FileSystemProvider should work with FastMCP server."""
|
|
(tmp_path / "greet.py").write_text(
|
|
"""\
|
|
from fastmcp.tools import tool
|
|
|
|
@tool
|
|
def greet(name: str) -> str:
|
|
'''Greet someone.'''
|
|
return f"Hello, {name}!"
|
|
"""
|
|
)
|
|
|
|
provider = FileSystemProvider(tmp_path)
|
|
mcp = FastMCP("TestServer", providers=[provider])
|
|
|
|
async with Client(mcp) as client:
|
|
# List tools
|
|
tools = await client.list_tools()
|
|
assert len(tools) == 1
|
|
assert tools[0].name == "greet"
|
|
|
|
# Call tool
|
|
result = await client.call_tool("greet", {"name": "World"})
|
|
assert "Hello, World!" in str(result)
|
|
|
|
async def test_provider_with_resources(self, tmp_path: Path):
|
|
"""FileSystemProvider should work with resources."""
|
|
(tmp_path / "config.py").write_text(
|
|
"""\
|
|
from fastmcp.resources import resource
|
|
|
|
@resource("config://app")
|
|
def get_config() -> str:
|
|
'''Get app config.'''
|
|
return '{"version": "1.0"}'
|
|
"""
|
|
)
|
|
|
|
provider = FileSystemProvider(tmp_path)
|
|
mcp = FastMCP("TestServer", providers=[provider])
|
|
|
|
async with Client(mcp) as client:
|
|
# List resources
|
|
resources = await client.list_resources()
|
|
assert len(resources) == 1
|
|
assert str(resources[0].uri) == "config://app"
|
|
|
|
# Read resource
|
|
result = await client.read_resource("config://app")
|
|
assert "1.0" in str(result)
|
|
|
|
async def test_provider_with_resource_templates(self, tmp_path: Path):
|
|
"""FileSystemProvider should work with resource templates."""
|
|
(tmp_path / "users.py").write_text(
|
|
"""\
|
|
from fastmcp.resources import resource
|
|
|
|
@resource("users://{user_id}/profile")
|
|
def get_profile(user_id: str) -> str:
|
|
'''Get user profile.'''
|
|
return f'{{"id": "{user_id}", "name": "User {user_id}"}}'
|
|
"""
|
|
)
|
|
|
|
provider = FileSystemProvider(tmp_path)
|
|
mcp = FastMCP("TestServer", providers=[provider])
|
|
|
|
async with Client(mcp) as client:
|
|
# List templates
|
|
templates = await client.list_resource_templates()
|
|
assert len(templates) == 1
|
|
|
|
# Read with parameter
|
|
result = await client.read_resource("users://123/profile")
|
|
assert "123" in str(result)
|
|
|
|
async def test_provider_with_prompts(self, tmp_path: Path):
|
|
"""FileSystemProvider should work with prompts."""
|
|
(tmp_path / "analyze.py").write_text(
|
|
"""\
|
|
from fastmcp.prompts import prompt
|
|
|
|
@prompt
|
|
def analyze(topic: str) -> str:
|
|
'''Analyze a topic.'''
|
|
return f"Please analyze: {topic}"
|
|
"""
|
|
)
|
|
|
|
provider = FileSystemProvider(tmp_path)
|
|
mcp = FastMCP("TestServer", providers=[provider])
|
|
|
|
async with Client(mcp) as client:
|
|
# List prompts
|
|
prompts = await client.list_prompts()
|
|
assert len(prompts) == 1
|
|
assert prompts[0].name == "analyze"
|
|
|
|
# Get prompt
|
|
result = await client.get_prompt("analyze", {"topic": "Python"})
|
|
assert "Python" in str(result)
|
|
|
|
async def test_nested_directory_structure(self, tmp_path: Path):
|
|
"""FileSystemProvider should work with nested directories."""
|
|
# Create nested structure
|
|
tools = tmp_path / "tools"
|
|
tools.mkdir()
|
|
(tools / "greet.py").write_text(
|
|
"""\
|
|
from fastmcp.tools import tool
|
|
|
|
@tool
|
|
def greet(name: str) -> str:
|
|
return f"Hello, {name}!"
|
|
"""
|
|
)
|
|
|
|
payments = tools / "payments"
|
|
payments.mkdir()
|
|
(payments / "charge.py").write_text(
|
|
"""\
|
|
from fastmcp.tools import tool
|
|
|
|
@tool
|
|
def charge(amount: float) -> str:
|
|
return f"Charged ${amount}"
|
|
"""
|
|
)
|
|
|
|
provider = FileSystemProvider(tmp_path)
|
|
mcp = FastMCP("TestServer", providers=[provider])
|
|
|
|
async with Client(mcp) as client:
|
|
tools_list = await client.list_tools()
|
|
assert len(tools_list) == 2
|
|
names = {t.name for t in tools_list}
|
|
assert names == {"greet", "charge"}
|
|
|
|
|
|
class TestFileSystemProviderVersioning:
|
|
"""Tests for version propagation through FileSystemProvider."""
|
|
|
|
async def test_versioned_tool_via_provider(self, tmp_path: Path):
|
|
"""FileSystemProvider should preserve tool version in list_tools output."""
|
|
(tmp_path / "versioned.py").write_text(
|
|
"""\
|
|
from fastmcp.tools import tool
|
|
|
|
@tool(version="1.0", description="v1 greet")
|
|
def greet(name: str) -> str:
|
|
return f"Hello, {name}!"
|
|
"""
|
|
)
|
|
|
|
provider = FileSystemProvider(tmp_path)
|
|
mcp = FastMCP("TestServer", providers=[provider])
|
|
|
|
async with Client(mcp) as client:
|
|
tools = await client.list_tools()
|
|
assert len(tools) == 1
|
|
assert tools[0].name == "greet"
|
|
meta = tools[0].meta
|
|
assert meta is not None
|
|
assert meta["fastmcp"]["version"] == "1.0"
|
|
|
|
async def test_versioned_resource_via_provider(self, tmp_path: Path):
|
|
"""FileSystemProvider should preserve resource version."""
|
|
(tmp_path / "versioned_resource.py").write_text(
|
|
"""\
|
|
from fastmcp.resources import resource
|
|
|
|
@resource("data://config", version="2.0", name="config", description="v2 config")
|
|
def config() -> str:
|
|
return '{"theme": "dark"}'
|
|
"""
|
|
)
|
|
|
|
provider = FileSystemProvider(tmp_path)
|
|
mcp = FastMCP("TestServer", providers=[provider])
|
|
|
|
async with Client(mcp) as client:
|
|
resources = await client.list_resources()
|
|
assert len(resources) == 1
|
|
assert resources[0].name == "config"
|
|
meta = resources[0].meta
|
|
assert meta is not None
|
|
assert meta["fastmcp"]["version"] == "2.0"
|
|
|
|
async def test_versioned_prompt_via_provider(self, tmp_path: Path):
|
|
"""FileSystemProvider should preserve prompt version."""
|
|
(tmp_path / "versioned_prompt.py").write_text(
|
|
"""\
|
|
from fastmcp.prompts import prompt
|
|
|
|
@prompt(name="summarize", version="1.0", description="v1 prompt")
|
|
def summarize(text: str) -> str:
|
|
return f"Summarize: {text}"
|
|
"""
|
|
)
|
|
|
|
provider = FileSystemProvider(tmp_path)
|
|
mcp = FastMCP("TestServer", providers=[provider])
|
|
|
|
async with Client(mcp) as client:
|
|
prompts = await client.list_prompts()
|
|
assert len(prompts) == 1
|
|
assert prompts[0].name == "summarize"
|
|
meta = prompts[0].meta
|
|
assert meta is not None
|
|
assert meta["fastmcp"]["version"] == "1.0"
|
|
|
|
async def test_multiple_tool_versions_via_provider(self, tmp_path: Path):
|
|
"""FileSystemProvider should handle multiple versions of the same tool."""
|
|
(tmp_path / "multi_version.py").write_text(
|
|
"""\
|
|
from fastmcp.tools import tool
|
|
|
|
@tool(name="add", version="1.0", description="v1 add")
|
|
def add_v1(x: int, y: int) -> int:
|
|
return x + y
|
|
|
|
@tool(name="add", version="2.0", description="v2 add with z")
|
|
def add_v2(x: int, y: int, z: int = 0) -> int:
|
|
return x + y + z
|
|
"""
|
|
)
|
|
|
|
provider = FileSystemProvider(tmp_path)
|
|
mcp = FastMCP("TestServer", providers=[provider])
|
|
|
|
async with Client(mcp) as client:
|
|
tools = await client.list_tools()
|
|
add_tools = [t for t in tools if t.name == "add"]
|
|
# list_tools deduplicates to the highest version
|
|
assert len(add_tools) == 1
|
|
meta = add_tools[0].meta
|
|
assert meta is not None
|
|
assert meta["fastmcp"]["version"] == "2.0"
|
|
assert meta["fastmcp"]["versions"] == ["2.0", "1.0"]
|