diff --git a/src/fastmcp/server/server.py b/src/fastmcp/server/server.py index fb652448b..6b7e13bf6 100644 --- a/src/fastmcp/server/server.py +++ b/src/fastmcp/server/server.py @@ -84,6 +84,7 @@ from fastmcp.utilities.logging import get_logger from fastmcp.utilities.types import FastMCPBaseModel, NotSet, NotSetT from fastmcp.utilities.versions import ( VersionSpec, + version_sort_key, ) if TYPE_CHECKING: @@ -584,6 +585,9 @@ class FastMCP( transforms (including session-level) have been applied. This ensures session transforms can override provider-level disables. + When the highest version is disabled and no explicit version was + requested, falls back to the next-highest enabled version. + Args: name: The tool name. version: Version filter (None returns highest version). @@ -597,9 +601,33 @@ class FastMCP( # Apply session transforms to single item tools = await apply_session_transforms([tool]) - if not tools or not is_enabled(tools[0]): + if tools and is_enabled(tools[0]): + return tools[0] + + # The highest version is disabled. If an explicit version was requested, + # respect the disable. Otherwise fall back to the next-highest enabled version. + if version is not None: return None - return tools[0] + + all_tools = [t for t in await super().list_tools() if t.name == name] + all_tools = list(await apply_session_transforms(all_tools)) + enabled = [t for t in all_tools if is_enabled(t)] + + skip_auth, token = _get_auth_context() + authorized: list[Tool] = [] + for t in enabled: + if not skip_auth and t.auth is not None: + ctx = AuthContext(token=token, component=t) + try: + if not await run_auth_checks(t.auth, ctx): + continue + except AuthorizationError: + continue + authorized.append(t) + + if not authorized: + return None + return cast(Tool, max(authorized, key=version_sort_key)) async def list_resources( self, *, run_middleware: bool = True @@ -681,6 +709,9 @@ class FastMCP( Overrides Provider.get_resource() to add visibility filtering after all transforms (including session-level) have been applied. + When the highest version is disabled and no explicit version was + requested, falls back to the next-highest enabled version. + Args: uri: The resource URI. version: Version filter (None returns highest version). @@ -694,9 +725,31 @@ class FastMCP( # Apply session transforms to single item resources = await apply_session_transforms([resource]) - if not resources or not is_enabled(resources[0]): + if resources and is_enabled(resources[0]): + return resources[0] + + if version is not None: return None - return resources[0] + + all_resources = [r for r in await super().list_resources() if str(r.uri) == uri] + all_resources = list(await apply_session_transforms(all_resources)) + enabled = [r for r in all_resources if is_enabled(r)] + + skip_auth, token = _get_auth_context() + authorized: list[Resource] = [] + for r in enabled: + if not skip_auth and r.auth is not None: + ctx = AuthContext(token=token, component=r) + try: + if not await run_auth_checks(r.auth, ctx): + continue + except AuthorizationError: + continue + authorized.append(r) + + if not authorized: + return None + return cast(Resource, max(authorized, key=version_sort_key)) async def list_resource_templates( self, *, run_middleware: bool = True @@ -780,6 +833,9 @@ class FastMCP( Overrides Provider.get_resource_template() to add visibility filtering after all transforms (including session-level) have been applied. + When the highest version is disabled and no explicit version was + requested, falls back to the next-highest enabled version. + Args: uri: The template URI. version: Version filter (None returns highest version). @@ -793,9 +849,35 @@ class FastMCP( # Apply session transforms to single item templates = await apply_session_transforms([template]) - if not templates or not is_enabled(templates[0]): + if templates and is_enabled(templates[0]): + return templates[0] + + if version is not None: return None - return templates[0] + + all_templates = [ + t + for t in await super().list_resource_templates() + if t.matches(uri) is not None + ] + all_templates = list(await apply_session_transforms(all_templates)) + enabled = [t for t in all_templates if is_enabled(t)] + + skip_auth, token = _get_auth_context() + authorized: list[ResourceTemplate] = [] + for t in enabled: + if not skip_auth and t.auth is not None: + ctx = AuthContext(token=token, component=t) + try: + if not await run_auth_checks(t.auth, ctx): + continue + except AuthorizationError: + continue + authorized.append(t) + + if not authorized: + return None + return cast(ResourceTemplate, max(authorized, key=version_sort_key)) async def list_prompts(self, *, run_middleware: bool = True) -> Sequence[Prompt]: """List all enabled prompts from providers. @@ -875,6 +957,9 @@ class FastMCP( Overrides Provider.get_prompt() to add visibility filtering after all transforms (including session-level) have been applied. + When the highest version is disabled and no explicit version was + requested, falls back to the next-highest enabled version. + Args: name: The prompt name. version: Version filter (None returns highest version). @@ -888,9 +973,31 @@ class FastMCP( # Apply session transforms to single item prompts = await apply_session_transforms([prompt]) - if not prompts or not is_enabled(prompts[0]): + if prompts and is_enabled(prompts[0]): + return prompts[0] + + if version is not None: return None - return prompts[0] + + all_prompts = [p for p in await super().list_prompts() if p.name == name] + all_prompts = list(await apply_session_transforms(all_prompts)) + enabled = [p for p in all_prompts if is_enabled(p)] + + skip_auth, token = _get_auth_context() + authorized: list[Prompt] = [] + for p in enabled: + if not skip_auth and p.auth is not None: + ctx = AuthContext(token=token, component=p) + try: + if not await run_auth_checks(p.auth, ctx): + continue + except AuthorizationError: + continue + authorized.append(p) + + if not authorized: + return None + return cast(Prompt, max(authorized, key=version_sort_key)) @overload async def call_tool( diff --git a/tests/server/versioning/test_visibility_version_fallback.py b/tests/server/versioning/test_visibility_version_fallback.py new file mode 100644 index 000000000..071c77d6e --- /dev/null +++ b/tests/server/versioning/test_visibility_version_fallback.py @@ -0,0 +1,376 @@ +"""Tests for version fallback when the highest version is disabled via visibility. + +Regression tests for https://github.com/jlowin/fastmcp/issues/3421: +When the latest version of a component is disabled, get_* methods should +fall back to the next-highest enabled version instead of returning None. +""" +# ruff: noqa: F811 # Intentional function redefinition for version testing + +from __future__ import annotations + +from mcp.server.auth.middleware.auth_context import auth_context_var +from mcp.server.auth.middleware.bearer_auth import AuthenticatedUser +from mcp.types import TextContent + +from fastmcp import FastMCP +from fastmcp.server.auth import AccessToken, require_scopes +from fastmcp.utilities.versions import VersionSpec + + +def _make_token(scopes: list[str] | None = None) -> AccessToken: + """Create a test access token.""" + return AccessToken( + token="test-token", + client_id="test-client", + scopes=scopes or [], + expires_at=None, + claims={}, + ) + + +def _set_token(token: AccessToken | None): + """Set the access token in the auth context var.""" + if token is None: + return auth_context_var.set(None) + return auth_context_var.set(AuthenticatedUser(token)) + + +class TestToolVersionFallback: + """Test that disabling the latest tool version falls back correctly.""" + + async def test_list_tools_shows_v1_when_v2_disabled(self): + """list_tools should show v1 when v2 is disabled.""" + mcp = FastMCP() + + @mcp.tool(version="1.0") + def calc() -> int: + return 1 + + @mcp.tool(version="2.0") + def calc() -> int: + return 2 + + mcp.disable(version=VersionSpec(eq="2.0")) + + tools = await mcp.list_tools() + assert len(tools) == 1 + assert tools[0].name == "calc" + assert tools[0].version == "1.0" + + async def test_get_tool_returns_v1_when_v2_disabled(self): + """get_tool should return v1 when v2 is disabled (core bug).""" + mcp = FastMCP() + + @mcp.tool(version="1.0") + def calc() -> int: + return 1 + + @mcp.tool(version="2.0") + def calc() -> int: + return 2 + + mcp.disable(version=VersionSpec(eq="2.0")) + + tool = await mcp.get_tool("calc") + assert tool is not None + assert tool.version == "1.0" + + async def test_call_tool_uses_v1_when_v2_disabled(self): + """call_tool should invoke v1 when v2 is disabled.""" + mcp = FastMCP() + + @mcp.tool(version="1.0") + def calc() -> int: + return 1 + + @mcp.tool(version="2.0") + def calc() -> int: + return 2 + + mcp.disable(version=VersionSpec(eq="2.0")) + + result = await mcp.call_tool("calc", {}) + first = result.content[0] + assert isinstance(first, TextContent) + assert first.text == "1" + + async def test_get_tool_explicit_disabled_version_returns_none(self): + """Requesting a specific disabled version should return None.""" + mcp = FastMCP() + + @mcp.tool(version="1.0") + def calc() -> int: + return 1 + + @mcp.tool(version="2.0") + def calc() -> int: + return 2 + + mcp.disable(version=VersionSpec(eq="2.0")) + + tool = await mcp.get_tool("calc", VersionSpec(eq="2.0")) + assert tool is None + + async def test_get_tool_all_versions_disabled_returns_none(self): + """When all versions are disabled, get_tool returns None.""" + mcp = FastMCP() + + @mcp.tool(version="1.0") + def calc() -> int: + return 1 + + @mcp.tool(version="2.0") + def calc() -> int: + return 2 + + mcp.disable(names={"calc"}) + + tool = await mcp.get_tool("calc") + assert tool is None + + async def test_get_tool_middle_version_fallback(self): + """Disabling v3 should fall back to v2, not v1.""" + 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 + + mcp.disable(version=VersionSpec(eq="3.0")) + + tool = await mcp.get_tool("calc") + assert tool is not None + assert tool.version == "2.0" + + +class TestResourceVersionFallback: + """Test that disabling the latest resource version falls back correctly.""" + + async def test_get_resource_returns_v1_when_v2_disabled(self): + """get_resource should return v1 when v2 is disabled.""" + mcp = FastMCP() + + @mcp.resource("data://info", version="1.0") + def info() -> str: + return "v1" + + @mcp.resource("data://info", version="2.0") + def info() -> str: + return "v2" + + mcp.disable(version=VersionSpec(eq="2.0")) + + resource = await mcp.get_resource("data://info") + assert resource is not None + assert resource.version == "1.0" + + async def test_get_resource_explicit_disabled_version_returns_none(self): + """Requesting a specific disabled resource version should return None.""" + mcp = FastMCP() + + @mcp.resource("data://info", version="1.0") + def info() -> str: + return "v1" + + @mcp.resource("data://info", version="2.0") + def info() -> str: + return "v2" + + mcp.disable(version=VersionSpec(eq="2.0")) + + resource = await mcp.get_resource("data://info", VersionSpec(eq="2.0")) + assert resource is None + + +class TestResourceTemplateVersionFallback: + """Test that disabling the latest template version falls back correctly.""" + + async def test_get_resource_template_returns_v1_when_v2_disabled(self): + """get_resource_template should return v1 when v2 is disabled.""" + mcp = FastMCP() + + @mcp.resource("data://items/{id}", version="1.0") + def item(id: str) -> str: + return f"v1-{id}" + + @mcp.resource("data://items/{id}", version="2.0") + def item(id: str) -> str: + return f"v2-{id}" + + mcp.disable(version=VersionSpec(eq="2.0")) + + template = await mcp.get_resource_template("data://items/{id}") + assert template is not None + assert template.version == "1.0" + + +class TestPromptVersionFallback: + """Test that disabling the latest prompt version falls back correctly.""" + + async def test_get_prompt_returns_v1_when_v2_disabled(self): + """get_prompt should return v1 when v2 is disabled.""" + mcp = FastMCP() + + @mcp.prompt(version="1.0") + def greet() -> str: + return "hello v1" + + @mcp.prompt(version="2.0") + def greet() -> str: + return "hello v2" + + mcp.disable(version=VersionSpec(eq="2.0")) + + prompt = await mcp.get_prompt("greet") + assert prompt is not None + assert prompt.version == "1.0" + + async def test_get_prompt_explicit_disabled_version_returns_none(self): + """Requesting a specific disabled prompt version should return None.""" + mcp = FastMCP() + + @mcp.prompt(version="1.0") + def greet() -> str: + return "hello v1" + + @mcp.prompt(version="2.0") + def greet() -> str: + return "hello v2" + + mcp.disable(version=VersionSpec(eq="2.0")) + + prompt = await mcp.get_prompt("greet", VersionSpec(eq="2.0")) + assert prompt is None + + +class TestFallbackRespectsAuth: + """Fallback to older versions must enforce auth checks. + + When the highest version is disabled and the code falls back to older + versions, those candidates must go through auth filtering. Otherwise + a protected v1 could be exposed to unauthenticated users when a + public v2 is disabled. + """ + + async def test_tool_fallback_respects_auth(self): + """Disabling v2 should not expose auth-protected v1 to unauthorized users.""" + mcp = FastMCP() + + @mcp.tool(version="1.0", auth=require_scopes("admin")) + def calc() -> int: + return 1 + + @mcp.tool(version="2.0") + def calc() -> int: + return 2 + + mcp.disable(version=VersionSpec(eq="2.0")) + + # Without an admin token, v1 should NOT be returned + tool = await mcp.get_tool("calc") + assert tool is None + + async def test_tool_fallback_allows_authorized_user(self): + """Fallback should return auth-protected v1 to authorized users.""" + mcp = FastMCP() + + @mcp.tool(version="1.0", auth=require_scopes("admin")) + def calc() -> int: + return 1 + + @mcp.tool(version="2.0") + def calc() -> int: + return 2 + + mcp.disable(version=VersionSpec(eq="2.0")) + + token = _make_token(scopes=["admin"]) + tok = _set_token(token) + try: + tool = await mcp.get_tool("calc") + assert tool is not None + assert tool.version == "1.0" + finally: + auth_context_var.reset(tok) + + async def test_resource_fallback_respects_auth(self): + """Disabling v2 should not expose auth-protected v1 resource.""" + mcp = FastMCP() + + @mcp.resource("data://info", version="1.0", auth=require_scopes("admin")) + def info() -> str: + return "v1" + + @mcp.resource("data://info", version="2.0") + def info() -> str: + return "v2" + + mcp.disable(version=VersionSpec(eq="2.0")) + + resource = await mcp.get_resource("data://info") + assert resource is None + + async def test_resource_template_fallback_respects_auth(self): + """Disabling v2 should not expose auth-protected v1 template.""" + mcp = FastMCP() + + @mcp.resource("data://items/{id}", version="1.0", auth=require_scopes("admin")) + def item(id: str) -> str: + return f"v1-{id}" + + @mcp.resource("data://items/{id}", version="2.0") + def item(id: str) -> str: + return f"v2-{id}" + + mcp.disable(version=VersionSpec(eq="2.0")) + + template = await mcp.get_resource_template("data://items/{id}") + assert template is None + + async def test_prompt_fallback_respects_auth(self): + """Disabling v2 should not expose auth-protected v1 prompt.""" + mcp = FastMCP() + + @mcp.prompt(version="1.0", auth=require_scopes("admin")) + def greet() -> str: + return "hello v1" + + @mcp.prompt(version="2.0") + def greet() -> str: + return "hello v2" + + mcp.disable(version=VersionSpec(eq="2.0")) + + prompt = await mcp.get_prompt("greet") + assert prompt is None + + async def test_fallback_skips_unauthorized_picks_next(self): + """When multiple fallback candidates exist, skip unauthorized ones.""" + mcp = FastMCP() + + @mcp.tool(version="1.0") + def calc() -> int: + return 1 + + @mcp.tool(version="2.0", auth=require_scopes("admin")) + def calc() -> int: + return 2 + + @mcp.tool(version="3.0") + def calc() -> int: + return 3 + + mcp.disable(version=VersionSpec(eq="3.0")) + + # v2 requires admin, so unauthorized user should get v1 + tool = await mcp.get_tool("calc") + assert tool is not None + assert tool.version == "1.0"