Refactor duplicated logic

src/fastmcp/contrib/component_manager/README.md

@@ -0,0 +1,125 @@
+# 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.component_manager import set_up_component_manager
+
+mcp = FastMCP("Component Manager", instructions="This is a test server with component manager.")
+set_up_component_manager(server=mcp)
+```
+
+---
+
+## 🔗 API Endpoints
+
+By default, 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
+```
+
+ * Works with template URIs too
+```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
+```
+
+---
+
+## ⚙️ 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("Component Manager", instructions="This is a test server with component manager.", auth=auth)
+set_up_component_manager(server=mcp, required_scopes=["tools:write", "tools: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
+```
+
+---
+
+## ⚙️ 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,7 @@
+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

@@ -14,11 +14,12 @@ from typing import Any
 from fastmcp.server.server import FastMCP
 
 def set_up_component_manager(
-    server: FastMCP, root_path: str = "/", required_scopes: list[str] | None = None
+    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
+        root_path: Path used to mount all component-related routes on the server
         required_scopes: Optional list of scopes required for these routes
     Returns:
         A list of routes or mounts for component management
@@ -47,13 +48,13 @@ def set_up_component_manager(
 
     if required_scopes is None:
         routes.extend(
-            build_component_manager_enpoints(route_configs, root_path)
+            build_component_manager_endpoints(route_configs, path)
         )        
     else:
-        if root_path != "/":
+        if path != "/":
             mounts.append(
                 build_component_manager_mount(
-                    route_configs, root_path, required_scopes
+                    route_configs, path, required_scopes
             ))
         else:
             mounts.append(
@@ -73,78 +74,52 @@ def set_up_component_manager(
     server._additional_http_routes.extend(mounts)
 
 
-def build_component_manager_enpoints(route_configs, root_path, required_scopes=None) -> list[Route]:
+def make_endpoint(action, component, config):
+    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:
+    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:
+        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]:
     component_management_routes: list[Route] = []
 
     for component in route_configs:
         config: dict[str, Any] = route_configs[component]
         for action in ["enable", "disable"]:
-
-            async def endpoint(
-                request: Request,
-                action: str = action,
-                component: str = component,
-                config: dict[str, Any] = config,
-            ):
-                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}",
-                    )
-
-            if required_scopes is not None and root_path in ["/tools", "/resources", "/prompts"]:
-                path = f"/{{{config['param']}}}/{action}"
-            else:
-                path = f"/{component}s/{{{config['param']}}}/{action}"
-
-            route = Route(path, endpoint=endpoint, methods=["POST"])
-            component_management_routes.append(route)
+            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:
     component_management_routes: list[Route] = []
 
     for component in route_configs:
         config: dict[str, Any] = route_configs[component]
         for action in ["enable", "disable"]:
-
-            async def endpoint(
-                request: Request,
-                action: str = action,
-                component: str = component,
-                config: dict[str, Any] = config,
-            ):
-                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}",
-                    )
-
-            if required_scopes is not None and root_path in ["/tools", "/resources", "/prompts"]:
-                path = f"/{{{config['param']}}}/{action}"
-            else:
-                path = f"/{component}s/{{{config['param']}}}/{action}"
-
-            route = Route(path, endpoint=endpoint, methods=["POST"])
-            component_management_routes.append(route)
+            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)
-    )
+        app=RequireAuthMiddleware(Starlette(routes=component_management_routes), required_scopes)
+    )

tests/contrib/test_component_manager.py

@@ -3,7 +3,7 @@ from starlette import status
 from starlette.testclient import TestClient
 
 from fastmcp import FastMCP
-from fastmcp.contrib.component_manager.component_manager import set_up_component_manager
+from fastmcp.contrib.component_manager import set_up_component_manager
 from fastmcp.server.auth.providers.bearer import BearerAuthProvider, RSAKeyPair