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

fix: route ResourcesAsTools/PromptsAsTools through server middleware (#3495)

Browse source 
* 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>
This commit is contained in:
Jeremiah Lowin 2026-03-15 16:35:54 -04:00 • committed by GitHub
commit f9ed06176a
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 177 additions and 83 deletions
Download patch file Download diff file Expand all files Collapse all files

1
CLAUDE.md
View file

@ -111,6 +111,7 @@ When modifying MCP functionality, changes typically need to be applied across al
- **Structure:** Headers form navigation guide, logical H2/H3 hierarchy
- **Content:** User-focused sections, motivate features (why) before mechanics (how)
- **Style:** Prose over code comments for important information
- **Docstrings:** FastMCP docstrings are automatically compiled into MDX documents. Use markdown (single backticks, fenced code blocks), not RST (no double backticks). Bare `{}` in examples will be interpreted as JSX — wrap in backticks instead.
## Critical Patterns

6
docs/servers/transforms/prompts-as-tools.mdx
View file

@ -21,7 +21,11 @@ This means any client that can call tools can now access prompts, even if the cl
## Basic Usage
Pass your server to `PromptsAsTools` when adding the transform. The transform queries that server for prompts whenever the generated tools are called.
Pass your FastMCP server to `PromptsAsTools` when adding the transform. The generated tools route through the server at runtime, which means all server middleware — auth, visibility, rate limiting — applies to prompt operations automatically, exactly as it would for direct `prompts/get` calls.
<Note>
`PromptsAsTools` (and `ResourcesAsTools`) should be applied to a FastMCP server instance, not a raw Provider. The generated tools call back into the server's middleware chain at runtime, so they need a server to route through. If you want to expose only a subset of prompts, create a dedicated FastMCP server for those prompts and apply the transform there.
</Note>
```python
from fastmcp import FastMCP

6
docs/servers/transforms/resources-as-tools.mdx
View file

@ -21,7 +21,11 @@ This means any client that can call tools can now access resources, even if the
## Basic Usage
Pass your server to `ResourcesAsTools` when adding the transform. The transform queries that server for resources whenever the generated tools are called.
Pass your FastMCP server to `ResourcesAsTools` when adding the transform. The generated tools route through the server at runtime, which means all server middleware — auth, visibility, rate limiting — applies to resource operations automatically, exactly as it would for direct `resources/read` calls.
<Note>
`ResourcesAsTools` (and `PromptsAsTools`) should be applied to a FastMCP server instance, not a raw Provider. The generated tools call back into the server's middleware chain at runtime, so they need a server to route through. If you want to expose only a subset of resources, create a dedicated FastMCP server for those resources and apply the transform there.
</Note>
```python
from fastmcp import FastMCP

58
src/fastmcp/server/transforms/prompts_as_tools.py
View file

@ -3,6 +3,10 @@
This transform generates tools for listing and getting prompts, enabling
clients that only support tools to access prompt functionality.
The generated tools route through `ctx.fastmcp` at runtime, so all server
middleware (auth, visibility, rate limiting, etc.) applies to prompt
operations exactly as it would for direct `prompts/get` calls.
Example:
```python
from fastmcp import FastMCP
@ -22,6 +26,7 @@ from typing import TYPE_CHECKING, Annotated, Any
from mcp.types import TextContent
from fastmcp.server.dependencies import get_context
from fastmcp.server.transforms import GetToolNext, Transform
from fastmcp.tools.tool import Tool
from fastmcp.utilities.versions import VersionSpec
@ -29,19 +34,20 @@ from fastmcp.utilities.versions import VersionSpec
if TYPE_CHECKING:
from fastmcp.server.providers.base import Provider
# Note: FastMCP imported inside tools to avoid circular import
class PromptsAsTools(Transform):
"""Transform that adds tools for listing and getting prompts.
Generates two tools:
- `list_prompts`: Lists all prompts from the provider
- `list_prompts`: Lists all prompts
- `get_prompt`: Gets a specific prompt with optional arguments
The transform captures a provider reference at construction and queries it
for prompts when the generated tools are called. When used with FastMCP,
the provider's auth and visibility filtering is automatically applied.
The generated tools route through the server at runtime, so auth,
middleware, and visibility apply automatically.
This transform should be applied to a FastMCP server instance, not
a raw Provider, because the generated tools need the server's
middleware chain for auth and visibility filtering.
Example:
```python
@ -52,12 +58,15 @@ class PromptsAsTools(Transform):
"""
def __init__(self, provider: Provider) -> None:
"""Initialize the transform with a provider reference.
from fastmcp.server.server import FastMCP
Args:
provider: The provider to query for prompts. Typically this is
the same FastMCP server the transform is added to.
"""
if not isinstance(provider, FastMCP):
raise TypeError(
"PromptsAsTools requires a FastMCP server instance, not a"
f" {type(provider).__name__}. The generated tools route through"
" the server's middleware chain at runtime for auth and"
" visibility. Pass your FastMCP server: PromptsAsTools(mcp)"
)
self._provider = provider
def __repr__(self) -> str:
@ -75,18 +84,14 @@ class PromptsAsTools(Transform):
self, name: str, call_next: GetToolNext, *, version: VersionSpec | None = None
) -> Tool | None:
"""Get a tool by name, including generated prompt tools."""
# Check if it's one of our generated tools
if name == "list_prompts":
return self._make_list_prompts_tool()
if name == "get_prompt":
return self._make_get_prompt_tool()
# Otherwise delegate to downstream
return await call_next(name, version=version)
def _make_list_prompts_tool(self) -> Tool:
"""Create the list_prompts tool."""
provider = self._provider
async def list_prompts() -> str:
"""List all available prompts.
@ -94,7 +99,8 @@ class PromptsAsTools(Transform):
Returns JSON with prompt metadata including name, description,
and optional arguments.
"""
prompts = await provider.list_prompts()
ctx = get_context()
prompts = await ctx.fastmcp.list_prompts()
result: list[dict[str, Any]] = []
for p in prompts:
@ -119,7 +125,6 @@ class PromptsAsTools(Transform):
def _make_get_prompt_tool(self) -> Tool:
"""Create the get_prompt tool."""
provider = self._provider
async def get_prompt(
name: Annotated[str, "The name of the prompt to get"],
@ -131,21 +136,11 @@ class PromptsAsTools(Transform):
"""Get a prompt by name with optional arguments.
Returns the rendered prompt as JSON with a messages array.
Arguments should be provided as a dict mapping argument names to values.
Arguments should be provided as a dict mapping argument names
to values.
"""
from fastmcp.server.server import FastMCP
# Use FastMCP.render_prompt() if available - runs middleware chain
if isinstance(provider, FastMCP):
result = await provider.render_prompt(name, arguments=arguments or {})
return _format_prompt_result(result)
# Fallback for plain providers - no middleware
prompt = await provider.get_prompt(name)
if prompt is None:
raise ValueError(f"Prompt not found: {name}")
result = await prompt._render(arguments or {})
ctx = get_context()
result = await ctx.fastmcp.render_prompt(name, arguments=arguments or {})
return _format_prompt_result(result)
return Tool.from_function(fn=get_prompt)
@ -162,7 +157,6 @@ def _format_prompt_result(result: Any) -> str:
if isinstance(msg.content, TextContent):
content = msg.content.text
else:
# Preserve structured content (e.g., EmbeddedResource) as dict
content = msg.content.model_dump(mode="json", exclude_none=True)
messages.append(

80
src/fastmcp/server/transforms/resources_as_tools.py
View file

@ -3,6 +3,10 @@
This transform generates tools for listing and reading resources, enabling
clients that only support tools to access resource functionality.
The generated tools route through `ctx.fastmcp` at runtime, so all server
middleware (auth, visibility, rate limiting, etc.) applies to resource
operations exactly as it would for direct `resources/read` calls.
Example:
```python
from fastmcp import FastMCP
@ -23,6 +27,7 @@ from typing import TYPE_CHECKING, Annotated, Any
from mcp.types import ToolAnnotations
from fastmcp.server.dependencies import get_context
from fastmcp.server.transforms import GetToolNext, Transform
from fastmcp.tools.tool import Tool
from fastmcp.utilities.versions import VersionSpec
@ -37,12 +42,15 @@ class ResourcesAsTools(Transform):
"""Transform that adds tools for listing and reading resources.
Generates two tools:
- `list_resources`: Lists all resources and templates from the provider
- `list_resources`: Lists all resources and templates
- `read_resource`: Reads a resource by URI
The transform captures a provider reference at construction and queries it
for resources when the generated tools are called. When used with FastMCP,
the provider's auth and visibility filtering is automatically applied.
The generated tools route through the server at runtime, so auth,
middleware, and visibility apply automatically.
This transform should be applied to a FastMCP server instance, not
a raw Provider, because the generated tools need the server's
middleware chain for auth and visibility filtering.
Example:
```python
@ -53,12 +61,15 @@ class ResourcesAsTools(Transform):
"""
def __init__(self, provider: Provider) -> None:
"""Initialize the transform with a provider reference.
from fastmcp.server.server import FastMCP
Args:
provider: The provider to query for resources. Typically this is
the same FastMCP server the transform is added to.
"""
if not isinstance(provider, FastMCP):
raise TypeError(
"ResourcesAsTools requires a FastMCP server instance, not a"
f" {type(provider).__name__}. The generated tools route through"
" the server's middleware chain at runtime for auth and"
" visibility. Pass your FastMCP server: ResourcesAsTools(mcp)"
)
self._provider = provider
def __repr__(self) -> str:
@ -76,31 +87,28 @@ class ResourcesAsTools(Transform):
self, name: str, call_next: GetToolNext, *, version: VersionSpec | None = None
) -> Tool | None:
"""Get a tool by name, including generated resource tools."""
# Check if it's one of our generated tools
if name == "list_resources":
return self._make_list_resources_tool()
if name == "read_resource":
return self._make_read_resource_tool()
# Otherwise delegate to downstream
return await call_next(name, version=version)
def _make_list_resources_tool(self) -> Tool:
"""Create the list_resources tool."""
provider = self._provider
async def list_resources() -> str:
"""List all available resources and resource templates.
Returns JSON with resource metadata. Static resources have a 'uri' field,
while templates have a 'uri_template' field with placeholders like {name}.
Returns JSON with resource metadata. Static resources have a
'uri' field, while templates have a 'uri_template' field with
placeholders like {name}.
"""
resources = await provider.list_resources()
templates = await provider.list_resource_templates()
ctx = get_context()
resources = await ctx.fastmcp.list_resources()
templates = await ctx.fastmcp.list_resource_templates()
result: list[dict[str, Any]] = []
# Static resources
for r in resources:
result.append(
{
@ -111,7 +119,6 @@ class ResourcesAsTools(Transform):
}
)
# Resource templates (URI contains placeholders like {name})
for t in templates:
result.append(
{
@ -127,40 +134,21 @@ class ResourcesAsTools(Transform):
def _make_read_resource_tool(self) -> Tool:
"""Create the read_resource tool."""
provider = self._provider
async def read_resource(
uri: Annotated[str, "The URI of the resource to read"],
) -> str:
"""Read a resource by its URI.
For static resources, provide the exact URI. For templated resources,
provide the URI with template parameters filled in.
For static resources, provide the exact URI. For templated
resources, provide the URI with template parameters filled in.
Returns the resource content as a string. Binary content is
base64-encoded.
"""
from fastmcp import FastMCP
# Use FastMCP.read_resource() if available - runs middleware chain
if isinstance(provider, FastMCP):
result = await provider.read_resource(uri)
return _format_result(result)
# Fallback for plain providers - no middleware
resource = await provider.get_resource(uri)
if resource is not None:
result = await resource._read()
return _format_result(result)
template = await provider.get_resource_template(uri)
if template is not None:
params = template.matches(uri)
if params is not None:
result = await template._read(uri, params)
return _format_result(result)
raise ValueError(f"Resource not found: {uri}")
ctx = get_context()
result = await ctx.fastmcp.read_resource(uri)
return _format_result(result)
return Tool.from_function(fn=read_resource, annotations=_DEFAULT_ANNOTATIONS)
@ -168,17 +156,15 @@ class ResourcesAsTools(Transform):
def _format_result(result: Any) -> str:
"""Format ResourceResult for tool output.
Single text content is returned as-is. Single binary content is base64-encoded.
Multiple contents are JSON-encoded with each item containing content and mime_type.
Single text content is returned as-is. Single binary content is
base64-encoded. Multiple contents are JSON-encoded.
"""
# result is a ResourceResult with .contents list
if len(result.contents) == 1:
content = result.contents[0].content
if isinstance(content, bytes):
return base64.b64encode(content).decode()
return content
# Multiple contents - JSON encode
return json.dumps(
[
{

105
tests/server/transforms/test_resources_as_tools.py
View file

@ -7,6 +7,8 @@ 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
@ -251,3 +253,106 @@ class TestResourcesAsToolsAnnotations:
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"})