mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-22 21:44:18 +02:00
Merge pull request #781 from jlowin/enabled
Support enable/disable for all FastMCP components (tools, prompts, resources, templates)
This commit is contained in:
Jeremiah Lowin
GitHub
commit
f455376bc9
17 changed files with 784 additions and 95 deletions
|
|
@ -53,6 +53,30 @@ product_search_tool = Tool.from_tool(
|
|||
|
||||
mcp.add_tool(product_search_tool)
|
||||
```
|
||||
|
||||
<Tip>
|
||||
When you transform a tool, the original tool remains registered on the server. To avoid confusing an LLM with two similar tools, you can disable the original one:
|
||||
|
||||
```python
|
||||
from fastmcp import FastMCP
|
||||
from fastmcp.tools import Tool
|
||||
|
||||
mcp = FastMCP()
|
||||
|
||||
# The original, generic tool
|
||||
@mcp.tool
|
||||
def search(query: str, category: str = "all") -> list[dict]:
|
||||
...
|
||||
|
||||
# Create a more domain-specific version
|
||||
product_search_tool = Tool.from_tool(search, ...)
|
||||
mcp.add_tool(product_search_tool)
|
||||
|
||||
# Disable the original tool
|
||||
search.disable()
|
||||
```
|
||||
</Tip>
|
||||
|
||||
Now, clients see a tool named `find_products` with a clear, domain-specific purpose and relevant tags, even though it still uses the original generic `search` function's logic.
|
||||
|
||||
### Parameters
|
||||
|
|
|
|||
|
|
@ -147,7 +147,32 @@ def data_analysis_prompt(
|
|||
- **`name`**: Sets the explicit prompt name exposed via MCP.
|
||||
- **`description`**: Provides the description exposed via MCP. If set, the function's docstring is ignored for this purpose.
|
||||
- **`tags`**: A set of strings used to categorize the prompt. Clients *might* use tags to filter or group available prompts.
|
||||
- **`enabled`**: A boolean to enable or disable the prompt (defaults to `True`). See [Disabling Prompts](#disabling-prompts) for more information.
|
||||
### Disabling Prompts
|
||||
|
||||
<VersionBadge version="2.8.0" />
|
||||
|
||||
You can control the visibility and availability of prompts by enabling or disabling them. Disabled prompts will not appear in the list of available prompts, and attempting to call a disabled prompt will result in an "Unknown prompt" error.
|
||||
|
||||
By default, all prompts are enabled. You can disable a prompt upon creation using the `enabled` parameter in the decorator:
|
||||
|
||||
```python
|
||||
@mcp.prompt(enabled=False)
|
||||
def experimental_prompt():
|
||||
"""This prompt is not ready for use."""
|
||||
return "This is an experimental prompt."
|
||||
```
|
||||
|
||||
You can also toggle a prompt's state programmatically after it has been created:
|
||||
|
||||
```python
|
||||
@mcp.prompt
|
||||
def seasonal_prompt(): return "Happy Holidays!"
|
||||
|
||||
# Disable and re-enable the prompt
|
||||
seasonal_prompt.disable()
|
||||
seasonal_prompt.enable()
|
||||
```
|
||||
### Asynchronous Prompts
|
||||
|
||||
FastMCP seamlessly supports both standard (`def`) and asynchronous (`async def`) functions as prompts.
|
||||
|
|
@ -191,6 +216,8 @@ async def generate_report_request(report_type: str, ctx: Context) -> str:
|
|||
|
||||
For full documentation on the Context object and all its capabilities, see the [Context documentation](/servers/context).
|
||||
|
||||
|
||||
|
||||
## Server Behavior
|
||||
|
||||
### Duplicate Prompts
|
||||
|
|
|
|||
|
|
@ -94,6 +94,33 @@ def get_application_status() -> dict:
|
|||
- **`description`**: Explanation of the resource (defaults to docstring).
|
||||
- **`mime_type`**: Specifies the content type (FastMCP often infers a default like `text/plain` or `application/json`, but explicit is better for non-text types).
|
||||
- **`tags`**: A set of strings for categorization, potentially used by clients for filtering.
|
||||
- **`enabled`**: A boolean to enable or disable the resource (defaults to `True`). See [Disabling Resources](#disabling-resources) for more information.
|
||||
|
||||
### Disabling Resources
|
||||
|
||||
<VersionBadge version="2.8.0" />
|
||||
|
||||
You can control the visibility and availability of resources and templates by enabling or disabling them. Disabled resources will not appear in the list of available resources or templates, and attempting to read a disabled resource will result in an "Unknown resource" error.
|
||||
|
||||
By default, all resources are enabled. You can disable a resource upon creation using the `enabled` parameter in the decorator:
|
||||
|
||||
```python
|
||||
@mcp.resource("data://secret", enabled=False)
|
||||
def get_secret_data():
|
||||
"""This resource is currently disabled."""
|
||||
return "Secret data"
|
||||
```
|
||||
|
||||
You can also toggle a resource's state programmatically after it has been created:
|
||||
|
||||
```python
|
||||
@mcp.resource("data://config")
|
||||
def get_config(): return {"version": 1}
|
||||
|
||||
# Disable and re-enable the resource
|
||||
get_config.disable()
|
||||
get_config.enable()
|
||||
```
|
||||
|
||||
### Accessing MCP Context
|
||||
|
||||
|
|
|
|||
|
|
@ -169,27 +169,58 @@ def search_products_implementation(query: str, category: str | None = None) -> l
|
|||
|
||||
- **`name`**: Sets the explicit tool name exposed via MCP.
|
||||
- **`description`**: Provides the description exposed via MCP. If set, the function's docstring is ignored for this purpose.
|
||||
- **`tags`**: A set of strings used to categorize the tool. Clients *might* use tags to filter or group available tools.
|
||||
- **`tags`**: A set of strings to categorize the tool. Clients *might* use tags to filter or group available tools.
|
||||
- **`enabled`**: A boolean to enable or disable the tool (defaults to `True`). See [Disabling Tools](#disabling-tools) for more information.
|
||||
- **`exclude_args`**: A list of argument names to exclude from the tool schema shown to the LLM. See [Excluding Arguments](#excluding-arguments) for more information.
|
||||
|
||||
### Excluding Arguments
|
||||
|
||||
- **`exclude_args`**:
|
||||
<VersionBadge version="2.6.0" />
|
||||
A list of argument names to exclude from the tool schema shown to the LLM. This is useful for arguments that are injected at runtime (such as `state`, `user_id`, or credentials) and should not be exposed to the LLM or client. Only arguments with default values can be excluded; attempting to exclude a required argument will raise an error.
|
||||
|
||||
<VersionBadge version="2.6.0" />
|
||||
|
||||
Example:
|
||||
You can exclude certain arguments from the tool schema shown to the LLM. This is useful for arguments that are injected at runtime (such as `state`, `user_id`, or credentials) and should not be exposed to the LLM or client. Only arguments with default values can be excluded; attempting to exclude a required argument will raise an error.
|
||||
|
||||
```python
|
||||
@mcp.tool(
|
||||
name="get_user_details",
|
||||
exclude_args=["user_id"]
|
||||
)
|
||||
def get_user_details(user_id: str = None) -> str:
|
||||
# user_id will be injected by the server, not provided by the LLM
|
||||
...
|
||||
```
|
||||
Example:
|
||||
|
||||
With this configuration, `user_id` will not appear in the tool's parameter schema, but can still be set by the server or framework at runtime.
|
||||
```python
|
||||
@mcp.tool(
|
||||
name="get_user_details",
|
||||
exclude_args=["user_id"]
|
||||
)
|
||||
def get_user_details(user_id: str = None) -> str:
|
||||
# user_id will be injected by the server, not provided by the LLM
|
||||
...
|
||||
```
|
||||
|
||||
With this configuration, `user_id` will not appear in the tool's parameter schema, but can still be set by the server or framework at runtime.
|
||||
|
||||
For more complex tool transformations, see [Transforming Tools](/patterns/tool-transformation).
|
||||
|
||||
### Disabling Tools
|
||||
|
||||
<VersionBadge version="2.8.0" />
|
||||
|
||||
You can control the visibility and availability of tools by enabling or disabling them. This is useful for feature flagging, maintenance, or dynamically changing the toolset available to a client. Disabled tools will not appear in the list of available tools returned by `list_tools`, and attempting to call a disabled tool will result in an "Unknown tool" error, just as if the tool did not exist.
|
||||
|
||||
By default, all tools are enabled. You can disable a tool upon creation using the `enabled` parameter in the decorator:
|
||||
|
||||
```python
|
||||
@mcp.tool(enabled=False)
|
||||
def maintenance_tool():
|
||||
"""This tool is currently under maintenance."""
|
||||
return "This tool is disabled."
|
||||
```
|
||||
|
||||
You can also toggle a tool's state programmatically after it has been created:
|
||||
|
||||
```python
|
||||
@mcp.tool
|
||||
def dynamic_tool():
|
||||
return "I am a dynamic tool."
|
||||
|
||||
# Disable and re-enable the tool
|
||||
dynamic_tool.disable()
|
||||
dynamic_tool.enable()
|
||||
```
|
||||
|
||||
### Async Tools
|
||||
|
||||
|
|
|
|||
|
|
@ -33,3 +33,7 @@ class ClientError(Exception):
|
|||
|
||||
class NotFoundError(Exception):
|
||||
"""Object not found."""
|
||||
|
||||
|
||||
class DisabledError(Exception):
|
||||
"""Object is disabled."""
|
||||
|
|
|
|||
|
|
@ -96,6 +96,7 @@ class Prompt(FastMCPComponent, ABC):
|
|||
name: str | None = None,
|
||||
description: str | None = None,
|
||||
tags: set[str] | None = None,
|
||||
enabled: bool | None = None,
|
||||
) -> FunctionPrompt:
|
||||
"""Create a Prompt from a function.
|
||||
|
||||
|
|
@ -106,7 +107,7 @@ class Prompt(FastMCPComponent, ABC):
|
|||
- A sequence of any of the above
|
||||
"""
|
||||
return FunctionPrompt.from_function(
|
||||
fn=fn, name=name, description=description, tags=tags
|
||||
fn=fn, name=name, description=description, tags=tags, enabled=enabled
|
||||
)
|
||||
|
||||
@abstractmethod
|
||||
|
|
@ -130,6 +131,7 @@ class FunctionPrompt(Prompt):
|
|||
name: str | None = None,
|
||||
description: str | None = None,
|
||||
tags: set[str] | None = None,
|
||||
enabled: bool | None = None,
|
||||
) -> FunctionPrompt:
|
||||
"""Create a Prompt from a function.
|
||||
|
||||
|
|
@ -195,6 +197,7 @@ class FunctionPrompt(Prompt):
|
|||
description=description,
|
||||
arguments=arguments,
|
||||
tags=tags or set(),
|
||||
enabled=enabled if enabled is not None else True,
|
||||
fn=fn,
|
||||
)
|
||||
|
||||
|
|
|
|||
|
|
@ -40,9 +40,11 @@ class PromptManager:
|
|||
|
||||
self.duplicate_behavior = duplicate_behavior
|
||||
|
||||
def get_prompt(self, key: str) -> Prompt | None:
|
||||
def get_prompt(self, key: str) -> Prompt:
|
||||
"""Get prompt by key."""
|
||||
return self._prompts.get(key)
|
||||
if key in self._prompts:
|
||||
return self._prompts[key]
|
||||
raise NotFoundError(f"Unknown prompt: {key}")
|
||||
|
||||
def get_prompts(self) -> dict[str, Prompt]:
|
||||
"""Get all registered prompts, indexed by registered key."""
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
)
|
||||
|
|
|
|||
|
|
@ -241,20 +241,20 @@ class FastMCPProxy(FastMCP):
|
|||
prompts[prompt_proxy.name] = prompt_proxy
|
||||
return prompts
|
||||
|
||||
async def _mcp_call_tool(
|
||||
async def _call_tool(
|
||||
self, key: str, arguments: dict[str, Any]
|
||||
) -> list[TextContent | ImageContent | EmbeddedResource]:
|
||||
try:
|
||||
result = await super()._mcp_call_tool(key, arguments)
|
||||
result = await super()._call_tool(key, arguments)
|
||||
return result
|
||||
except NotFoundError:
|
||||
async with self.client:
|
||||
result = await self.client.call_tool(key, arguments)
|
||||
return result
|
||||
|
||||
async def _mcp_read_resource(self, uri: AnyUrl | str) -> list[ReadResourceContents]:
|
||||
async def _read_resource(self, uri: AnyUrl | str) -> list[ReadResourceContents]:
|
||||
try:
|
||||
result = await super()._mcp_read_resource(uri)
|
||||
result = await super()._read_resource(uri)
|
||||
return result
|
||||
except NotFoundError:
|
||||
async with self.client:
|
||||
|
|
@ -270,11 +270,11 @@ class FastMCPProxy(FastMCP):
|
|||
ReadResourceContents(content=content, mime_type=resource[0].mimeType)
|
||||
]
|
||||
|
||||
async def _mcp_get_prompt(
|
||||
async def _get_prompt(
|
||||
self, name: str, arguments: dict[str, Any] | None = None
|
||||
) -> GetPromptResult:
|
||||
try:
|
||||
result = await super()._mcp_get_prompt(name, arguments)
|
||||
result = await super()._get_prompt(name, arguments)
|
||||
return result
|
||||
except NotFoundError:
|
||||
async with self.client:
|
||||
|
|
|
|||
|
|
@ -44,7 +44,7 @@ from starlette.routing import BaseRoute, Route
|
|||
import fastmcp
|
||||
import fastmcp.server
|
||||
import fastmcp.settings
|
||||
from fastmcp.exceptions import NotFoundError
|
||||
from fastmcp.exceptions import DisabledError, NotFoundError
|
||||
from fastmcp.prompts import Prompt, PromptManager
|
||||
from fastmcp.prompts.prompt import FunctionPrompt
|
||||
from fastmcp.resources import Resource, ResourceManager
|
||||
|
|
@ -291,6 +291,12 @@ class FastMCP(Generic[LifespanResultT]):
|
|||
self._cache.set("resources", resources)
|
||||
return resources
|
||||
|
||||
async def get_resource(self, key: str) -> Resource:
|
||||
resources = await self.get_resources()
|
||||
if key not in resources:
|
||||
raise NotFoundError(f"Unknown resource: {key}")
|
||||
return resources[key]
|
||||
|
||||
async def get_resource_templates(self) -> dict[str, ResourceTemplate]:
|
||||
"""Get all registered resource templates, indexed by registered key."""
|
||||
if (
|
||||
|
|
@ -311,6 +317,12 @@ class FastMCP(Generic[LifespanResultT]):
|
|||
self._cache.set("resource_templates", templates)
|
||||
return templates
|
||||
|
||||
async def get_resource_template(self, key: str) -> ResourceTemplate:
|
||||
templates = await self.get_resource_templates()
|
||||
if key not in templates:
|
||||
raise NotFoundError(f"Unknown resource template: {key}")
|
||||
return templates[key]
|
||||
|
||||
async def get_prompts(self) -> dict[str, Prompt]:
|
||||
"""
|
||||
List all available prompts.
|
||||
|
|
@ -330,6 +342,12 @@ class FastMCP(Generic[LifespanResultT]):
|
|||
self._cache.set("prompts", prompts)
|
||||
return prompts
|
||||
|
||||
async def get_prompt(self, key: str) -> Prompt:
|
||||
prompts = await self.get_prompts()
|
||||
if key not in prompts:
|
||||
raise NotFoundError(f"Unknown prompt: {key}")
|
||||
return prompts[key]
|
||||
|
||||
def custom_route(
|
||||
self,
|
||||
path: str,
|
||||
|
|
@ -381,7 +399,9 @@ class FastMCP(Generic[LifespanResultT]):
|
|||
|
||||
"""
|
||||
tools = await self.get_tools()
|
||||
return [tool.to_mcp_tool(name=key) for key, tool in tools.items()]
|
||||
return [
|
||||
tool.to_mcp_tool(name=key) for key, tool in tools.items() if tool.enabled
|
||||
]
|
||||
|
||||
async def _mcp_list_resources(self) -> list[MCPResource]:
|
||||
"""
|
||||
|
|
@ -391,7 +411,9 @@ class FastMCP(Generic[LifespanResultT]):
|
|||
"""
|
||||
resources = await self.get_resources()
|
||||
return [
|
||||
resource.to_mcp_resource(uri=key) for key, resource in resources.items()
|
||||
resource.to_mcp_resource(uri=key)
|
||||
for key, resource in resources.items()
|
||||
if resource.enabled
|
||||
]
|
||||
|
||||
async def _mcp_list_resource_templates(self) -> list[MCPResourceTemplate]:
|
||||
|
|
@ -404,6 +426,7 @@ class FastMCP(Generic[LifespanResultT]):
|
|||
return [
|
||||
template.to_mcp_template(uriTemplate=key)
|
||||
for key, template in templates.items()
|
||||
if template.enabled
|
||||
]
|
||||
|
||||
async def _mcp_list_prompts(self) -> list[MCPPrompt]:
|
||||
|
|
@ -413,12 +436,19 @@ class FastMCP(Generic[LifespanResultT]):
|
|||
|
||||
"""
|
||||
prompts = await self.get_prompts()
|
||||
return [prompt.to_mcp_prompt(name=key) for key, prompt in prompts.items()]
|
||||
return [
|
||||
prompt.to_mcp_prompt(name=key)
|
||||
for key, prompt in prompts.items()
|
||||
if prompt.enabled
|
||||
]
|
||||
|
||||
async def _mcp_call_tool(
|
||||
self, key: str, arguments: dict[str, Any]
|
||||
) -> list[TextContent | ImageContent | EmbeddedResource]:
|
||||
"""Handle MCP 'callTool' requests.
|
||||
"""
|
||||
Handle MCP 'callTool' requests.
|
||||
|
||||
Delegates to _call_tool, which should be overridden by FastMCP subclasses.
|
||||
|
||||
Args:
|
||||
key: The name of the tool to call
|
||||
|
|
@ -431,43 +461,109 @@ class FastMCP(Generic[LifespanResultT]):
|
|||
|
||||
# Create and use context for the entire call
|
||||
with fastmcp.server.context.Context(fastmcp=self):
|
||||
# Get tool, checking first from our tools, then from the mounted servers
|
||||
if self._tool_manager.has_tool(key):
|
||||
return await self._tool_manager.call_tool(key, arguments)
|
||||
try:
|
||||
return await self._call_tool(key, arguments)
|
||||
except DisabledError:
|
||||
# convert to NotFoundError to avoid leaking tool presence
|
||||
raise NotFoundError(f"Unknown tool: {key}")
|
||||
except NotFoundError:
|
||||
# standardize NotFound message
|
||||
raise NotFoundError(f"Unknown tool: {key}")
|
||||
|
||||
# Check mounted servers to see if they have the tool
|
||||
for server in self._mounted_servers.values():
|
||||
if server.match_tool(key):
|
||||
tool_key = server.strip_tool_prefix(key)
|
||||
return await server.server._mcp_call_tool(tool_key, arguments)
|
||||
async def _call_tool(
|
||||
self, key: str, arguments: dict[str, Any]
|
||||
) -> list[TextContent | ImageContent | EmbeddedResource]:
|
||||
"""
|
||||
Call a tool with raw MCP arguments. FastMCP subclasses should override
|
||||
this method, not _mcp_call_tool.
|
||||
|
||||
raise NotFoundError(f"Unknown tool: {key}")
|
||||
Args:
|
||||
key: The name of the tool to call arguments: Arguments to pass to
|
||||
the tool
|
||||
|
||||
Returns:
|
||||
List of MCP Content objects containing the tool results
|
||||
"""
|
||||
|
||||
# Get tool, checking first from our tools, then from the mounted servers
|
||||
if self._tool_manager.has_tool(key):
|
||||
tool = self._tool_manager.get_tool(key)
|
||||
if not tool.enabled:
|
||||
raise DisabledError(f"Tool {key!r} is disabled")
|
||||
return await self._tool_manager.call_tool(key, arguments)
|
||||
|
||||
# Check mounted servers to see if they have the tool
|
||||
for server in self._mounted_servers.values():
|
||||
if server.match_tool(key):
|
||||
tool_key = server.strip_tool_prefix(key)
|
||||
return await server.server._call_tool(tool_key, arguments)
|
||||
|
||||
raise NotFoundError(f"Unknown tool: {key!r}")
|
||||
|
||||
async def _mcp_read_resource(self, uri: AnyUrl | str) -> list[ReadResourceContents]:
|
||||
"""
|
||||
Handle MCP 'readResource' requests.
|
||||
|
||||
Delegates to _read_resource, which should be overridden by FastMCP subclasses.
|
||||
"""
|
||||
logger.debug("Read resource: %s", uri)
|
||||
|
||||
with fastmcp.server.context.Context(fastmcp=self):
|
||||
try:
|
||||
return await self._read_resource(uri)
|
||||
except DisabledError:
|
||||
# convert to NotFoundError to avoid leaking resource presence
|
||||
raise NotFoundError(f"Unknown resource: {str(uri)!r}")
|
||||
except NotFoundError:
|
||||
# standardize NotFound message
|
||||
raise NotFoundError(f"Unknown resource: {str(uri)!r}")
|
||||
|
||||
async def _read_resource(self, uri: AnyUrl | str) -> list[ReadResourceContents]:
|
||||
"""
|
||||
Read a resource by URI, in the format expected by the low-level MCP
|
||||
server.
|
||||
"""
|
||||
with fastmcp.server.context.Context(fastmcp=self):
|
||||
if self._resource_manager.has_resource(uri):
|
||||
resource = await self._resource_manager.get_resource(uri)
|
||||
content = await self._resource_manager.read_resource(uri)
|
||||
return [
|
||||
ReadResourceContents(
|
||||
content=content,
|
||||
mime_type=resource.mime_type,
|
||||
)
|
||||
]
|
||||
if self._resource_manager.has_resource(uri):
|
||||
resource = await self._resource_manager.get_resource(uri)
|
||||
if not resource.enabled:
|
||||
raise DisabledError(f"Resource {str(uri)!r} is disabled")
|
||||
content = await self._resource_manager.read_resource(uri)
|
||||
return [
|
||||
ReadResourceContents(
|
||||
content=content,
|
||||
mime_type=resource.mime_type,
|
||||
)
|
||||
]
|
||||
else:
|
||||
for server in self._mounted_servers.values():
|
||||
if server.match_resource(str(uri)):
|
||||
new_uri = server.strip_resource_prefix(str(uri))
|
||||
return await server.server._mcp_read_resource(new_uri)
|
||||
else:
|
||||
for server in self._mounted_servers.values():
|
||||
if server.match_resource(str(uri)):
|
||||
new_uri = server.strip_resource_prefix(str(uri))
|
||||
return await server.server._mcp_read_resource(new_uri)
|
||||
else:
|
||||
raise NotFoundError(f"Unknown resource: {uri}")
|
||||
raise NotFoundError(f"Unknown resource: {uri}")
|
||||
|
||||
async def _mcp_get_prompt(
|
||||
self, name: str, arguments: dict[str, Any] | None = None
|
||||
) -> GetPromptResult:
|
||||
"""
|
||||
Handle MCP 'getPrompt' requests.
|
||||
|
||||
Delegates to _get_prompt, which should be overridden by FastMCP subclasses.
|
||||
"""
|
||||
logger.debug("Get prompt: %s with %s", name, arguments)
|
||||
|
||||
with fastmcp.server.context.Context(fastmcp=self):
|
||||
try:
|
||||
return await self._get_prompt(name, arguments)
|
||||
except DisabledError:
|
||||
# convert to NotFoundError to avoid leaking prompt presence
|
||||
raise NotFoundError(f"Unknown prompt: {name}")
|
||||
except NotFoundError:
|
||||
# standardize NotFound message
|
||||
raise NotFoundError(f"Unknown prompt: {name}")
|
||||
|
||||
async def _get_prompt(
|
||||
self, name: str, arguments: dict[str, Any] | None = None
|
||||
) -> GetPromptResult:
|
||||
"""Handle MCP 'getPrompt' requests.
|
||||
|
||||
|
|
@ -480,19 +576,20 @@ class FastMCP(Generic[LifespanResultT]):
|
|||
"""
|
||||
logger.debug("Get prompt: %s with %s", name, arguments)
|
||||
|
||||
# Create and use context for the entire call
|
||||
with fastmcp.server.context.Context(fastmcp=self):
|
||||
# Get prompt, checking first from our prompts, then from the mounted servers
|
||||
if self._prompt_manager.has_prompt(name):
|
||||
return await self._prompt_manager.render_prompt(name, arguments)
|
||||
# Get prompt, checking first from our prompts, then from the mounted servers
|
||||
if self._prompt_manager.has_prompt(name):
|
||||
prompt = self._prompt_manager.get_prompt(name)
|
||||
if not prompt.enabled:
|
||||
raise DisabledError(f"Prompt {name!r} is disabled")
|
||||
return await self._prompt_manager.render_prompt(name, arguments)
|
||||
|
||||
# Check mounted servers to see if they have the prompt
|
||||
for server in self._mounted_servers.values():
|
||||
if server.match_prompt(name):
|
||||
prompt_name = server.strip_prompt_prefix(name)
|
||||
return await server.server._mcp_get_prompt(prompt_name, arguments)
|
||||
# Check mounted servers to see if they have the prompt
|
||||
for server in self._mounted_servers.values():
|
||||
if server.match_prompt(name):
|
||||
prompt_name = server.strip_prompt_prefix(name)
|
||||
return await server.server._mcp_get_prompt(prompt_name, arguments)
|
||||
|
||||
raise NotFoundError(f"Unknown prompt: {name}")
|
||||
raise NotFoundError(f"Unknown prompt: {name}")
|
||||
|
||||
def add_tool(self, tool: Tool) -> None:
|
||||
"""Add a tool to the server.
|
||||
|
|
@ -528,6 +625,7 @@ class FastMCP(Generic[LifespanResultT]):
|
|||
tags: set[str] | None = None,
|
||||
annotations: ToolAnnotations | dict[str, Any] | None = None,
|
||||
exclude_args: list[str] | None = None,
|
||||
enabled: bool | None = None,
|
||||
) -> FunctionTool: ...
|
||||
|
||||
@overload
|
||||
|
|
@ -540,6 +638,7 @@ class FastMCP(Generic[LifespanResultT]):
|
|||
tags: set[str] | None = None,
|
||||
annotations: ToolAnnotations | dict[str, Any] | None = None,
|
||||
exclude_args: list[str] | None = None,
|
||||
enabled: bool | None = None,
|
||||
) -> Callable[[AnyFunction], FunctionTool]: ...
|
||||
|
||||
def tool(
|
||||
|
|
@ -551,6 +650,7 @@ class FastMCP(Generic[LifespanResultT]):
|
|||
tags: set[str] | None = None,
|
||||
annotations: ToolAnnotations | dict[str, Any] | None = None,
|
||||
exclude_args: list[str] | None = None,
|
||||
enabled: bool | None = None,
|
||||
) -> Callable[[AnyFunction], FunctionTool] | FunctionTool:
|
||||
"""Decorator to register a tool.
|
||||
|
||||
|
|
@ -567,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
|
||||
|
|
@ -624,6 +725,7 @@ class FastMCP(Generic[LifespanResultT]):
|
|||
annotations=annotations,
|
||||
exclude_args=exclude_args,
|
||||
serializer=self._tool_serializer,
|
||||
enabled=enabled,
|
||||
)
|
||||
self.add_tool(tool)
|
||||
return tool
|
||||
|
|
@ -652,6 +754,7 @@ class FastMCP(Generic[LifespanResultT]):
|
|||
tags=tags,
|
||||
annotations=annotations,
|
||||
exclude_args=exclude_args,
|
||||
enabled=enabled,
|
||||
)
|
||||
|
||||
def add_resource(self, resource: Resource, key: str | None = None) -> None:
|
||||
|
|
@ -718,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.
|
||||
|
||||
|
|
@ -740,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")
|
||||
|
|
@ -804,6 +909,7 @@ class FastMCP(Generic[LifespanResultT]):
|
|||
description=description,
|
||||
mime_type=mime_type,
|
||||
tags=tags,
|
||||
enabled=enabled,
|
||||
)
|
||||
self.add_template(template)
|
||||
return template
|
||||
|
|
@ -815,6 +921,7 @@ class FastMCP(Generic[LifespanResultT]):
|
|||
description=description,
|
||||
mime_type=mime_type,
|
||||
tags=tags,
|
||||
enabled=enabled,
|
||||
)
|
||||
self.add_resource(resource)
|
||||
return resource
|
||||
|
|
@ -843,6 +950,7 @@ class FastMCP(Generic[LifespanResultT]):
|
|||
name: str | None = None,
|
||||
description: str | None = None,
|
||||
tags: set[str] | None = None,
|
||||
enabled: bool | None = None,
|
||||
) -> FunctionPrompt: ...
|
||||
|
||||
@overload
|
||||
|
|
@ -853,6 +961,7 @@ class FastMCP(Generic[LifespanResultT]):
|
|||
name: str | None = None,
|
||||
description: str | None = None,
|
||||
tags: set[str] | None = None,
|
||||
enabled: bool | None = None,
|
||||
) -> Callable[[AnyFunction], FunctionPrompt]: ...
|
||||
|
||||
def prompt(
|
||||
|
|
@ -862,6 +971,7 @@ class FastMCP(Generic[LifespanResultT]):
|
|||
name: str | None = None,
|
||||
description: str | None = None,
|
||||
tags: set[str] | None = None,
|
||||
enabled: bool | None = None,
|
||||
) -> Callable[[AnyFunction], FunctionPrompt] | FunctionPrompt:
|
||||
"""Decorator to register a prompt.
|
||||
|
||||
|
|
@ -878,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
|
||||
|
|
@ -953,6 +1064,7 @@ class FastMCP(Generic[LifespanResultT]):
|
|||
name=prompt_name,
|
||||
description=description,
|
||||
tags=tags,
|
||||
enabled=enabled,
|
||||
)
|
||||
self.add_prompt(prompt)
|
||||
|
||||
|
|
@ -980,6 +1092,7 @@ class FastMCP(Generic[LifespanResultT]):
|
|||
name=prompt_name,
|
||||
description=description,
|
||||
tags=tags,
|
||||
enabled=enabled,
|
||||
)
|
||||
|
||||
async def run_stdio_async(self) -> None:
|
||||
|
|
|
|||
|
|
@ -62,6 +62,7 @@ class Tool(FastMCPComponent, ABC):
|
|||
annotations: ToolAnnotations | None = None,
|
||||
exclude_args: list[str] | None = None,
|
||||
serializer: Callable[[Any], str] | None = None,
|
||||
enabled: bool | None = None,
|
||||
) -> FunctionTool:
|
||||
"""Create a Tool from a function."""
|
||||
return FunctionTool.from_function(
|
||||
|
|
@ -72,6 +73,7 @@ class Tool(FastMCPComponent, ABC):
|
|||
annotations=annotations,
|
||||
exclude_args=exclude_args,
|
||||
serializer=serializer,
|
||||
enabled=enabled,
|
||||
)
|
||||
|
||||
@abstractmethod
|
||||
|
|
@ -92,6 +94,7 @@ class Tool(FastMCPComponent, ABC):
|
|||
tags: set[str] | None = None,
|
||||
annotations: ToolAnnotations | None = None,
|
||||
serializer: Callable[[Any], str] | None = None,
|
||||
enabled: bool | None = None,
|
||||
) -> TransformedTool:
|
||||
from fastmcp.tools.tool_transform import TransformedTool
|
||||
|
||||
|
|
@ -104,6 +107,7 @@ class Tool(FastMCPComponent, ABC):
|
|||
tags=tags,
|
||||
annotations=annotations,
|
||||
serializer=serializer,
|
||||
enabled=enabled,
|
||||
)
|
||||
|
||||
|
||||
|
|
@ -120,6 +124,7 @@ class FunctionTool(Tool):
|
|||
annotations: ToolAnnotations | None = None,
|
||||
exclude_args: list[str] | None = None,
|
||||
serializer: Callable[[Any], str] | None = None,
|
||||
enabled: bool | None = None,
|
||||
) -> FunctionTool:
|
||||
"""Create a Tool from a function."""
|
||||
|
||||
|
|
@ -136,6 +141,7 @@ class FunctionTool(Tool):
|
|||
tags=tags or set(),
|
||||
annotations=annotations,
|
||||
serializer=serializer,
|
||||
enabled=enabled if enabled is not None else True,
|
||||
)
|
||||
|
||||
async def run(
|
||||
|
|
|
|||
|
|
@ -267,6 +267,7 @@ class TransformedTool(Tool):
|
|||
transform_args: dict[str, ArgTransform] | None = None,
|
||||
annotations: ToolAnnotations | None = None,
|
||||
serializer: Callable[[Any], str] | None = None,
|
||||
enabled: bool | None = None,
|
||||
) -> TransformedTool:
|
||||
"""Create a transformed tool from a parent tool.
|
||||
|
||||
|
|
@ -399,6 +400,7 @@ class TransformedTool(Tool):
|
|||
annotations=annotations or tool.annotations,
|
||||
serializer=serializer or tool.serializer,
|
||||
transform_args=transform_args,
|
||||
enabled=enabled if enabled is not None else True,
|
||||
)
|
||||
|
||||
return transformed_tool
|
||||
|
|
|
|||
|
|
@ -32,6 +32,11 @@ class FastMCPComponent(FastMCPBaseModel):
|
|||
description="Tags for the component.",
|
||||
)
|
||||
|
||||
enabled: bool = Field(
|
||||
default=True,
|
||||
description="Whether the component is enabled.",
|
||||
)
|
||||
|
||||
def __eq__(self, other: object) -> bool:
|
||||
if type(self) is not type(other):
|
||||
return False
|
||||
|
|
@ -39,4 +44,12 @@ 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."""
|
||||
self.enabled = True
|
||||
|
||||
def disable(self) -> None:
|
||||
"""Disable the component."""
|
||||
self.enabled = False
|
||||
|
|
|
|||
|
|
@ -178,7 +178,9 @@ class TestResources:
|
|||
assert json.loads(result[0].text) == USERS # type: ignore[attr-defined]
|
||||
|
||||
async def test_read_resource_returns_none_if_not_found(self, proxy_server):
|
||||
with pytest.raises(McpError, match="Unknown resource: resource://nonexistent"):
|
||||
with pytest.raises(
|
||||
McpError, match="Unknown resource: 'resource://nonexistent'"
|
||||
):
|
||||
async with Client(proxy_server) as client:
|
||||
await client.read_resource("resource://nonexistent")
|
||||
|
||||
|
|
|
|||
|
|
@ -699,6 +699,102 @@ class TestToolContextInjection:
|
|||
assert result[0].text == "3" # type: ignore[attr-defined]
|
||||
|
||||
|
||||
class TestToolEnabled:
|
||||
async def test_toggle_enabled(self):
|
||||
mcp = FastMCP()
|
||||
|
||||
@mcp.tool
|
||||
def sample_tool(x: int) -> int:
|
||||
return x * 2
|
||||
|
||||
assert sample_tool.enabled
|
||||
|
||||
tool = await mcp.get_tool("sample_tool")
|
||||
assert tool.enabled
|
||||
|
||||
tool.disable()
|
||||
|
||||
assert not tool.enabled
|
||||
assert not sample_tool.enabled
|
||||
|
||||
tool.enable()
|
||||
assert tool.enabled
|
||||
assert sample_tool.enabled
|
||||
|
||||
async def test_tool_disabled_in_decorator(self):
|
||||
mcp = FastMCP()
|
||||
|
||||
@mcp.tool(enabled=False)
|
||||
def sample_tool(x: int) -> int:
|
||||
return x * 2
|
||||
|
||||
async with Client(mcp) as client:
|
||||
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()
|
||||
|
||||
@mcp.tool(enabled=False)
|
||||
def sample_tool(x: int) -> int:
|
||||
return x * 2
|
||||
|
||||
sample_tool.enable()
|
||||
|
||||
async with Client(mcp) as client:
|
||||
tools = await client.list_tools()
|
||||
assert len(tools) == 1
|
||||
|
||||
async def test_tool_toggle_disabled(self):
|
||||
mcp = FastMCP()
|
||||
|
||||
@mcp.tool
|
||||
def sample_tool(x: int) -> int:
|
||||
return x * 2
|
||||
|
||||
sample_tool.disable()
|
||||
|
||||
async with Client(mcp) as client:
|
||||
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()
|
||||
|
||||
@mcp.tool
|
||||
def sample_tool(x: int) -> int:
|
||||
return x * 2
|
||||
|
||||
tool = await mcp.get_tool("sample_tool")
|
||||
assert tool.enabled
|
||||
|
||||
sample_tool.disable()
|
||||
|
||||
async with Client(mcp) as client:
|
||||
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()
|
||||
|
||||
@mcp.tool(enabled=False)
|
||||
def sample_tool(x: int) -> int:
|
||||
return x * 2
|
||||
|
||||
with pytest.raises(Exception, match="Unknown tool"):
|
||||
async with Client(mcp) as client:
|
||||
await client.call_tool("sample_tool", {"x": 5})
|
||||
|
||||
|
||||
class TestResource:
|
||||
async def test_text_resource(self):
|
||||
mcp = FastMCP()
|
||||
|
|
@ -783,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
|
||||
|
|
@ -1034,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."""
|
||||
|
||||
|
|
@ -1220,6 +1505,102 @@ class TestPrompts:
|
|||
assert prompt.tags == {"example", "test-tag"}
|
||||
|
||||
|
||||
class TestPromptEnabled:
|
||||
async def test_toggle_enabled(self):
|
||||
mcp = FastMCP()
|
||||
|
||||
@mcp.prompt
|
||||
def sample_prompt() -> str:
|
||||
return "Hello, world!"
|
||||
|
||||
assert sample_prompt.enabled
|
||||
|
||||
prompt = await mcp.get_prompt("sample_prompt")
|
||||
assert prompt.enabled
|
||||
|
||||
prompt.disable()
|
||||
|
||||
assert not prompt.enabled
|
||||
assert not sample_prompt.enabled
|
||||
|
||||
prompt.enable()
|
||||
assert prompt.enabled
|
||||
assert sample_prompt.enabled
|
||||
|
||||
async def test_prompt_disabled_in_decorator(self):
|
||||
mcp = FastMCP()
|
||||
|
||||
@mcp.prompt(enabled=False)
|
||||
def sample_prompt() -> str:
|
||||
return "Hello, world!"
|
||||
|
||||
async with Client(mcp) as client:
|
||||
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()
|
||||
|
||||
@mcp.prompt(enabled=False)
|
||||
def sample_prompt() -> str:
|
||||
return "Hello, world!"
|
||||
|
||||
sample_prompt.enable()
|
||||
|
||||
async with Client(mcp) as client:
|
||||
prompts = await client.list_prompts()
|
||||
assert len(prompts) == 1
|
||||
|
||||
async def test_prompt_toggle_disabled(self):
|
||||
mcp = FastMCP()
|
||||
|
||||
@mcp.prompt
|
||||
def sample_prompt() -> str:
|
||||
return "Hello, world!"
|
||||
|
||||
sample_prompt.disable()
|
||||
|
||||
async with Client(mcp) as client:
|
||||
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()
|
||||
|
||||
@mcp.prompt
|
||||
def sample_prompt() -> str:
|
||||
return "Hello, world!"
|
||||
|
||||
prompt = await mcp.get_prompt("sample_prompt")
|
||||
assert prompt.enabled
|
||||
|
||||
sample_prompt.disable()
|
||||
|
||||
async with Client(mcp) as client:
|
||||
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()
|
||||
|
||||
@mcp.prompt(enabled=False)
|
||||
def sample_prompt() -> str:
|
||||
return "Hello, world!"
|
||||
|
||||
with pytest.raises(McpError, match="Unknown prompt"):
|
||||
async with Client(mcp) as client:
|
||||
await client.get_prompt("sample_prompt")
|
||||
|
||||
|
||||
class TestPromptContext:
|
||||
async def test_prompt_context(self):
|
||||
mcp = FastMCP()
|
||||
|
|
|
|||
|
|
@ -9,6 +9,7 @@ from typing_extensions import TypedDict
|
|||
|
||||
from fastmcp import FastMCP
|
||||
from fastmcp.client.client import Client
|
||||
from fastmcp.exceptions import ToolError
|
||||
from fastmcp.tools import Tool, forward, forward_raw
|
||||
from fastmcp.tools.tool import FunctionTool
|
||||
from fastmcp.tools.tool_transform import ArgTransform, TransformedTool
|
||||
|
|
@ -51,7 +52,7 @@ async def test_tool_defaults_are_maintained_on_unmapped_args(add_tool):
|
|||
add_tool, transform_args={"old_x": ArgTransform(name="new_x")}
|
||||
)
|
||||
result = await new_tool.run(arguments={"new_x": 1})
|
||||
assert result[0].text == "11" # type: ignore
|
||||
assert result[0].text == "11" # type: ignore[attr-defined]
|
||||
|
||||
|
||||
async def test_tool_defaults_are_maintained_on_mapped_args(add_tool):
|
||||
|
|
@ -59,7 +60,7 @@ async def test_tool_defaults_are_maintained_on_mapped_args(add_tool):
|
|||
add_tool, transform_args={"old_y": ArgTransform(name="new_y")}
|
||||
)
|
||||
result = await new_tool.run(arguments={"old_x": 1})
|
||||
assert result[0].text == "11" # type: ignore
|
||||
assert result[0].text == "11" # type: ignore[attr-defined]
|
||||
|
||||
|
||||
def test_tool_change_arg_name(add_tool):
|
||||
|
|
@ -86,7 +87,7 @@ async def test_tool_drop_arg(add_tool):
|
|||
)
|
||||
assert sorted(new_tool.parameters["properties"]) == ["old_x"]
|
||||
result = await new_tool.run(arguments={"old_x": 1})
|
||||
assert result[0].text == "11" # type: ignore
|
||||
assert result[0].text == "11" # type: ignore[attr-defined]
|
||||
|
||||
|
||||
async def test_dropped_args_error_if_provided(add_tool):
|
||||
|
|
@ -108,7 +109,7 @@ async def test_hidden_arg_with_constant_default(add_tool):
|
|||
assert sorted(new_tool.parameters["properties"]) == ["old_x"]
|
||||
# Should pass old_x=5 and old_y=20 to parent
|
||||
result = await new_tool.run(arguments={"old_x": 5})
|
||||
assert result[0].text == "25" # type: ignore
|
||||
assert result[0].text == "25" # type: ignore[attr-defined]
|
||||
|
||||
|
||||
async def test_hidden_arg_without_default_uses_parent_default(add_tool):
|
||||
|
|
@ -120,7 +121,7 @@ async def test_hidden_arg_without_default_uses_parent_default(add_tool):
|
|||
assert sorted(new_tool.parameters["properties"]) == ["old_x"]
|
||||
# Should pass old_x=3 and let parent use its default old_y=10
|
||||
result = await new_tool.run(arguments={"old_x": 3})
|
||||
assert result[0].text == "13" # type: ignore
|
||||
assert result[0].text == "13" # type: ignore[attr-defined]
|
||||
|
||||
|
||||
async def test_mixed_hidden_args_with_custom_function(add_tool):
|
||||
|
|
@ -145,7 +146,7 @@ async def test_mixed_hidden_args_with_custom_function(add_tool):
|
|||
assert sorted(new_tool.parameters["properties"]) == ["visible_x"]
|
||||
# Should pass visible_x=7 as old_x=7 and old_y=25 to parent
|
||||
result = await new_tool.run(arguments={"visible_x": 7})
|
||||
assert result[0].text == "32" # type: ignore
|
||||
assert result[0].text == "32" # type: ignore[attr-defined]
|
||||
|
||||
|
||||
async def test_hide_required_param_without_default_raises_error():
|
||||
|
|
@ -183,7 +184,7 @@ async def test_hide_required_param_with_user_default_works():
|
|||
assert sorted(new_tool.parameters["properties"]) == ["optional_param"]
|
||||
# Should pass required_param=5 and optional_param=20 to parent
|
||||
result = await new_tool.run(arguments={"optional_param": 20})
|
||||
assert result[0].text == "25" # type: ignore
|
||||
assert result[0].text == "25" # type: ignore[attr-defined]
|
||||
|
||||
|
||||
async def test_forward_with_argument_mapping(add_tool):
|
||||
|
|
@ -202,7 +203,7 @@ async def test_forward_with_argument_mapping(add_tool):
|
|||
)
|
||||
|
||||
result = await new_tool.run(arguments={"new_x": 2, "new_y": 3})
|
||||
assert result[0].text == "5" # type: ignore
|
||||
assert result[0].text == "5" # type: ignore[attr-defined]
|
||||
|
||||
|
||||
async def test_forward_with_incorrect_args_raises_error(add_tool):
|
||||
|
|
@ -242,7 +243,7 @@ async def test_forward_raw_without_argument_mapping(add_tool):
|
|||
)
|
||||
|
||||
result = await new_tool.run(arguments={"new_x": 2, "new_y": 3})
|
||||
assert result[0].text == "5" # type: ignore
|
||||
assert result[0].text == "5" # type: ignore[attr-defined]
|
||||
|
||||
|
||||
async def test_custom_fn_with_kwargs_and_no_transform_args(add_tool):
|
||||
|
|
@ -252,7 +253,7 @@ async def test_custom_fn_with_kwargs_and_no_transform_args(add_tool):
|
|||
|
||||
new_tool = Tool.from_tool(add_tool, transform_fn=custom_fn)
|
||||
result = await new_tool.run(arguments={"extra": 1, "old_x": 2, "old_y": 3})
|
||||
assert result[0].text == "6" # type: ignore
|
||||
assert result[0].text == "6" # type: ignore[attr-defined]
|
||||
assert new_tool.parameters["required"] == IsList(
|
||||
"extra", "old_x", check_order=False
|
||||
)
|
||||
|
|
@ -269,7 +270,7 @@ async def test_fn_with_kwargs_passes_through_original_args(add_tool):
|
|||
|
||||
new_tool = Tool.from_tool(add_tool, transform_fn=custom_fn)
|
||||
result = await new_tool.run(arguments={"new_y": 2, "old_y": 3})
|
||||
assert result[0].text == "5" # type: ignore
|
||||
assert result[0].text == "5" # type: ignore[attr-defined]
|
||||
|
||||
|
||||
async def test_fn_with_kwargs_receives_transformed_arg_names(add_tool):
|
||||
|
|
@ -287,7 +288,7 @@ async def test_fn_with_kwargs_receives_transformed_arg_names(add_tool):
|
|||
transform_args={"old_x": ArgTransform(name="new_x")},
|
||||
)
|
||||
result = await new_tool.run(arguments={"new_x": 2, "old_y": 3})
|
||||
assert result[0].text == "5" # type: ignore
|
||||
assert result[0].text == "5" # type: ignore[attr-defined]
|
||||
|
||||
|
||||
async def test_fn_with_kwargs_handles_partial_explicit_args(add_tool):
|
||||
|
|
@ -307,7 +308,7 @@ async def test_fn_with_kwargs_handles_partial_explicit_args(add_tool):
|
|||
result = await new_tool.run(
|
||||
arguments={"new_x": 3, "old_y": 7, "some_other_param": "test"}
|
||||
)
|
||||
assert result[0].text == "10" # type: ignore
|
||||
assert result[0].text == "10" # type: ignore[attr-defined]
|
||||
|
||||
|
||||
async def test_fn_with_kwargs_mixed_mapped_and_unmapped_args(add_tool):
|
||||
|
|
@ -325,7 +326,7 @@ async def test_fn_with_kwargs_mixed_mapped_and_unmapped_args(add_tool):
|
|||
transform_args={"old_x": ArgTransform(name="new_x")},
|
||||
) # only map 'a'
|
||||
result = await new_tool.run(arguments={"new_x": 1, "old_y": 5})
|
||||
assert result[0].text == "6" # type: ignore
|
||||
assert result[0].text == "6" # type: ignore[attr-defined]
|
||||
|
||||
|
||||
async def test_fn_with_kwargs_dropped_args_not_in_kwargs(add_tool):
|
||||
|
|
@ -468,7 +469,7 @@ async def test_tool_transform_chaining(add_tool):
|
|||
tool2 = Tool.from_tool(tool1, transform_args={"x": ArgTransform(name="final_x")})
|
||||
|
||||
result = await tool2.run(arguments={"final_x": 5})
|
||||
assert result[0].text == "15" # type: ignore
|
||||
assert result[0].text == "15" # type: ignore[attr-defined]
|
||||
|
||||
# Transform tool1 with custom function that handles all parameters
|
||||
async def custom(final_x: int, **kwargs) -> str:
|
||||
|
|
@ -479,7 +480,7 @@ async def test_tool_transform_chaining(add_tool):
|
|||
tool1, transform_fn=custom, transform_args={"x": ArgTransform(name="final_x")}
|
||||
)
|
||||
result = await tool3.run(arguments={"final_x": 3, "old_y": 5})
|
||||
assert result[0].text == "custom 8" # type: ignore
|
||||
assert result[0].text == "custom 8" # type: ignore[attr-defined]
|
||||
|
||||
|
||||
class MyModel(BaseModel):
|
||||
|
|
@ -634,7 +635,7 @@ async def test_arg_transform_precedence_over_function_with_kwargs():
|
|||
# Test it works at runtime
|
||||
result = await tool.run(arguments={"y": "test"})
|
||||
# Should use ArgTransform default of 42
|
||||
assert "42: test" in result[0].text # type: ignore
|
||||
assert "42: test" in result[0].text # type: ignore[attr-defined]
|
||||
|
||||
|
||||
def test_arg_transform_combined_attributes():
|
||||
|
|
@ -691,8 +692,8 @@ async def test_arg_transform_type_precedence_runtime():
|
|||
|
||||
# Test it works with string input
|
||||
result = await tool.run(arguments={"x": "5", "y": 3})
|
||||
assert "String input '5'" in result[0].text # type: ignore
|
||||
assert "result: 8" in result[0].text # type: ignore
|
||||
assert "String input '5'" in result[0].text # type: ignore[attr-defined]
|
||||
assert "result: 8" in result[0].text # type: ignore[attr-defined]
|
||||
|
||||
|
||||
class TestProxy:
|
||||
|
|
@ -727,7 +728,7 @@ class TestProxy:
|
|||
async with Client(proxy_server) as client:
|
||||
# The tool should be registered with its transformed name
|
||||
result = await client.call_tool("add_transformed", {"new_x": 1, "old_y": 2})
|
||||
assert result[0].text == "3" # type: ignore
|
||||
assert result[0].text == "3" # type: ignore[attr-defined]
|
||||
|
||||
|
||||
async def test_arg_transform_default_factory():
|
||||
|
|
@ -750,7 +751,7 @@ async def test_arg_transform_default_factory():
|
|||
|
||||
# Should work without providing timestamp (gets value from factory)
|
||||
result = await new_tool.run(arguments={"x": 42})
|
||||
assert result[0].text == "42_12345.0" # type: ignore
|
||||
assert result[0].text == "42_12345.0" # type: ignore[attr-defined]
|
||||
|
||||
|
||||
async def test_arg_transform_default_factory_called_each_time():
|
||||
|
|
@ -778,11 +779,11 @@ async def test_arg_transform_default_factory_called_each_time():
|
|||
|
||||
# First call
|
||||
result1 = await new_tool.run(arguments={"x": 1})
|
||||
assert result1[0].text == "1_1" # type: ignore
|
||||
assert result1[0].text == "1_1" # type: ignore[attr-defined]
|
||||
|
||||
# Second call should get a different value
|
||||
result2 = await new_tool.run(arguments={"x": 2})
|
||||
assert result2[0].text == "2_2" # type: ignore
|
||||
assert result2[0].text == "2_2" # type: ignore[attr-defined]
|
||||
|
||||
|
||||
async def test_arg_transform_hidden_with_default_factory():
|
||||
|
|
@ -807,7 +808,7 @@ async def test_arg_transform_hidden_with_default_factory():
|
|||
|
||||
# Should pass hidden request_id with factory value
|
||||
result = await new_tool.run(arguments={"x": 42})
|
||||
assert result[0].text == "42_req_123" # type: ignore
|
||||
assert result[0].text == "42_req_123" # type: ignore[attr-defined]
|
||||
|
||||
|
||||
async def test_arg_transform_default_and_factory_raises_error():
|
||||
|
|
@ -942,3 +943,47 @@ async def test_arg_transform_hide_and_required_raises_error():
|
|||
ValueError, match="Cannot specify both 'hide=True' and 'required=True'"
|
||||
):
|
||||
ArgTransform(hide=True, required=True)
|
||||
|
||||
|
||||
class TestEnableDisable:
|
||||
async def test_transform_disabled_tool(self):
|
||||
"""
|
||||
Tests that a transformed tool can run even if the parent tool is disabled
|
||||
"""
|
||||
mcp = FastMCP()
|
||||
|
||||
@mcp.tool(enabled=False)
|
||||
def add(x: int, y: int = 10) -> int:
|
||||
return x + y
|
||||
|
||||
new_add = Tool.from_tool(add, name="new_add")
|
||||
mcp.add_tool(new_add)
|
||||
|
||||
assert new_add.enabled
|
||||
|
||||
async with Client(mcp) as client:
|
||||
tools = await client.list_tools()
|
||||
assert {tool.name for tool in tools} == {"new_add"}
|
||||
|
||||
result = await client.call_tool("new_add", {"x": 1, "y": 2})
|
||||
assert result[0].text == "3" # type: ignore[attr-defined]
|
||||
|
||||
with pytest.raises(ToolError):
|
||||
await client.call_tool("add", {"x": 1, "y": 2})
|
||||
|
||||
async def test_disable_transformed_tool(self):
|
||||
mcp = FastMCP()
|
||||
|
||||
@mcp.tool(enabled=False)
|
||||
def add(x: int, y: int = 10) -> int:
|
||||
return x + y
|
||||
|
||||
new_add = Tool.from_tool(add, name="new_add", enabled=False)
|
||||
mcp.add_tool(new_add)
|
||||
|
||||
async with Client(mcp) as client:
|
||||
tools = await client.list_tools()
|
||||
assert len(tools) == 0
|
||||
|
||||
with pytest.raises(ToolError):
|
||||
await client.call_tool("new_add", {"x": 1, "y": 2})
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue