fix: resolve URL path truncation in SSE transport for proxied servers

Author: SOURABHMISHRA5221

src/mcp/server/sse.py

@@ -1,13 +1,19 @@
 """
-SSE Server Transport Module
+SSE Server Transport Module - Fixed Version
 
 This module implements a Server-Sent Events (SSE) transport layer for MCP servers.
+Fixes the URL path joining issue when using subpaths/proxied servers.
 
 Example usage:
-```
-    # Create an SSE transport at an endpoint
+```python
+    # Option 1: Create an SSE transport with absolute path (leading slash)
+    # This treats "/messages/" as absolute within the app
     sse = SseServerTransport("/messages/")
 
+    # Option 2: Create an SSE transport with relative path (no leading slash)
+    # This treats "messages/" as relative to the root path - RECOMMENDED for proxied servers
+    sse = SseServerTransport("messages/")
+
     # Create Starlette routes for SSE and message handling
     routes = [
         Route("/sse", endpoint=handle_sse, methods=["GET"]),
@@ -30,6 +36,15 @@ async def handle_sse(request):
     uvicorn.run(starlette_app, host="127.0.0.1", port=port)
 ```
 
+Path behavior examples:
+- With root_path="" and endpoint="/messages/": Final path = "/messages/"
+- With root_path="" and endpoint="messages/": Final path = "/messages/"
+- With root_path="/api" and endpoint="/messages/": Final path = "/api/messages/"
+- With root_path="/api" and endpoint="messages/": Final path = "/api/messages/"
+
+For servers behind proxies or mounted at subpaths, use the relative path format
+(without leading slash) to ensure proper URL joining with urllib.parse.urljoin().
+
 Note: The handle_sse function must return a Response to avoid a "TypeError: 'NoneType'
 object is not callable" error when client disconnects. The example above returns
 an empty Response() after the SSE connection ends to fix this.
@@ -84,7 +99,7 @@ def __init__(self, endpoint: str, security_settings: TransportSecuritySettings |
 
         Args:
             endpoint: A relative path where messages should be posted
-                    (e.g., "/messages/").
+                    (e.g., "/messages/" or "messages/").
             security_settings: Optional security settings for DNS rebinding protection.
 
         Note:
@@ -96,6 +111,9 @@ def __init__(self, endpoint: str, security_settings: TransportSecuritySettings |
             3. Portability: The same endpoint configuration works across different
                environments (development, staging, production)
 
+        The endpoint path handling has been updated to work correctly with urllib.parse.urljoin()
+        when servers are behind proxies or mounted at subpaths.
+
         Raises:
             ValueError: If the endpoint is a full URL instead of a relative path
         """
@@ -105,19 +123,49 @@ def __init__(self, endpoint: str, security_settings: TransportSecuritySettings |
         # Validate that endpoint is a relative path and not a full URL
         if "://" in endpoint or endpoint.startswith("//") or "?" in endpoint or "#" in endpoint:
             raise ValueError(
-                f"Given endpoint: {endpoint} is not a relative path (e.g., '/messages/'), "
-                "expecting a relative path (e.g., '/messages/')."
+                f"Given endpoint: {endpoint} is not a relative path (e.g., '/messages/' or 'messages/'), "
+                "expecting a relative path (e.g., '/messages/' or 'messages/')."
             )
 
-        # Ensure endpoint starts with a forward slash
-        if not endpoint.startswith("/"):
-            endpoint = "/" + endpoint
-
+        # Handle leading slash more intelligently
+        # Remove automatic leading slash enforcement to support proper URL joining
+        # Store the endpoint as-is, allowing both "/messages/" and "messages/" formats
         self._endpoint = endpoint
+        
         self._read_stream_writers = {}
         self._security = TransportSecurityMiddleware(security_settings)
         logger.debug(f"SseServerTransport initialized with endpoint: {endpoint}")
 
+    def _build_message_path(self, root_path: str) -> str:
+        """
+        Helper method to properly construct the message path
+        
+        This method handles the path construction logic that was causing issues
+        with urllib.parse.urljoin() when servers are proxied or mounted at subpaths.
+        
+        Args:
+            root_path: The root path from ASGI scope (e.g., "" or "/api_prefix")
+            
+        Returns:
+            The properly constructed path for client message posting
+        """
+        # Clean up the root path
+        clean_root_path = root_path.rstrip("/")
+        
+        # If endpoint starts with "/", it's meant to be absolute within the app
+        # If endpoint doesn't start with "/", it's meant to be relative to root_path
+        if self._endpoint.startswith("/"):
+            # Absolute path within the app - just concatenate
+            full_path = clean_root_path + self._endpoint
+        else:
+            # Relative path - ensure proper joining
+            if clean_root_path:
+                full_path = clean_root_path + "/" + self._endpoint
+            else:
+                full_path = "/" + self._endpoint
+                
+        return full_path
+
     @asynccontextmanager
     async def connect_sse(self, scope: Scope, receive: Receive, send: Send):
         if scope["type"] != "http":
@@ -145,17 +193,9 @@ async def connect_sse(self, scope: Scope, receive: Receive, send: Send):
         self._read_stream_writers[session_id] = read_stream_writer
         logger.debug(f"Created new session with ID: {session_id}")
 
-        # Determine the full path for the message endpoint to be sent to the client.
-        # scope['root_path'] is the prefix where the current Starlette app
-        # instance is mounted.
-        # e.g., "" if top-level, or "/api_prefix" if mounted under "/api_prefix".
+        # Use the new helper method for proper path construction
         root_path = scope.get("root_path", "")
-
-        # self._endpoint is the path *within* this app, e.g., "/messages".
-        # Concatenating them gives the full absolute path from the server root.
-        # e.g., "" + "/messages" -> "/messages"
-        # e.g., "/api_prefix" + "/messages" -> "/api_prefix/messages"
-        full_message_path_for_client = root_path.rstrip("/") + self._endpoint
+        full_message_path_for_client = self._build_message_path(root_path)
 
         # This is the URI (path + query) the client will use to POST messages.
         client_post_uri_data = f"{quote(full_message_path_for_client)}?session_id={session_id.hex}"
@@ -246,4 +286,4 @@ async def handle_post_message(self, scope: Scope, receive: Receive, send: Send)
         logger.debug(f"Sending session message to writer: {session_message}")
         response = Response("Accepted", status_code=202)
         await response(scope, receive, send)
-        await writer.send(session_message)
+        await writer.send(session_message)

tests/server/test_sse_security.py

@@ -291,3 +291,34 @@ async def test_sse_security_post_valid_content_type(server_port: int):
     finally:
         process.terminate()
         process.join()
+
+
+@pytest.mark.anyio
+async def test_endpoint_validation_rejects_absolute_urls():
+    """Test that SseServerTransport properly validates endpoint format."""
+    # These should all raise ValueError due to being absolute URLs or having invalid characters
+    invalid_endpoints = [
+        "http://example.com/messages/",
+        "https://example.com/messages/", 
+        "//example.com/messages/",
+        "/messages/?query=test",
+        "/messages/#fragment",
+    ]
+    
+    for invalid_endpoint in invalid_endpoints:
+        with pytest.raises(ValueError, match="is not a relative path"):
+            SseServerTransport(invalid_endpoint)
+    
+    # These should all be valid - endpoint is stored as-is (no automatic normalization)
+    valid_endpoints_and_expected = [
+        ("/messages/", "/messages/"),  # Absolute path format
+        ("messages/", "messages/"),    # Relative path format  
+        ("/api/v1/messages/", "/api/v1/messages/"),
+        ("api/v1/messages/", "api/v1/messages/"),
+    ]
+    
+    for valid_endpoint, expected_stored_value in valid_endpoints_and_expected:
+        # Should not raise an exception
+        transport = SseServerTransport(valid_endpoint)
+        # Endpoint should be stored exactly as provided (no normalization)
+        assert transport._endpoint == expected_stored_value

tests/shared/test_sse.py

@@ -487,9 +487,9 @@ def test_sse_message_id_coercion():
 @pytest.mark.parametrize(
     "endpoint, expected_result",
     [
-        # Valid endpoints - should normalize and work
+        # These should all be valid - endpoint is stored as-is (no automatic normalization)
         ("/messages/", "/messages/"),
-        ("messages/", "/messages/"),
+        ("messages/", "messages/"), 
         ("/", "/"),
         # Invalid endpoints - should raise ValueError
         ("http://example.com/messages/", ValueError),
@@ -506,7 +506,6 @@ def test_sse_server_transport_endpoint_validation(endpoint: str, expected_result
         with pytest.raises(expected_result, match="is not a relative path.*expecting a relative path"):
             SseServerTransport(endpoint)
     else:
-        # Test valid endpoints that should normalize correctly
+        # Endpoint should be stored exactly as provided (no normalization)
         sse = SseServerTransport(endpoint)
         assert sse._endpoint == expected_result
-        assert sse._endpoint.startswith("/")