Merge pull request #336 from jlowin/invalid-resource

Prevent invalid resource URIs

src/fastmcp/server/server.py

@@ -14,6 +14,7 @@ from typing import TYPE_CHECKING, Any, Generic, Literal
 
 import anyio
 import httpx
+import pydantic
 import uvicorn
 from mcp.server.auth.middleware.auth_context import AuthContextMiddleware
 from mcp.server.auth.middleware.bearer_auth import (
@@ -39,7 +40,7 @@ from mcp.types import Prompt as MCPPrompt
 from mcp.types import Resource as MCPResource
 from mcp.types import ResourceTemplate as MCPResourceTemplate
 from mcp.types import Tool as MCPTool
-from pydantic.networks import AnyUrl
+from pydantic import AnyUrl
 from starlette.applications import Starlette
 from starlette.middleware import Middleware
 from starlette.middleware.authentication import AuthenticationMiddleware
@@ -88,6 +89,8 @@ class MountedServer:
         if prompt_separator is None:
             prompt_separator = "_"
 
+        _validate_resource_prefix(f"{prefix}{resource_separator}")
+
         self.server = server
         self.prefix = prefix
         self.tool_separator = tool_separator
@@ -1074,6 +1077,7 @@ class FastMCP(Generic[LifespanResultT]):
 
         # Import resources and templates from the mounted server
         resource_prefix = f"{prefix}{resource_separator}"
+        _validate_resource_prefix(resource_prefix)
         for key, resource in (await server.get_resources()).items():
             self._resource_manager.add_resource(resource, key=f"{resource_prefix}{key}")
         for key, template in (await server.get_resource_templates()).items():
@@ -1131,3 +1135,13 @@ class FastMCP(Generic[LifespanResultT]):
         from fastmcp.server.proxy import FastMCPProxy
 
         return FastMCPProxy(client=client, **settings)
+
+
+def _validate_resource_prefix(prefix: str) -> None:
+    valid_resource = "resource://path/to/resource"
+    try:
+        AnyUrl(f"{prefix}{valid_resource}")
+    except pydantic.ValidationError as e:
+        raise ValueError(
+            f"Resource prefix or separator would result in an invalid resource URI: {e}"
+        )

tests/server/test_import_server.py

@@ -1,6 +1,7 @@
 import json
 from urllib.parse import quote
 
+import pytest
 from mcp.types import TextContent, TextResourceContents
 
 from fastmcp.client.client import Client
@@ -391,3 +392,25 @@ async def test_import_with_proxy_resource_templates():
         user_data = json.loads(result[0].text)
         assert user_data["name"] == "John Doe"
         assert user_data["email"] == "john@example.com"
+
+
+async def test_import_invalid_resource_prefix():
+    main_app = FastMCP("MainApp")
+    api_app = FastMCP("APIApp")
+
+    with pytest.raises(
+        ValueError,
+        match="Resource prefix or separator would result in an invalid resource URI",
+    ):
+        await main_app.import_server("api_sub", api_app)
+
+
+async def test_import_invalid_resource_separator():
+    main_app = FastMCP("MainApp")
+    api_app = FastMCP("APIApp")
+
+    with pytest.raises(
+        ValueError,
+        match="Resource prefix or separator would result in an invalid resource URI",
+    ):
+        await main_app.import_server("api", api_app, resource_separator="_")

tests/server/test_mount.py

@@ -59,6 +59,26 @@ class TestBasicMount:
         assert isinstance(result[0], TextContent)
         assert result[0].text == "Hello, World!"
 
+    async def test_mount_invalid_resource_prefix(self):
+        main_app = FastMCP("MainApp")
+        api_app = FastMCP("APIApp")
+
+        with pytest.raises(
+            ValueError,
+            match="Resource prefix or separator would result in an invalid resource URI",
+        ):
+            main_app.mount("api_sub", api_app)
+
+    async def test_mount_invalid_resource_separator(self):
+        main_app = FastMCP("MainApp")
+        api_app = FastMCP("APIApp")
+
+        with pytest.raises(
+            ValueError,
+            match="Resource prefix or separator would result in an invalid resource URI",
+        ):
+            main_app.mount("api", api_app, resource_separator="_")
+
     async def test_unmount_server(self):
         """Test unmounting a server removes access to its tools."""
         main_app = FastMCP("MainApp")