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

Merge pull request #976 from gorocode/feature/component-manager

Browse source 
Component Manager
This commit is contained in:
Jeremiah Lowin 2025-06-27 20:43:18 -04:00 • committed by GitHub
commit 05104aa3a8
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 1387 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

170
src/fastmcp/contrib/component_manager/README.md Normal file
View file

@ -0,0 +1,170 @@
# Component Manager – Contrib Module for FastMCP
The **Component Manager** provides a unified API for enabling and disabling tools, resources, and prompts at runtime in a FastMCP server. This module is useful for dynamic control over which components are active, enabling advanced features like feature toggling, admin interfaces, or automation workflows.
---
## 🔧 Features
- Enable/disable **tools**, **resources**, and **prompts** via HTTP endpoints.
- Supports **local** and **mounted (server)** components.
- Customizable **API root path**.
- Optional **Auth scopes** for secured access.
- Fully integrates with FastMCP with minimal configuration.
---
## 📦 Installation
This module is part of the `fastmcp.contrib` package. No separate installation is required if you're already using **FastMCP**.
---
## 🚀 Usage
### Basic Setup
```python
from fastmcp import FastMCP
from fastmcp.contrib.component_manager import set_up_component_manager
mcp = FastMCP(name="Component Manager", instructions="This is a test server with component manager.")
set_up_component_manager(server=mcp)
```
---
## 🔗 API Endpoints
All endpoints are registered at `/` by default, or under the custom path if one is provided.
### Tools
```http
POST /tools/{tool_name}/enable
POST /tools/{tool_name}/disable
```
### Resources
```http
POST /resources/{uri:path}/enable
POST /resources/{uri:path}/disable
```
* Supports template URIs as well
```http
POST /resources/example://test/{id}/enable
POST /resources/example://test/{id}/disable
```
### Prompts
```http
POST /prompts/{prompt_name}/enable
POST /prompts/{prompt_name}/disable
```
---
#### 🧪 Example Response
```http
HTTP/1.1 200 OK
Content-Type: application/json
{
"message": "Disabled tool: example_tool"
}
```
---
## ⚙️ Configuration Options
### Custom Root Path
To mount the API under a different path:
```python
set_up_component_manager(server=mcp, path="/admin")
```
### Securing Endpoints with Auth Scopes
If your server uses authentication:
```python
mcp = FastMCP(name="Component Manager", instructions="This is a test server with component manager.", auth=auth)
set_up_component_manager(server=mcp, required_scopes=["write", "read"])
```
---
## 🧪 Example: Enabling a Tool with Curl
```bash
curl -X POST \
-H "Authorization: Bearer YOUR_TOKEN_HERE" \
-H "Content-Type: application/json" \
http://localhost:8001/tools/example_tool/enable
```
---
## 🧱 Working with Mounted Servers
You can also combine different configurations when working with mounted servers — for example, using different scopes:
```python
mcp = FastMCP(name="Component Manager", instructions="This is a test server with component manager.", auth=auth)
set_up_component_manager(server=mcp, required_scopes=["mcp:write"])
mounted = FastMCP(name="Component Manager", instructions="This is a test server with component manager.", auth=auth)
set_up_component_manager(server=mounted, required_scopes=["mounted:write"])
mcp.mount(server=mounted, prefix="mo")
```
This allows you to grant different levels of access:
```bash
# Accessing the main server gives you control over both local and mounted components
curl -X POST \
-H "Authorization: Bearer YOUR_TOKEN_HERE" \
-H "Content-Type: application/json" \
http://localhost:8001/tools/mo_example_tool/enable
# Accessing the mounted server gives you control only over its own components
curl -X POST \
-H "Authorization: Bearer YOUR_TOKEN_HERE" \
-H "Content-Type: application/json" \
http://localhost:8002/tools/example_tool/enable
```
---
## ⚙️ How It Works
- `set_up_component_manager()` registers API routes for tools, resources, and prompts.
- The `ComponentService` class exposes async methods to enable/disable components.
- Each endpoint returns a success message in JSON or a 404 error if the component isn't found.
---
## 🧩 Extending
You can subclass `ComponentService` for custom behavior or mount its routes elsewhere as needed.
---
## Maintenance Notice
This module is not officially maintained by the core FastMCP team. It is an independent extension developed by [gorocode](https://github.com/gorocode).
If you encounter any issues or wish to contribute, please feel free to open an issue or submit a pull request, and kindly notify me. I'd love to stay up to date.
## 📄 License
This module follows the license of the main [FastMCP](https://github.com/jlowin/fastmcp) project.

4
src/fastmcp/contrib/component_manager/__init__.py Normal file
View file

@ -0,0 +1,4 @@
from .component_manager import set_up_component_manager
from .component_service import ComponentService
__all__ = ["set_up_component_manager", "ComponentService"]

186
src/fastmcp/contrib/component_manager/component_manager.py Normal file
View file

@ -0,0 +1,186 @@
"""
Routes and helpers for managing tools, resources, and prompts in FastMCP.
Provides endpoints for enabling/disabling components via HTTP, with optional authentication scopes.
"""
from typing import Any
from mcp.server.auth.middleware.bearer_auth import RequireAuthMiddleware
from starlette.applications import Starlette
from starlette.exceptions import HTTPException as StarletteHTTPException
from starlette.requests import Request
from starlette.responses import JSONResponse
from starlette.routing import Mount, Route
from fastmcp.contrib.component_manager.component_service import ComponentService
from fastmcp.exceptions import NotFoundError
from fastmcp.server.server import FastMCP
def set_up_component_manager(
server: FastMCP, path: str = "/", required_scopes: list[str] | None = None
):
"""Set up routes for enabling/disabling tools, resources, and prompts.
Args:
server: The FastMCP server instance
path: Path used to mount all component-related routes on the server
required_scopes: Optional list of scopes required for these routes. Applies only if authentication is enabled.
"""
service = ComponentService(server)
routes: list[Route] = []
mounts: list[Mount] = []
route_configs = {
"tool": {
"param": "tool_name",
"enable": service._enable_tool,
"disable": service._disable_tool,
},
"resource": {
"param": "uri:path",
"enable": service._enable_resource,
"disable": service._disable_resource,
},
"prompt": {
"param": "prompt_name",
"enable": service._enable_prompt,
"disable": service._disable_prompt,
},
}
if required_scopes is None:
routes.extend(build_component_manager_endpoints(route_configs, path))
else:
if path != "/":
mounts.append(
build_component_manager_mount(route_configs, path, required_scopes)
)
else:
mounts.append(
build_component_manager_mount(
{"tool": route_configs["tool"]}, "/tools", required_scopes
)
)
mounts.append(
build_component_manager_mount(
{"resource": route_configs["resource"]},
"/resources",
required_scopes,
)
)
mounts.append(
build_component_manager_mount(
{"prompt": route_configs["prompt"]}, "/prompts", required_scopes
)
)
server._additional_http_routes.extend(routes)
server._additional_http_routes.extend(mounts)
def make_endpoint(action, component, config):
"""
Factory for creating Starlette endpoint functions for enabling/disabling a component.
Args:
action: 'enable' or 'disable'
component: The component type (e.g., 'tool', 'resource', or 'prompt')
config: Dict with param and handler functions for the component
Returns:
An async endpoint function for Starlette.
"""
async def endpoint(request: Request):
name = request.path_params[config["param"].split(":")[0]]
try:
await config[action](name)
return JSONResponse(
{"message": f"{action.capitalize()}d {component}: {name}"}
)
except NotFoundError:
raise StarletteHTTPException(
status_code=404,
detail=f"Unknown {component}: {name}",
)
return endpoint
def make_route(action, component, config, required_scopes, root_path) -> Route:
"""
Creates a Starlette Route for enabling/disabling a component.
Args:
action: 'enable' or 'disable'
component: The component type
config: Dict with param and handler functions
required_scopes: Optional list of required auth scopes
root_path: The base path for the route
Returns:
A Starlette Route object.
"""
endpoint = make_endpoint(action, component, config)
if required_scopes is not None and root_path in [
"/tools",
"/resources",
"/prompts",
]:
path = f"/{{{config['param']}}}/{action}"
else:
if root_path != "/" and required_scopes is None:
path = f"{root_path}/{component}s/{{{config['param']}}}/{action}"
else:
path = f"/{component}s/{{{config['param']}}}/{action}"
return Route(path, endpoint=endpoint, methods=["POST"])
def build_component_manager_endpoints(
route_configs, root_path, required_scopes=None
) -> list[Route]:
"""
Build a list of Starlette Route objects for all components/actions.
Args:
route_configs: Dict describing component types and their handlers
root_path: The base path for the routes
required_scopes: Optional list of required auth scopes
Returns:
List of Starlette Route objects for component management.
"""
component_management_routes: list[Route] = []
for component in route_configs:
config: dict[str, Any] = route_configs[component]
for action in ["enable", "disable"]:
component_management_routes.append(
make_route(action, component, config, required_scopes, root_path)
)
return component_management_routes
def build_component_manager_mount(route_configs, root_path, required_scopes) -> Mount:
"""
Build a Starlette Mount with authentication for component management routes.
Args:
route_configs: Dict describing component types and their handlers
root_path: The base path for the mount
required_scopes: List of required auth scopes
Returns:
A Starlette Mount object with authentication middleware.
"""
component_management_routes: list[Route] = []
for component in route_configs:
config: dict[str, Any] = route_configs[component]
for action in ["enable", "disable"]:
component_management_routes.append(
make_route(action, component, config, required_scopes, root_path)
)
return Mount(
f"{root_path}",
app=RequireAuthMiddleware(
Starlette(routes=component_management_routes), required_scopes
),
)

225
src/fastmcp/contrib/component_manager/component_service.py Normal file
View file

@ -0,0 +1,225 @@
"""
ComponentService: Provides async management of tools, resources, and prompts for FastMCP servers.
Handles enabling/disabling components both locally and across mounted servers.
"""
from fastmcp.exceptions import NotFoundError
from fastmcp.prompts.prompt import Prompt
from fastmcp.resources.resource import Resource
from fastmcp.resources.template import ResourceTemplate
from fastmcp.server.server import FastMCP, has_resource_prefix, remove_resource_prefix
from fastmcp.tools.tool import Tool
from fastmcp.utilities.logging import get_logger
logger = get_logger(__name__)
class ComponentService:
"""Service for managing components like tools, resources, and prompts."""
def __init__(self, server: FastMCP):
self._server = server
self._tool_manager = server._tool_manager
self._resource_manager = server._resource_manager
self._prompt_manager = server._prompt_manager
async def _enable_tool(self, key: str) -> Tool:
"""Handle 'enableTool' requests.
Args:
key: The key of the tool to enable
Returns:
The tool that was enabled
"""
logger.debug("Enabling tool: %s", key)
# 1. Check local tools first. The server will have already applied its filter.
if key in self._server._tool_manager._tools:
tool: Tool = await self._server.get_tool(key)
tool.enable()
return tool
# 2. Check mounted servers using the filtered protocol path.
for mounted in reversed(self._tool_manager._mounted_servers):
if mounted.prefix:
if key.startswith(f"{mounted.prefix}_"):
tool_key = key.removeprefix(f"{mounted.prefix}_")
mounted_service = ComponentService(mounted.server)
tool = await mounted_service._enable_tool(tool_key)
return tool
else:
continue
raise NotFoundError(f"Unknown tool: {key}")
async def _disable_tool(self, key: str) -> Tool:
"""Handle 'disableTool' requests.
Args:
key: The key of the tool to disable
Returns:
The tool that was disabled
"""
logger.debug("Disable tool: %s", key)
# 1. Check local tools first. The server will have already applied its filter.
if key in self._server._tool_manager._tools:
tool: Tool = await self._server.get_tool(key)
tool.disable()
return tool
# 2. Check mounted servers using the filtered protocol path.
for mounted in reversed(self._tool_manager._mounted_servers):
if mounted.prefix:
if key.startswith(f"{mounted.prefix}_"):
tool_key = key.removeprefix(f"{mounted.prefix}_")
mounted_service = ComponentService(mounted.server)
tool = await mounted_service._disable_tool(tool_key)
return tool
else:
continue
raise NotFoundError(f"Unknown tool: {key}")
async def _enable_resource(self, key: str) -> Resource | ResourceTemplate:
"""Handle 'enableResource' requests.
Args:
key: The key of the resource to enable
Returns:
The resource that was enabled
"""
logger.debug("Enabling resource: %s", key)
# 1. Check local resources first. The server will have already applied its filter.
if key in self._resource_manager._resources:
resource: Resource = await self._server.get_resource(key)
resource.enable()
return resource
if key in self._resource_manager._templates:
template: ResourceTemplate = await self._server.get_resource_template(key)
template.enable()
return template
# 2. Check mounted servers using the filtered protocol path.
for mounted in reversed(self._resource_manager._mounted_servers):
if mounted.prefix:
if has_resource_prefix(
key,
mounted.prefix,
mounted.resource_prefix_format,
):
key = remove_resource_prefix(
key,
mounted.prefix,
mounted.resource_prefix_format,
)
mounted_service = ComponentService(mounted.server)
mounted_resource: (
Resource | ResourceTemplate
) = await mounted_service._enable_resource(key)
return mounted_resource
else:
continue
raise NotFoundError(f"Unknown resource: {key}")
async def _disable_resource(self, key: str) -> Resource | ResourceTemplate:
"""Handle 'disableResource' requests.
Args:
key: The key of the resource to disable
Returns:
The resource that was disabled
"""
logger.debug("Disable resource: %s", key)
# 1. Check local resources first. The server will have already applied its filter.
if key in self._resource_manager._resources:
resource: Resource = await self._server.get_resource(key)
resource.disable()
return resource
if key in self._resource_manager._templates:
template: ResourceTemplate = await self._server.get_resource_template(key)
template.disable()
return template
# 2. Check mounted servers using the filtered protocol path.
for mounted in reversed(self._resource_manager._mounted_servers):
if mounted.prefix:
if has_resource_prefix(
key,
mounted.prefix,
mounted.resource_prefix_format,
):
key = remove_resource_prefix(
key,
mounted.prefix,
mounted.resource_prefix_format,
)
mounted_service = ComponentService(mounted.server)
mounted_resource: (
Resource | ResourceTemplate
) = await mounted_service._disable_resource(key)
return mounted_resource
else:
continue
raise NotFoundError(f"Unknown resource: {key}")
async def _enable_prompt(self, key: str) -> Prompt:
"""Handle 'enablePrompt' requests.
Args:
key: The key of the prompt to enable
Returns:
The prompt that was enable
"""
logger.debug("Enabling prompt: %s", key)
# 1. Check local prompts first. The server will have already applied its filter.
if key in self._server._prompt_manager._prompts:
prompt: Prompt = await self._server.get_prompt(key)
prompt.enable()
return prompt
# 2. Check mounted servers using the filtered protocol path.
for mounted in reversed(self._prompt_manager._mounted_servers):
if mounted.prefix:
if key.startswith(f"{mounted.prefix}_"):
prompt_key = key.removeprefix(f"{mounted.prefix}_")
mounted_service = ComponentService(mounted.server)
prompt = await mounted_service._enable_prompt(prompt_key)
return prompt
else:
continue
raise NotFoundError(f"Unknown prompt: {key}")
async def _disable_prompt(self, key: str) -> Prompt:
"""Handle 'disablePrompt' requests.
Args:
key: The key of the prompt to disable
Returns:
The prompt that was disabled
"""
# 1. Check local prompts first. The server will have already applied its filter.
if key in self._server._prompt_manager._prompts:
prompt: Prompt = await self._server.get_prompt(key)
prompt.disable()
return prompt
# 2. Check mounted servers using the filtered protocol path.
for mounted in reversed(self._prompt_manager._mounted_servers):
if mounted.prefix:
if key.startswith(f"{mounted.prefix}_"):
prompt_key = key.removeprefix(f"{mounted.prefix}_")
mounted_service = ComponentService(mounted.server)
prompt = await mounted_service._disable_prompt(prompt_key)
return prompt
else:
continue
raise NotFoundError(f"Unknown prompt: {key}")

59
src/fastmcp/contrib/component_manager/example.py Normal file
View file

@ -0,0 +1,59 @@
from fastmcp import FastMCP
from fastmcp.contrib.component_manager import set_up_component_manager
from fastmcp.server.auth.providers.bearer import BearerAuthProvider, RSAKeyPair
key_pair = RSAKeyPair.generate()
auth = BearerAuthProvider(
public_key=key_pair.public_key,
issuer="https://dev.example.com",
audience="my-dev-server",
required_scopes=["mcp:read"],
)
# Build main server
mcp_token = key_pair.create_token(
subject="dev-user",
issuer="https://dev.example.com",
audience="my-dev-server",
scopes=["mcp:write", "mcp:read"],
)
mcp = FastMCP(
name="Component Manager",
instructions="This is a test server with component manager.",
auth=auth,
)
# Set up main server component manager
set_up_component_manager(server=mcp, required_scopes=["mcp:write"])
# Build mounted server
mounted_token = key_pair.create_token(
subject="dev-user",
issuer="https://dev.example.com",
audience="my-dev-server",
scopes=["mounted:write", "mcp:read"],
)
mounted = FastMCP(
name="Component Manager",
instructions="This is a test server with component manager.",
auth=auth,
)
# Set up mounted server component manager
set_up_component_manager(server=mounted, required_scopes=["mounted:write"])
# Mount
mcp.mount(server=mounted, prefix="mo")
@mcp.resource("resource://greeting")
def get_greeting() -> str:
"""Provides a simple greeting message."""
return "Hello from FastMCP Resources!"
@mounted.tool("greeting")
def get_info() -> str:
"""Provides a simple info."""
return "You are using component manager contrib module!"

743
tests/contrib/test_component_manager.py Normal file
View file

@ -0,0 +1,743 @@
import pytest
from starlette import status
from starlette.testclient import TestClient
from fastmcp import FastMCP
from fastmcp.contrib.component_manager import set_up_component_manager
from fastmcp.server.auth.providers.bearer import BearerAuthProvider, RSAKeyPair
class TestComponentManagementRoutes:
"""Test the component management routes for tools, resources, and prompts."""
@pytest.fixture
def mounted_mcp(self):
"""Create a FastMCP server with a mounted sub-server and a tool, resource, and prompt on the sub-server."""
mounted_mcp = FastMCP("SubServer")
@mounted_mcp.tool()
def mounted_tool() -> str:
"""Test tool for tool management routes."""
return "mounted_tool_result"
@mounted_mcp.resource("data://mounted_resource")
def mounted_resource() -> str:
"""Test resource for tool management routes."""
return "mounted_resource_result"
# Add a test resource
@mounted_mcp.resource("data://mounted_resource/{id}")
def test_template(id: str) -> dict:
"""Test template for tool management routes."""
return {"id": id, "value": "data"}
@mounted_mcp.prompt()
def mounted_prompt() -> str:
"""Test prompt for tool management routes."""
return "mounted_prompt_result"
return mounted_mcp
@pytest.fixture
def mcp(self, mounted_mcp):
"""Create a FastMCP server with test tools, resources, and prompts."""
mcp = FastMCP("TestServer")
mcp.mount(mounted_mcp, prefix="sub")
set_up_component_manager(server=mcp)
# Add a test tool
@mcp.tool
def test_tool() -> str:
"""Test tool for tool management routes."""
return "test_tool_result"
# Add a test resource
@mcp.resource("data://test_resource")
def test_resource() -> str:
"""Test resource for tool management routes."""
return "test_resource_result"
# Add a test resource
@mcp.resource("data://test_resource/{id}")
def test_template(id: str) -> dict:
"""Test template for tool management routes."""
return {"id": id, "value": "data"}
# Add a test prompt
@mcp.prompt
def test_prompt() -> str:
"""Test prompt for tool management routes."""
return "test_prompt_result"
return mcp
@pytest.fixture
def client(self, mcp):
"""Create a test client for the FastMCP server."""
return TestClient(mcp.http_app())
async def test_enable_tool_route(self, client, mcp):
"""Test enabling a tool via the HTTP route."""
# First disable the tool
tool = await mcp._tool_manager.get_tool("test_tool")
tool.enabled = False
# Enable the tool via the HTTP route
response = client.post("/tools/test_tool/enable")
assert response.status_code == status.HTTP_200_OK
assert response.json() == {"message": "Enabled tool: test_tool"}
# Verify the tool is enabled
tool = await mcp._tool_manager.get_tool("test_tool")
assert tool.enabled is True
async def test_disable_tool_route(self, client, mcp):
"""Test disabling a tool via the HTTP route."""
# First ensure the tool is enabled
tool = await mcp._tool_manager.get_tool("test_tool")
tool.enabled = True
# Disable the tool via the HTTP route
response = client.post("/tools/test_tool/disable")
assert response.status_code == status.HTTP_200_OK
assert response.json() == {"message": "Disabled tool: test_tool"}
# Verify the tool is disabled
tool = await mcp._tool_manager.get_tool("test_tool")
assert tool.enabled is False
async def test_enable_resource_route(self, client, mcp):
"""Test enabling a resource via the HTTP route."""
# First disable the resource
resource = await mcp._resource_manager.get_resource("data://test_resource")
resource.enabled = False
# Enable the resource via the HTTP route
response = client.post("/resources/data://test_resource/enable")
assert response.status_code == status.HTTP_200_OK
assert response.json() == {"message": "Enabled resource: data://test_resource"}
# Verify the resource is enabled
resource = await mcp._resource_manager.get_resource("data://test_resource")
assert resource.enabled is True
async def test_disable_resource_route(self, client, mcp):
"""Test disabling a resource via the HTTP route."""
# First ensure the resource is enabled
resource = await mcp._resource_manager.get_resource("data://test_resource")
resource.enabled = True
# Disable the resource via the HTTP route
response = client.post("/resources/data://test_resource/disable")
assert response.status_code == status.HTTP_200_OK
assert response.json() == {"message": "Disabled resource: data://test_resource"}
# Verify the resource is disabled
resource = await mcp._resource_manager.get_resource("data://test_resource")
assert resource.enabled is False
async def test_enable_template_route(self, client, mcp):
"""Test enabling a resource on a mounted server via the parent server's HTTP route."""
key = "data://test_resource/{id}"
resource = mcp._resource_manager._templates[key]
resource.enabled = False
response = client.post("/resources/data://test_resource/{id}/enable")
assert response.status_code == status.HTTP_200_OK
assert response.json() == {
"message": "Enabled resource: data://test_resource/{id}"
}
assert resource.enabled is True
async def test_disable_template_route(self, client, mcp):
"""Test disabling a resource on a mounted server via the parent server's HTTP route."""
key = "data://test_resource/{id}"
resource = mcp._resource_manager._templates[key]
resource.enabled = True
response = client.post("/resources/data://test_resource/{id}/disable")
assert response.status_code == status.HTTP_200_OK
assert response.json() == {
"message": "Disabled resource: data://test_resource/{id}"
}
assert resource.enabled is False
async def test_enable_prompt_route(self, client, mcp):
"""Test enabling a prompt via the HTTP route."""
# First disable the prompt
prompt = await mcp._prompt_manager.get_prompt("test_prompt")
prompt.enabled = False
# Enable the prompt via the HTTP route
response = client.post("/prompts/test_prompt/enable")
assert response.status_code == status.HTTP_200_OK
assert response.json() == {"message": "Enabled prompt: test_prompt"}
# Verify the prompt is enabled
prompt = await mcp._prompt_manager.get_prompt("test_prompt")
assert prompt.enabled is True
async def test_disable_prompt_route(self, client, mcp):
"""Test disabling a prompt via the HTTP route."""
# First ensure the prompt is enabled
prompt = await mcp._prompt_manager.get_prompt("test_prompt")
prompt.enabled = True
# Disable the prompt via the HTTP route
response = client.post("/prompts/test_prompt/disable")
assert response.status_code == status.HTTP_200_OK
assert response.json() == {"message": "Disabled prompt: test_prompt"}
# Verify the prompt is disabled
prompt = await mcp._prompt_manager.get_prompt("test_prompt")
assert prompt.enabled is False
async def test_enable_tool_route_on_mounted_server(self, client, mounted_mcp):
"""Test enabling a tool on a mounted server via the parent server's HTTP route."""
# Disable the tool on the sub-server
sub_tool = await mounted_mcp._tool_manager.get_tool("mounted_tool")
sub_tool.enabled = False
# Enable via parent
response = client.post("/tools/sub_mounted_tool/enable")
assert response.status_code == status.HTTP_200_OK
assert response.json() == {"message": "Enabled tool: sub_mounted_tool"}
# Confirm disabled on sub-server
assert sub_tool.enabled is True
async def test_disable_tool_route_on_mounted_server(self, client, mounted_mcp):
"""Test disabling a tool on a mounted server via the parent server's HTTP route."""
# Enable the tool on the sub-server
sub_tool = await mounted_mcp._tool_manager.get_tool("mounted_tool")
sub_tool.enabled = True
# Disable via parent
response = client.post("/tools/sub_mounted_tool/disable")
assert response.status_code == status.HTTP_200_OK
assert response.json() == {"message": "Disabled tool: sub_mounted_tool"}
# Confirm disabled on sub-server
assert sub_tool.enabled is False
async def test_enable_resource_route_on_mounted_server(self, client, mounted_mcp):
"""Test enabling a resource on a mounted server via the parent server's HTTP route."""
resource = await mounted_mcp._resource_manager.get_resource(
"data://mounted_resource"
)
resource.enabled = False
response = client.post("/resources/data://sub/mounted_resource/enable")
assert response.status_code == status.HTTP_200_OK
assert response.json() == {
"message": "Enabled resource: data://sub/mounted_resource"
}
resource = await mounted_mcp._resource_manager.get_resource(
"data://mounted_resource"
)
assert resource.enabled is True
async def test_disable_resource_route_on_mounted_server(self, client, mounted_mcp):
"""Test disabling a resource on a mounted server via the parent server's HTTP route."""
resource = await mounted_mcp._resource_manager.get_resource(
"data://mounted_resource"
)
resource.enabled = True
response = client.post("/resources/data://sub/mounted_resource/disable")
assert response.status_code == status.HTTP_200_OK
assert response.json() == {
"message": "Disabled resource: data://sub/mounted_resource"
}
resource = await mounted_mcp._resource_manager.get_resource(
"data://mounted_resource"
)
assert resource.enabled is False
async def test_enable_template_route_on_mounted_server(self, client, mounted_mcp):
"""Test enabling a resource on a mounted server via the parent server's HTTP route."""
key = "data://mounted_resource/{id}"
resource = mounted_mcp._resource_manager._templates[key]
resource.enabled = False
response = client.post("/resources/data://sub/mounted_resource/{id}/enable")
assert response.status_code == status.HTTP_200_OK
assert response.json() == {
"message": "Enabled resource: data://sub/mounted_resource/{id}"
}
assert resource.enabled is True
async def test_disable_template_route_on_mounted_server(self, client, mounted_mcp):
"""Test disabling a resource on a mounted server via the parent server's HTTP route."""
key = "data://mounted_resource/{id}"
resource = mounted_mcp._resource_manager._templates[key]
resource.enabled = True
response = client.post("/resources/data://sub/mounted_resource/{id}/disable")
assert response.status_code == status.HTTP_200_OK
assert response.json() == {
"message": "Disabled resource: data://sub/mounted_resource/{id}"
}
assert resource.enabled is False
async def test_enable_prompt_route_on_mounted_server(self, client, mounted_mcp):
"""Test enabling a prompt on a mounted server via the parent server's HTTP route."""
prompt = await mounted_mcp._prompt_manager.get_prompt("mounted_prompt")
prompt.enabled = False
response = client.post("/prompts/sub_mounted_prompt/enable")
assert response.status_code == status.HTTP_200_OK
assert response.json() == {"message": "Enabled prompt: sub_mounted_prompt"}
prompt = await mounted_mcp._prompt_manager.get_prompt("mounted_prompt")
assert prompt.enabled is True
async def test_disable_prompt_route_on_mounted_server(self, client, mounted_mcp):
"""Test disabling a prompt on a mounted server via the parent server's HTTP route."""
prompt = await mounted_mcp._prompt_manager.get_prompt("mounted_prompt")
prompt.enabled = True
response = client.post("/prompts/sub_mounted_prompt/disable")
assert response.status_code == status.HTTP_200_OK
assert response.json() == {"message": "Disabled prompt: sub_mounted_prompt"}
prompt = await mounted_mcp._prompt_manager.get_prompt("mounted_prompt")
assert prompt.enabled is False
def test_enable_nonexistent_tool(self, client):
"""Test enabling a non-existent tool returns 404."""
response = client.post("/tools/nonexistent_tool/enable")
assert response.status_code == status.HTTP_404_NOT_FOUND
assert response.text == "Unknown tool: nonexistent_tool"
def test_disable_nonexistent_tool(self, client):
"""Test disabling a non-existent tool returns 404."""
response = client.post("/tools/nonexistent_tool/disable")
assert response.status_code == status.HTTP_404_NOT_FOUND
assert response.text == "Unknown tool: nonexistent_tool"
def test_enable_nonexistent_resource(self, client):
"""Test enabling a non-existent resource returns 404."""
response = client.post("/resources/nonexistent://resource/enable")
assert response.status_code == status.HTTP_404_NOT_FOUND
assert response.text == "Unknown resource: nonexistent://resource"
def test_disable_nonexistent_resource(self, client):
"""Test disabling a non-existent resource returns 404."""
response = client.post("/resources/nonexistent://resource/disable")
assert response.status_code == status.HTTP_404_NOT_FOUND
assert response.text == "Unknown resource: nonexistent://resource"
def test_enable_nonexistent_prompt(self, client):
"""Test enabling a non-existent prompt returns 404."""
response = client.post("/prompts/nonexistent_prompt/enable")
assert response.status_code == status.HTTP_404_NOT_FOUND
assert response.text == "Unknown prompt: nonexistent_prompt"
def test_disable_nonexistent_prompt(self, client):
"""Test disabling a non-existent prompt returns 404."""
response = client.post("/prompts/nonexistent_prompt/disable")
assert response.status_code == status.HTTP_404_NOT_FOUND
assert response.text == "Unknown prompt: nonexistent_prompt"
class TestAuthComponentManagementRoutes:
"""Test the component management routes with authentication for tools, resources, and prompts."""
def setup_method(self):
"""Set up test fixtures."""
# Generate a key pair and create an auth provider
key_pair = RSAKeyPair.generate()
self.auth = BearerAuthProvider(
public_key=key_pair.public_key,
issuer="https://dev.example.com",
audience="my-dev-server",
)
self.mcp = FastMCP("TestServerWithAuth", auth=self.auth)
set_up_component_manager(
server=self.mcp, required_scopes=["tool:write", "tool:read"]
)
self.token = key_pair.create_token(
subject="dev-user",
issuer="https://dev.example.com",
audience="my-dev-server",
scopes=["tool:write", "tool:read"],
)
self.token_without_scopes = key_pair.create_token(
subject="dev-user",
issuer="https://dev.example.com",
audience="my-dev-server",
scopes=["tool:read"],
)
# Add test components
@self.mcp.tool
def test_tool() -> str:
"""Test tool for auth testing."""
return "test_tool_result"
@self.mcp.resource("data://test_resource")
def test_resource() -> str:
"""Test resource for auth testing."""
return "test_resource_result"
@self.mcp.prompt
def test_prompt() -> str:
"""Test prompt for auth testing."""
return "test_prompt_result"
# Create test client
self.client = TestClient(self.mcp.http_app())
async def test_unauthorized_enable_tool(self):
"""Test that unauthenticated requests to enable a tool are rejected."""
tool = await self.mcp._tool_manager.get_tool("test_tool")
tool.enabled = False
response = self.client.post("/tools/test_tool/enable")
assert response.status_code == 401
assert tool.enabled is False
async def test_authorized_enable_tool(self):
"""Test that authenticated requests to enable a tool are allowed."""
tool = await self.mcp._tool_manager.get_tool("test_tool")
tool.enabled = False
response = self.client.post(
"/tools/test_tool/enable", headers={"Authorization": "Bearer " + self.token}
)
assert response.status_code == 200
assert response.json() == {"message": "Enabled tool: test_tool"}
assert tool.enabled is True
async def test_unauthorized_disable_tool(self):
"""Test that unauthenticated requests to disable a tool are rejected."""
tool = await self.mcp._tool_manager.get_tool("test_tool")
tool.enabled = True
response = self.client.post("/tools/test_tool/disable")
assert response.status_code == 401
assert tool.enabled is True
async def test_authorized_disable_tool(self):
"""Test that authenticated requests to disable a tool are allowed."""
tool = await self.mcp._tool_manager.get_tool("test_tool")
tool.enabled = True
response = self.client.post(
"/tools/test_tool/disable",
headers={"Authorization": "Bearer " + self.token},
)
assert response.status_code == 200
assert response.json() == {"message": "Disabled tool: test_tool"}
assert tool.enabled is False
async def test_forbidden_enable_tool(self):
"""Test that unauthenticated requests to enable a resource are rejected."""
tool = await self.mcp._tool_manager.get_tool("test_tool")
tool.enabled = False
response = self.client.post(
"/tools/test_tool/enable",
headers={"Authorization": "Bearer " + self.token_without_scopes},
)
assert response.status_code == 403
assert tool.enabled is False
async def test_authorized_enable_resource(self):
"""Test that authenticated requests to enable a resource are allowed."""
resource = await self.mcp._resource_manager.get_resource("data://test_resource")
resource.enabled = False
response = self.client.post(
"/resources/data://test_resource/enable",
headers={"Authorization": "Bearer " + self.token},
)
assert response.status_code == 200
assert response.json() == {"message": "Enabled resource: data://test_resource"}
assert resource.enabled is True
async def test_unauthorized_disable_resource(self):
"""Test that unauthenticated requests to disable a resource are rejected."""
resource = await self.mcp._resource_manager.get_resource("data://test_resource")
resource.enabled = True
response = self.client.post("/resources/data://test_resource/disable")
assert response.status_code == 401
assert resource.enabled is True
async def test_forbidden_enable_resource(self):
"""Test that unauthenticated requests to enable a resource are rejected."""
resource = await self.mcp._resource_manager.get_resource("data://test_resource")
resource.enabled = False
response = self.client.post(
"/resources/data://test_resource/disable",
headers={"Authorization": "Bearer " + self.token_without_scopes},
)
assert response.status_code == 403
assert resource.enabled is False
async def test_authorized_disable_resource(self):
"""Test that authenticated requests to disable a resource are allowed."""
resource = await self.mcp._resource_manager.get_resource("data://test_resource")
resource.enabled = True
response = self.client.post(
"/resources/data://test_resource/disable",
headers={"Authorization": "Bearer " + self.token},
)
assert response.status_code == 200
assert response.json() == {"message": "Disabled resource: data://test_resource"}
assert resource.enabled is False
async def test_unauthorized_enable_prompt(self):
"""Test that unauthenticated requests to enable a prompt are rejected."""
prompt = await self.mcp._prompt_manager.get_prompt("test_prompt")
prompt.enabled = False
response = self.client.post("/prompts/test_prompt/enable")
assert response.status_code == 401
assert prompt.enabled is False
async def test_authorized_enable_prompt(self):
"""Test that authenticated requests to enable a prompt are allowed."""
prompt = await self.mcp._prompt_manager.get_prompt("test_prompt")
prompt.enabled = False
response = self.client.post(
"/prompts/test_prompt/enable",
headers={"Authorization": "Bearer " + self.token},
)
assert response.status_code == 200
assert response.json() == {"message": "Enabled prompt: test_prompt"}
assert prompt.enabled is True
async def test_unauthorized_disable_prompt(self):
"""Test that unauthenticated requests to disable a prompt are rejected."""
prompt = await self.mcp._prompt_manager.get_prompt("test_prompt")
prompt.enabled = True
response = self.client.post("/prompts/test_prompt/disable")
assert response.status_code == 401
assert prompt.enabled is True
async def test_forbidden_disable_prompt(self):
"""Test that unauthenticated requests to enable a resource are rejected."""
prompt = await self.mcp._prompt_manager.get_prompt("test_prompt")
prompt.enabled = True
response = self.client.post(
"/prompts/test_prompt/disable",
headers={"Authorization": "Bearer " + self.token_without_scopes},
)
assert response.status_code == 403
assert prompt.enabled is True
async def test_authorized_disable_prompt(self):
"""Test that authenticated requests to disable a prompt are allowed."""
prompt = await self.mcp._prompt_manager.get_prompt("test_prompt")
prompt.enabled = True
response = self.client.post(
"/prompts/test_prompt/disable",
headers={"Authorization": "Bearer " + self.token},
)
assert response.status_code == 200
assert response.json() == {"message": "Disabled prompt: test_prompt"}
assert prompt.enabled is False
class TestComponentManagerWithPath:
"""Test component manager routes when mounted at a custom path."""
@pytest.fixture
def mcp_with_path(self):
mcp = FastMCP("TestServerWithPath")
set_up_component_manager(server=mcp, path="/test")
@mcp.tool
def test_tool() -> str:
return "test_tool_result"
@mcp.resource("data://test_resource")
def test_resource() -> str:
return "test_resource_result"
@mcp.prompt
def test_prompt() -> str:
return "test_prompt_result"
return mcp
@pytest.fixture
def client_with_path(self, mcp_with_path):
return TestClient(mcp_with_path.http_app())
@pytest.mark.asyncio
async def test_enable_tool_route_with_path(self, client_with_path, mcp_with_path):
tool = await mcp_with_path._tool_manager.get_tool("test_tool")
tool.enabled = False
response = client_with_path.post("/test/tools/test_tool/enable")
assert response.status_code == status.HTTP_200_OK
assert response.json() == {"message": "Enabled tool: test_tool"}
tool = await mcp_with_path._tool_manager.get_tool("test_tool")
assert tool.enabled is True
@pytest.mark.asyncio
async def test_disable_resource_route_with_path(
self, client_with_path, mcp_with_path
):
resource = await mcp_with_path._resource_manager.get_resource(
"data://test_resource"
)
resource.enabled = True
response = client_with_path.post("/test/resources/data://test_resource/disable")
assert response.status_code == status.HTTP_200_OK
assert response.json() == {"message": "Disabled resource: data://test_resource"}
resource = await mcp_with_path._resource_manager.get_resource(
"data://test_resource"
)
assert resource.enabled is False
@pytest.mark.asyncio
async def test_enable_prompt_route_with_path(self, client_with_path, mcp_with_path):
prompt = await mcp_with_path._prompt_manager.get_prompt("test_prompt")
prompt.enabled = False
response = client_with_path.post("/test/prompts/test_prompt/enable")
assert response.status_code == status.HTTP_200_OK
assert response.json() == {"message": "Enabled prompt: test_prompt"}
prompt = await mcp_with_path._prompt_manager.get_prompt("test_prompt")
assert prompt.enabled is True
class TestComponentManagerWithPathAuth:
"""Test component manager routes with auth when mounted at a custom path."""
def setup_method(self):
# Generate a key pair and create an auth provider
key_pair = RSAKeyPair.generate()
self.auth = BearerAuthProvider(
public_key=key_pair.public_key,
issuer="https://dev.example.com",
audience="my-dev-server",
required_scopes=["tool:write", "tool:read"],
)
self.mcp = FastMCP("TestServerWithPathAuth", auth=self.auth)
set_up_component_manager(
server=self.mcp, path="/test", required_scopes=["tool:write", "tool:read"]
)
self.token = key_pair.create_token(
subject="dev-user",
issuer="https://dev.example.com",
audience="my-dev-server",
scopes=["tool:read", "tool:write"],
)
self.token_without_scopes = key_pair.create_token(
subject="dev-user",
issuer="https://dev.example.com",
audience="my-dev-server",
scopes=[],
)
@self.mcp.tool
def test_tool() -> str:
return "test_tool_result"
@self.mcp.resource("data://test_resource")
def test_resource() -> str:
return "test_resource_result"
@self.mcp.prompt
def test_prompt() -> str:
return "test_prompt_result"
self.client = TestClient(self.mcp.http_app())
@pytest.mark.asyncio
async def test_unauthorized_enable_tool(self):
tool = await self.mcp._tool_manager.get_tool("test_tool")
tool.enabled = False
response = self.client.post("/test/tools/test_tool/enable")
assert response.status_code == 401
assert tool.enabled is False
@pytest.mark.asyncio
async def test_forbidden_enable_tool(self):
tool = await self.mcp._tool_manager.get_tool("test_tool")
tool.enabled = False
response = self.client.post(
"/test/tools/test_tool/enable",
headers={"Authorization": "Bearer " + self.token_without_scopes},
)
assert response.status_code == 403
assert tool.enabled is False
@pytest.mark.asyncio
async def test_authorized_enable_tool(self):
tool = await self.mcp._tool_manager.get_tool("test_tool")
tool.enabled = False
response = self.client.post(
"/test/tools/test_tool/enable",
headers={"Authorization": "Bearer " + self.token},
)
assert response.status_code == 200
assert response.json() == {"message": "Enabled tool: test_tool"}
tool = await self.mcp._tool_manager.get_tool("test_tool")
assert tool.enabled is True
@pytest.mark.asyncio
async def test_unauthorized_disable_resource(self):
resource = await self.mcp._resource_manager.get_resource("data://test_resource")
resource.enabled = True
response = self.client.post("/test/resources/data://test_resource/disable")
assert response.status_code == 401
assert resource.enabled is True
@pytest.mark.asyncio
async def test_forbidden_disable_resource(self):
resource = await self.mcp._resource_manager.get_resource("data://test_resource")
resource.enabled = True
response = self.client.post(
"/test/resources/data://test_resource/disable",
headers={"Authorization": "Bearer " + self.token_without_scopes},
)
assert response.status_code == 403
assert resource.enabled is True
@pytest.mark.asyncio
async def test_authorized_disable_resource(self):
resource = await self.mcp._resource_manager.get_resource("data://test_resource")
resource.enabled = True
response = self.client.post(
"/test/resources/data://test_resource/disable",
headers={"Authorization": "Bearer " + self.token},
)
assert response.status_code == 200
assert response.json() == {"message": "Disabled resource: data://test_resource"}
resource = await self.mcp._resource_manager.get_resource("data://test_resource")
assert resource.enabled is False
@pytest.mark.asyncio
async def test_unauthorized_enable_prompt(self):
prompt = await self.mcp._prompt_manager.get_prompt("test_prompt")
prompt.enabled = False
response = self.client.post("/test/prompts/test_prompt/enable")
assert response.status_code == 401
assert prompt.enabled is False
@pytest.mark.asyncio
async def test_forbidden_enable_prompt(self):
prompt = await self.mcp._prompt_manager.get_prompt("test_prompt")
prompt.enabled = False
response = self.client.post(
"/test/prompts/test_prompt/enable",
headers={"Authorization": "Bearer " + self.token_without_scopes},
)
assert response.status_code == 403
assert prompt.enabled is False
@pytest.mark.asyncio
async def test_authorized_enable_prompt(self):
prompt = await self.mcp._prompt_manager.get_prompt("test_prompt")
prompt.enabled = False
response = self.client.post(
"/test/prompts/test_prompt/enable",
headers={"Authorization": "Bearer " + self.token},
)
assert response.status_code == 200
assert response.json() == {"message": "Enabled prompt: test_prompt"}
prompt = await self.mcp._prompt_manager.get_prompt("test_prompt")
assert prompt.enabled is True