sse endpoint validation + test

Author: AishwaryaKalloli

src/mcp/server/sse.py

@@ -78,7 +78,7 @@ def __init__(self, endpoint: str) -> None:
         messages to the relative path given.
 
         Args:
-            endpoint: A relative path where messages should be posted 
+            endpoint: A relative path where messages should be posted
                     (e.g., "/messages/").
 
         Note:
@@ -97,9 +97,10 @@ def __init__(self, endpoint: str) -> None:
         super().__init__()
 
         # Validate that endpoint is a relative path and not a full URL
-        if "://" in endpoint or endpoint.startswith("//"):
+        if "://" in endpoint or endpoint.startswith("//") or "?" in endpoint or "#" in endpoint:
             raise ValueError(
-                "Endpoint must be a relative path (e.g., '/messages/'), not a full URL."
+                f"Given endpoint: {endpoint} is not a relative path (e.g., '/messages/'), \
+                             expecting a relative path(e.g., '/messages/')."
             )
 
         # Ensure endpoint starts with a forward slash

tests/shared/test_sse.py

@@ -464,3 +464,31 @@ def test_sse_message_id_coercion():
     json_message = '{"jsonrpc": "2.0", "id": "123", "method": "ping", "params": null}'
     msg = types.JSONRPCMessage.model_validate_json(json_message)
     assert msg == snapshot(types.JSONRPCMessage(root=types.JSONRPCRequest(method="ping", jsonrpc="2.0", id=123)))
+
+
+@pytest.mark.parametrize(
+    "endpoint, expected_result",
+    [
+        # Valid endpoints - should normalize and work
+        ("/messages/", "/messages/"),
+        ("messages/", "/messages/"),
+        ("/", "/"),
+        # Invalid endpoints - should raise ValueError
+        ("http://example.com/messages/", ValueError),
+        ("//example.com/messages/", ValueError),
+        ("ftp://example.com/messages/", ValueError),
+        ("/messages/?param=value", ValueError),
+        ("/messages/#fragment", ValueError),
+    ],
+)
+def test_sse_server_transport_endpoint_validation(endpoint: str, expected_result: str | type[Exception]):
+    """Test that SseServerTransport properly validates and normalizes endpoints."""
+    if isinstance(expected_result, type) and issubclass(expected_result, Exception):
+        # Test invalid endpoints that should raise an exception
+        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
+        sse = SseServerTransport(endpoint)
+        assert sse._endpoint == expected_result
+        assert sse._endpoint.startswith("/")