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

Fix get_* returning None when latest version is disabled (#3439)

Browse source 
* Fix get_* returning None when latest version is disabled (#3421)

When a visibility transform disabled the highest version of a component,
get_tool/get_resource/get_resource_template/get_prompt returned None
instead of falling back to the next-highest enabled version. The list_*
path already worked correctly because deduplication runs after visibility
filtering. The get_* path now falls back to listing all versions and
picking the highest enabled one when the top version is disabled.

* Apply auth checks in version fallback paths

The fallback code in get_tool, get_resource, get_resource_template, and
get_prompt bypassed auth filtering when falling back to older versions
after the highest version was disabled. This could expose auth-protected
older versions to unauthorized users.
This commit is contained in:
Jeremiah Lowin 2026-03-07 11:55:10 -05:00 • committed by GitHub
commit 9f8347dbaf
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 491 additions and 8 deletions
Download patch file Download diff file Expand all files Collapse all files

123
src/fastmcp/server/server.py
View file

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

376
tests/server/versioning/test_visibility_version_fallback.py Normal file
View file

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