mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-10 23:59:10 +02:00
f9ed06176a* fix: enforce auth/visibility in ResourcesAsTools and PromptsAsTools for non-FastMCP providers
🤖 Co-authored-by: Claude <noreply@anthropic.com>
* fix: honor stdio auth bypass and correct transform ordering in provider wrappers
Co-authored-by: Claude <noreply@anthropic.com>
* fix: move context/dependencies imports into function to break circular import
* fix: route ResourcesAsTools/PromptsAsTools through ctx.fastmcp
Instead of manually reimplementing auth, visibility, and session
transforms in the transform layer, tool functions now call
ctx.fastmcp.read_resource() / ctx.fastmcp.render_prompt() which
routes through the server's full middleware chain. This matches
the pattern CodeMode uses with ctx.fastmcp.call_tool().
The isinstance(provider, FastMCP) branching is removed entirely.
* feat: add _scope parameter for provider-scoped listing
AggregateProvider can now filter which child providers to query when
listing components. ResourcesAsTools and PromptsAsTools use this to
scope listings to their configured provider while still routing
through ctx.fastmcp for full middleware coverage.
The scope matching walks wrapped providers, so a
WrappedProvider(Namespace, inner=MyProvider) matches if MyProvider
is in the scope list.
* test: add coverage for ResourcesAsTools scoped to a sub-server
* fix: delegate to super() when _scope is None, add AggregateProvider to scope matching
* simplify: remove _scope machinery, route everything through ctx.fastmcp
Reverts the _scope parameter from Provider/AggregateProvider/Server.
ResourcesAsTools and PromptsAsTools now simply route through
ctx.fastmcp for all operations. Apply to a FastMCP server instance
for proper auth/visibility/middleware coverage.
Tests rewritten to use FastMCP server directly instead of raw providers.
* warn when ResourcesAsTools/PromptsAsTools is applied to a non-FastMCP provider
* docs: explain that ResourcesAsTools/PromptsAsTools should wrap a FastMCP server
* raise TypeError instead of warning when applied to non-FastMCP provider
---------
Co-authored-by: Claude <noreply@anthropic.com>
358 lines
12 KiB
Python
358 lines
12 KiB
Python
"""Tests for ResourcesAsTools transform."""
|
|
|
|
import base64
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from fastmcp import FastMCP
|
|
from fastmcp.client import Client
|
|
from fastmcp.exceptions import ToolError
|
|
from fastmcp.server.auth import AuthContext
|
|
from fastmcp.server.transforms import ResourcesAsTools
|
|
|
|
|
|
class TestResourcesAsToolsBasic:
|
|
"""Test basic ResourcesAsTools functionality."""
|
|
|
|
async def test_adds_list_resources_tool(self):
|
|
"""Transform adds list_resources tool."""
|
|
mcp = FastMCP("Test")
|
|
mcp.add_transform(ResourcesAsTools(mcp))
|
|
|
|
async with Client(mcp) as client:
|
|
tools = await client.list_tools()
|
|
tool_names = {t.name for t in tools}
|
|
assert "list_resources" in tool_names
|
|
|
|
async def test_adds_read_resource_tool(self):
|
|
"""Transform adds read_resource tool."""
|
|
mcp = FastMCP("Test")
|
|
mcp.add_transform(ResourcesAsTools(mcp))
|
|
|
|
async with Client(mcp) as client:
|
|
tools = await client.list_tools()
|
|
tool_names = {t.name for t in tools}
|
|
assert "read_resource" in tool_names
|
|
|
|
async def test_preserves_existing_tools(self):
|
|
"""Transform preserves existing tools."""
|
|
mcp = FastMCP("Test")
|
|
|
|
@mcp.tool
|
|
def my_tool() -> str:
|
|
return "result"
|
|
|
|
mcp.add_transform(ResourcesAsTools(mcp))
|
|
|
|
async with Client(mcp) as client:
|
|
tools = await client.list_tools()
|
|
tool_names = {t.name for t in tools}
|
|
assert "my_tool" in tool_names
|
|
assert "list_resources" in tool_names
|
|
assert "read_resource" in tool_names
|
|
|
|
|
|
class TestListResourcesTool:
|
|
"""Test the list_resources tool."""
|
|
|
|
async def test_lists_static_resources(self):
|
|
"""list_resources returns static resources with uri."""
|
|
mcp = FastMCP("Test")
|
|
|
|
@mcp.resource("config://app")
|
|
def app_config() -> str:
|
|
return "config data"
|
|
|
|
mcp.add_transform(ResourcesAsTools(mcp))
|
|
|
|
async with Client(mcp) as client:
|
|
result = await client.call_tool("list_resources", {})
|
|
resources = json.loads(result.data)
|
|
|
|
assert len(resources) == 1
|
|
assert resources[0]["uri"] == "config://app"
|
|
assert resources[0]["name"] == "app_config"
|
|
|
|
async def test_lists_resource_templates(self):
|
|
"""list_resources returns templates with uri_template."""
|
|
mcp = FastMCP("Test")
|
|
|
|
@mcp.resource("file://{path}")
|
|
def read_file(path: str) -> str:
|
|
return f"content of {path}"
|
|
|
|
mcp.add_transform(ResourcesAsTools(mcp))
|
|
|
|
async with Client(mcp) as client:
|
|
result = await client.call_tool("list_resources", {})
|
|
resources = json.loads(result.data)
|
|
|
|
assert len(resources) == 1
|
|
assert resources[0]["uri_template"] == "file://{path}"
|
|
assert "uri" not in resources[0]
|
|
|
|
async def test_lists_both_resources_and_templates(self):
|
|
"""list_resources returns both static and templated resources."""
|
|
mcp = FastMCP("Test")
|
|
|
|
@mcp.resource("config://app")
|
|
def app_config() -> str:
|
|
return "config"
|
|
|
|
@mcp.resource("file://{path}")
|
|
def read_file(path: str) -> str:
|
|
return f"content of {path}"
|
|
|
|
mcp.add_transform(ResourcesAsTools(mcp))
|
|
|
|
async with Client(mcp) as client:
|
|
result = await client.call_tool("list_resources", {})
|
|
resources = json.loads(result.data)
|
|
|
|
assert len(resources) == 2
|
|
# One has uri, one has uri_template
|
|
uris = [r.get("uri") for r in resources if r.get("uri")]
|
|
templates = [
|
|
r.get("uri_template") for r in resources if r.get("uri_template")
|
|
]
|
|
assert uris == ["config://app"]
|
|
assert templates == ["file://{path}"]
|
|
|
|
async def test_empty_when_no_resources(self):
|
|
"""list_resources returns empty list when no resources exist."""
|
|
mcp = FastMCP("Test")
|
|
mcp.add_transform(ResourcesAsTools(mcp))
|
|
|
|
async with Client(mcp) as client:
|
|
result = await client.call_tool("list_resources", {})
|
|
assert json.loads(result.data) == []
|
|
|
|
|
|
class TestReadResourceTool:
|
|
"""Test the read_resource tool."""
|
|
|
|
async def test_reads_static_resource(self):
|
|
"""read_resource reads a static resource by URI."""
|
|
mcp = FastMCP("Test")
|
|
|
|
@mcp.resource("config://app")
|
|
def app_config() -> str:
|
|
return "my config data"
|
|
|
|
mcp.add_transform(ResourcesAsTools(mcp))
|
|
|
|
async with Client(mcp) as client:
|
|
result = await client.call_tool("read_resource", {"uri": "config://app"})
|
|
assert result.data == "my config data"
|
|
|
|
async def test_reads_templated_resource(self):
|
|
"""read_resource reads a templated resource with parameters."""
|
|
mcp = FastMCP("Test")
|
|
|
|
@mcp.resource("user://{user_id}/profile")
|
|
def user_profile(user_id: str) -> str:
|
|
return f"Profile for user {user_id}"
|
|
|
|
mcp.add_transform(ResourcesAsTools(mcp))
|
|
|
|
async with Client(mcp) as client:
|
|
result = await client.call_tool(
|
|
"read_resource", {"uri": "user://123/profile"}
|
|
)
|
|
assert result.data == "Profile for user 123"
|
|
|
|
async def test_error_on_unknown_resource(self):
|
|
"""read_resource raises error for unknown URI."""
|
|
from fastmcp.exceptions import ToolError
|
|
|
|
mcp = FastMCP("Test")
|
|
mcp.add_transform(ResourcesAsTools(mcp))
|
|
|
|
async with Client(mcp) as client:
|
|
with pytest.raises(ToolError, match="Unknown resource"):
|
|
await client.call_tool("read_resource", {"uri": "unknown://resource"})
|
|
|
|
async def test_reads_binary_as_base64(self):
|
|
"""read_resource returns binary content as base64."""
|
|
mcp = FastMCP("Test")
|
|
|
|
@mcp.resource("data://binary", mime_type="application/octet-stream")
|
|
def binary_data() -> bytes:
|
|
return b"\x00\x01\x02\x03"
|
|
|
|
mcp.add_transform(ResourcesAsTools(mcp))
|
|
|
|
async with Client(mcp) as client:
|
|
result = await client.call_tool("read_resource", {"uri": "data://binary"})
|
|
# Should be base64 encoded
|
|
decoded = base64.b64decode(result.data)
|
|
assert decoded == b"\x00\x01\x02\x03"
|
|
|
|
|
|
class TestResourcesAsToolsWithNamespace:
|
|
"""Test ResourcesAsTools combined with other transforms."""
|
|
|
|
async def test_works_with_namespace_on_provider(self):
|
|
"""ResourcesAsTools works when provider has Namespace transform."""
|
|
from fastmcp.server.providers import FastMCPProvider
|
|
from fastmcp.server.transforms import Namespace
|
|
|
|
sub = FastMCP("Sub")
|
|
|
|
@sub.resource("config://app")
|
|
def app_config() -> str:
|
|
return "sub config"
|
|
|
|
main = FastMCP("Main")
|
|
provider = FastMCPProvider(sub)
|
|
provider.add_transform(Namespace("sub"))
|
|
main.add_provider(provider)
|
|
main.add_transform(ResourcesAsTools(main))
|
|
|
|
async with Client(main) as client:
|
|
result = await client.call_tool("list_resources", {})
|
|
resources = json.loads(result.data)
|
|
|
|
# Resource should have namespaced URI
|
|
assert len(resources) == 1
|
|
assert resources[0]["uri"] == "config://sub/app"
|
|
|
|
|
|
class TestResourcesAsToolsRepr:
|
|
"""Test ResourcesAsTools repr."""
|
|
|
|
def test_repr(self):
|
|
"""Transform has useful repr."""
|
|
mcp = FastMCP("Test")
|
|
transform = ResourcesAsTools(mcp)
|
|
assert "ResourcesAsTools" in repr(transform)
|
|
|
|
|
|
class TestResourcesAsToolsAnnotations:
|
|
"""Test ToolAnnotations on generated resource tools."""
|
|
|
|
async def test_list_resources_is_read_only(self):
|
|
"""list_resources is annotated as read-only by default."""
|
|
mcp = FastMCP("Test")
|
|
mcp.add_transform(ResourcesAsTools(mcp))
|
|
|
|
async with Client(mcp) as client:
|
|
tools = await client.list_tools()
|
|
tool = next(t for t in tools if t.name == "list_resources")
|
|
assert tool.annotations is not None
|
|
assert tool.annotations.readOnlyHint is True
|
|
|
|
async def test_read_resource_is_read_only(self):
|
|
"""read_resource is annotated as read-only by default."""
|
|
mcp = FastMCP("Test")
|
|
mcp.add_transform(ResourcesAsTools(mcp))
|
|
|
|
async with Client(mcp) as client:
|
|
tools = await client.list_tools()
|
|
tool = next(t for t in tools if t.name == "read_resource")
|
|
assert tool.annotations is not None
|
|
assert tool.annotations.readOnlyHint is True
|
|
|
|
|
|
def _deny_all(ctx: AuthContext) -> bool:
|
|
"""Auth check that always denies access."""
|
|
return False
|
|
|
|
|
|
class TestResourcesAsToolsAuthOnServer:
|
|
"""Auth checks work when using ResourcesAsTools on a FastMCP server."""
|
|
|
|
async def test_auth_protected_resources_hidden_from_list(self):
|
|
"""Auth-protected resources are filtered from list_resources tool."""
|
|
mcp = FastMCP("Test")
|
|
|
|
@mcp.resource("test://open")
|
|
def open_resource() -> str:
|
|
return "open content"
|
|
|
|
@mcp.resource("test://protected", auth=_deny_all)
|
|
def protected_resource() -> str:
|
|
return "protected content"
|
|
|
|
mcp.add_transform(ResourcesAsTools(mcp))
|
|
|
|
async with Client(mcp) as client:
|
|
result = await client.call_tool("list_resources", {})
|
|
items = json.loads(result.data)
|
|
uris = [r.get("uri") for r in items if r.get("uri")]
|
|
assert "test://open" in uris
|
|
assert "test://protected" not in uris
|
|
|
|
async def test_auth_protected_resource_cannot_be_read(self):
|
|
"""Auth-protected resources cannot be read via read_resource tool."""
|
|
mcp = FastMCP("Test")
|
|
|
|
@mcp.resource("test://protected", auth=_deny_all)
|
|
def protected_resource() -> str:
|
|
return "protected content"
|
|
|
|
mcp.add_transform(ResourcesAsTools(mcp))
|
|
|
|
async with Client(mcp) as client:
|
|
with pytest.raises(ToolError):
|
|
await client.call_tool("read_resource", {"uri": "test://protected"})
|
|
|
|
async def test_open_resource_still_accessible(self):
|
|
"""Non-auth-protected resources can still be read."""
|
|
mcp = FastMCP("Test")
|
|
|
|
@mcp.resource("test://open")
|
|
def open_resource() -> str:
|
|
return "open content"
|
|
|
|
@mcp.resource("test://protected", auth=_deny_all)
|
|
def protected_resource() -> str:
|
|
return "protected content"
|
|
|
|
mcp.add_transform(ResourcesAsTools(mcp))
|
|
|
|
async with Client(mcp) as client:
|
|
result = await client.call_tool("read_resource", {"uri": "test://open"})
|
|
assert result.data == "open content"
|
|
|
|
|
|
class TestResourcesAsToolsVisibilityOnServer:
|
|
"""Visibility filtering works when using ResourcesAsTools on a server."""
|
|
|
|
async def test_disabled_resources_hidden_from_list(self):
|
|
"""Disabled resources are not listed via list_resources tool."""
|
|
mcp = FastMCP("Test")
|
|
|
|
@mcp.resource("test://public")
|
|
def public_resource() -> str:
|
|
return "public content"
|
|
|
|
@mcp.resource("test://secret")
|
|
def secret_resource() -> str:
|
|
return "secret content"
|
|
|
|
mcp.disable(names={"test://secret"})
|
|
mcp.add_transform(ResourcesAsTools(mcp))
|
|
|
|
async with Client(mcp) as client:
|
|
result = await client.call_tool("list_resources", {})
|
|
items = json.loads(result.data)
|
|
uris = [r.get("uri") for r in items if r.get("uri")]
|
|
assert "test://public" in uris
|
|
assert "test://secret" not in uris
|
|
|
|
async def test_disabled_resource_cannot_be_read(self):
|
|
"""Disabled resources cannot be read via read_resource tool."""
|
|
mcp = FastMCP("Test")
|
|
|
|
@mcp.resource("test://secret")
|
|
def secret_resource() -> str:
|
|
return "secret content"
|
|
|
|
mcp.disable(names={"test://secret"})
|
|
mcp.add_transform(ResourcesAsTools(mcp))
|
|
|
|
async with Client(mcp) as client:
|
|
with pytest.raises(ToolError):
|
|
await client.call_tool("read_resource", {"uri": "test://secret"})
|