fix: return 405 for pre-session GET the server won't serve as SSE

src/mcp/server/streamable_http.py

@@ -656,11 +656,29 @@ async def _handle_get_request(self, request: Request, send: Send) -> None:
         This allows the server to communicate to the client without the client
         first sending data via HTTP POST. The server can send JSON-RPC requests
         and notifications on this stream.
+
+        Per MCP spec: "The server MUST either return Content-Type: text/event-stream
+        in response to this HTTP GET, or else return HTTP 405 Method Not Allowed."
         """
         writer = self._read_stream_writer
         if writer is None:  # pragma: no cover
             raise ValueError("No read stream writer available. Ensure connect() is called first.")
 
+        # Per MCP spec, pre-session GETs that cannot be served as SSE must return
+        # 405 Method Not Allowed. A GET without a session ID in stateful mode
+        # cannot establish an SSE stream because no session exists yet.
+        if self.mcp_session_id and not self._get_session_id(request):
+            headers = {
+                "Allow": "GET, POST, DELETE",
+            }
+            response = self._create_error_response(
+                "Method Not Allowed: GET requires an established session",
+                HTTPStatus.METHOD_NOT_ALLOWED,
+                headers=headers,
+            )
+            await response(request.scope, request.receive, send)
+            return
+
         # Validate Accept header - must include text/event-stream
         _, has_sse = check_accept_headers(request)
 
@@ -842,7 +860,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/interaction/_requirements.py

@@ -3094,6 +3094,16 @@ def __post_init__(self) -> None:
         transports=("streamable-http",),
         note="Only observable over HTTP: 405 is an HTTP status code.",
     ),
+    "hosting:http:pre-session-get-405": Requirement(
+        source=f"{SPEC_BASE_URL}/basic/transports#receiving-messages-from-the-server",
+        behavior=(
+            "A GET without a session ID in stateful mode returns 405 Method Not Allowed. "
+            "Per MCP spec: 'The server MUST either return Content-Type: text/event-stream in response "
+            "to this HTTP GET, or else return HTTP 405 Method Not Allowed.'"
+        ),
+        transports=("streamable-http",),
+        note="Only observable over HTTP: 405 is an HTTP status code. Fixes issue #3102.",
+    ),
     "hosting:http:no-broadcast": Requirement(
         source=f"{SPEC_BASE_URL}/basic/transports#multiple-connections",
         behavior=(

tests/interaction/transports/test_hosting_http.py

@@ -95,6 +95,50 @@ async def test_unsupported_http_methods_return_405() -> None:
     assert (patch.status_code, patch.headers.get("allow")) == snapshot((405, "GET, POST, DELETE"))
 
 
+@requirement("hosting:http:pre-session-get-405")
+async def test_pre_session_get_returns_405() -> None:
+    """A GET without a session ID in stateful mode returns 405 Method Not Allowed.
+
+    Per MCP spec: "The server MUST either return Content-Type: text/event-stream in response
+    to this HTTP GET, or else return HTTP 405 Method Not Allowed." A pre-session GET cannot
+    establish an SSE stream, so 405 is the spec-mandated response.
+
+    See: https://github.com/modelcontextprotocol/python-sdk/issues/3102
+    """
+    async with mounted_app(_server()) as (http, _):
+        # Test with various Accept headers - all should return 405 for pre-session GET
+        # Accept: */* (wildcard)
+        response_wildcard = await http.get("/mcp", headers={"accept": "*/*", "mcp-protocol-version": "2025-11-25"})
+        # Accept: application/json (no SSE)
+        response_json = await http.get(
+            "/mcp", headers={"accept": "application/json", "mcp-protocol-version": "2025-11-25"}
+        )
+        # No Accept header at all
+        response_no_accept = await http.get("/mcp", headers={"mcp-protocol-version": "2025-11-25"})
+        # Accept: text/event-stream (correct, but still no session)
+        response_sse = await http.get(
+            "/mcp", headers={"accept": "text/event-stream", "mcp-protocol-version": "2025-11-25"}
+        )
+
+    # All pre-session GETs must return 405 with Allow header listing supported methods
+    assert (response_wildcard.status_code, response_wildcard.headers.get("allow")) == snapshot(
+        (405, "GET, POST, DELETE")
+    )
+    assert "Method Not Allowed" in response_wildcard.json()["error"]["message"]
+    assert response_wildcard.json()["error"]["code"] == -32600  # INVALID_REQUEST
+
+    assert (response_json.status_code, response_json.headers.get("allow")) == snapshot((405, "GET, POST, DELETE"))
+    assert "Method Not Allowed" in response_json.json()["error"]["message"]
+
+    assert (response_no_accept.status_code, response_no_accept.headers.get("allow")) == snapshot(
+        (405, "GET, POST, DELETE")
+    )
+    assert "Method Not Allowed" in response_no_accept.json()["error"]["message"]
+
+    assert (response_sse.status_code, response_sse.headers.get("allow")) == snapshot((405, "GET, POST, DELETE"))
+    assert "Method Not Allowed" in response_sse.json()["error"]["message"]
+
+
 @requirement("hosting:http:accept-406")
 async def test_missing_accept_media_types_return_406() -> None:
     """A POST whose Accept header lacks both required types, or a GET lacking text/event-stream, returns 406."""

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}  # pragma: no cover
+
+        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/server/test_streamable_http_security.py

@@ -124,7 +124,8 @@ async def test_streamable_http_security_get_request() -> None:
         assert response.text == "Invalid Host header"
 
         response = await client.get("/", headers={"Accept": "text/event-stream", "Host": "127.0.0.1"})
-        # An allowed host passes security and fails on session validation instead.
-        assert response.status_code == 400
+        # An allowed host passes security but fails because GET requires an established session.
+        # Per MCP spec, pre-session GETs return 405 Method Not Allowed (issue #3102).
+        assert response.status_code == 405
         body = response.json()
-        assert "Missing session ID" in body["error"]["message"]
+        assert "Method Not Allowed" in body["error"]["message"]

tests/shared/test_streamable_http.py

@@ -2283,3 +2283,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}  # pragma: no cover
+
+    # 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}  # pragma: no cover
+
+    # 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