Updates based on PR Feedback

This commit is contained in:
strawgate 2025-04-16 23:30:09 -05:00
commit 5de40ac300
6 changed files with 376 additions and 32 deletions

9
src/contrib/README.md Normal file
View file

@ -0,0 +1,9 @@
# FastMCP Contrib Modules
This directory holds community-contributed modules for FastMCP. These modules extend FastMCP's functionality but are not officially maintained by the core team.
**Guarantees:**
* Modules in `contrib` may have different testing requirements or stability guarantees compared to the core library.
* Changes to the core FastMCP library might break modules in `contrib` without explicit warnings in the main changelog.
Use these modules at your own discretion. Contributions are welcome, but please include tests and documentation.

View file

@ -0,0 +1,39 @@
# MCP Mixin
This module provides the `MCPMixin` base class and associated decorators (`@mcp_tool`, `@mcp_resource`, `@mcp_prompt`).
It allows developers to easily define classes whose methods can be registered as tools, resources, or prompts with a `FastMCP` server instance using the `register_all()`, `register_tools()`, `register_resources()`, or `register_prompts()` methods provided by the mixin.
## Usage
Inherit from `MCPMixin` and use the decorators on the methods you want to register.
```python
from fastmcp import FastMCP
from contrib.mcp_mixin.mcp_mixin import MCPMixin, mcp_tool, mcp_resource
class MyComponent(MCPMixin):
@mcp_tool(name="my_tool", description="Does something cool.")
def tool_method(self):
return "Tool executed!"
@mcp_resource(uri="component://data")
def resource_method(self):
return {"data": "some data"}
mcp_server = FastMCP()
component = MyComponent()
# Register all decorated methods with a prefix
# Useful if you will have multiple instantiated objects of the same class
# and want to avoid name collisions.
component.register_all(mcp_server, prefix="my_comp")
# Register without a prefix
# component.register_all(mcp_server)
# Now 'my_comp_my_tool' tool and 'my_comp+component://data' resource are registered (if prefix used)
# Or 'my_tool' and 'component://data' are registered (if no prefix used)
```
The `prefix` argument in registration methods is optional. If omitted, methods are registered with their original decorated names/URIs. Individual separators (`tools_separator`, `resources_separator`, `prompts_separator`) can also be provided to `register_all` to change the separator for specific types.

View file

@ -1,17 +1,17 @@
"""Sample code for FastMCP."""
"""Sample code for FastMCP using MCPMixin."""
from src.fastmcp import FastMCP
from src.fastmcp.utilities.registerable import (
McpRegisterable,
from contrib.mcp_mixin.mcp_mixin import (
MCPMixin,
mcp_prompt,
mcp_resource,
mcp_tool,
)
from fastmcp import FastMCP
mcp = FastMCP()
class Sample(McpRegisterable):
class Sample(MCPMixin):
def __init__(self, name):
self.name = name
@ -39,7 +39,11 @@ second_sample.register_all(mcp_server=mcp, prefix="second")
def main():
mcp.run("sse")
print("MCP Server running with registered components...")
print("Tools:", list(mcp.get_tools().keys()))
print("Resources:", list(mcp.get_resources().keys()))
print("Prompts:", [p.name for p in mcp.list_prompts()])
mcp.run()
if __name__ == "__main__":

View file

@ -1,15 +1,19 @@
"""Provides a base class and decorators for easy registration of class methods with FastMCP."""
"""Provides a base mixin class and decorators for easy registration of class methods with FastMCP."""
from collections.abc import Callable
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from ..server import FastMCP
from fastmcp.server import FastMCP
_MCP_REGISTRATION_TOOL_ATTR = "_mcp_tool_registration"
_MCP_REGISTRATION_RESOURCE_ATTR = "_mcp_resource_registration"
_MCP_REGISTRATION_PROMPT_ATTR = "_mcp_prompt_registration"
_DEFAULT_SEPARATOR_TOOL = "_"
_DEFAULT_SEPARATOR_RESOURCE = "+"
_DEFAULT_SEPARATOR_PROMPT = "_"
def mcp_tool(
name: str | None = None,
@ -80,81 +84,125 @@ def mcp_prompt(
return decorator
class McpRegisterable:
"""Base class for objects that can register tools, resources, and prompts
class MCPMixin:
"""Base mixin class for objects that can register tools, resources, and prompts
with a FastMCP server instance using decorators.
This mixin provides methods like `register_all`, `register_tools`, etc.,
which iterate over the methods of the inheriting class, find methods
decorated with `@mcp_tool`, `@mcp_resource`, or `@mcp_prompt`, and
register them with the provided FastMCP server instance.
"""
def _get_methods_to_register(self, registration_type: str):
"""Retrieves all registration info for the specified type."""
"""Retrieves all methods marked for a specific registration type."""
return [
(
getattr(self, method_name),
getattr(getattr(self, method_name), registration_type).copy(),
)
for method_name in dir(self)
if hasattr(getattr(self, method_name), registration_type)
if callable(getattr(self, method_name))
and hasattr(getattr(self, method_name), registration_type)
]
def register_tools(self, mcp_server: "FastMCP", prefix: str | None = None) -> None:
def register_tools(
self,
mcp_server: "FastMCP",
prefix: str | None = None,
separator: str = _DEFAULT_SEPARATOR_TOOL,
) -> None:
"""Registers all methods marked with @mcp_tool with the FastMCP server.
Args:
mcp_server: The FastMCP server instance to register tools with.
prefix: Optional prefix to prepend to tool names. If provided, the
final name will be f"{prefix}{separator}{original_name}".
separator: The separator string used between prefix and original name.
Defaults to '_'.
"""
for method, registration_info in self._get_methods_to_register(
_MCP_REGISTRATION_TOOL_ATTR
):
if prefix:
registration_info["name"] = f"{prefix}_{registration_info['name']}"
registration_info["name"] = (
f"{prefix}{separator}{registration_info['name']}"
)
mcp_server.add_tool(fn=method, **registration_info)
def register_resources(
self, mcp_server: "FastMCP", prefix: str | None = None
self,
mcp_server: "FastMCP",
prefix: str | None = None,
separator: str = _DEFAULT_SEPARATOR_RESOURCE,
) -> None:
"""Registers all methods marked with @mcp_resource with the FastMCP server.
Args:
mcp_server: The FastMCP server instance to register resources with.
prefix: Optional prefix to prepend to resource names and URIs. If provided,
the final name will be f"{prefix}{separator}{original_name}" and the
final URI will be f"{prefix}{separator}{original_uri}".
separator: The separator string used between prefix and original name/URI.
Defaults to '+'.
"""
for method, registration_info in self._get_methods_to_register(
_MCP_REGISTRATION_RESOURCE_ATTR
):
if prefix:
registration_info["name"] = f"{prefix}_{registration_info['name']}"
registration_info["uri"] = f"{prefix}+{registration_info['uri']}"
registration_info["name"] = (
f"{prefix}{separator}{registration_info['name']}"
)
registration_info["uri"] = (
f"{prefix}{separator}{registration_info['uri']}"
)
mcp_server.add_resource_fn(fn=method, **registration_info)
def register_prompts(
self, mcp_server: "FastMCP", prefix: str | None = None
self,
mcp_server: "FastMCP",
prefix: str | None = None,
separator: str = _DEFAULT_SEPARATOR_PROMPT,
) -> None:
"""Registers all methods marked with @mcp_prompt with the FastMCP server.
Args:
mcp_server: The FastMCP server instance to register prompts with.
prefix: Optional prefix to prepend to prompt names. If provided, the
final name will be f"{prefix}{separator}{original_name}".
separator: The separator string used between prefix and original name.
Defaults to '_'.
"""
for method, registration_info in self._get_methods_to_register(
_MCP_REGISTRATION_PROMPT_ATTR
):
if prefix:
registration_info["name"] = f"{prefix}_{registration_info['name']}"
registration_info["name"] = (
f"{prefix}{separator}{registration_info['name']}"
)
mcp_server.add_prompt(fn=method, **registration_info)
def register_all(
self,
mcp_server: "FastMCP",
prefix: str | None = None,
tools_prefix: str | None = None,
resources_prefix: str | None = None,
prompts_prefix: str | None = None,
tool_separator: str = _DEFAULT_SEPARATOR_TOOL,
resource_separator: str = _DEFAULT_SEPARATOR_RESOURCE,
prompt_separator: str = _DEFAULT_SEPARATOR_PROMPT,
) -> None:
"""Registers all marked tools, resources, and prompts."""
self.register_tools(mcp_server, prefix=tools_prefix or prefix)
self.register_resources(mcp_server, prefix=resources_prefix or prefix)
self.register_prompts(mcp_server, prefix=prompts_prefix or prefix)
"""Registers all marked tools, resources, and prompts with the server.
This method calls `register_tools`, `register_resources`, and `register_prompts`
internally, passing the provided prefix and separators.
Args:
mcp_server: The FastMCP server instance to register with.
prefix: Optional prefix applied to all registered items unless overridden
by a specific separator argument.
tool_separator: Separator for tool names (defaults to '_').
resource_separator: Separator for resource names/URIs (defaults to '+').
prompt_separator: Separator for prompt names (defaults to '_').
"""
self.register_tools(mcp_server, prefix=prefix, separator=tool_separator)
self.register_resources(mcp_server, prefix=prefix, separator=resource_separator)
self.register_prompts(mcp_server, prefix=prefix, separator=prompt_separator)

View file

@ -0,0 +1 @@
# This file makes Python treat the directory as a package.

View file

@ -0,0 +1,243 @@
"""Tests for the MCPMixin class."""
import pytest
from contrib.mcp_mixin.mcp_mixin import (
_DEFAULT_SEPARATOR_PROMPT,
_DEFAULT_SEPARATOR_RESOURCE,
_DEFAULT_SEPARATOR_TOOL,
MCPMixin,
mcp_prompt,
mcp_resource,
mcp_tool,
)
from fastmcp import FastMCP
class TestMCPMixin:
"""Test suite for MCPMixin functionality."""
def test_initialization(self):
"""Test that a class inheriting MCPMixin can be initialized."""
class MyMixin(MCPMixin):
pass
instance = MyMixin()
assert instance is not None
# --- Tool Registration Tests ---
@pytest.mark.parametrize(
"prefix, separator, expected_key, unexpected_key",
[
(
None,
_DEFAULT_SEPARATOR_TOOL,
"sample_tool",
f"None{_DEFAULT_SEPARATOR_TOOL}sample_tool",
),
(
"pref",
_DEFAULT_SEPARATOR_TOOL,
f"pref{_DEFAULT_SEPARATOR_TOOL}sample_tool",
"sample_tool",
),
(
"pref",
"-",
"pref-sample_tool",
f"pref{_DEFAULT_SEPARATOR_TOOL}sample_tool",
),
],
ids=["No prefix", "Default separator", "Custom separator"],
)
def test_tool_registration(self, prefix, separator, expected_key, unexpected_key):
"""Test tool registration with prefix and separator variations."""
mcp = FastMCP()
class MyToolMixin(MCPMixin):
@mcp_tool()
def sample_tool(self):
pass
instance = MyToolMixin()
instance.register_tools(mcp, prefix=prefix, separator=separator)
registered_tools = mcp.get_tools()
assert expected_key in registered_tools
assert unexpected_key not in registered_tools
@pytest.mark.parametrize(
"prefix, separator, expected_uri_key, expected_name, unexpected_uri_key",
[
(
None,
_DEFAULT_SEPARATOR_RESOURCE,
"test://resource",
"sample_resource",
f"None{_DEFAULT_SEPARATOR_RESOURCE}test://resource",
),
(
"pref",
_DEFAULT_SEPARATOR_RESOURCE,
f"pref{_DEFAULT_SEPARATOR_RESOURCE}test://resource",
f"pref{_DEFAULT_SEPARATOR_RESOURCE}sample_resource",
"test://resource",
),
(
"pref",
"fff",
"prefffftest://resource",
"preffffsample_resource",
f"pref{_DEFAULT_SEPARATOR_RESOURCE}test://resource",
),
],
ids=["No prefix", "Default separator", "Custom separator"],
)
def test_resource_registration(
self, prefix, separator, expected_uri_key, expected_name, unexpected_uri_key
):
"""Test resource registration with prefix and separator variations."""
mcp = FastMCP()
class MyResourceMixin(MCPMixin):
@mcp_resource(uri="test://resource")
def sample_resource(self):
pass
instance = MyResourceMixin()
instance.register_resources(mcp, prefix=prefix, separator=separator)
registered_resources = mcp.get_resources()
assert expected_uri_key in registered_resources
assert registered_resources[expected_uri_key].name == expected_name
assert unexpected_uri_key not in registered_resources
@pytest.mark.parametrize(
"prefix, separator, expected_name, unexpected_name",
[
(
None,
_DEFAULT_SEPARATOR_PROMPT,
"sample_prompt",
f"None{_DEFAULT_SEPARATOR_PROMPT}sample_prompt",
),
(
"pref",
_DEFAULT_SEPARATOR_PROMPT,
f"pref{_DEFAULT_SEPARATOR_PROMPT}sample_prompt",
"sample_prompt",
),
(
"pref",
":",
"pref:sample_prompt",
f"pref{_DEFAULT_SEPARATOR_PROMPT}sample_prompt",
),
],
ids = ["No prefix", "Default separator", "Custom separator"]
)
def test_prompt_registration(
self, prefix, separator, expected_name, unexpected_name
):
"""Test prompt registration with prefix and separator variations."""
mcp = FastMCP()
class MyPromptMixin(MCPMixin):
@mcp_prompt()
def sample_prompt(self):
pass
instance = MyPromptMixin()
instance.register_prompts(mcp, prefix=prefix, separator=separator)
registered_prompt_names = {p.name for p in mcp.list_prompts()}
assert expected_name in registered_prompt_names
assert unexpected_name not in registered_prompt_names
def test_register_all_no_prefix(self):
"""Test register_all method registers all types without a prefix."""
mcp = FastMCP()
class MyFullMixin(MCPMixin):
@mcp_tool()
def tool_all(self):
pass
@mcp_resource(uri="res://all")
def resource_all(self):
pass
@mcp_prompt()
def prompt_all(self):
pass
instance = MyFullMixin()
instance.register_all(mcp)
assert "tool_all" in mcp.get_tools()
assert "res://all" in mcp.get_resources()
assert "prompt_all" in {p.name for p in mcp.list_prompts()}
def test_register_all_with_prefix_default_separators(self):
"""Test register_all method registers all types with a prefix and default separators."""
mcp = FastMCP()
class MyFullMixinPrefixed(MCPMixin):
@mcp_tool()
def tool_all_p(self):
pass
@mcp_resource(uri="res://all_p")
def resource_all_p(self):
pass
@mcp_prompt()
def prompt_all_p(self):
pass
instance = MyFullMixinPrefixed()
instance.register_all(mcp, prefix="all")
assert f"all{_DEFAULT_SEPARATOR_TOOL}tool_all_p" in mcp.get_tools()
assert f"all{_DEFAULT_SEPARATOR_RESOURCE}res://all_p" in mcp.get_resources()
assert f"all{_DEFAULT_SEPARATOR_PROMPT}prompt_all_p" in {
p.name for p in mcp.list_prompts()
}
def test_register_all_with_prefix_custom_separators(self):
"""Test register_all method registers all types with a prefix and custom separators."""
mcp = FastMCP()
class MyFullMixinCustomSep(MCPMixin):
@mcp_tool()
def tool_cust(self):
pass
@mcp_resource(uri="res://cust")
def resource_cust(self):
pass
@mcp_prompt()
def prompt_cust(self):
pass
instance = MyFullMixinCustomSep()
instance.register_all(
mcp,
prefix="cust",
tool_separator="-",
resource_separator="::",
prompt_separator=".",
)
assert "cust-tool_cust" in mcp.get_tools()
assert "cust::res://cust" in mcp.get_resources()
assert "cust.prompt_cust" in {p.name for p in mcp.list_prompts()}
# Check default separators weren't used
assert f"cust{_DEFAULT_SEPARATOR_TOOL}tool_cust" not in mcp.get_tools()
assert f"cust{_DEFAULT_SEPARATOR_RESOURCE}res://cust" not in mcp.get_resources()
assert f"cust{_DEFAULT_SEPARATOR_PROMPT}prompt_cust" not in {
p.name for p in mcp.list_prompts()
}