Passed context to runtime tools generator

Author: Nikita Bellini

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(

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)

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"