mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-10-09 22:43:20 +02:00
786a4cb10eKeeps filesystem discovery under the reload lock through AnyIO cancellation; the supported cancellation regression and neighboring reload behavior are verified. Fixes #4133 Co-authored-by: sherxlg-gif <298523066+sherxlg-gif@users.noreply.github.com> Co-authored-by: NEEDI <sherxlg@gmail.com>
771 lines
24 KiB
Python
771 lines
24 KiB
Python
"""Tests for FileSystemProvider."""
|
|
|
|
import asyncio
|
|
import threading
|
|
import time
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
import anyio
|
|
import pytest
|
|
from anyio.abc import TaskStatus
|
|
from anyio.to_thread import run_sync
|
|
|
|
from fastmcp import FastMCP
|
|
from fastmcp.client import Client
|
|
from fastmcp.server.providers import FileSystemProvider
|
|
from fastmcp.server.providers.filesystem_discovery import (
|
|
DiscoveryResult,
|
|
discover_and_import,
|
|
)
|
|
|
|
|
|
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")
|
|
assert provider.failed_files == {}
|
|
|
|
def test_provider_exposes_failed_files(self, tmp_path: Path):
|
|
"""Provider should expose import failures from the initial load."""
|
|
broken_file = tmp_path / "broken.py"
|
|
broken_file.write_text("raise RuntimeError('broken import')")
|
|
|
|
provider = FileSystemProvider(tmp_path)
|
|
|
|
assert set(provider.failed_files) == {broken_file}
|
|
assert "broken import" in provider.failed_files[broken_file]
|
|
|
|
# The public value is a snapshot, not the provider's mutable state.
|
|
failures = provider.failed_files
|
|
assert isinstance(failures, dict)
|
|
failures.clear()
|
|
assert set(provider.failed_files) == {broken_file}
|
|
|
|
def test_provider_exposes_registration_failures(
|
|
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
):
|
|
"""Provider should expose registration failures alongside import failures."""
|
|
tool_file = tmp_path / "tool.py"
|
|
tool_file.write_text(
|
|
"""\
|
|
from fastmcp.tools import tool
|
|
|
|
@tool
|
|
def my_tool() -> str:
|
|
return "tool"
|
|
"""
|
|
)
|
|
|
|
def fail_registration(self: FileSystemProvider, component: object) -> None:
|
|
raise RuntimeError("registration failed")
|
|
|
|
monkeypatch.setattr(
|
|
FileSystemProvider, "_register_component", fail_registration
|
|
)
|
|
|
|
provider = FileSystemProvider(tmp_path)
|
|
|
|
assert provider.failed_files == {tool_file: "RuntimeError: registration failed"}
|
|
|
|
async def test_failed_files_refresh_on_reload(self, tmp_path: Path):
|
|
"""Reload mode should expose the latest load failures."""
|
|
broken_file = tmp_path / "broken.py"
|
|
broken_file.write_text("raise RuntimeError('broken import')")
|
|
provider = FileSystemProvider(tmp_path, reload=True)
|
|
|
|
assert set(provider.failed_files) == {broken_file}
|
|
|
|
broken_file.unlink()
|
|
assert await provider.list_tools() == []
|
|
assert provider.failed_files == {}
|
|
|
|
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 read
|
|
(tmp_path / "tool2.py").write_text(
|
|
"""\
|
|
from fastmcp.tools import tool
|
|
|
|
@tool
|
|
def added() -> str:
|
|
return "added"
|
|
"""
|
|
)
|
|
|
|
# With reload=True, reading triggers a re-scan
|
|
tools = await provider._list_tools()
|
|
assert len(tools) == 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._list_tools()
|
|
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._list_tools()
|
|
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._list_tools()
|
|
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._list_tools()
|
|
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 TestFileSystemProviderReloadRace:
|
|
"""Test that concurrent readers don't see empty components during reload."""
|
|
|
|
async def test_cancelled_reload_waits_for_discovery(self, tmp_path: Path) -> None:
|
|
"""Cancellation must not release the lock while discovery is still running."""
|
|
(tmp_path / "tool.py").write_text(
|
|
"from fastmcp.tools import tool\n"
|
|
"@tool\n"
|
|
"def my_tool() -> str:\n"
|
|
" return 'hello'\n"
|
|
)
|
|
provider = FileSystemProvider(tmp_path, reload=True)
|
|
discovery_started = threading.Event()
|
|
finish_discovery = threading.Event()
|
|
reload_exited = anyio.Event()
|
|
|
|
def slow_discovery(root: Path) -> DiscoveryResult:
|
|
discovery_started.set()
|
|
assert finish_discovery.wait(timeout=2)
|
|
return discover_and_import(root)
|
|
|
|
async def reload(
|
|
*, task_status: TaskStatus[anyio.CancelScope] = anyio.TASK_STATUS_IGNORED
|
|
) -> None:
|
|
with anyio.CancelScope() as scope:
|
|
task_status.started(scope)
|
|
await provider.list_tools()
|
|
reload_exited.set()
|
|
|
|
async def read() -> None:
|
|
tools = await provider.list_tools()
|
|
assert [tool.name for tool in tools] == ["my_tool"]
|
|
|
|
with patch(
|
|
"fastmcp.server.providers.filesystem.discover_and_import",
|
|
side_effect=slow_discovery,
|
|
) as discovery:
|
|
async with anyio.create_task_group() as tasks:
|
|
try:
|
|
scope = await tasks.start(reload)
|
|
assert await run_sync(discovery_started.wait, 2)
|
|
scope.cancel()
|
|
tasks.start_soon(read)
|
|
await anyio.wait_all_tasks_blocked()
|
|
exited_during_discovery = reload_exited.is_set()
|
|
finally:
|
|
finish_discovery.set()
|
|
|
|
assert not exited_during_discovery
|
|
assert reload_exited.is_set()
|
|
assert discovery.call_count == 1
|
|
|
|
async def test_concurrent_reader_never_sees_empty(self, tmp_path: Path):
|
|
"""A reader during reload should see either old or new components, never empty."""
|
|
(tmp_path / "tool.py").write_text(
|
|
"""\
|
|
from fastmcp.tools import tool
|
|
|
|
@tool
|
|
def my_tool() -> str:
|
|
return "hello"
|
|
"""
|
|
)
|
|
|
|
provider = FileSystemProvider(tmp_path, reload=True)
|
|
assert len(await provider._list_tools()) == 1
|
|
|
|
observed_empty = False
|
|
|
|
async def reader():
|
|
nonlocal observed_empty
|
|
for _ in range(20):
|
|
tools = await provider._list_tools()
|
|
if len(tools) == 0:
|
|
observed_empty = True
|
|
await asyncio.sleep(0)
|
|
|
|
async def reloader():
|
|
for _ in range(5):
|
|
provider._loaded = False
|
|
await provider._list_tools() # triggers reload
|
|
await asyncio.sleep(0)
|
|
|
|
await asyncio.gather(reader(), reloader())
|
|
assert not observed_empty, "Reader saw empty components during reload"
|
|
|
|
|
|
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"}
|
|
|
|
async def test_mounted_servers_with_same_package_and_module_name(
|
|
self, tmp_path: Path
|
|
):
|
|
"""Two providers sharing a package/module path should not collide.
|
|
|
|
Reproduces the exact scenario from the bug report: both providers have a
|
|
`components/tools.py` package. Without isolation the second provider
|
|
reuses the first's cached module and `list_tools` returns the first
|
|
server's tool twice.
|
|
"""
|
|
for server_name, tool_name in [
|
|
("server1", "server1_tool"),
|
|
("server2", "server2_tool"),
|
|
]:
|
|
components = tmp_path / server_name / "components"
|
|
components.mkdir(parents=True)
|
|
(components / "__init__.py").write_text("")
|
|
(components / "tools.py").write_text(
|
|
f"""\
|
|
from fastmcp.tools import tool
|
|
|
|
@tool
|
|
def {tool_name}() -> str:
|
|
return "{tool_name}"
|
|
"""
|
|
)
|
|
|
|
server1 = FastMCP(
|
|
"server1",
|
|
providers=[FileSystemProvider(tmp_path / "server1" / "components")],
|
|
)
|
|
server2 = FastMCP(
|
|
"server2",
|
|
providers=[FileSystemProvider(tmp_path / "server2" / "components")],
|
|
)
|
|
parent = FastMCP("parent")
|
|
parent.mount(server1)
|
|
parent.mount(server2)
|
|
|
|
tools = await parent.list_tools()
|
|
assert {t.name for t in tools} == {"server1_tool", "server2_tool"}
|
|
|
|
result = await parent.call_tool("server2_tool", {})
|
|
assert "server2_tool" in str(result)
|
|
|
|
async def test_mounted_servers_with_same_package_different_module(
|
|
self, tmp_path: Path
|
|
):
|
|
"""Same package name but different module filenames must coexist.
|
|
|
|
The package directory name (`components`) collides while the leaf module
|
|
differs (`a.py` vs `b.py`). The shared package must not resolve to the
|
|
first provider's directory, which would drop the second provider's tool.
|
|
A relative import inside each module exercises that the isolated tree is
|
|
a real package.
|
|
"""
|
|
for server_name, module_name, tool_name in [
|
|
("server1", "a", "server1_tool"),
|
|
("server2", "b", "server2_tool"),
|
|
]:
|
|
components = tmp_path / server_name / "components"
|
|
components.mkdir(parents=True)
|
|
(components / "__init__.py").write_text("")
|
|
(components / "_shared.py").write_text(f'LABEL = "{tool_name}"\n')
|
|
(components / f"{module_name}.py").write_text(
|
|
f"""\
|
|
from fastmcp.tools import tool
|
|
|
|
from ._shared import LABEL
|
|
|
|
@tool
|
|
def {tool_name}() -> str:
|
|
return LABEL
|
|
"""
|
|
)
|
|
|
|
server1 = FastMCP(
|
|
"server1",
|
|
providers=[FileSystemProvider(tmp_path / "server1" / "components")],
|
|
)
|
|
server2 = FastMCP(
|
|
"server2",
|
|
providers=[FileSystemProvider(tmp_path / "server2" / "components")],
|
|
)
|
|
parent = FastMCP("parent")
|
|
parent.mount(server1)
|
|
parent.mount(server2)
|
|
|
|
tools = await parent.list_tools()
|
|
assert {t.name for t in tools} == {"server1_tool", "server2_tool"}
|
|
|
|
result = await parent.call_tool("server2_tool", {})
|
|
assert "server2_tool" in str(result)
|
|
|
|
|
|
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"]
|