deps: Port to FastMCP 3.2 (#452)

Upgrade FastMCP to >= 3.2.4.

Changes:

 * Use built-in support for mcp-apps in FastMCP to simplify app
   and resource handling.
 * Remove awful workarounds to get access to to session from
   on_initialize() middleware.
 * Use mcp.get_tool() rather than mcp.get_tools()

There is a bug in FastMCP 3.2.4 where visibility=["app"] tools
are not listed or callable - monkeypatch FastMCP to temporarily
work around that. (Fix at PrefectHQ/fastmcp#4112)

Author: owtaylor

pyproject.toml

@@ -16,7 +16,7 @@ requires-python = ">=3.10"
 # no longer getting security fixes.
 dependencies = [
     "asyncssh[bcrypt] >= 2.22.0",
-    "fastmcp >=2.14.4, <2.14.6",
+    "fastmcp >= 3.2.4",
     "litellm>=1.80.16",
     "pydantic-settings >= 2.12.0",
     "pydantic >= 2.12.5",

src/linux_mcp_server/mcp_app.py

@@ -1,4 +1,3 @@
 RUN_SCRIPT_APP_URI = "ui://run_script_readonly_with_mcp_app/run-script-app.html"
-ALLOWED_UI_RESOURCE_URIS = set([RUN_SCRIPT_APP_URI])
 MCP_APP_MIME_TYPE = "text/html;profile=mcp-app"
 MCP_UI_EXTENSION = "io.modelcontextprotocol/ui"

src/linux_mcp_server/server.py

@@ -5,23 +5,17 @@
 
 from importlib import resources
 from pathlib import Path
-from types import CellType
-from types import CodeType
 
 from fastmcp import FastMCP
+from fastmcp.resources import ResourceContent
+from fastmcp.resources import ResourceResult
 from fastmcp.server.dependencies import get_access_token
 from fastmcp.server.middleware import Middleware
 from fastmcp.server.middleware import MiddlewareContext
 from fastmcp.server.middleware.middleware import CallNext
-from mcp import ServerSession
-from mcp.types import BlobResourceContents
 from mcp.types import InitializeRequest
 from mcp.types import InitializeRequestParams
 from mcp.types import InitializeResult
-from mcp.types import ReadResourceRequest
-from mcp.types import ReadResourceResult
-from mcp.types import ServerResult
-from mcp.types import TextResourceContents
 
 import linux_mcp_server
 
@@ -31,12 +25,26 @@
 from linux_mcp_server.config import CONFIG
 from linux_mcp_server.config import Toolset
 from linux_mcp_server.config import Transport
-from linux_mcp_server.mcp_app import ALLOWED_UI_RESOURCE_URIS
 from linux_mcp_server.mcp_app import MCP_APP_MIME_TYPE
 from linux_mcp_server.mcp_app import MCP_UI_EXTENSION
+from linux_mcp_server.mcp_app import RUN_SCRIPT_APP_URI
 from linux_mcp_server.toolset import get_toolset
 
 
+def monkeypatch_fastmcp_for_app_visibility():
+    # fastmcp 3.2.4 has a bug where tools defined with
+    # visibility=["app"] aren't returned by tools/list
+    # https://github.com/PrefectHQ/fastmcp/issues/4088
+    # https://github.com/PrefectHQ/fastmcp/pull/4112
+    import fastmcp.server.server as m
+
+    if hasattr(m, "_is_model_visible"):
+        m._is_model_visible = lambda _tool: True
+
+
+monkeypatch_fastmcp_for_app_visibility()
+
+
 logger = logging.getLogger("linux-mcp-server")
 
 INSTRUCTIONS_FIXED = """You have access to predefined commands that inspect the system. They run standard Linux utilities and return formatted results.
@@ -158,81 +166,55 @@
 toolset = get_toolset(CONFIG.toolset.value)
 assert toolset is not None, f"Toolset not found in registry: {CONFIG.toolset}"
 
-kwargs = {}
-if toolset.include_tags:
-    kwargs["include_tags"] = toolset.include_tags
-if toolset.exclude_tags:
-    kwargs["exclude_tags"] = toolset.exclude_tags
-
 if CONFIG.toolset != Toolset.FIXED and CONFIG.gatekeeper_model is None:
     logger.error("LINUX_MCP_GATEKEEPER_MODEL not set, this is needed for run_script tools")
     sys.exit(1)
 
 # Create auth provider if configured
 auth_provider = create_auth_provider()
 
-mcp = FastMCP(
-    "linux-mcp-server", instructions=instructions, version=linux_mcp_server.__version__, auth=auth_provider, **kwargs
-)
-
+mcp = FastMCP("linux-mcp-server", instructions=instructions, version=linux_mcp_server.__version__, auth=auth_provider)
 
-_low_level_server = mcp._mcp_server
-_original_resource_request_handler = _low_level_server.request_handlers[ReadResourceRequest]
-
-
-async def _read_resource_with_meta(req: ReadResourceRequest):
-    uri = str(req.params.uri)
-    fallback_contents: list[TextResourceContents | BlobResourceContents] = [
-        TextResourceContents(uri=req.params.uri, mimeType="text/plain", text="Resource not found")
-    ]
-
-    if uri.startswith("ui://"):
-        if uri in ALLOWED_UI_RESOURCE_URIS:
-            filename = uri.split("/")[-1]
-
-            # Try ui_resources first (wheel install)
-            ui_resources_path = resources.files(linux_mcp_server).joinpath("ui_resources")
-            resource_file = ui_resources_path.joinpath(filename)
-            logger.debug(f"Checking for UI resource at: {resource_file}")
-
-            # Check if we need to fall back to mcp-app/dist (editable install)
-            if not resource_file.is_file():
-                package_path = Path(linux_mcp_server.__file__).parent
-                repo_root = package_path.parent.parent
-                mcp_app_dist = repo_root / "mcp-app" / "dist" / filename
-                logger.debug(f"Checking for UI resource at: {mcp_app_dist}")
-
-                if mcp_app_dist.exists():
-                    resource_file = mcp_app_dist
-                else:
-                    logger.error(f"UI resource not found: {filename}")
-                    raise FileNotFoundError(f"Resource {filename} not found")
-
-            # Read the file
-            try:
-                html = resource_file.read_text()
-                logger.info(f"Serving UI resource from: {resource_file}")
-            except Exception as e:
-                logger.error(f"Failed to read UI resource from {resource_file}: {e}")
-                raise
-
-            content = TextResourceContents.model_validate(
-                {
-                    "uri": uri,
-                    "mimeType": MCP_APP_MIME_TYPE,
-                    "text": html,
-                }
-            )
+if toolset.include_tags:
+    mcp.enable(tags=toolset.include_tags, only=True)
+else:
+    mcp.enable()
 
-            return ServerResult(ReadResourceResult(contents=[content]))
-    else:
-        if _original_resource_request_handler:
-            return await _original_resource_request_handler(req)
+if toolset.exclude_tags:
+    mcp.disable(tags=toolset.exclude_tags)
 
-    return ServerResult(ReadResourceResult(contents=fallback_contents))
 
+@mcp.resource(
+    RUN_SCRIPT_APP_URI,
+    tags={"run_script"},
+)
+def run_script_app_html() -> ResourceResult:
+    filename = "run-script-app.html"
+
+    # Try ui_resources first (wheel install)
+    ui_resources_path = resources.files(linux_mcp_server).joinpath("ui_resources")
+    resource_file = ui_resources_path.joinpath(filename)
+    logger.debug(f"Checking for UI resource at: {resource_file}")
+    # Check if we need to fall back to mcp-app/dist (editable install)
+    if not resource_file.is_file():
+        package_path = Path(linux_mcp_server.__file__).parent
+        repo_root = package_path.parent.parent
+        mcp_app_dist = repo_root / "mcp-app" / "dist" / filename
+        logger.debug(f"Checking for UI resource at: {mcp_app_dist}")
+        if mcp_app_dist.exists():
+            resource_file = mcp_app_dist
+        else:
+            logger.error(f"UI resource not found: {filename}")
+            raise FileNotFoundError(f"Resource {filename} not found")
+    # Read the file
+    try:
+        html = resource_file.read_text()
+        logger.info(f"Serving UI resource from: {resource_file}")
+    except Exception as e:
+        logger.error(f"Failed to read UI resource from {resource_file}: {e}")
+        raise
 
-_low_level_server.request_handlers[ReadResourceRequest] = _read_resource_with_meta
+    return ResourceResult(contents=[ResourceContent(html, mime_type=MCP_APP_MIME_TYPE)])
 
 
 from linux_mcp_server.tools import *  # noqa: E402, F403
@@ -334,31 +316,15 @@ async def on_initialize(
         # away in the ServerSession object, so we need to modify that based
         # on whether we'll use mcp-apps with the client making the InitializeRequest.
 
+        assert context.fastmcp_context
+        session = context.fastmcp_context.session
+
         if _use_mcp_app_for_client(context.message.params):
-            # Getting the ServerSession object is easy for FastMCP 3.x - it's
-            # just context.fastcmp_context.session, but the property getter
-            # will raise RuntimeError for FastMCP 2.x, so we check _session instead.
-            assert context.fastmcp_context is not None, "fastmcp_context should be set in on_initialize"
-            session: ServerSession | None = getattr(context.fastmcp_context, "_session", None)
-            if session is None:
-                # FastMCP 2.x - let's pull out the hacks! call_next is a closure within a method
-                # of fastmcp.server.low_level.MiddlewareServerSession. The "self" variable used
-                # in the closure is what we need. Assuming CPython, we can dig and and get it!
-                code: CodeType | None = getattr(call_next, "__code__", None)
-                closure: tuple[CellType, ...] | None = getattr(call_next, "__closure__", None)
-                if code and closure:
-                    # co_freevars gives us the names of the variables captured in __closure__
-                    closure_dict = dict(zip(code.co_freevars, [c.cell_contents for c in closure]))
-                    session = closure_dict.get("self")
-
-            if session and isinstance(session, ServerSession):
-                instructions = session._init_options.instructions
-                if instructions:
-                    session._init_options.instructions = instructions.replace(
-                        "run_script_with_confirmation", "run_script_interactive"
-                    )
-            else:
-                logger.warning("Unable to get ServerSession to update instructions for mcp-apps")
+            instructions = session._init_options.instructions
+            if instructions:
+                session._init_options.instructions = instructions.replace(
+                    "run_script_with_confirmation", "run_script_interactive"
+                )
 
         return await call_next(context)
 

src/linux_mcp_server/tools/run_script.py

@@ -7,8 +7,9 @@
 from dataclasses import dataclass
 
 from fastmcp import Context
+from fastmcp.apps import AppConfig
 from fastmcp.exceptions import ToolError
-from fastmcp.tools.tool import ToolResult
+from fastmcp.tools import ToolResult
 from mcp.types import ContentBlock
 from mcp.types import TextContent
 from mcp.types import ToolAnnotations
@@ -234,7 +235,7 @@ class ExecuteScriptResult:
 @mcp.tool(
     tags={"run_script", "hidden_from_model"},
     description="Execute a script; this is only available to the our mcp-app",
-    meta={"ui": {"visibility": ["app"]}},
+    app=AppConfig(visibility=["app"]),
 )
 @log_tool_call
 @disallow_local_execution_in_containers
@@ -271,7 +272,7 @@ async def execute_script(
 @mcp.tool(
     tags={"run_script", "hidden_from_model"},
     description="Reject a script; this is only available to the our mcp-app",
-    meta={"ui": {"visibility": ["app"]}},
+    app=AppConfig(visibility=["app"]),
 )
 @log_tool_call
 @disallow_local_execution_in_containers
@@ -287,7 +288,7 @@ async def reject_script(
     description=RUN_SCRIPT_INTERACTIVE_DESCRIPTION,
     annotations=ToolAnnotations(destructiveHint=True),
     output_schema=RunScriptInteractiveResult.model_json_schema(),
-    meta={"ui": {"resourceUri": RUN_SCRIPT_APP_URI}},
+    app=AppConfig(resourceUri=RUN_SCRIPT_APP_URI),
 )
 @log_tool_call
 @disallow_local_execution_in_containers
@@ -359,7 +360,7 @@ async def run_script_interactive(
     tags={"run_script", "hidden_from_model"},
     title="Get the execution state with request ID",
     description="Get the execution state with request ID",
-    meta={"ui": {"visibility": ["app"]}},
+    app=AppConfig(visibility=["app"]),
 )
 @log_tool_call
 @disallow_local_execution_in_containers

tests/test_middleware.py

@@ -59,62 +59,24 @@ async def test_skips_modification_when_disabled(self, middleware, mock_context,
         mock_session = mocker.Mock(spec=ServerSession)
         mock_session._init_options = mocker.Mock()
         mock_session._init_options.instructions = "Use run_script_with_confirmation for changes"
-        mock_context.fastmcp_context._session = mock_session
+        mock_context.fastmcp_context.session = mock_session
 
         await middleware.on_initialize(mock_context, mocker.AsyncMock(return_value=mocker.Mock()))
 
         assert mock_session._init_options.instructions == "Use run_script_with_confirmation for changes"
 
-    async def test_modifies_instructions_fastmcp_3x(self, middleware, mock_context, mocker):
-        # Test instruction modification with FastMCP 3.x
+    async def test_modifies_instructions(self, middleware, mock_context, mocker):
         mocker.patch.object(server_module, "_use_mcp_app_for_client", return_value=True)
 
         mock_session = mocker.Mock(spec=ServerSession)
         mock_session._init_options = mocker.Mock()
         mock_session._init_options.instructions = "Use run_script_with_confirmation for changes"
-        mock_context.fastmcp_context._session = mock_session
+        mock_context.fastmcp_context.session = mock_session
 
         await middleware.on_initialize(mock_context, mocker.AsyncMock(return_value=mocker.Mock()))
 
         assert mock_session._init_options.instructions == "Use run_script_interactive for changes"
 
-    async def test_modifies_instructions_fastmcp_2x(self, middleware, mock_context, mocker):
-        # Test instruction modification with FastMCP 2.x via closure extraction
-        mocker.patch.object(server_module, "_use_mcp_app_for_client", return_value=True)
-        mock_context.fastmcp_context._session = None
-
-        mock_session = mocker.Mock(spec=ServerSession)
-        mock_session._init_options = mocker.Mock()
-        mock_session._init_options.instructions = "Use run_script_with_confirmation for changes"
-
-        def make_call_next(session_obj):
-            self = session_obj
-
-            async def call_next_func(_ctx):
-                _ = self
-                return mocker.Mock()
-
-            return call_next_func
-
-        await middleware.on_initialize(mock_context, make_call_next(mock_session))
-
-        assert mock_session._init_options.instructions == "Use run_script_interactive for changes"
-
-    async def test_handles_extraction_failure(self, middleware, mock_context, mocker):
-        # Test graceful handling when session extraction fails
-        mocker.patch.object(server_module, "_use_mcp_app_for_client", return_value=True)
-        mock_logger = mocker.patch.object(server_module, "logger")
-
-        mock_context.fastmcp_context._session = None
-        call_next = mocker.AsyncMock(return_value=mocker.Mock())
-        call_next.__code__ = mocker.Mock()
-        call_next.__code__.co_freevars = []
-        call_next.__closure__ = None
-
-        await middleware.on_initialize(mock_context, call_next)
-
-        mock_logger.warning.assert_called_once()
-
 
 class TestDynamicDiscoveryMiddlewareOnListTools:
     @pytest.fixture

tests/test_tool_schemas.py

@@ -5,25 +5,6 @@
 from linux_mcp_server.server import mcp
 
 
-@pytest.fixture(scope="module")
-async def mcp_tools() -> dict:
-    """Fetch all MCP tools using the public API."""
-    return await mcp.get_tools()
-
-
-@pytest.fixture
-def tool_properties(mcp_tools: dict):
-    """Get the properties dict for a tool's parameters schema."""
-
-    def _tool_properties(tool_name: str) -> dict:
-        tool = mcp_tools.get(tool_name)
-        if tool is None:
-            raise ValueError(f"Tool '{tool_name}' not found")  # pragma: no cover
-        return tool.parameters.get("properties", {})
-
-    return _tool_properties
-
-
 class TestToolSchemaExamples:
     """Verify parameters have examples for LLM guidance."""
 
@@ -42,8 +23,10 @@ class TestToolSchemaExamples:
             ("read_file", "path"),
         ],
     )
-    def test_parameter_has_examples(self, tool_name: str, param_name: str, tool_properties) -> None:
-        props = tool_properties(tool_name)
+    async def test_parameter_has_examples(self, tool_name: str, param_name: str) -> None:
+        tool = await mcp.get_tool(tool_name)
+        assert tool
+        props = tool.parameters.get("properties", {})
 
         assert param_name in props, f"Parameter '{param_name}' not found in {tool_name}"
         assert "examples" in props[param_name], f"Parameter '{param_name}' in {tool_name} missing examples"