ceos/fastmcp
1
0
Fork 0
mirror of https://github.com/PrefectHQ/fastmcp.git synced 2026-08-09 15:19:10 +02:00
Code Issues Projects Releases Packages Wiki Activity

Add missing documentation

Browse source
This commit is contained in:
Goro 2025-06-27 22:17:44 +02:00
commit 7d252e6da4
6 changed files with 221 additions and 49 deletions
Download patch file Download diff file Expand all files Collapse all files

53
src/fastmcp/contrib/component_manager/README.md
View file

@ -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.

5
src/fastmcp/contrib/component_manager/__init__.py
View file

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

108
src/fastmcp/contrib/component_manager/component_manager.py
View file

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

38
src/fastmcp/contrib/component_manager/component_service.py
View file

@ -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:

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

@ -0,0 +1,59 @@
from fastmcp import FastMCP
from fastmcp.contrib.component_manager import set_up_component_manager
from fastmcp.server.auth.providers.bearer import BearerAuthProvider, RSAKeyPair
key_pair = RSAKeyPair.generate()
auth = BearerAuthProvider(
public_key=key_pair.public_key,
issuer="https://dev.example.com",
audience="my-dev-server",
required_scopes=["mcp:read"],
)
# Build main server
mcp_token = key_pair.create_token(
subject="dev-user",
issuer="https://dev.example.com",
audience="my-dev-server",
scopes=["mcp:write", "mcp:read"],
)
mcp = FastMCP(
"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!"

9
tests/contrib/test_component_manager.py
View file

@ -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
assert prompt.enabled is False