Implemented runtime tools

README.md

@@ -28,6 +28,7 @@
     - [Resources](#resources)
     - [Tools](#tools)
       - [Structured Output](#structured-output)
+      - [Runtime Tools](#runtime-tools)
     - [Prompts](#prompts)
     - [Images](#images)
     - [Context](#context)
@@ -382,6 +383,44 @@ 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
+from mcp.server.fastmcp.server import Context
+
+
+async def runtime_mcp_tools_generator(ctx: Context) -> 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}
+
+    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(
+    name="Weather Service", runtime_mcp_tools_generator=runtime_mcp_tools_generator
+)
+```
+
 ### Prompts
 
 Prompts are reusable templates that help LLMs interact with your server effectively:

src/mcp/server/fastmcp/server.py

@@ -140,6 +140,8 @@ def __init__(
         event_store: EventStore | None = None,
         *,
         tools: list[Tool] | None = None,
+        runtime_mcp_tools_generator: Callable[[Context[ServerSession, object, Request]], Awaitable[list[Tool]]]
+        | None = None,
         **settings: Any,
     ):
         self.settings = Settings(**settings)
@@ -172,6 +174,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 +248,18 @@ 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:
+            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}):
+            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 +286,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(context)
+            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]:

tests/server/fastmcp/test_runtime_tools.py

@@ -0,0 +1,91 @@
+"""Integration tests for runtime tools functionality."""
+
+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
+
+
+@pytest.mark.anyio
+async def test_runtime_tools():
+    """Test that runtime tools work correctly."""
+
+    async def runtime_mcp_tools_generator(ctx: Context) -> list[Tool]:
+        """Generate runtime tools."""
+
+        def runtime_tool_1(message: str):
+            return message
+
+        def runtime_tool_2(message: str):
+            return message
+
+        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)
+
+    # Static tool
+    @mcp.tool(description="Static tool")
+    def static_tool(message: str) -> str:
+        return message
+
+    # Start server and connect client without authorization
+    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"
+
+        # 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"