diff --git a/src/fastmcp/contrib/component_manager/README.md b/src/fastmcp/contrib/component_manager/README.md index b7b1dccec..32cd4121d 100644 --- a/src/fastmcp/contrib/component_manager/README.md +++ b/src/fastmcp/contrib/component_manager/README.md @@ -26,7 +26,7 @@ This module is part of the `fastmcp.contrib` package. No separate installation i ```python 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 mcp = FastMCP("Component Manager", instructions="This is a test server with component manager.") set_up_component_manager(server=mcp) @@ -36,7 +36,7 @@ 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. +All endpoints are registered at `/` by default, or under the custom path if one is provided. ### Tools @@ -52,7 +52,7 @@ POST /resources/{uri:path}/enable POST /resources/{uri:path}/disable ``` - * Works with template URIs too + * Supports template URIs as well ```http POST /resources/example://test/{id}/enable POST /resources/example://test/{id}/disable @@ -63,6 +63,19 @@ POST /resources/example://test/{id}/disable ```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" +} + ``` --- @@ -83,7 +96,7 @@ 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"]) +set_up_component_manager(server=mcp, required_scopes=["write", "read"]) ``` --- @@ -99,6 +112,38 @@ curl -X POST \ --- +## ๐Ÿงฑ Working with Mounted Servers + +You can also combine different configurations when working with mounted servers โ€” for example, using different scopes: + +```python +mcp = FastMCP("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("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. diff --git a/src/fastmcp/contrib/component_manager/__init__.py b/src/fastmcp/contrib/component_manager/__init__.py index edc85e475..6bb6c89ba 100644 --- a/src/fastmcp/contrib/component_manager/__init__.py +++ b/src/fastmcp/contrib/component_manager/__init__.py @@ -1,7 +1,4 @@ from .component_manager import set_up_component_manager from .component_service import ComponentService -__all__ = [ - "set_up_component_manager", - "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 index 6fc20eace..a5d0259fd 100644 --- a/src/fastmcp/contrib/component_manager/component_manager.py +++ b/src/fastmcp/contrib/component_manager/component_manager.py @@ -1,28 +1,30 @@ +""" +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 mcp.server.auth.middleware.bearer_auth import RequireAuthMiddleware -from typing import Any - 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 - 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 + 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) @@ -47,34 +49,46 @@ def set_up_component_manager( } if required_scopes is None: - routes.extend( - build_component_manager_endpoints(route_configs, path) - ) + routes.extend(build_component_manager_endpoints(route_configs, path)) else: if path != "/": mounts.append( - build_component_manager_mount( - route_configs, path, required_scopes - )) + 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 - )) + {"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]] @@ -88,38 +102,82 @@ def make_endpoint(action, component, config): 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"]: + 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]: + + +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)) + 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)) + 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 + ), ) diff --git a/src/fastmcp/contrib/component_manager/component_service.py b/src/fastmcp/contrib/component_manager/component_service.py index 7077d8d2a..777834f52 100644 --- a/src/fastmcp/contrib/component_manager/component_service.py +++ b/src/fastmcp/contrib/component_manager/component_service.py @@ -1,16 +1,22 @@ +""" +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.tools.tool import Tool - -from fastmcp.utilities.logging import get_logger 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: + + +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 @@ -33,7 +39,7 @@ class ComponentService: 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: @@ -63,7 +69,7 @@ class ComponentService: 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: @@ -112,11 +118,13 @@ class ComponentService: mounted.resource_prefix_format, ) mounted_service = ComponentService(mounted.server) - mounted_resource: Resource | ResourceTemplate = await mounted_service._enable_resource(key) + mounted_resource: ( + Resource | ResourceTemplate + ) = await mounted_service._enable_resource(key) mounted_resource.enable() return mounted_resource else: - continue + continue raise NotFoundError(f"Unknown resource: {key}") async def _disable_resource(self, key: str) -> Resource | ResourceTemplate: @@ -154,11 +162,13 @@ class ComponentService: mounted.resource_prefix_format, ) mounted_service = ComponentService(mounted.server) - mounted_resource: Resource | ResourceTemplate = await mounted_service._disable_resource(key) + mounted_resource: ( + Resource | ResourceTemplate + ) = await mounted_service._disable_resource(key) mounted_resource.disable() return mounted_resource else: - continue + continue raise NotFoundError(f"Unknown resource: {key}") async def _enable_prompt(self, key: str) -> Prompt: @@ -177,7 +187,7 @@ class ComponentService: 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: @@ -190,7 +200,7 @@ class ComponentService: else: continue raise NotFoundError(f"Unknown prompt: {key}") - + async def _disable_prompt(self, key: str) -> Prompt: """Handle 'disablePrompt' requests. @@ -206,7 +216,7 @@ class ComponentService: 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: diff --git a/src/fastmcp/contrib/component_manager/example.py b/src/fastmcp/contrib/component_manager/example.py new file mode 100644 index 000000000..e05455446 --- /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( + "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( + "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 index 644890ea0..e2b43c36d 100644 --- a/tests/contrib/test_component_manager.py +++ b/tests/contrib/test_component_manager.py @@ -44,6 +44,7 @@ class TestComponentManagementRoutes: 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: @@ -345,7 +346,9 @@ class TestAuthComponentManagementRoutes: audience="my-dev-server", ) self.mcp = FastMCP("TestServerWithAuth", auth=self.auth) - set_up_component_manager(server=self.mcp, required_scopes=["tool:write", "tool:read"]) + 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", @@ -445,7 +448,7 @@ class TestAuthComponentManagementRoutes: 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") @@ -534,4 +537,4 @@ class TestAuthComponentManagementRoutes: ) assert response.status_code == 200 assert response.json() == {"message": "Disabled prompt: test_prompt"} - assert prompt.enabled is False \ No newline at end of file + assert prompt.enabled is False