From ae0ec1690afa1d2711068ea35ce8458231373401 Mon Sep 17 00:00:00 2001 From: Nikita Bellini Date: Fri, 4 Jul 2025 17:09:24 +0200 Subject: [PATCH 1/5] Implemented runtime tools --- README.md | 30 +++++++++ src/mcp/server/fastmcp/server.py | 19 ++++++ tests/server/fastmcp/test_runtime_tools.py | 73 ++++++++++++++++++++++ 3 files changed, 122 insertions(+) create mode 100644 tests/server/fastmcp/test_runtime_tools.py diff --git a/README.md b/README.md index f2b001231..cae543812 100644 --- a/README.md +++ b/README.md @@ -371,6 +371,36 @@ def get_temperature(city: str) -> float: # Returns: {"result": 22.5} ``` +#### Runtime tools + +It is also possible to define tools at runtime, allowing for dynamic modification of the available tools, for example, to display specific tools based on the user making the request. This is done passing a function dedicated to the tools generation: + +```python +from mcp.server.fastmcp import FastMCP +from mcp.server.fastmcp.tools.base import Tool + + +async def runtime_mcp_tools_generator() -> list[Tool]: + """Generate runtime tools.""" + + def list_cities() -> list[str]: + """Get a list of cities""" + return ["London", "Paris", "Tokyo"] + # Returns: {"result": ["London", "Paris", "Tokyo"]} + + def get_temperature(city: str) -> float: + """Get temperature as a simple float""" + return 22.5 + # Returns: {"result": 22.5} + + return [Tool.from_function(list_cities), Tool.from_function(get_temperature)] + + +mcp = FastMCP( + name="Weather Service", runtime_mcp_tools_generator=runtime_mcp_tools_generator +) +``` + ### Prompts Prompts are reusable templates that help LLMs interact with your server effectively: diff --git a/src/mcp/server/fastmcp/server.py b/src/mcp/server/fastmcp/server.py index 705d9f494..16a1b8307 100644 --- a/src/mcp/server/fastmcp/server.py +++ b/src/mcp/server/fastmcp/server.py @@ -140,6 +140,7 @@ def __init__( event_store: EventStore | None = None, *, tools: list[Tool] | None = None, + runtime_mcp_tools_generator: Callable[[], Awaitable[list[Tool]]] | None = None, **settings: Any, ): self.settings = Settings(**settings) @@ -172,6 +173,7 @@ def __init__( self._custom_starlette_routes: list[Route] = [] self.dependencies = self.settings.dependencies self._session_manager: StreamableHTTPSessionManager | None = None + self._runtime_mcp_tools_generator = runtime_mcp_tools_generator # Set up MCP protocol handlers self._setup_handlers() @@ -245,6 +247,14 @@ def _setup_handlers(self) -> None: async def list_tools(self) -> list[MCPTool]: """List all available tools.""" tools = self._tool_manager.list_tools() + + if self._runtime_mcp_tools_generator: + tools.extend(await self._runtime_mcp_tools_generator()) + + # Check if there are no duplicated tools + if len(tools) != len(set([tool.name for tool in tools])): + raise Exception("There are duplicated tools. Check the for tools with the same name both static and generated at runtime.") + return [ MCPTool( name=info.name, @@ -271,6 +281,15 @@ def get_context(self) -> Context[ServerSession, object, Request]: async def call_tool(self, name: str, arguments: dict[str, Any]) -> Sequence[ContentBlock] | dict[str, Any]: """Call a tool by name with arguments.""" context = self.get_context() + + # Try to call a runtime tool + if self._runtime_mcp_tools_generator: + runtime_tools = await self._runtime_mcp_tools_generator() + for tool in runtime_tools: + if tool.name == name: + return await tool.run(arguments=arguments, context=context, convert_result=True) + + # Call a static tool if the runtime tool has not been called return await self._tool_manager.call_tool(name, arguments, context=context, convert_result=True) async def list_resources(self) -> list[MCPResource]: diff --git a/tests/server/fastmcp/test_runtime_tools.py b/tests/server/fastmcp/test_runtime_tools.py new file mode 100644 index 000000000..d6b9deba6 --- /dev/null +++ b/tests/server/fastmcp/test_runtime_tools.py @@ -0,0 +1,73 @@ +"""Integration tests for runtime tools functionality.""" + +import pytest + +from mcp.server.fastmcp import FastMCP +from mcp.shared.memory import create_connected_server_and_client_session +from mcp.server.fastmcp.tools.base import Tool +from mcp.types import TextContent + +@pytest.mark.anyio +async def test_runtime_tools(): + """Test that runtime tools work correctly.""" + async def runtime_mcp_tools_generator() -> list[Tool]: + """Generate runtime tools.""" + def runtime_tool_1(message: str): + return message + + def runtime_tool_2(message: str): + return message + + return [ + Tool.from_function(runtime_tool_1), + Tool.from_function(runtime_tool_2) + ] + + # Create server with various tool configurations, both static and runtime + mcp = FastMCP(name="RuntimeToolsTestServer", runtime_mcp_tools_generator=runtime_mcp_tools_generator) + + # Static tool + @mcp.tool(description="Static tool") + def static_tool(message: str) -> str: + return message + + # Start server and connect client + async with create_connected_server_and_client_session(mcp._mcp_server) as client: + await client.initialize() + + # List tools + tools_result = await client.list_tools() + tool_names = {tool.name: tool for tool in tools_result.tools} + + # Verify both tools + assert "static_tool" in tool_names + assert "runtime_tool_1" in tool_names + assert "runtime_tool_2" in tool_names + + # Check static tool + result = await client.call_tool("static_tool", {"message": "This is a test"}) + assert len(result.content) == 1 + content = result.content[0] + assert isinstance(content, TextContent) + assert content.text == "This is a test" + + # Check runtime tool 1 + result = await client.call_tool("runtime_tool_1", {"message": "This is a test"}) + assert len(result.content) == 1 + content = result.content[0] + assert isinstance(content, TextContent) + assert content.text == "This is a test" + + # Check runtime tool 2 + result = await client.call_tool("runtime_tool_2", {"message": "This is a test"}) + assert len(result.content) == 1 + content = result.content[0] + assert isinstance(content, TextContent) + assert content.text == "This is a test" + + # Check non existing tool + result = await client.call_tool("non_existing_tool", {"message": "This is a test"}) + assert len(result.content) == 1 + content = result.content[0] + assert isinstance(content, TextContent) + assert content.text == "Unknown tool: non_existing_tool" \ No newline at end of file From 8fec9d23ed039b8b8bfc329af132d6fee9de4345 Mon Sep 17 00:00:00 2001 From: Nikita Bellini Date: Fri, 4 Jul 2025 18:13:11 +0200 Subject: [PATCH 2/5] Format --- src/mcp/server/fastmcp/server.py | 4 +++- tests/server/fastmcp/test_runtime_tools.py | 12 ++++++------ 2 files changed, 9 insertions(+), 7 deletions(-) diff --git a/src/mcp/server/fastmcp/server.py b/src/mcp/server/fastmcp/server.py index 16a1b8307..88fec8907 100644 --- a/src/mcp/server/fastmcp/server.py +++ b/src/mcp/server/fastmcp/server.py @@ -253,7 +253,9 @@ async def list_tools(self) -> list[MCPTool]: # Check if there are no duplicated tools if len(tools) != len(set([tool.name for tool in tools])): - raise Exception("There are duplicated tools. Check the for tools with the same name both static and generated at runtime.") + raise Exception( + "There are duplicated tools. Check the for tools with the same name both static and generated at runtime." + ) return [ MCPTool( diff --git a/tests/server/fastmcp/test_runtime_tools.py b/tests/server/fastmcp/test_runtime_tools.py index d6b9deba6..898bb97c2 100644 --- a/tests/server/fastmcp/test_runtime_tools.py +++ b/tests/server/fastmcp/test_runtime_tools.py @@ -7,21 +7,21 @@ from mcp.server.fastmcp.tools.base import Tool from mcp.types import TextContent + @pytest.mark.anyio async def test_runtime_tools(): """Test that runtime tools work correctly.""" + async def runtime_mcp_tools_generator() -> list[Tool]: """Generate runtime tools.""" + def runtime_tool_1(message: str): return message - + def runtime_tool_2(message: str): return message - return [ - Tool.from_function(runtime_tool_1), - Tool.from_function(runtime_tool_2) - ] + return [Tool.from_function(runtime_tool_1), Tool.from_function(runtime_tool_2)] # Create server with various tool configurations, both static and runtime mcp = FastMCP(name="RuntimeToolsTestServer", runtime_mcp_tools_generator=runtime_mcp_tools_generator) @@ -70,4 +70,4 @@ def static_tool(message: str) -> str: assert len(result.content) == 1 content = result.content[0] assert isinstance(content, TextContent) - assert content.text == "Unknown tool: non_existing_tool" \ No newline at end of file + assert content.text == "Unknown tool: non_existing_tool" From d028116ed6d3e382190be4dbf12765590cacde36 Mon Sep 17 00:00:00 2001 From: Nikita Bellini Date: Fri, 4 Jul 2025 18:22:31 +0200 Subject: [PATCH 3/5] Format --- src/mcp/server/fastmcp/server.py | 5 +++-- tests/server/fastmcp/test_runtime_tools.py | 2 +- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/src/mcp/server/fastmcp/server.py b/src/mcp/server/fastmcp/server.py index 88fec8907..2e6b8e0a9 100644 --- a/src/mcp/server/fastmcp/server.py +++ b/src/mcp/server/fastmcp/server.py @@ -252,9 +252,10 @@ async def list_tools(self) -> list[MCPTool]: tools.extend(await self._runtime_mcp_tools_generator()) # Check if there are no duplicated tools - if len(tools) != len(set([tool.name for tool in tools])): + if len(tools) != len({tool.name for tool in tools}): raise Exception( - "There are duplicated tools. Check the for tools with the same name both static and generated at runtime." + "There are duplicated tools. Check the for tools" + "with the same name both static and generated at runtime." ) return [ diff --git a/tests/server/fastmcp/test_runtime_tools.py b/tests/server/fastmcp/test_runtime_tools.py index 898bb97c2..3b28c2253 100644 --- a/tests/server/fastmcp/test_runtime_tools.py +++ b/tests/server/fastmcp/test_runtime_tools.py @@ -3,8 +3,8 @@ import pytest from mcp.server.fastmcp import FastMCP -from mcp.shared.memory import create_connected_server_and_client_session from mcp.server.fastmcp.tools.base import Tool +from mcp.shared.memory import create_connected_server_and_client_session from mcp.types import TextContent From b5e4fccad040c8bc7f35d21c1c100ff52c1b630f Mon Sep 17 00:00:00 2001 From: Nikita Bellini Date: Fri, 4 Jul 2025 19:58:59 +0200 Subject: [PATCH 4/5] Fixed documentation --- README.md | 1 + 1 file changed, 1 insertion(+) diff --git a/README.md b/README.md index 6ed36abd1..bcf6e6131 100644 --- a/README.md +++ b/README.md @@ -28,6 +28,7 @@ - [Resources](#resources) - [Tools](#tools) - [Structured Output](#structured-output) + - [Runtime Tools](#runtime-tools) - [Prompts](#prompts) - [Images](#images) - [Context](#context) From e9f76500a10230f803d5e9618c554dbd0b3cb363 Mon Sep 17 00:00:00 2001 From: Nikita Bellini Date: Mon, 7 Jul 2025 13:40:44 +0200 Subject: [PATCH 5/5] Passed context to runtime tools generator --- README.md | 12 +++++++++-- src/mcp/server/fastmcp/server.py | 8 +++++--- tests/server/fastmcp/test_runtime_tools.py | 24 +++++++++++++++++++--- 3 files changed, 36 insertions(+), 8 deletions(-) diff --git a/README.md b/README.md index bcf6e6131..a861c03ef 100644 --- a/README.md +++ b/README.md @@ -390,9 +390,10 @@ It is also possible to define tools at runtime, allowing for dynamic modificatio ```python from mcp.server.fastmcp import FastMCP from mcp.server.fastmcp.tools.base import Tool +from mcp.server.fastmcp.server import Context -async def runtime_mcp_tools_generator() -> list[Tool]: +async def runtime_mcp_tools_generator(ctx: Context) -> list[Tool]: """Generate runtime tools.""" def list_cities() -> list[str]: @@ -405,7 +406,14 @@ async def runtime_mcp_tools_generator() -> list[Tool]: return 22.5 # Returns: {"result": 22.5} - return [Tool.from_function(list_cities), Tool.from_function(get_temperature)] + tools = [Tool.from_function(list_cities)] + + # Tool added only after authorization + request = ctx.request_context.request + if request and request.header.get("Authorization") == "Bearer auth_token_123": + tools.append(Tool.from_function(get_temperature)) + + return tools mcp = FastMCP( diff --git a/src/mcp/server/fastmcp/server.py b/src/mcp/server/fastmcp/server.py index 2e6b8e0a9..853bb7028 100644 --- a/src/mcp/server/fastmcp/server.py +++ b/src/mcp/server/fastmcp/server.py @@ -140,7 +140,8 @@ def __init__( event_store: EventStore | None = None, *, tools: list[Tool] | None = None, - runtime_mcp_tools_generator: Callable[[], Awaitable[list[Tool]]] | None = None, + runtime_mcp_tools_generator: Callable[[Context[ServerSession, object, Request]], Awaitable[list[Tool]]] + | None = None, **settings: Any, ): self.settings = Settings(**settings) @@ -249,7 +250,8 @@ async def list_tools(self) -> list[MCPTool]: tools = self._tool_manager.list_tools() if self._runtime_mcp_tools_generator: - tools.extend(await self._runtime_mcp_tools_generator()) + context = self.get_context() + tools.extend(await self._runtime_mcp_tools_generator(context)) # Check if there are no duplicated tools if len(tools) != len({tool.name for tool in tools}): @@ -287,7 +289,7 @@ async def call_tool(self, name: str, arguments: dict[str, Any]) -> Sequence[Cont # Try to call a runtime tool if self._runtime_mcp_tools_generator: - runtime_tools = await self._runtime_mcp_tools_generator() + runtime_tools = await self._runtime_mcp_tools_generator(context) for tool in runtime_tools: if tool.name == name: return await tool.run(arguments=arguments, context=context, convert_result=True) diff --git a/tests/server/fastmcp/test_runtime_tools.py b/tests/server/fastmcp/test_runtime_tools.py index 3b28c2253..e71f5fe2c 100644 --- a/tests/server/fastmcp/test_runtime_tools.py +++ b/tests/server/fastmcp/test_runtime_tools.py @@ -3,6 +3,7 @@ import pytest from mcp.server.fastmcp import FastMCP +from mcp.server.fastmcp.server import Context from mcp.server.fastmcp.tools.base import Tool from mcp.shared.memory import create_connected_server_and_client_session from mcp.types import TextContent @@ -12,7 +13,7 @@ async def test_runtime_tools(): """Test that runtime tools work correctly.""" - async def runtime_mcp_tools_generator() -> list[Tool]: + async def runtime_mcp_tools_generator(ctx: Context) -> list[Tool]: """Generate runtime tools.""" def runtime_tool_1(message: str): @@ -21,7 +22,17 @@ def runtime_tool_1(message: str): def runtime_tool_2(message: str): return message - return [Tool.from_function(runtime_tool_1), Tool.from_function(runtime_tool_2)] + def runtime_tool_3(message: str): + return message + + tools = [Tool.from_function(runtime_tool_1), Tool.from_function(runtime_tool_2)] + + # Tool added only after authorization + request = ctx.request_context.request + if request and request.header.get("Authorization") == "Bearer test_auth": + tools.append(Tool.from_function(runtime_tool_3)) + + return tools # Create server with various tool configurations, both static and runtime mcp = FastMCP(name="RuntimeToolsTestServer", runtime_mcp_tools_generator=runtime_mcp_tools_generator) @@ -31,7 +42,7 @@ def runtime_tool_2(message: str): def static_tool(message: str) -> str: return message - # Start server and connect client + # Start server and connect client without authorization async with create_connected_server_and_client_session(mcp._mcp_server) as client: await client.initialize() @@ -71,3 +82,10 @@ def static_tool(message: str) -> str: content = result.content[0] assert isinstance(content, TextContent) assert content.text == "Unknown tool: non_existing_tool" + + # Check not authorized tool + result = await client.call_tool("runtime_tool_3", {"message": "This is a test"}) + assert len(result.content) == 1 + content = result.content[0] + assert isinstance(content, TextContent) + assert content.text == "Unknown tool: runtime_tool_3"