fix: raise clear error for server-to-client requests in stateless mode

src/mcp/server/session.py

@@ -93,6 +93,7 @@ def __init__(
         stateless: bool = False,
     ) -> None:
         super().__init__(read_stream, write_stream, types.ClientRequest, types.ClientNotification)
+        self._stateless = stateless
         self._initialization_state = (
             InitializationState.Initialized if stateless else InitializationState.NotInitialized
         )
@@ -156,6 +157,26 @@ def check_client_capability(self, capability: types.ClientCapabilities) -> bool:
 
         return True
 
+    def _require_stateful_mode(self, feature_name: str) -> None:
+        """Raise an error if trying to use a feature that requires stateful mode.
+
+        Server-to-client requests (sampling, elicitation, list_roots) are not
+        supported in stateless HTTP mode because there is no persistent connection
+        for bidirectional communication.
+
+        Args:
+            feature_name: Name of the feature being used (for error message)
+
+        Raises:
+            RuntimeError: If the session is in stateless mode
+        """
+        if self._stateless:
+            raise RuntimeError(
+                f"Cannot use {feature_name} in stateless HTTP mode. "
+                "Stateless mode does not support server-to-client requests. "
+                "Use stateful mode (stateless_http=False) to enable this feature."
+            )
+
     async def _receive_loop(self) -> None:
         async with self._incoming_message_stream_writer:
             await super()._receive_loop()
@@ -311,7 +332,9 @@ async def create_message(
         Raises:
             McpError: If tools are provided but client doesn't support them.
             ValueError: If tool_use or tool_result message structure is invalid.
+            RuntimeError: If called in stateless HTTP mode.
         """
+        self._require_stateful_mode("sampling")
         client_caps = self._client_params.capabilities if self._client_params else None
         validate_sampling_tools(client_caps, tools, tool_choice)
         validate_tool_use_result_messages(messages)
@@ -349,6 +372,7 @@ async def create_message(
 
     async def list_roots(self) -> types.ListRootsResult:
         """Send a roots/list request."""
+        self._require_stateful_mode("list_roots")
         return await self.send_request(
             types.ServerRequest(types.ListRootsRequest()),
             types.ListRootsResult,
@@ -391,7 +415,11 @@ async def elicit_form(
 
         Returns:
             The client's response with form data
+
+        Raises:
+            RuntimeError: If called in stateless HTTP mode.
         """
+        self._require_stateful_mode("elicitation")
         return await self.send_request(
             types.ServerRequest(
                 types.ElicitRequest(
@@ -425,7 +453,11 @@ async def elicit_url(
 
         Returns:
             The client's response indicating acceptance, decline, or cancellation
+
+        Raises:
+            RuntimeError: If called in stateless HTTP mode.
         """
+        self._require_stateful_mode("elicitation")
         return await self.send_request(
             types.ServerRequest(
                 types.ElicitRequest(

tests/server/test_stateless_mode.py

@@ -0,0 +1,263 @@
+"""Tests for stateless HTTP mode limitations.
+
+Stateless HTTP mode does not support server-to-client requests because there
+is no persistent connection for bidirectional communication. These tests verify
+that appropriate errors are raised when attempting to use unsupported features.
+
+See: https://github.com/modelcontextprotocol/python-sdk/issues/1097
+"""
+
+import anyio
+import pytest
+
+import mcp.types as types
+from mcp.server.models import InitializationOptions
+from mcp.server.session import ServerSession
+from mcp.shared.message import SessionMessage
+from mcp.types import ServerCapabilities
+
+
+def create_test_streams():
+    """Create memory streams for testing."""
+    server_to_client_send, server_to_client_receive = anyio.create_memory_object_stream[SessionMessage](1)
+    client_to_server_send, client_to_server_receive = anyio.create_memory_object_stream[SessionMessage | Exception](1)
+    return (
+        server_to_client_send,
+        server_to_client_receive,
+        client_to_server_send,
+        client_to_server_receive,
+    )
+
+
+def create_init_options():
+    """Create default initialization options for testing."""
+    return InitializationOptions(
+        server_name="test",
+        server_version="0.1.0",
+        capabilities=ServerCapabilities(),
+    )
+
+
+@pytest.mark.anyio
+async def test_list_roots_fails_in_stateless_mode():
+    """Test that list_roots raises RuntimeError in stateless mode."""
+    (
+        server_to_client_send,
+        server_to_client_receive,
+        client_to_server_send,
+        client_to_server_receive,
+    ) = create_test_streams()
+
+    async with (
+        client_to_server_send,
+        client_to_server_receive,
+        server_to_client_send,
+        server_to_client_receive,
+    ):
+        async with ServerSession(
+            client_to_server_receive,
+            server_to_client_send,
+            create_init_options(),
+            stateless=True,
+        ) as session:
+            with pytest.raises(RuntimeError) as exc_info:
+                await session.list_roots()
+
+            assert "stateless HTTP mode" in str(exc_info.value)
+            assert "list_roots" in str(exc_info.value)
+
+
+@pytest.mark.anyio
+async def test_create_message_fails_in_stateless_mode():
+    """Test that create_message raises RuntimeError in stateless mode."""
+    (
+        server_to_client_send,
+        server_to_client_receive,
+        client_to_server_send,
+        client_to_server_receive,
+    ) = create_test_streams()
+
+    async with (
+        client_to_server_send,
+        client_to_server_receive,
+        server_to_client_send,
+        server_to_client_receive,
+    ):
+        async with ServerSession(
+            client_to_server_receive,
+            server_to_client_send,
+            create_init_options(),
+            stateless=True,
+        ) as session:
+            with pytest.raises(RuntimeError) as exc_info:
+                await session.create_message(
+                    messages=[
+                        types.SamplingMessage(
+                            role="user",
+                            content=types.TextContent(type="text", text="hello"),
+                        )
+                    ],
+                    max_tokens=100,
+                )
+
+            assert "stateless HTTP mode" in str(exc_info.value)
+            assert "sampling" in str(exc_info.value)
+
+
+@pytest.mark.anyio
+async def test_elicit_form_fails_in_stateless_mode():
+    """Test that elicit_form raises RuntimeError in stateless mode."""
+    (
+        server_to_client_send,
+        server_to_client_receive,
+        client_to_server_send,
+        client_to_server_receive,
+    ) = create_test_streams()
+
+    async with (
+        client_to_server_send,
+        client_to_server_receive,
+        server_to_client_send,
+        server_to_client_receive,
+    ):
+        async with ServerSession(
+            client_to_server_receive,
+            server_to_client_send,
+            create_init_options(),
+            stateless=True,
+        ) as session:
+            with pytest.raises(RuntimeError) as exc_info:
+                await session.elicit_form(
+                    message="Please provide input",
+                    requestedSchema={"type": "object", "properties": {}},
+                )
+
+            assert "stateless HTTP mode" in str(exc_info.value)
+            assert "elicitation" in str(exc_info.value)
+
+
+@pytest.mark.anyio
+async def test_elicit_url_fails_in_stateless_mode():
+    """Test that elicit_url raises RuntimeError in stateless mode."""
+    (
+        server_to_client_send,
+        server_to_client_receive,
+        client_to_server_send,
+        client_to_server_receive,
+    ) = create_test_streams()
+
+    async with (
+        client_to_server_send,
+        client_to_server_receive,
+        server_to_client_send,
+        server_to_client_receive,
+    ):
+        async with ServerSession(
+            client_to_server_receive,
+            server_to_client_send,
+            create_init_options(),
+            stateless=True,
+        ) as session:
+            with pytest.raises(RuntimeError) as exc_info:
+                await session.elicit_url(
+                    message="Please authenticate",
+                    url="https://example.com/auth",
+                    elicitation_id="test-123",
+                )
+
+            assert "stateless HTTP mode" in str(exc_info.value)
+            assert "elicitation" in str(exc_info.value)
+
+
+@pytest.mark.anyio
+async def test_elicit_deprecated_fails_in_stateless_mode():
+    """Test that the deprecated elicit method also fails in stateless mode."""
+    (
+        server_to_client_send,
+        server_to_client_receive,
+        client_to_server_send,
+        client_to_server_receive,
+    ) = create_test_streams()
+
+    async with (
+        client_to_server_send,
+        client_to_server_receive,
+        server_to_client_send,
+        server_to_client_receive,
+    ):
+        async with ServerSession(
+            client_to_server_receive,
+            server_to_client_send,
+            create_init_options(),
+            stateless=True,
+        ) as session:
+            with pytest.raises(RuntimeError) as exc_info:
+                await session.elicit(
+                    message="Please provide input",
+                    requestedSchema={"type": "object", "properties": {}},
+                )
+
+            assert "stateless HTTP mode" in str(exc_info.value)
+            assert "elicitation" in str(exc_info.value)
+
+
+@pytest.mark.anyio
+async def test_require_stateful_mode_does_not_raise_in_stateful_mode():
+    """Test that _require_stateful_mode does not raise in stateful mode."""
+    (
+        server_to_client_send,
+        server_to_client_receive,
+        client_to_server_send,
+        client_to_server_receive,
+    ) = create_test_streams()
+
+    async with (
+        client_to_server_send,
+        client_to_server_receive,
+        server_to_client_send,
+        server_to_client_receive,
+    ):
+        async with ServerSession(
+            client_to_server_receive,
+            server_to_client_send,
+            create_init_options(),
+            stateless=False,  # Stateful mode
+        ) as session:
+            # These should not raise - the check passes in stateful mode
+            session._require_stateful_mode("list_roots")
+            session._require_stateful_mode("sampling")
+            session._require_stateful_mode("elicitation")
+
+
+@pytest.mark.anyio
+async def test_stateless_error_message_is_actionable():
+    """Test that the error message provides actionable guidance."""
+    (
+        server_to_client_send,
+        server_to_client_receive,
+        client_to_server_send,
+        client_to_server_receive,
+    ) = create_test_streams()
+
+    async with (
+        client_to_server_send,
+        client_to_server_receive,
+        server_to_client_send,
+        server_to_client_receive,
+    ):
+        async with ServerSession(
+            client_to_server_receive,
+            server_to_client_send,
+            create_init_options(),
+            stateless=True,
+        ) as session:
+            with pytest.raises(RuntimeError) as exc_info:
+                await session.list_roots()
+
+            error_message = str(exc_info.value)
+            # Should mention it's stateless mode
+            assert "stateless HTTP mode" in error_message
+            # Should explain why it doesn't work
+            assert "server-to-client requests" in error_message
+            # Should tell user how to fix it
+            assert "stateless_http=False" in error_message