fix: reject pre-session GET in the session manager before a session is created

A pre-session GET (one without a session ID header in stateful mode) was
being handled AFTER a transport and session were already registered. The
transport's _handle_get_request correctly returned 405, but by then an
unused session existed indefinitely (with default no idle timeout).

Restructured to reject pre-session GETs at the manager layer BEFORE any
transport creation:

1. Manager now checks for GET without session ID first
2. Security validation (DNS rebinding protection) runs before the 405
   so malformed/attack requests get their proper 421 response
3. Only after security passes does it return 405 Method Not Allowed

The transport-level check remains as defensive code for standalone
transport use (the class is public). Tests now cover both paths:
- Manager path: test_pre_session_get_rejected_without_creating_transport
- Standalone path: test_standalone_transport_pre_session_get_returns_405

Also added test_standalone_transport_get_with_wrong_session_returns_404
to cover the session ID mismatch validation (removed pragma: no cover).

Addresses review finding: #3129

Author: Veerendra Kumar

src/mcp/server/streamable_http.py

@@ -894,7 +894,7 @@ async def _validate_session(self, request: Request, send: Send) -> bool:
             return False
 
         # If session ID doesn't match, return error
-        if request_session_id != self.mcp_session_id:  # pragma: no cover
+        if request_session_id != self.mcp_session_id:
             response = self._create_error_response(
                 "Not Found: Invalid or expired session ID",
                 HTTPStatus.NOT_FOUND,

src/mcp/server/streamable_http_manager.py

@@ -253,6 +253,38 @@ async def _handle_stateful_request(self, scope: Scope, receive: Receive, send: S
         request = Request(scope, receive)
         request_mcp_session_id = request.headers.get(MCP_SESSION_ID_HEADER)
 
+        # Per MCP spec, a GET without a session ID cannot establish an SSE stream in
+        # stateful mode because no session exists yet. Reject with 405 BEFORE creating
+        # any transport or session to avoid leaking resources. Security validation
+        # (DNS rebinding protection) happens first so malformed/attack requests get
+        # their proper 421 response rather than our 405.
+        if request.method == "GET" and request_mcp_session_id is None:
+            # Run security validation first if enabled
+            if self.security_settings is not None:
+                from mcp.server.transport_security import TransportSecurityMiddleware
+
+                security = TransportSecurityMiddleware(self.security_settings)
+                error_response = await security.validate_request(request, is_post=False)
+                if error_response:
+                    await error_response(scope, receive, send)
+                    return
+
+            response = Response(
+                JSONRPCError(
+                    jsonrpc="2.0",
+                    id=None,
+                    error=ErrorData(
+                        code=INVALID_REQUEST,
+                        message="Method Not Allowed: GET requires an established session",
+                    ),
+                ).model_dump_json(by_alias=True, exclude_unset=True),
+                status_code=405,
+                media_type="application/json",
+                headers={"Allow": "GET, POST, DELETE"},
+            )
+            await response(scope, receive, send)
+            return
+
         user = scope.get("user")
         requestor = authorization_context(user) if isinstance(user, AuthenticatedUser) else None
 

tests/server/test_streamable_http_manager.py

@@ -746,3 +746,49 @@ async def test_anonymous_session_accepts_anonymous_requests(
     session_id = await _open_session(manager, None)
 
     assert await _request_session(manager, session_id, None) != 404
+
+
+@pytest.mark.anyio
+async def test_pre_session_get_rejected_without_creating_transport() -> None:
+    """A GET without a session ID returns 405 before any transport or session is created."""
+    app = Server("test-pre-session-get")
+    manager = StreamableHTTPSessionManager(app=app)
+
+    async with manager.run():
+        sent_messages: list[Message] = []
+        response_body = b""
+
+        async def mock_send(message: Message) -> None:
+            nonlocal response_body
+            sent_messages.append(message)
+            if message["type"] == "http.response.body":
+                response_body += message.get("body", b"")
+
+        scope: Scope = {
+            "type": "http",
+            "method": "GET",
+            "path": "/mcp",
+            "headers": [(b"accept", b"text/event-stream")],
+        }
+
+        async def mock_receive() -> Message:
+            return {"type": "http.request", "body": b"", "more_body": False}
+
+        await manager.handle_request(scope, mock_receive, mock_send)
+
+        # Should return 405 Method Not Allowed
+        response_start = next(msg for msg in sent_messages if msg["type"] == "http.response.start")
+        assert response_start["status"] == 405
+
+        # Verify Allow header is present
+        headers = dict(response_start.get("headers", []))
+        assert headers.get(b"allow") == b"GET, POST, DELETE"
+
+        # Verify JSON-RPC error format
+        error_data = json.loads(response_body)
+        assert error_data["jsonrpc"] == "2.0"
+        assert error_data["id"] is None
+        assert "Method Not Allowed" in error_data["error"]["message"]
+
+        # Most importantly: no session was created
+        assert manager._server_instances == {}

tests/shared/test_streamable_http.py

@@ -2264,3 +2264,106 @@ async def asgi_receive() -> Message:
     assert body_chunks[-1] == {"type": "http.response.body", "body": b"", "more_body": False}
     assert "Error in standalone SSE writer" not in caplog.text
     assert "Error in standalone SSE response" not in caplog.text
+
+
+@pytest.mark.anyio
+async def test_standalone_transport_pre_session_get_returns_405() -> None:
+    """A stateful transport (mcp_session_id set) rejects a GET without session ID with 405.
+
+    This tests the transport-level defensive check that prevents pre-session GETs from
+    establishing an SSE stream. When used via the session manager, the manager rejects
+    these requests first; this check ensures the transport is also safe for standalone use.
+    """
+    # Create a transport with a session ID (stateful mode)
+    transport = StreamableHTTPServerTransport(
+        mcp_session_id="test-session-id",
+        security_settings=TransportSecuritySettings(enable_dns_rebinding_protection=False),
+    )
+
+    # Set up the read stream writer so handle_request doesn't fail
+    read_stream_writer, read_stream = create_context_streams[SessionMessage | Exception](0)
+    transport._read_stream_writer = read_stream_writer  # pyright: ignore[reportPrivateUsage]
+
+    sent: list[Message] = []
+
+    async def asgi_send(message: Message) -> None:
+        sent.append(message)
+
+    async def asgi_receive() -> Message:
+        return {"type": "http.request", "body": b"", "more_body": False}
+
+    # Send a GET request without a session ID header
+    scope: Scope = {
+        "type": "http",
+        "method": "GET",
+        "path": "/mcp",
+        "query_string": b"",
+        "headers": [(b"accept", b"text/event-stream")],
+    }
+
+    async with read_stream_writer, read_stream:
+        await transport.handle_request(scope, asgi_receive, asgi_send)
+
+    # Verify 405 Method Not Allowed response
+    response_start = sent[0]
+    assert response_start["type"] == "http.response.start"
+    assert response_start["status"] == 405
+
+    # Verify Allow header
+    headers = dict(response_start.get("headers", []))
+    assert headers.get(b"allow") == b"GET, POST, DELETE"
+
+    # Verify response body contains the error message
+    body = b"".join(m.get("body", b"") for m in sent if m["type"] == "http.response.body")
+    assert b"Method Not Allowed" in body
+    assert b"GET requires an established session" in body
+
+
+@pytest.mark.anyio
+async def test_standalone_transport_get_with_wrong_session_returns_404() -> None:
+    """A stateful transport rejects a GET with a mismatched session ID with 404.
+
+    This tests the transport-level session validation for standalone transport use.
+    When used via the session manager, the manager validates session IDs first.
+    """
+    # Create a transport with a specific session ID
+    transport = StreamableHTTPServerTransport(
+        mcp_session_id="correct-session-id",
+        security_settings=TransportSecuritySettings(enable_dns_rebinding_protection=False),
+    )
+
+    # Set up the read stream writer so handle_request doesn't fail
+    read_stream_writer, read_stream = create_context_streams[SessionMessage | Exception](0)
+    transport._read_stream_writer = read_stream_writer  # pyright: ignore[reportPrivateUsage]
+
+    sent: list[Message] = []
+
+    async def asgi_send(message: Message) -> None:
+        sent.append(message)
+
+    async def asgi_receive() -> Message:
+        return {"type": "http.request", "body": b"", "more_body": False}
+
+    # Send a GET request with a WRONG session ID header
+    scope: Scope = {
+        "type": "http",
+        "method": "GET",
+        "path": "/mcp",
+        "query_string": b"",
+        "headers": [
+            (b"accept", b"text/event-stream"),
+            (b"mcp-session-id", b"wrong-session-id"),
+        ],
+    }
+
+    async with read_stream_writer, read_stream:
+        await transport.handle_request(scope, asgi_receive, asgi_send)
+
+    # Verify 404 Not Found response (session validation failed)
+    response_start = sent[0]
+    assert response_start["type"] == "http.response.start"
+    assert response_start["status"] == 404
+
+    # Verify response body contains the error message
+    body = b"".join(m.get("body", b"") for m in sent if m["type"] == "http.response.body")
+    assert b"Invalid or expired session ID" in body