mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-28 02:10:38 +02:00
78ab933dc5
FastMCPProvider now calls _get_* methods instead of get_* to ensure nested server transforms are applied during lookups. Also converts string versions to VersionSpec in MCP handlers.
1235 lines
39 KiB
Python
1235 lines
39 KiB
Python
"""Tests for component versioning functionality."""
|
|
# ruff: noqa: F811 # Intentional function redefinition for version testing
|
|
|
|
from __future__ import annotations
|
|
|
|
from mcp.types import TextContent
|
|
|
|
from fastmcp import FastMCP
|
|
from fastmcp.utilities.versions import (
|
|
VersionKey,
|
|
VersionSpec,
|
|
compare_versions,
|
|
is_version_greater,
|
|
)
|
|
|
|
|
|
class TestVersionKey:
|
|
"""Tests for VersionKey comparison class."""
|
|
|
|
def test_none_sorts_lowest(self):
|
|
"""None (unversioned) should sort lower than any version."""
|
|
assert VersionKey(None) < VersionKey("1.0")
|
|
assert VersionKey(None) < VersionKey("0.1")
|
|
assert VersionKey(None) < VersionKey("anything")
|
|
|
|
def test_none_equals_none(self):
|
|
"""Two None versions should be equal."""
|
|
assert VersionKey(None) == VersionKey(None)
|
|
assert not (VersionKey(None) < VersionKey(None))
|
|
assert not (VersionKey(None) > VersionKey(None))
|
|
|
|
def test_pep440_versions_compared_semantically(self):
|
|
"""Valid PEP 440 versions should compare semantically."""
|
|
assert VersionKey("1.0") < VersionKey("2.0")
|
|
assert VersionKey("1.0") < VersionKey("1.1")
|
|
assert VersionKey("1.9") < VersionKey("1.10") # Semantic, not string
|
|
assert VersionKey("2") < VersionKey("10") # Semantic, not string
|
|
|
|
def test_v_prefix_stripped(self):
|
|
"""Versions with 'v' prefix should be handled correctly."""
|
|
assert VersionKey("v1.0") == VersionKey("1.0")
|
|
assert VersionKey("v2.0") > VersionKey("v1.0")
|
|
|
|
def test_string_fallback_for_invalid_versions(self):
|
|
"""Invalid PEP 440 versions should fall back to string comparison."""
|
|
# Dates are not valid PEP 440
|
|
assert VersionKey("2024-01-01") < VersionKey("2025-01-01")
|
|
# String comparison (lexicographic)
|
|
assert VersionKey("alpha") < VersionKey("beta")
|
|
|
|
def test_pep440_sorts_before_strings(self):
|
|
"""PEP 440 versions sort before invalid string versions."""
|
|
# "1.0" is valid PEP 440, "not-semver" is not
|
|
assert VersionKey("1.0") < VersionKey("not-semver")
|
|
assert VersionKey("999.0") < VersionKey("aaa") # PEP 440 < string
|
|
|
|
def test_repr(self):
|
|
"""Test string representation."""
|
|
assert repr(VersionKey("1.0")) == "VersionKey('1.0')"
|
|
assert repr(VersionKey(None)) == "VersionKey(None)"
|
|
|
|
|
|
class TestVersionFunctions:
|
|
"""Tests for version comparison functions."""
|
|
|
|
def test_compare_versions(self):
|
|
"""Test compare_versions function."""
|
|
assert compare_versions("1.0", "2.0") == -1
|
|
assert compare_versions("2.0", "1.0") == 1
|
|
assert compare_versions("1.0", "1.0") == 0
|
|
assert compare_versions(None, "1.0") == -1
|
|
assert compare_versions("1.0", None) == 1
|
|
assert compare_versions(None, None) == 0
|
|
|
|
def test_is_version_greater(self):
|
|
"""Test is_version_greater function."""
|
|
assert is_version_greater("2.0", "1.0")
|
|
assert not is_version_greater("1.0", "2.0")
|
|
assert not is_version_greater("1.0", "1.0")
|
|
assert is_version_greater("1.0", None)
|
|
assert not is_version_greater(None, "1.0")
|
|
|
|
|
|
class TestComponentVersioning:
|
|
"""Tests for versioning in FastMCP components."""
|
|
|
|
async def test_tool_with_version(self):
|
|
"""Tool version should be reflected in key."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.tool(version="2.0")
|
|
def my_tool(x: int) -> int:
|
|
return x * 2
|
|
|
|
tools = await mcp.get_tools()
|
|
assert len(tools) == 1
|
|
assert tools[0].name == "my_tool"
|
|
assert tools[0].version == "2.0"
|
|
assert tools[0].key == "tool:my_tool@2.0"
|
|
|
|
async def test_tool_without_version(self):
|
|
"""Tool without version should have @ sentinel in key but empty version."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.tool
|
|
def my_tool(x: int) -> int:
|
|
return x * 2
|
|
|
|
tools = await mcp.get_tools()
|
|
assert len(tools) == 1
|
|
assert tools[0].version is None
|
|
# Keys always have @ sentinel for unambiguous parsing
|
|
assert tools[0].key == "tool:my_tool@"
|
|
|
|
async def test_tool_version_as_int(self):
|
|
"""Tool version as int should be coerced to string."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.tool(version=2)
|
|
def my_tool(x: int) -> int:
|
|
return x * 2
|
|
|
|
tools = await mcp.get_tools()
|
|
assert len(tools) == 1
|
|
assert tools[0].version == "2"
|
|
assert tools[0].key == "tool:my_tool@2"
|
|
|
|
async def test_tool_version_zero_is_truthy(self):
|
|
"""Version 0 should become "0" (truthy string), not empty."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.tool(version=0)
|
|
def my_tool(x: int) -> int:
|
|
return x * 2
|
|
|
|
tools = await mcp.get_tools()
|
|
assert len(tools) == 1
|
|
assert tools[0].version == "0"
|
|
assert tools[0].key == "tool:my_tool@0" # Not "tool:my_tool@"
|
|
|
|
async def test_multiple_tool_versions_deduplicated(self):
|
|
"""Multiple versions of same tool should deduplicate to highest."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.tool(version="1.0")
|
|
def add(x: int, y: int) -> int:
|
|
return x + y
|
|
|
|
@mcp.tool(version="2.0")
|
|
def add(x: int, y: int, z: int = 0) -> int:
|
|
return x + y + z
|
|
|
|
tools = await mcp.get_tools()
|
|
# Should only show the highest version
|
|
assert len(tools) == 1
|
|
assert tools[0].name == "add"
|
|
assert tools[0].version == "2.0"
|
|
|
|
async def test_call_tool_invokes_highest_version(self):
|
|
"""Calling a tool by name should invoke the highest version."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.tool(version="1.0")
|
|
def add(x: int, y: int) -> int:
|
|
return x + y
|
|
|
|
@mcp.tool(version="2.0")
|
|
def add(x: int, y: int) -> int:
|
|
return (x + y) * 10 # Different behavior to distinguish
|
|
|
|
result = await mcp.call_tool("add", {"x": 1, "y": 2})
|
|
# Should invoke v2.0 which multiplies by 10
|
|
assert isinstance(result.content[0], TextContent)
|
|
assert result.content[0].text == "30"
|
|
|
|
async def test_mixing_versioned_and_unversioned_rejected(self):
|
|
"""Cannot mix versioned and unversioned tools with the same name."""
|
|
import pytest
|
|
|
|
mcp = FastMCP()
|
|
|
|
@mcp.tool
|
|
def my_tool() -> str:
|
|
return "unversioned"
|
|
|
|
# Adding versioned tool when unversioned exists should fail
|
|
with pytest.raises(ValueError, match="versioned.*unversioned"):
|
|
|
|
@mcp.tool(version="1.0")
|
|
def my_tool() -> str:
|
|
return "v1.0"
|
|
|
|
async def test_mixing_unversioned_after_versioned_rejected(self):
|
|
"""Cannot add unversioned tool when versioned exists."""
|
|
import pytest
|
|
|
|
mcp = FastMCP()
|
|
|
|
@mcp.tool(version="1.0")
|
|
def my_tool() -> str:
|
|
return "v1.0"
|
|
|
|
# Adding unversioned tool when versioned exists should fail
|
|
with pytest.raises(ValueError, match="unversioned.*versioned"):
|
|
|
|
@mcp.tool
|
|
def my_tool() -> str:
|
|
return "unversioned"
|
|
|
|
async def test_resource_with_version(self):
|
|
"""Resource version should work like tool version."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.resource("file://config", version="1.0")
|
|
def config_v1() -> str:
|
|
return "config v1"
|
|
|
|
@mcp.resource("file://config", version="2.0")
|
|
def config_v2() -> str:
|
|
return "config v2"
|
|
|
|
resources = await mcp.get_resources()
|
|
assert len(resources) == 1
|
|
assert resources[0].version == "2.0"
|
|
|
|
async def test_prompt_with_version(self):
|
|
"""Prompt version should work like tool version."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.prompt(version="1.0")
|
|
def greet(name: str) -> str:
|
|
return f"Hello, {name}!"
|
|
|
|
@mcp.prompt(version="2.0")
|
|
def greet(name: str) -> str:
|
|
return f"Greetings, {name}!"
|
|
|
|
prompts = await mcp.get_prompts()
|
|
assert len(prompts) == 1
|
|
assert prompts[0].version == "2.0"
|
|
|
|
|
|
class TestVersionSorting:
|
|
"""Tests for version sorting behavior."""
|
|
|
|
async def test_semantic_version_sorting(self):
|
|
"""Versions should sort semantically, not lexicographically."""
|
|
mcp = FastMCP()
|
|
|
|
# Add versions out of order
|
|
@mcp.tool(version="1")
|
|
def count() -> int:
|
|
return 1
|
|
|
|
@mcp.tool(version="10")
|
|
def count() -> int:
|
|
return 10
|
|
|
|
@mcp.tool(version="2")
|
|
def count() -> int:
|
|
return 2
|
|
|
|
tools = await mcp.get_tools()
|
|
# Should keep v10 as highest (semantic: 10 > 2 > 1)
|
|
assert len(tools) == 1
|
|
assert tools[0].version == "10"
|
|
|
|
result = await mcp.call_tool("count", {})
|
|
assert isinstance(result.content[0], TextContent)
|
|
assert result.content[0].text == "10"
|
|
|
|
async def test_semver_sorting(self):
|
|
"""Full semver versions should sort correctly."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.tool(version="1.2.3")
|
|
def info() -> str:
|
|
return "1.2.3"
|
|
|
|
@mcp.tool(version="1.2.10")
|
|
def info() -> str:
|
|
return "1.2.10"
|
|
|
|
@mcp.tool(version="1.10.1")
|
|
def info() -> str:
|
|
return "1.10.1"
|
|
|
|
tools = await mcp.get_tools()
|
|
assert len(tools) == 1
|
|
# 1.10.1 > 1.2.10 > 1.2.3 (semantic)
|
|
assert tools[0].version == "1.10.1"
|
|
|
|
async def test_v_prefix_normalized(self):
|
|
"""Versions with 'v' prefix should compare correctly."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.tool(version="v1.0")
|
|
def calc() -> int:
|
|
return 1
|
|
|
|
@mcp.tool(version="v2.0")
|
|
def calc() -> int:
|
|
return 2
|
|
|
|
tools = await mcp.get_tools()
|
|
assert len(tools) == 1
|
|
assert tools[0].version == "v2.0"
|
|
|
|
|
|
class TestMountedServerVersioning:
|
|
"""Tests for versioning in mounted servers (FastMCPProvider)."""
|
|
|
|
async def test_mounted_tool_preserves_version(self):
|
|
"""Mounted tools should preserve their version info."""
|
|
child = FastMCP("Child")
|
|
|
|
@child.tool(version="2.0")
|
|
def add(x: int, y: int) -> int:
|
|
return x + y
|
|
|
|
parent = FastMCP("Parent")
|
|
parent.mount(child, "child")
|
|
|
|
tools = await parent.get_tools()
|
|
assert len(tools) == 1
|
|
assert tools[0].name == "child_add"
|
|
assert tools[0].version == "2.0"
|
|
|
|
async def test_mounted_resource_preserves_version(self):
|
|
"""Mounted resources should preserve their version info."""
|
|
child = FastMCP("Child")
|
|
|
|
@child.resource("file://config", version="1.5")
|
|
def config() -> str:
|
|
return "config data"
|
|
|
|
parent = FastMCP("Parent")
|
|
parent.mount(child, "child")
|
|
|
|
resources = await parent.get_resources()
|
|
assert len(resources) == 1
|
|
assert resources[0].version == "1.5"
|
|
|
|
async def test_mounted_prompt_preserves_version(self):
|
|
"""Mounted prompts should preserve their version info."""
|
|
child = FastMCP("Child")
|
|
|
|
@child.prompt(version="3.0")
|
|
def greet(name: str) -> str:
|
|
return f"Hello, {name}!"
|
|
|
|
parent = FastMCP("Parent")
|
|
parent.mount(child, "child")
|
|
|
|
prompts = await parent.get_prompts()
|
|
assert len(prompts) == 1
|
|
assert prompts[0].name == "child_greet"
|
|
assert prompts[0].version == "3.0"
|
|
|
|
async def test_mounted_get_tool_with_version(self):
|
|
"""Should be able to get specific version from mounted server."""
|
|
child = FastMCP("Child")
|
|
|
|
@child.tool(version="1.0")
|
|
def calc() -> int:
|
|
return 1
|
|
|
|
@child.tool(version="2.0")
|
|
def calc() -> int:
|
|
return 2
|
|
|
|
parent = FastMCP("Parent")
|
|
parent.mount(child, "child")
|
|
|
|
# Get highest version (default)
|
|
tool = await parent.get_tool("child_calc")
|
|
assert tool is not None
|
|
assert tool.version == "2.0"
|
|
|
|
# Get specific version
|
|
tool_v1 = await parent.get_tool("child_calc", VersionSpec(eq="1.0"))
|
|
assert tool_v1 is not None
|
|
assert tool_v1.version == "1.0"
|
|
|
|
async def test_mounted_multiple_versions_deduplicates(self):
|
|
"""Mounted server with multiple versions should show only highest."""
|
|
child = FastMCP("Child")
|
|
|
|
@child.tool(version="1.0")
|
|
def my_tool() -> str:
|
|
return "v1"
|
|
|
|
@child.tool(version="3.0")
|
|
def my_tool() -> str:
|
|
return "v3"
|
|
|
|
@child.tool(version="2.0")
|
|
def my_tool() -> str:
|
|
return "v2"
|
|
|
|
parent = FastMCP("Parent")
|
|
parent.mount(child, "child")
|
|
|
|
tools = await parent.get_tools()
|
|
assert len(tools) == 1
|
|
assert tools[0].version == "3.0"
|
|
|
|
async def test_mounted_call_tool_uses_highest_version(self):
|
|
"""Calling mounted tool should use highest version."""
|
|
child = FastMCP("Child")
|
|
|
|
@child.tool(version="1.0")
|
|
def double(x: int) -> int:
|
|
return x * 2
|
|
|
|
@child.tool(version="2.0")
|
|
def double(x: int) -> int:
|
|
return x * 2 + 100 # Different behavior
|
|
|
|
parent = FastMCP("Parent")
|
|
parent.mount(child, "child")
|
|
|
|
result = await parent.call_tool("child_double", {"x": 5})
|
|
# Should use v2.0 which adds 100
|
|
assert isinstance(result.content[0], TextContent)
|
|
assert result.content[0].text == "110"
|
|
|
|
|
|
class TestVersionFilter:
|
|
"""Tests for VersionFilter transform."""
|
|
|
|
async def test_version_lt_filters_high_versions(self):
|
|
"""VersionFilter(version_lt='3.0') hides v3+, shows v1 and v2."""
|
|
from fastmcp.server.transforms import VersionFilter
|
|
|
|
mcp = FastMCP()
|
|
|
|
@mcp.tool(version="1.0")
|
|
def calc() -> int:
|
|
return 1
|
|
|
|
@mcp.tool(version="2.0")
|
|
def calc() -> int:
|
|
return 2
|
|
|
|
@mcp.tool(version="3.0")
|
|
def calc() -> int:
|
|
return 3
|
|
|
|
# Without filter, should show v3 (highest)
|
|
tools = await mcp.get_tools()
|
|
assert len(tools) == 1
|
|
assert tools[0].version == "3.0"
|
|
|
|
# With filter, should show v2 (highest below 3.0)
|
|
mcp.add_transform(VersionFilter(version_lt="3.0"))
|
|
tools = await mcp.get_tools()
|
|
assert len(tools) == 1
|
|
assert tools[0].version == "2.0"
|
|
|
|
async def test_version_gte_filters_low_versions(self):
|
|
"""VersionFilter(version_gte='2.0') hides v1, shows v2 and v3."""
|
|
from fastmcp.server.transforms import VersionFilter
|
|
|
|
mcp = FastMCP()
|
|
|
|
@mcp.tool(version="1.0")
|
|
def add(x: int) -> int:
|
|
return x + 1
|
|
|
|
@mcp.tool(version="2.0")
|
|
def add(x: int) -> int:
|
|
return x + 2
|
|
|
|
@mcp.tool(version="3.0")
|
|
def add(x: int) -> int:
|
|
return x + 3
|
|
|
|
mcp.add_transform(VersionFilter(version_gte="2.0"))
|
|
|
|
# Should show v3 (highest >= 2.0)
|
|
tools = await mcp.get_tools()
|
|
assert len(tools) == 1
|
|
assert tools[0].version == "3.0"
|
|
|
|
# Can request specific versions in range (use _get_tool to apply transforms)
|
|
tool_v2 = await mcp._get_tool("add", VersionSpec(eq="2.0"))
|
|
assert tool_v2 is not None
|
|
assert tool_v2.version == "2.0"
|
|
|
|
# Cannot request version outside range - returns None
|
|
assert await mcp._get_tool("add", VersionSpec(eq="1.0")) is None
|
|
|
|
async def test_version_range(self):
|
|
"""VersionFilter(version_gte='2.0', version_lt='3.0') shows only v2.x."""
|
|
from fastmcp.server.transforms import VersionFilter
|
|
|
|
mcp = FastMCP()
|
|
|
|
@mcp.tool(version="1.0")
|
|
def calc() -> int:
|
|
return 1
|
|
|
|
@mcp.tool(version="2.0")
|
|
def calc() -> int:
|
|
return 2
|
|
|
|
@mcp.tool(version="2.5")
|
|
def calc() -> int:
|
|
return 25
|
|
|
|
@mcp.tool(version="3.0")
|
|
def calc() -> int:
|
|
return 3
|
|
|
|
mcp.add_transform(VersionFilter(version_gte="2.0", version_lt="3.0"))
|
|
|
|
# Should show v2.5 (highest in range)
|
|
tools = await mcp.get_tools()
|
|
assert len(tools) == 1
|
|
assert tools[0].version == "2.5"
|
|
|
|
# Can request specific versions in range (use _get_tool to apply transforms)
|
|
tool_v2 = await mcp._get_tool("calc", VersionSpec(eq="2.0"))
|
|
assert tool_v2 is not None
|
|
assert tool_v2.version == "2.0"
|
|
|
|
# Versions outside range are not accessible - return None
|
|
assert await mcp._get_tool("calc", VersionSpec(eq="1.0")) is None
|
|
assert await mcp._get_tool("calc", VersionSpec(eq="3.0")) is None
|
|
|
|
async def test_unversioned_always_passes(self):
|
|
"""Unversioned components pass through any filter."""
|
|
from fastmcp.server.transforms import VersionFilter
|
|
|
|
mcp = FastMCP()
|
|
|
|
@mcp.tool
|
|
def unversioned_tool() -> str:
|
|
return "unversioned"
|
|
|
|
@mcp.tool(version="5.0")
|
|
def versioned_tool() -> str:
|
|
return "v5"
|
|
|
|
# Filter that would exclude v5.0
|
|
mcp.add_transform(VersionFilter(version_lt="3.0"))
|
|
|
|
tools = await mcp.get_tools()
|
|
names = [t.name for t in tools]
|
|
assert "unversioned_tool" in names
|
|
assert "versioned_tool" not in names
|
|
|
|
async def test_date_versions(self):
|
|
"""Works with date-based versions like '2025-01-15'."""
|
|
from fastmcp.server.transforms import VersionFilter
|
|
|
|
mcp = FastMCP()
|
|
|
|
@mcp.tool(version="2025-01-01")
|
|
def report() -> str:
|
|
return "jan"
|
|
|
|
@mcp.tool(version="2025-06-01")
|
|
def report() -> str:
|
|
return "jun"
|
|
|
|
@mcp.tool(version="2025-12-01")
|
|
def report() -> str:
|
|
return "dec"
|
|
|
|
# Q1 API: before April
|
|
mcp.add_transform(VersionFilter(version_lt="2025-04-01"))
|
|
|
|
tools = await mcp.get_tools()
|
|
assert len(tools) == 1
|
|
assert tools[0].version == "2025-01-01"
|
|
|
|
async def test_get_tool_respects_filter(self):
|
|
"""get_tool() returns None if highest version is filtered out."""
|
|
|
|
from fastmcp.server.transforms import VersionFilter
|
|
|
|
mcp = FastMCP()
|
|
|
|
@mcp.tool(version="5.0")
|
|
def only_v5() -> str:
|
|
return "v5"
|
|
|
|
mcp.add_transform(VersionFilter(version_lt="3.0"))
|
|
|
|
# Tool exists but is filtered out - returns None (use _get_tool to apply transforms)
|
|
assert await mcp._get_tool("only_v5") is None
|
|
|
|
async def test_must_specify_at_least_one(self):
|
|
"""VersionFilter() with no args raises ValueError."""
|
|
import pytest
|
|
|
|
from fastmcp.server.transforms import VersionFilter
|
|
|
|
with pytest.raises(ValueError, match="At least one of"):
|
|
VersionFilter()
|
|
|
|
async def test_resources_filtered(self):
|
|
"""Resources are filtered by version."""
|
|
from fastmcp.server.transforms import VersionFilter
|
|
|
|
mcp = FastMCP()
|
|
|
|
@mcp.resource("file://config", version="1.0")
|
|
def config_v1() -> str:
|
|
return "v1"
|
|
|
|
@mcp.resource("file://config", version="2.0")
|
|
def config_v2() -> str:
|
|
return "v2"
|
|
|
|
mcp.add_transform(VersionFilter(version_lt="2.0"))
|
|
|
|
resources = await mcp.get_resources()
|
|
assert len(resources) == 1
|
|
assert resources[0].version == "1.0"
|
|
|
|
async def test_prompts_filtered(self):
|
|
"""Prompts are filtered by version."""
|
|
from fastmcp.server.transforms import VersionFilter
|
|
|
|
mcp = FastMCP()
|
|
|
|
@mcp.prompt(version="1.0")
|
|
def greet(name: str) -> str:
|
|
return f"Hi {name}"
|
|
|
|
@mcp.prompt(version="2.0")
|
|
def greet(name: str) -> str:
|
|
return f"Hello {name}"
|
|
|
|
mcp.add_transform(VersionFilter(version_lt="2.0"))
|
|
|
|
prompts = await mcp.get_prompts()
|
|
assert len(prompts) == 1
|
|
assert prompts[0].version == "1.0"
|
|
|
|
async def test_repr(self):
|
|
"""Test VersionFilter string representation."""
|
|
from fastmcp.server.transforms import VersionFilter
|
|
|
|
f1 = VersionFilter(version_lt="3.0")
|
|
assert repr(f1) == "VersionFilter(version_lt='3.0')"
|
|
|
|
f2 = VersionFilter(version_gte="2.0", version_lt="3.0")
|
|
assert repr(f2) == "VersionFilter(version_gte='2.0', version_lt='3.0')"
|
|
|
|
f3 = VersionFilter(version_gte="1.0")
|
|
assert repr(f3) == "VersionFilter(version_gte='1.0')"
|
|
|
|
|
|
class TestVersionMixingValidation:
|
|
"""Tests for versioned/unversioned mixing prevention."""
|
|
|
|
async def test_resource_mixing_rejected(self):
|
|
"""Cannot mix versioned and unversioned resources with the same URI."""
|
|
import pytest
|
|
|
|
mcp = FastMCP()
|
|
|
|
@mcp.resource("file://config", version="1.0")
|
|
def config_v1() -> str:
|
|
return "v1"
|
|
|
|
with pytest.raises(ValueError, match="unversioned.*versioned"):
|
|
|
|
@mcp.resource("file://config")
|
|
def config_unversioned() -> str:
|
|
return "unversioned"
|
|
|
|
async def test_prompt_mixing_rejected(self):
|
|
"""Cannot mix versioned and unversioned prompts with the same name."""
|
|
import pytest
|
|
|
|
mcp = FastMCP()
|
|
|
|
@mcp.prompt
|
|
def greet(name: str) -> str:
|
|
return f"Hello, {name}!"
|
|
|
|
with pytest.raises(ValueError, match="versioned.*unversioned"):
|
|
|
|
@mcp.prompt(version="1.0")
|
|
def greet(name: str) -> str:
|
|
return f"Hi, {name}!"
|
|
|
|
async def test_multiple_versions_allowed(self):
|
|
"""Multiple versioned components with same name are allowed."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.tool(version="1.0")
|
|
def calc() -> int:
|
|
return 1
|
|
|
|
@mcp.tool(version="2.0")
|
|
def calc() -> int:
|
|
return 2
|
|
|
|
@mcp.tool(version="3.0")
|
|
def calc() -> int:
|
|
return 3
|
|
|
|
# All versioned - this should work
|
|
tools = await mcp.get_tools()
|
|
assert len(tools) == 1
|
|
assert tools[0].version == "3.0"
|
|
|
|
|
|
class TestMountedVersionFiltering:
|
|
"""Tests for version filtering with mounted servers (FastMCPProvider).
|
|
|
|
Note: For mounted servers, list_* methods show what the child exposes (already
|
|
deduplicated to highest version). get_* methods support range filtering via
|
|
VersionSpec propagation to FastMCPProvider.
|
|
"""
|
|
|
|
async def test_mounted_get_tool_with_range_filter(self):
|
|
"""FastMCPProvider.get_tool applies range filtering from VersionSpec."""
|
|
from fastmcp.server.providers.fastmcp_provider import FastMCPProvider
|
|
from fastmcp.utilities.versions import VersionSpec
|
|
|
|
child = FastMCP("Child")
|
|
|
|
@child.tool(version="2.0")
|
|
def calc() -> int:
|
|
return 2
|
|
|
|
provider = FastMCPProvider(child)
|
|
|
|
# Without range spec, should return the tool
|
|
tool = await provider.get_tool("calc")
|
|
assert tool is not None
|
|
assert tool.version == "2.0"
|
|
|
|
# With range spec that excludes v2.0, should return None
|
|
tool = await provider.get_tool("calc", version=VersionSpec(lt="2.0"))
|
|
assert tool is None
|
|
|
|
# With range spec that includes v2.0, should return the tool
|
|
tool = await provider.get_tool("calc", version=VersionSpec(gte="2.0"))
|
|
assert tool is not None
|
|
assert tool.version == "2.0"
|
|
|
|
async def test_mounted_get_resource_with_range_filter(self):
|
|
"""FastMCPProvider.get_resource applies range filtering from VersionSpec."""
|
|
from fastmcp.server.providers.fastmcp_provider import FastMCPProvider
|
|
from fastmcp.utilities.versions import VersionSpec
|
|
|
|
child = FastMCP("Child")
|
|
|
|
@child.resource("file://data/", version="2.0")
|
|
def data() -> str:
|
|
return "data"
|
|
|
|
provider = FastMCPProvider(child)
|
|
|
|
# Without range spec, should return the resource
|
|
resource = await provider.get_resource("file://data/")
|
|
assert resource is not None
|
|
assert resource.version == "2.0"
|
|
|
|
# With range spec that excludes v2.0, should return None
|
|
resource = await provider.get_resource(
|
|
"file://data/", version=VersionSpec(lt="2.0")
|
|
)
|
|
assert resource is None
|
|
|
|
async def test_mounted_get_prompt_with_range_filter(self):
|
|
"""FastMCPProvider.get_prompt applies range filtering from VersionSpec."""
|
|
from fastmcp.server.providers.fastmcp_provider import FastMCPProvider
|
|
from fastmcp.utilities.versions import VersionSpec
|
|
|
|
child = FastMCP("Child")
|
|
|
|
@child.prompt(version="2.0")
|
|
def greet(name: str) -> str:
|
|
return f"Hello {name}"
|
|
|
|
provider = FastMCPProvider(child)
|
|
|
|
# Without range spec, should return the prompt
|
|
prompt = await provider.get_prompt("greet")
|
|
assert prompt is not None
|
|
assert prompt.version == "2.0"
|
|
|
|
# With range spec that excludes v2.0, should return None
|
|
prompt = await provider.get_prompt("greet", version=VersionSpec(lt="2.0"))
|
|
assert prompt is None
|
|
|
|
async def test_mounted_unversioned_passes_version_filter(self):
|
|
"""Unversioned components in mounted servers pass through version filters."""
|
|
from fastmcp.server.transforms import VersionFilter
|
|
|
|
child = FastMCP("Child")
|
|
|
|
@child.tool
|
|
def unversioned_tool() -> str:
|
|
return "unversioned"
|
|
|
|
parent = FastMCP("Parent")
|
|
parent.mount(child, "child")
|
|
parent.add_transform(VersionFilter(version_lt="3.0"))
|
|
|
|
# Unversioned should pass through
|
|
tools = await parent.get_tools()
|
|
assert len(tools) == 1
|
|
assert tools[0].name == "child_unversioned_tool"
|
|
assert tools[0].version is None
|
|
|
|
async def test_version_filter_filters_out_high_mounted_version(self):
|
|
"""VersionFilter hides mounted components outside the range."""
|
|
from fastmcp.server.transforms import VersionFilter
|
|
|
|
child = FastMCP("Child")
|
|
|
|
@child.tool(version="5.0")
|
|
def high_version_tool() -> int:
|
|
return 5
|
|
|
|
parent = FastMCP("Parent")
|
|
parent.mount(child, "child")
|
|
parent.add_transform(VersionFilter(version_lt="3.0"))
|
|
|
|
# v5.0 is outside the filter range, so it should be hidden
|
|
tools = await parent.get_tools()
|
|
assert len(tools) == 0
|
|
|
|
# _get_tool should also return None (respects filter, applies transforms)
|
|
assert await parent._get_tool("child_high_version_tool") is None
|
|
|
|
|
|
class TestMountedRangeFiltering:
|
|
"""Tests for version range filtering with mounted servers."""
|
|
|
|
async def test_mounted_lower_version_selected_by_filter(self):
|
|
"""When parent has filter <2.0 and child has v1.0+v3.0, should get v1.0."""
|
|
from fastmcp.server.transforms import VersionFilter
|
|
|
|
child = FastMCP("Child")
|
|
|
|
@child.tool(version="1.0")
|
|
def calc() -> int:
|
|
return 1
|
|
|
|
@child.tool(version="3.0")
|
|
def calc() -> int:
|
|
return 3
|
|
|
|
parent = FastMCP("Parent")
|
|
parent.mount(child, "child")
|
|
parent.add_transform(VersionFilter(version_lt="2.0"))
|
|
|
|
# Should return v1.0 (the highest version that matches <2.0)
|
|
# Use _get_tool to apply transforms
|
|
tool = await parent._get_tool("child_calc")
|
|
assert tool is not None
|
|
assert tool.version == "1.0"
|
|
|
|
async def test_explicit_version_honored_within_filter_range(self):
|
|
"""Explicit version="1.0" request should work within filter range."""
|
|
from fastmcp.server.transforms import VersionFilter
|
|
|
|
child = FastMCP("Child")
|
|
|
|
@child.tool(version="1.0")
|
|
def calc() -> int:
|
|
return 1
|
|
|
|
@child.tool(version="2.0")
|
|
def calc() -> int:
|
|
return 2
|
|
|
|
@child.tool(version="3.0")
|
|
def calc() -> int:
|
|
return 3
|
|
|
|
parent = FastMCP("Parent")
|
|
parent.mount(child, "child")
|
|
parent.add_transform(VersionFilter(version_gte="1.0", version_lt="3.0"))
|
|
|
|
# Request specific version within range (use _get_tool to apply transforms)
|
|
tool = await parent._get_tool("child_calc", VersionSpec(eq="1.0"))
|
|
assert tool is not None
|
|
assert tool.version == "1.0"
|
|
|
|
# Request version outside range should return None
|
|
result = await parent._get_tool("child_calc", VersionSpec(eq="3.0"))
|
|
assert result is None
|
|
|
|
|
|
class TestUnversionedExemption:
|
|
"""Tests confirming unversioned components bypass version filters."""
|
|
|
|
async def test_unversioned_bypasses_version_filter(self):
|
|
"""Unversioned components pass through any VersionFilter - by design."""
|
|
from fastmcp.server.transforms import VersionFilter
|
|
|
|
mcp = FastMCP()
|
|
|
|
@mcp.tool
|
|
def unversioned_tool() -> str:
|
|
return "unversioned"
|
|
|
|
@mcp.tool(version="5.0")
|
|
def versioned_tool() -> str:
|
|
return "v5"
|
|
|
|
# Filter that would exclude v5.0
|
|
mcp.add_transform(VersionFilter(version_lt="3.0"))
|
|
|
|
tools = await mcp.get_tools()
|
|
names = [t.name for t in tools]
|
|
|
|
# Unversioned passes through (exempt from filtering)
|
|
assert "unversioned_tool" in names
|
|
# Versioned is filtered out
|
|
assert "versioned_tool" not in names
|
|
|
|
async def test_unversioned_returned_for_exact_version_request(self):
|
|
"""Requesting exact version of unversioned tool returns the tool."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.tool
|
|
def my_tool() -> str:
|
|
return "unversioned"
|
|
|
|
# Even with explicit version request, unversioned tool is returned
|
|
# (it's the only version that exists, and unversioned matches any spec)
|
|
tool = await mcp.get_tool("my_tool", VersionSpec(eq="1.0"))
|
|
assert tool is not None
|
|
assert tool.version is None
|
|
|
|
async def test_unversioned_matches_any_version_spec(self):
|
|
"""VersionSpec.matches(None) returns True for any spec."""
|
|
from fastmcp.utilities.versions import VersionSpec
|
|
|
|
# Unversioned matches exact version specs
|
|
assert VersionSpec(eq="1.0").matches(None) is True
|
|
|
|
# Unversioned matches range specs
|
|
assert VersionSpec(gte="1.0", lt="3.0").matches(None) is True
|
|
|
|
# Unversioned matches open specs
|
|
assert VersionSpec(lt="5.0").matches(None) is True
|
|
assert VersionSpec(gte="1.0").matches(None) is True
|
|
|
|
|
|
class TestVersionValidation:
|
|
"""Tests for version string validation."""
|
|
|
|
async def test_version_with_at_symbol_rejected(self):
|
|
"""Version strings containing '@' should be rejected."""
|
|
import pytest
|
|
from pydantic import ValidationError
|
|
|
|
mcp = FastMCP()
|
|
|
|
with pytest.raises(ValidationError, match="cannot contain '@'"):
|
|
|
|
@mcp.tool(version="1.0@beta")
|
|
def my_tool() -> str:
|
|
return "test"
|
|
|
|
|
|
class TestVersionMetadata:
|
|
"""Tests for version metadata exposure in list operations."""
|
|
|
|
async def test_tool_versions_in_meta(self):
|
|
"""List tools should include versions list in meta."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.tool(version="1.0")
|
|
def add(x: int, y: int) -> int: # noqa: F811
|
|
return x + y
|
|
|
|
@mcp.tool(version="2.0")
|
|
def add(x: int, y: int) -> int: # noqa: F811
|
|
return x + y
|
|
|
|
tools = await mcp.get_tools()
|
|
assert len(tools) == 1
|
|
|
|
tool = tools[0]
|
|
meta = tool.get_meta()
|
|
assert meta["fastmcp"]["version"] == "2.0"
|
|
assert meta["fastmcp"]["versions"] == ["2.0", "1.0"]
|
|
|
|
async def test_resource_versions_in_meta(self):
|
|
"""List resources should include versions list in meta."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.resource("data://config", version="1.0")
|
|
def config_v1() -> str: # noqa: F811
|
|
return "v1"
|
|
|
|
@mcp.resource("data://config", version="2.0")
|
|
def config_v2() -> str: # noqa: F811
|
|
return "v2"
|
|
|
|
resources = await mcp.get_resources()
|
|
assert len(resources) == 1
|
|
|
|
resource = resources[0]
|
|
meta = resource.get_meta()
|
|
assert meta["fastmcp"]["version"] == "2.0"
|
|
assert meta["fastmcp"]["versions"] == ["2.0", "1.0"]
|
|
|
|
async def test_prompt_versions_in_meta(self):
|
|
"""List prompts should include versions list in meta."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.prompt(version="1.0")
|
|
def greet() -> str: # noqa: F811
|
|
return "Hello v1"
|
|
|
|
@mcp.prompt(version="2.0")
|
|
def greet() -> str: # noqa: F811
|
|
return "Hello v2"
|
|
|
|
prompts = await mcp.get_prompts()
|
|
assert len(prompts) == 1
|
|
|
|
prompt = prompts[0]
|
|
meta = prompt.get_meta()
|
|
assert meta["fastmcp"]["version"] == "2.0"
|
|
assert meta["fastmcp"]["versions"] == ["2.0", "1.0"]
|
|
|
|
async def test_unversioned_no_versions_list(self):
|
|
"""Unversioned components should not have versions list in meta."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.tool
|
|
def simple() -> str:
|
|
return "simple"
|
|
|
|
tools = await mcp.get_tools()
|
|
assert len(tools) == 1
|
|
|
|
tool = tools[0]
|
|
meta = tool.get_meta()
|
|
assert "versions" not in meta.get("fastmcp", {})
|
|
|
|
|
|
class TestVersionedCalls:
|
|
"""Tests for calling specific component versions."""
|
|
|
|
async def test_call_tool_with_version(self):
|
|
"""call_tool should use specified version."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.tool(version="1.0")
|
|
def calculate(x: int, y: int) -> int: # noqa: F811
|
|
return x + y
|
|
|
|
@mcp.tool(version="2.0")
|
|
def calculate(x: int, y: int) -> int: # noqa: F811
|
|
return x * y
|
|
|
|
# Default: highest version (2.0, multiplication)
|
|
result = await mcp.call_tool("calculate", {"x": 3, "y": 4})
|
|
assert result.structured_content is not None
|
|
assert result.structured_content["result"] == 12
|
|
|
|
# Explicit v1.0 (addition)
|
|
result = await mcp.call_tool(
|
|
"calculate", {"x": 3, "y": 4}, version=VersionSpec(eq="1.0")
|
|
)
|
|
assert result.structured_content is not None
|
|
assert result.structured_content["result"] == 7
|
|
|
|
# Explicit v2.0 (multiplication)
|
|
result = await mcp.call_tool(
|
|
"calculate", {"x": 3, "y": 4}, version=VersionSpec(eq="2.0")
|
|
)
|
|
assert result.structured_content is not None
|
|
assert result.structured_content["result"] == 12
|
|
|
|
async def test_read_resource_with_version(self):
|
|
"""read_resource should use specified version."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.resource("data://config", version="1.0")
|
|
def config() -> str: # noqa: F811
|
|
return "config v1"
|
|
|
|
@mcp.resource("data://config", version="2.0")
|
|
def config() -> str: # noqa: F811
|
|
return "config v2"
|
|
|
|
# Default: highest version
|
|
result = await mcp.read_resource("data://config")
|
|
assert result.contents[0].content == "config v2"
|
|
|
|
# Explicit v1.0
|
|
result = await mcp.read_resource("data://config", version=VersionSpec(eq="1.0"))
|
|
assert result.contents[0].content == "config v1"
|
|
|
|
async def test_render_prompt_with_version(self):
|
|
"""render_prompt should use specified version."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.prompt(version="1.0")
|
|
def greet() -> str: # noqa: F811
|
|
return "Hello from v1"
|
|
|
|
@mcp.prompt(version="2.0")
|
|
def greet() -> str: # noqa: F811
|
|
return "Hello from v2"
|
|
|
|
# Default: highest version
|
|
result = await mcp.render_prompt("greet")
|
|
content = result.messages[0].content
|
|
assert isinstance(content, TextContent) and content.text == "Hello from v2"
|
|
|
|
# Explicit v1.0
|
|
result = await mcp.render_prompt("greet", version=VersionSpec(eq="1.0"))
|
|
content = result.messages[0].content
|
|
assert isinstance(content, TextContent) and content.text == "Hello from v1"
|
|
|
|
async def test_call_tool_invalid_version_not_found(self):
|
|
"""Calling with non-existent version should raise NotFoundError."""
|
|
import pytest
|
|
|
|
from fastmcp.exceptions import NotFoundError
|
|
|
|
mcp = FastMCP()
|
|
|
|
@mcp.tool(version="1.0")
|
|
def mytool() -> str:
|
|
return "v1"
|
|
|
|
with pytest.raises(NotFoundError):
|
|
await mcp.call_tool("mytool", {}, version=VersionSpec(eq="999.0"))
|
|
|
|
|
|
class TestClientVersionSelection:
|
|
"""Tests for client-side version selection via the version parameter.
|
|
|
|
Version selection flows through request-level _meta, not arguments.
|
|
"""
|
|
|
|
import pytest
|
|
|
|
@pytest.mark.parametrize(
|
|
"version,expected",
|
|
[
|
|
(None, 10), # Default: highest version (2.0) -> 5 * 2
|
|
("1.0", 6), # v1.0 -> 5 + 1
|
|
("2.0", 10), # v2.0 -> 5 * 2
|
|
],
|
|
)
|
|
async def test_call_tool_version_selection(
|
|
self, version: str | None, expected: int
|
|
):
|
|
"""Client.call_tool routes to correct version via request meta."""
|
|
from fastmcp import Client
|
|
|
|
mcp = FastMCP()
|
|
|
|
@mcp.tool(version="1.0")
|
|
def calc(x: int) -> int: # noqa: F811
|
|
return x + 1
|
|
|
|
@mcp.tool(version="2.0")
|
|
def calc(x: int) -> int: # noqa: F811
|
|
return x * 2
|
|
|
|
async with Client(mcp) as client:
|
|
result = await client.call_tool("calc", {"x": 5}, version=version)
|
|
assert result.data == expected
|
|
|
|
@pytest.mark.parametrize(
|
|
"version,expected",
|
|
[
|
|
(None, "Hello world from v2"), # Default: highest version
|
|
("1.0", "Hello world from v1"),
|
|
("2.0", "Hello world from v2"),
|
|
],
|
|
)
|
|
async def test_get_prompt_version_selection(
|
|
self, version: str | None, expected: str
|
|
):
|
|
"""Client.get_prompt routes to correct version via request meta."""
|
|
from fastmcp import Client
|
|
|
|
mcp = FastMCP()
|
|
|
|
@mcp.prompt(version="1.0")
|
|
def greet(name: str) -> str: # noqa: F811
|
|
return f"Hello {name} from v1"
|
|
|
|
@mcp.prompt(version="2.0")
|
|
def greet(name: str) -> str: # noqa: F811
|
|
return f"Hello {name} from v2"
|
|
|
|
async with Client(mcp) as client:
|
|
result = await client.get_prompt(
|
|
"greet", {"name": "world"}, version=version
|
|
)
|
|
content = result.messages[0].content
|
|
assert isinstance(content, TextContent) and content.text == expected
|
|
|
|
@pytest.mark.parametrize(
|
|
"version,expected",
|
|
[
|
|
(None, "v2 data"), # Default: highest version
|
|
("1.0", "v1 data"),
|
|
("2.0", "v2 data"),
|
|
],
|
|
)
|
|
async def test_read_resource_version_selection(
|
|
self, version: str | None, expected: str
|
|
):
|
|
"""Client.read_resource routes to correct version via request meta."""
|
|
from fastmcp import Client
|
|
|
|
mcp = FastMCP()
|
|
|
|
@mcp.resource("data://info", version="1.0")
|
|
def info_v1() -> str: # noqa: F811
|
|
return "v1 data"
|
|
|
|
@mcp.resource("data://info", version="2.0")
|
|
def info_v2() -> str: # noqa: F811
|
|
return "v2 data"
|
|
|
|
async with Client(mcp) as client:
|
|
result = await client.read_resource("data://info", version=version)
|
|
assert result[0].text == expected # type: ignore[union-attr]
|