diff --git a/src/fastmcp/contrib/component_manager/README.md b/src/fastmcp/contrib/component_manager/README.md new file mode 100644 index 000000000..efdeb3b7b --- /dev/null +++ b/src/fastmcp/contrib/component_manager/README.md @@ -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. \ No newline at end of file diff --git a/src/fastmcp/contrib/component_manager/__init__.py b/src/fastmcp/contrib/component_manager/__init__.py new file mode 100644 index 000000000..6bb6c89ba --- /dev/null +++ b/src/fastmcp/contrib/component_manager/__init__.py @@ -0,0 +1,4 @@ +from .component_manager import set_up_component_manager +from .component_service import ComponentService + +__all__ = ["set_up_component_manager", "ComponentService"] diff --git a/src/fastmcp/contrib/component_manager/component_manager.py b/src/fastmcp/contrib/component_manager/component_manager.py new file mode 100644 index 000000000..01a24eff0 --- /dev/null +++ b/src/fastmcp/contrib/component_manager/component_manager.py @@ -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 + ), + ) diff --git a/src/fastmcp/contrib/component_manager/component_service.py b/src/fastmcp/contrib/component_manager/component_service.py new file mode 100644 index 000000000..d8b1736f3 --- /dev/null +++ b/src/fastmcp/contrib/component_manager/component_service.py @@ -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}") diff --git a/src/fastmcp/contrib/component_manager/example.py b/src/fastmcp/contrib/component_manager/example.py new file mode 100644 index 000000000..845c374ff --- /dev/null +++ b/src/fastmcp/contrib/component_manager/example.py @@ -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!" diff --git a/tests/contrib/test_component_manager.py b/tests/contrib/test_component_manager.py new file mode 100644 index 000000000..8fea3c8bc --- /dev/null +++ b/tests/contrib/test_component_manager.py @@ -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