Support enabled/disabled resources and templates

This commit is contained in:
Jeremiah Lowin 2025-06-10 10:44:46 -04:00
commit 9c25cc6454
5 changed files with 226 additions and 4 deletions

View file

@ -52,6 +52,7 @@ class Resource(FastMCPComponent, abc.ABC):
description: str | None = None,
mime_type: str | None = None,
tags: set[str] | None = None,
enabled: bool | None = None,
) -> FunctionResource:
return FunctionResource.from_function(
fn=fn,
@ -60,6 +61,7 @@ class Resource(FastMCPComponent, abc.ABC):
description=description,
mime_type=mime_type,
tags=tags,
enabled=enabled,
)
@field_validator("mime_type", mode="before")
@ -124,6 +126,7 @@ class FunctionResource(Resource):
description: str | None = None,
mime_type: str | None = None,
tags: set[str] | None = None,
enabled: bool | None = None,
) -> FunctionResource:
"""Create a FunctionResource from a function."""
if isinstance(uri, str):
@ -135,6 +138,7 @@ class FunctionResource(Resource):
description=description or fn.__doc__,
mime_type=mime_type or "text/plain",
tags=tags or set(),
enabled=enabled if enabled is not None else True,
)
async def read(self) -> str | bytes:

View file

@ -70,6 +70,7 @@ class ResourceTemplate(FastMCPComponent):
description: str | None = None,
mime_type: str | None = None,
tags: set[str] | None = None,
enabled: bool | None = None,
) -> FunctionResourceTemplate:
return FunctionResourceTemplate.from_function(
fn=fn,
@ -78,6 +79,7 @@ class ResourceTemplate(FastMCPComponent):
description=description,
mime_type=mime_type,
tags=tags,
enabled=enabled,
)
@field_validator("mime_type", mode="before")
@ -113,6 +115,7 @@ class ResourceTemplate(FastMCPComponent):
description=self.description,
mime_type=self.mime_type,
tags=self.tags,
enabled=self.enabled,
)
def to_mcp_template(self, **overrides: Any) -> MCPResourceTemplate:
@ -155,6 +158,7 @@ class FunctionResourceTemplate(ResourceTemplate):
description: str | None = None,
mime_type: str | None = None,
tags: set[str] | None = None,
enabled: bool | None = None,
) -> FunctionResourceTemplate:
"""Create a template from a function."""
from fastmcp.server.context import Context
@ -237,4 +241,5 @@ class FunctionResourceTemplate(ResourceTemplate):
fn=fn,
parameters=parameters,
tags=tags or set(),
enabled=enabled if enabled is not None else True,
)

View file

@ -667,11 +667,12 @@ class FastMCP(Generic[LifespanResultT]):
Args:
name_or_fn: Either a function (when used as @tool), a string name, or None
name: Optional name for the tool (keyword-only, alternative to name_or_fn)
description: Optional description of what the tool does
tags: Optional set of tags for categorizing the tool
annotations: Optional annotations about the tool's behavior
annotations: Optional annotations about the tool's behavior (e.g. {"is_async": True})
exclude_args: Optional list of argument names to exclude from the tool schema
name: Optional name for the tool (keyword-only, alternative to name_or_fn)
enabled: Optional boolean to enable or disable the tool
Example:
@server.tool
@ -820,6 +821,7 @@ class FastMCP(Generic[LifespanResultT]):
description: str | None = None,
mime_type: str | None = None,
tags: set[str] | None = None,
enabled: bool | None = None,
) -> Callable[[AnyFunction], Resource | ResourceTemplate]:
"""Decorator to register a function as a resource.
@ -842,6 +844,7 @@ class FastMCP(Generic[LifespanResultT]):
description: Optional description of the resource
mime_type: Optional MIME type for the resource
tags: Optional set of tags for categorizing the resource
enabled: Optional boolean to enable or disable the resource
Example:
@server.resource("resource://my-resource")
@ -906,6 +909,7 @@ class FastMCP(Generic[LifespanResultT]):
description=description,
mime_type=mime_type,
tags=tags,
enabled=enabled,
)
self.add_template(template)
return template
@ -917,6 +921,7 @@ class FastMCP(Generic[LifespanResultT]):
description=description,
mime_type=mime_type,
tags=tags,
enabled=enabled,
)
self.add_resource(resource)
return resource
@ -983,9 +988,10 @@ class FastMCP(Generic[LifespanResultT]):
Args:
name_or_fn: Either a function (when used as @prompt), a string name, or None
name: Optional name for the prompt (keyword-only, alternative to name_or_fn)
description: Optional description of what the prompt does
tags: Optional set of tags for categorizing the prompt
name: Optional name for the prompt (keyword-only, alternative to name_or_fn)
enabled: Optional boolean to enable or disable the prompt
Example:
@server.prompt

View file

@ -44,7 +44,7 @@ class FastMCPComponent(FastMCPBaseModel):
return self.model_dump() == other.model_dump()
def __repr__(self) -> str:
return f"{self.__class__.__name__}(name={self.name!r}, description={self.description!r}, tags={self.tags})"
return f"{self.__class__.__name__}(name={self.name!r}, description={self.description!r}, tags={self.tags}, enabled={self.enabled})"
def enable(self) -> None:
"""Enable the component."""

View file

@ -732,6 +732,9 @@ class TestToolEnabled:
tools = await client.list_tools()
assert len(tools) == 0
with pytest.raises(ToolError, match="Unknown tool"):
await client.call_tool("sample_tool", {"x": 5})
async def test_tool_toggle_enabled(self):
mcp = FastMCP()
@ -758,6 +761,9 @@ class TestToolEnabled:
tools = await client.list_tools()
assert len(tools) == 0
with pytest.raises(ToolError, match="Unknown tool"):
await client.call_tool("sample_tool", {"x": 5})
async def test_get_tool_and_disable(self):
mcp = FastMCP()
@ -774,6 +780,9 @@ class TestToolEnabled:
result = await client.list_tools()
assert len(result) == 0
with pytest.raises(ToolError, match="Unknown tool"):
await client.call_tool("sample_tool", {"x": 5})
async def test_cant_call_disabled_tool(self):
mcp = FastMCP()
@ -870,6 +879,102 @@ class TestResourceContext:
assert result[0].text == "1" # type: ignore[attr-defined]
class TestResourceEnabled:
async def test_toggle_enabled(self):
mcp = FastMCP()
@mcp.resource("resource://data")
def sample_resource() -> str:
return "Hello, world!"
assert sample_resource.enabled
resource = await mcp.get_resource("resource://data")
assert resource.enabled
resource.disable()
assert not resource.enabled
assert not sample_resource.enabled
resource.enable()
assert resource.enabled
assert sample_resource.enabled
async def test_resource_disabled_in_decorator(self):
mcp = FastMCP()
@mcp.resource("resource://data", enabled=False)
def sample_resource() -> str:
return "Hello, world!"
async with Client(mcp) as client:
resources = await client.list_resources()
assert len(resources) == 0
with pytest.raises(McpError, match="Unknown resource"):
await client.read_resource(AnyUrl("resource://data"))
async def test_resource_toggle_enabled(self):
mcp = FastMCP()
@mcp.resource("resource://data", enabled=False)
def sample_resource() -> str:
return "Hello, world!"
sample_resource.enable()
async with Client(mcp) as client:
resources = await client.list_resources()
assert len(resources) == 1
async def test_resource_toggle_disabled(self):
mcp = FastMCP()
@mcp.resource("resource://data")
def sample_resource() -> str:
return "Hello, world!"
sample_resource.disable()
async with Client(mcp) as client:
resources = await client.list_resources()
assert len(resources) == 0
with pytest.raises(McpError, match="Unknown resource"):
await client.read_resource(AnyUrl("resource://data"))
async def test_get_resource_and_disable(self):
mcp = FastMCP()
@mcp.resource("resource://data")
def sample_resource() -> str:
return "Hello, world!"
resource = await mcp.get_resource("resource://data")
assert resource.enabled
sample_resource.disable()
async with Client(mcp) as client:
result = await client.list_resources()
assert len(result) == 0
with pytest.raises(McpError, match="Unknown resource"):
await client.read_resource(AnyUrl("resource://data"))
async def test_cant_read_disabled_resource(self):
mcp = FastMCP()
@mcp.resource("resource://data", enabled=False)
def sample_resource() -> str:
return "Hello, world!"
with pytest.raises(McpError, match="Unknown resource"):
async with Client(mcp) as client:
await client.read_resource(AnyUrl("resource://data"))
class TestResourceTemplates:
async def test_resource_with_params_not_in_uri(self):
"""Test that a resource with function parameters raises an error if the URI
@ -1121,6 +1226,99 @@ class TestResourceTemplateContext:
assert result[0].text.startswith("Resource template: test 1") # type: ignore[attr-defined]
class TestResourceTemplateEnabled:
async def test_toggle_enabled(self):
mcp = FastMCP()
@mcp.resource("resource://{param}")
def sample_template(param: str) -> str:
return f"Template: {param}"
assert sample_template.enabled
template = await mcp.get_resource_template("resource://{param}")
assert template.enabled
template.disable()
assert not template.enabled
assert not sample_template.enabled
template.enable()
assert template.enabled
assert sample_template.enabled
async def test_template_disabled_in_decorator(self):
mcp = FastMCP()
@mcp.resource("resource://{param}", enabled=False)
def sample_template(param: str) -> str:
return f"Template: {param}"
async with Client(mcp) as client:
templates = await client.list_resource_templates()
assert len(templates) == 0
with pytest.raises(McpError, match="Unknown resource"):
await client.read_resource(AnyUrl("resource://test"))
async def test_template_toggle_enabled(self):
mcp = FastMCP()
@mcp.resource("resource://{param}", enabled=False)
def sample_template(param: str) -> str:
return f"Template: {param}"
sample_template.enable()
async with Client(mcp) as client:
templates = await client.list_resource_templates()
assert len(templates) == 1
async def test_template_toggle_disabled(self):
mcp = FastMCP()
@mcp.resource("resource://{param}")
def sample_template(param: str) -> str:
return f"Template: {param}"
sample_template.disable()
async with Client(mcp) as client:
templates = await client.list_resource_templates()
assert len(templates) == 0
async def test_get_template_and_disable(self):
mcp = FastMCP()
@mcp.resource("resource://{param}")
def sample_template(param: str) -> str:
return f"Template: {param}"
template = await mcp.get_resource_template("resource://{param}")
assert template.enabled
sample_template.disable()
async with Client(mcp) as client:
result = await client.list_resource_templates()
assert len(result) == 0
with pytest.raises(McpError, match="Unknown resource"):
await client.read_resource(AnyUrl("resource://test"))
async def test_cant_read_disabled_template(self):
mcp = FastMCP()
@mcp.resource("resource://{param}", enabled=False)
def sample_template(param: str) -> str:
return f"Template: {param}"
with pytest.raises(McpError, match="Unknown resource"):
async with Client(mcp) as client:
await client.read_resource(AnyUrl("resource://test"))
class TestPrompts:
"""Test prompt functionality in FastMCP server."""
@ -1340,6 +1538,9 @@ class TestPromptEnabled:
prompts = await client.list_prompts()
assert len(prompts) == 0
with pytest.raises(McpError, match="Unknown prompt"):
await client.get_prompt("sample_prompt")
async def test_prompt_toggle_enabled(self):
mcp = FastMCP()
@ -1366,6 +1567,9 @@ class TestPromptEnabled:
prompts = await client.list_prompts()
assert len(prompts) == 0
with pytest.raises(McpError, match="Unknown prompt"):
await client.get_prompt("sample_prompt")
async def test_get_prompt_and_disable(self):
mcp = FastMCP()
@ -1382,6 +1586,9 @@ class TestPromptEnabled:
result = await client.list_prompts()
assert len(result) == 0
with pytest.raises(McpError, match="Unknown prompt"):
await client.get_prompt("sample_prompt")
async def test_cant_get_disabled_prompt(self):
mcp = FastMCP()