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

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.

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"]

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
+        ),
+    )

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}")

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!"

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