Merge branch 'main' into ihrpr/shttp

Author: ihrpr

README.md

@@ -334,7 +334,7 @@ mcp = FastMCP("My App",
 )
 ```
 
-See [OAuthServerProvider](mcp/server/auth/provider.py) for more details.
+See [OAuthServerProvider](src/mcp/server/auth/provider.py) for more details.
 
 ## Running Your Server
 
@@ -410,6 +410,43 @@ app = Starlette(
 app.router.routes.append(Host('mcp.acme.corp', app=mcp.sse_app()))
 ```
 
+When mounting multiple MCP servers under different paths, you can configure the mount path in several ways:
+
+```python
+from starlette.applications import Starlette
+from starlette.routing import Mount
+from mcp.server.fastmcp import FastMCP
+
+# Create multiple MCP servers
+github_mcp = FastMCP("GitHub API")
+browser_mcp = FastMCP("Browser")
+curl_mcp = FastMCP("Curl")
+search_mcp = FastMCP("Search")
+
+# Method 1: Configure mount paths via settings (recommended for persistent configuration)
+github_mcp.settings.mount_path = "/github"
+browser_mcp.settings.mount_path = "/browser"
+
+# Method 2: Pass mount path directly to sse_app (preferred for ad-hoc mounting)
+# This approach doesn't modify the server's settings permanently
+
+# Create Starlette app with multiple mounted servers
+app = Starlette(
+    routes=[
+        # Using settings-based configuration
+        Mount("/github", app=github_mcp.sse_app()),
+        Mount("/browser", app=browser_mcp.sse_app()),
+        # Using direct mount path parameter
+        Mount("/curl", app=curl_mcp.sse_app("/curl")),
+        Mount("/search", app=search_mcp.sse_app("/search")),
+    ]
+)
+
+# Method 3: For direct execution, you can also pass the mount path to run()
+if __name__ == "__main__":
+    search_mcp.run(transport="sse", mount_path="/search")
+```
+
 For more information on mounting applications in Starlette, see the [Starlette documentation](https://www.starlette.io/routing/#submounting-routes).
 
 ## Examples

src/mcp/server/auth/middleware/bearer_auth.py

@@ -34,8 +34,15 @@ def __init__(
         self.provider = provider
 
     async def authenticate(self, conn: HTTPConnection):
-        auth_header = conn.headers.get("Authorization")
-        if not auth_header or not auth_header.startswith("Bearer "):
+        auth_header = next(
+            (
+                conn.headers.get(key)
+                for key in conn.headers
+                if key.lower() == "authorization"
+            ),
+            None,
+        )
+        if not auth_header or not auth_header.lower().startswith("bearer "):
             return None
 
         token = auth_header[7:]  # Remove "Bearer " prefix

src/mcp/server/fastmcp/server.py

@@ -89,6 +89,7 @@ class Settings(BaseSettings, Generic[LifespanResultT]):
     # HTTP settings
     host: str = "0.0.0.0"
     port: int = 8000
+    mount_path: str = "/"  # Mount path (e.g. "/github", defaults to root path)
     sse_path: str = "/sse"
     message_path: str = "/messages/"
     streamable_http_path: str = "/mcp"
@@ -190,12 +191,15 @@ def instructions(self) -> str | None:
         return self._mcp_server.instructions
 
     def run(
-        self, transport: Literal["stdio", "sse", "streamable-http"] = "stdio"
+        self,
+        transport: Literal["stdio", "sse", "streamable-http"] = "stdio",
+        mount_path: str | None = None,
     ) -> None:
         """Run the FastMCP server. Note this is a synchronous function.
 
         Args:
             transport: Transport protocol to use ("stdio", "sse", or "streamable-http")
+            mount_path: Optional mount path for SSE transport
         """
         TRANSPORTS = Literal["stdio", "sse", "streamable-http"]
         if transport not in TRANSPORTS.__args__:  # type: ignore
@@ -205,7 +209,7 @@ def run(
             case "stdio":
                 anyio.run(self.run_stdio_async)
             case "sse":
-                anyio.run(self.run_sse_async)
+                anyio.run(lambda: self.run_sse_async(mount_path))
             case "streamable-http":
                 anyio.run(self.run_streamable_http_async)
 
@@ -568,11 +572,11 @@ async def run_stdio_async(self) -> None:
                 self._mcp_server.create_initialization_options(),
             )
 
-    async def run_sse_async(self) -> None:
+    async def run_sse_async(self, mount_path: str | None = None) -> None:
         """Run the server using SSE transport."""
         import uvicorn
 
-        starlette_app = self.sse_app()
+        starlette_app = self.sse_app(mount_path)
 
         config = uvicorn.Config(
             starlette_app,
@@ -598,14 +602,51 @@ async def run_streamable_http_async(self) -> None:
         server = uvicorn.Server(config)
         await server.serve()
 
-    def sse_app(self) -> Starlette:
+    def _normalize_path(self, mount_path: str, endpoint: str) -> str:
+        """
+        Combine mount path and endpoint to return a normalized path.
+
+        Args:
+            mount_path: The mount path (e.g. "/github" or "/")
+            endpoint: The endpoint path (e.g. "/messages/")
+
+        Returns:
+            Normalized path (e.g. "/github/messages/")
+        """
+        # Special case: root path
+        if mount_path == "/":
+            return endpoint
+
+        # Remove trailing slash from mount path
+        if mount_path.endswith("/"):
+            mount_path = mount_path[:-1]
+
+        # Ensure endpoint starts with slash
+        if not endpoint.startswith("/"):
+            endpoint = "/" + endpoint
+
+        # Combine paths
+        return mount_path + endpoint
+
+    def sse_app(self, mount_path: str | None = None) -> Starlette:
         """Return an instance of the SSE server app."""
         from starlette.middleware import Middleware
         from starlette.routing import Mount, Route
 
+        # Update mount_path in settings if provided
+        if mount_path is not None:
+            self.settings.mount_path = mount_path
+
+        # Create normalized endpoint considering the mount path
+        normalized_message_endpoint = self._normalize_path(
+            self.settings.mount_path, self.settings.message_path
+        )
+
         # Set up auth context and dependencies
 
-        sse = SseServerTransport(self.settings.message_path)
+        sse = SseServerTransport(
+            normalized_message_endpoint,
+        )
 
         async def handle_sse(scope: Scope, receive: Receive, send: Send):
             # Add client ID from auth context into request context if available

tests/server/auth/middleware/test_bearer_auth.py

@@ -7,6 +7,7 @@
 
 import pytest
 from starlette.authentication import AuthCredentials
+from starlette.datastructures import Headers
 from starlette.exceptions import HTTPException
 from starlette.requests import Request
 from starlette.types import Message, Receive, Scope, Send
@@ -221,6 +222,66 @@ async def test_token_without_expiry(
         assert user.access_token == no_expiry_access_token
         assert user.scopes == ["read", "write"]
 
+    async def test_lowercase_bearer_prefix(
+        self,
+        mock_oauth_provider: OAuthAuthorizationServerProvider[Any, Any, Any],
+        valid_access_token: AccessToken,
+    ):
+        """Test with lowercase 'bearer' prefix in Authorization header"""
+        backend = BearerAuthBackend(provider=mock_oauth_provider)
+        add_token_to_provider(mock_oauth_provider, "valid_token", valid_access_token)
+        headers = Headers({"Authorization": "bearer valid_token"})
+        scope = {"type": "http", "headers": headers.raw}
+        request = Request(scope)
+        result = await backend.authenticate(request)
+        assert result is not None
+        credentials, user = result
+        assert isinstance(credentials, AuthCredentials)
+        assert isinstance(user, AuthenticatedUser)
+        assert credentials.scopes == ["read", "write"]
+        assert user.display_name == "test_client"
+        assert user.access_token == valid_access_token
+
+    async def test_mixed_case_bearer_prefix(
+        self,
+        mock_oauth_provider: OAuthAuthorizationServerProvider[Any, Any, Any],
+        valid_access_token: AccessToken,
+    ):
+        """Test with mixed 'BeArEr' prefix in Authorization header"""
+        backend = BearerAuthBackend(provider=mock_oauth_provider)
+        add_token_to_provider(mock_oauth_provider, "valid_token", valid_access_token)
+        headers = Headers({"authorization": "BeArEr valid_token"})
+        scope = {"type": "http", "headers": headers.raw}
+        request = Request(scope)
+        result = await backend.authenticate(request)
+        assert result is not None
+        credentials, user = result
+        assert isinstance(credentials, AuthCredentials)
+        assert isinstance(user, AuthenticatedUser)
+        assert credentials.scopes == ["read", "write"]
+        assert user.display_name == "test_client"
+        assert user.access_token == valid_access_token
+
+    async def test_mixed_case_authorization_header(
+        self,
+        mock_oauth_provider: OAuthAuthorizationServerProvider[Any, Any, Any],
+        valid_access_token: AccessToken,
+    ):
+        """Test authentication with mixed 'Authorization' header."""
+        backend = BearerAuthBackend(provider=mock_oauth_provider)
+        add_token_to_provider(mock_oauth_provider, "valid_token", valid_access_token)
+        headers = Headers({"AuThOrIzAtIoN": "BeArEr valid_token"})
+        scope = {"type": "http", "headers": headers.raw}
+        request = Request(scope)
+        result = await backend.authenticate(request)
+        assert result is not None
+        credentials, user = result
+        assert isinstance(credentials, AuthCredentials)
+        assert isinstance(user, AuthenticatedUser)
+        assert credentials.scopes == ["read", "write"]
+        assert user.display_name == "test_client"
+        assert user.access_token == valid_access_token
+
 
 @pytest.mark.anyio
 class TestRequireAuthMiddleware:

tests/server/fastmcp/test_server.py

@@ -1,9 +1,11 @@
 import base64
 from pathlib import Path
 from typing import TYPE_CHECKING
+from unittest.mock import patch
 
 import pytest
 from pydantic import AnyUrl
+from starlette.routing import Mount, Route
 
 from mcp.server.fastmcp import Context, FastMCP
 from mcp.server.fastmcp.prompts.base import EmbeddedResource, Message, UserMessage
@@ -31,6 +33,97 @@ async def test_create_server(self):
         assert mcp.name == "FastMCP"
         assert mcp.instructions == "Server instructions"
 
+    @pytest.mark.anyio
+    async def test_normalize_path(self):
+        """Test path normalization for mount paths."""
+        mcp = FastMCP()
+
+        # Test root path
+        assert mcp._normalize_path("/", "/messages/") == "/messages/"
+
+        # Test path with trailing slash
+        assert mcp._normalize_path("/github/", "/messages/") == "/github/messages/"
+
+        # Test path without trailing slash
+        assert mcp._normalize_path("/github", "/messages/") == "/github/messages/"
+
+        # Test endpoint without leading slash
+        assert mcp._normalize_path("/github", "messages/") == "/github/messages/"
+
+        # Test both with trailing/leading slashes
+        assert mcp._normalize_path("/api/", "/v1/") == "/api/v1/"
+
+    @pytest.mark.anyio
+    async def test_sse_app_with_mount_path(self):
+        """Test SSE app creation with different mount paths."""
+        # Test with default mount path
+        mcp = FastMCP()
+        with patch.object(
+            mcp, "_normalize_path", return_value="/messages/"
+        ) as mock_normalize:
+            mcp.sse_app()
+            # Verify _normalize_path was called with correct args
+            mock_normalize.assert_called_once_with("/", "/messages/")
+
+        # Test with custom mount path in settings
+        mcp = FastMCP()
+        mcp.settings.mount_path = "/custom"
+        with patch.object(
+            mcp, "_normalize_path", return_value="/custom/messages/"
+        ) as mock_normalize:
+            mcp.sse_app()
+            # Verify _normalize_path was called with correct args
+            mock_normalize.assert_called_once_with("/custom", "/messages/")
+
+        # Test with mount_path parameter
+        mcp = FastMCP()
+        with patch.object(
+            mcp, "_normalize_path", return_value="/param/messages/"
+        ) as mock_normalize:
+            mcp.sse_app(mount_path="/param")
+            # Verify _normalize_path was called with correct args
+            mock_normalize.assert_called_once_with("/param", "/messages/")
+
+    @pytest.mark.anyio
+    async def test_starlette_routes_with_mount_path(self):
+        """Test that Starlette routes are correctly configured with mount path."""
+        # Test with mount path in settings
+        mcp = FastMCP()
+        mcp.settings.mount_path = "/api"
+        app = mcp.sse_app()
+
+        # Find routes by type
+        sse_routes = [r for r in app.routes if isinstance(r, Route)]
+        mount_routes = [r for r in app.routes if isinstance(r, Mount)]
+
+        # Verify routes exist
+        assert len(sse_routes) == 1, "Should have one SSE route"
+        assert len(mount_routes) == 1, "Should have one mount route"
+
+        # Verify path values
+        assert sse_routes[0].path == "/sse", "SSE route path should be /sse"
+        assert (
+            mount_routes[0].path == "/messages"
+        ), "Mount route path should be /messages"
+
+        # Test with mount path as parameter
+        mcp = FastMCP()
+        app = mcp.sse_app(mount_path="/param")
+
+        # Find routes by type
+        sse_routes = [r for r in app.routes if isinstance(r, Route)]
+        mount_routes = [r for r in app.routes if isinstance(r, Mount)]
+
+        # Verify routes exist
+        assert len(sse_routes) == 1, "Should have one SSE route"
+        assert len(mount_routes) == 1, "Should have one mount route"
+
+        # Verify path values
+        assert sse_routes[0].path == "/sse", "SSE route path should be /sse"
+        assert (
+            mount_routes[0].path == "/messages"
+        ), "Mount route path should be /messages"
+
     @pytest.mark.anyio
     async def test_non_ascii_description(self):
         """Test that FastMCP handles non-ASCII characters in descriptions correctly"""
@@ -518,8 +611,6 @@ async def async_tool(x: int, ctx: Context) -> str:
 
     @pytest.mark.anyio
     async def test_context_logging(self):
-        from unittest.mock import patch
-
         import mcp.server.session
 
         """Test that context logging methods work."""