Add tests to cover SSE polling reconnection code paths

Add tests for previously uncovered lines in streamable_http client
and server to achieve full test coverage:

- Client _get_next_reconnection_delay with server retry values
- Client resume_stream early returns and error handling
- Server _create_priming_event with various configurations
- Server close_sse_stream including edge cases

Github-Issue:#1654

Author: felixweinberger

tests/shared/test_streamable_http.py

@@ -1606,3 +1606,254 @@ async def bad_client():
             assert isinstance(result, InitializeResult)
             tools = await session.list_tools()
             assert tools.tools
+
+
+# =============================================================================
+# SEP-1699: SSE Polling Support Tests
+# =============================================================================
+
+
+@pytest.mark.anyio
+async def test_reconnection_delay_with_server_retry():
+    """Test _get_next_reconnection_delay uses server-provided retry value."""
+    from mcp.client.streamable_http import (
+        StreamableHTTPReconnectionOptions,
+        StreamableHTTPTransport,
+    )
+
+    transport = StreamableHTTPTransport(
+        "http://localhost:8000",
+        reconnection_options=StreamableHTTPReconnectionOptions(
+            initial_reconnection_delay=1.0,
+            max_reconnection_delay=30.0,
+            reconnection_delay_grow_factor=2.0,
+            max_retries=5,
+        ),
+    )
+
+    # Without server retry, should use exponential backoff
+    delay_0 = transport._get_next_reconnection_delay(0)
+    assert delay_0 == 1.0  # initial_delay * 2^0 = 1.0
+
+    delay_1 = transport._get_next_reconnection_delay(1)
+    assert delay_1 == 2.0  # initial_delay * 2^1 = 2.0
+
+    delay_2 = transport._get_next_reconnection_delay(2)
+    assert delay_2 == 4.0  # initial_delay * 2^2 = 4.0
+
+    # Should cap at max_reconnection_delay
+    delay_large = transport._get_next_reconnection_delay(10)
+    assert delay_large == 30.0  # capped at max
+
+    # Set server-provided retry value
+    transport._server_retry_seconds = 5.0
+
+    # Should now use server-provided value regardless of attempt
+    assert transport._get_next_reconnection_delay(0) == 5.0
+    assert transport._get_next_reconnection_delay(5) == 5.0
+    assert transport._get_next_reconnection_delay(100) == 5.0
+
+
+@pytest.mark.anyio
+async def test_create_priming_event_with_event_store():
+    """Test _create_priming_event generates correct event when event store configured."""
+    event_store = SimpleEventStore()
+
+    transport = StreamableHTTPServerTransport(
+        mcp_session_id="test-session",
+        event_store=event_store,
+        retry_interval=5000,  # 5 seconds in ms
+    )
+
+    # Create priming event
+    priming_event = await transport._create_priming_event("stream-123")
+
+    assert priming_event is not None
+    assert "id" in priming_event
+    assert priming_event["id"] == "1"  # First event ID from SimpleEventStore
+    assert priming_event["data"] == ""  # Empty data for priming
+    assert priming_event["retry"] == "5000"
+
+
+@pytest.mark.anyio
+async def test_create_priming_event_without_retry_interval():
+    """Test _create_priming_event without retry interval configured."""
+    event_store = SimpleEventStore()
+
+    transport = StreamableHTTPServerTransport(
+        mcp_session_id="test-session",
+        event_store=event_store,
+        # No retry_interval
+    )
+
+    priming_event = await transport._create_priming_event("stream-456")
+
+    assert priming_event is not None
+    assert "id" in priming_event
+    assert priming_event["data"] == ""
+    assert "retry" not in priming_event  # No retry field
+
+
+@pytest.mark.anyio
+async def test_create_priming_event_without_event_store():
+    """Test _create_priming_event returns None without event store."""
+    transport = StreamableHTTPServerTransport(
+        mcp_session_id="test-session",
+        # No event_store
+    )
+
+    priming_event = await transport._create_priming_event("stream-789")
+
+    assert priming_event is None
+
+
+@pytest.mark.anyio
+async def test_close_sse_stream():
+    """Test close_sse_stream closes the stream and cleans up."""
+    transport = StreamableHTTPServerTransport(
+        mcp_session_id="test-session",
+    )
+
+    # Manually add a stream to _request_streams
+    send_stream, recv_stream = anyio.create_memory_object_stream[EventMessage](0)
+    transport._request_streams["request-123"] = (send_stream, recv_stream)
+
+    assert "request-123" in transport._request_streams
+
+    # Close the stream
+    await transport.close_sse_stream("request-123")
+
+    # Stream should be removed
+    assert "request-123" not in transport._request_streams
+
+
+@pytest.mark.anyio
+async def test_close_sse_stream_nonexistent():
+    """Test close_sse_stream handles nonexistent stream gracefully."""
+    transport = StreamableHTTPServerTransport(
+        mcp_session_id="test-session",
+    )
+
+    # Should not raise even if stream doesn't exist
+    await transport.close_sse_stream("nonexistent-stream")
+
+
+@pytest.mark.anyio
+async def test_close_sse_stream_already_closed():
+    """Test close_sse_stream handles already-closed streams gracefully."""
+    transport = StreamableHTTPServerTransport(
+        mcp_session_id="test-session",
+    )
+
+    # Manually add a stream and close it before calling close_sse_stream
+    send_stream, recv_stream = anyio.create_memory_object_stream[EventMessage](0)
+    transport._request_streams["request-456"] = (send_stream, recv_stream)
+
+    # Close streams manually first
+    await send_stream.aclose()
+    await recv_stream.aclose()
+
+    # Should handle gracefully without error
+    await transport.close_sse_stream("request-456")
+
+    # Stream should still be removed from dict
+    assert "request-456" not in transport._request_streams
+
+
+@pytest.mark.anyio
+async def test_resume_stream_without_session_id():
+    """Test resume_stream returns early without session ID."""
+    from mcp.client.streamable_http import StreamableHTTPTransport
+
+    transport = StreamableHTTPTransport("http://localhost:8000")
+    assert transport.session_id is None
+
+    # Create a dummy stream writer
+    read_stream_writer, read_stream_reader = anyio.create_memory_object_stream(0)
+
+    async with httpx.AsyncClient() as client:
+        # Should return early without making request
+        await transport.resume_stream(
+            client,
+            read_stream_writer,
+            "event-123",
+        )
+
+    # Clean up streams to avoid resource warnings
+    await read_stream_writer.aclose()
+    await read_stream_reader.aclose()
+
+    # No errors should occur
+
+
+@pytest.mark.anyio
+async def test_resume_stream_with_405_response(basic_server: None, basic_server_url: str):
+    """Test resume_stream handles 405 Method Not Allowed gracefully."""
+    from mcp.client.streamable_http import StreamableHTTPTransport
+
+    transport = StreamableHTTPTransport(f"{basic_server_url}/mcp")
+
+    # First establish a session via initialization
+    async with streamablehttp_client(f"{basic_server_url}/mcp") as (
+        read_stream,
+        write_stream,
+        get_session_id,
+    ):
+        async with ClientSession(read_stream, write_stream) as session:
+            await session.initialize()
+            transport.session_id = get_session_id()
+
+    # Now try to resume with the session - server might return 405
+    read_stream_writer, read_stream_reader = anyio.create_memory_object_stream(0)
+
+    async with httpx.AsyncClient() as client:
+        # This should handle the 405 gracefully
+        await transport.resume_stream(
+            client,
+            read_stream_writer,
+            "nonexistent-event-id",
+        )
+
+    # Clean up streams to avoid resource warnings
+    await read_stream_writer.aclose()
+    await read_stream_reader.aclose()
+
+
+@pytest.mark.anyio
+async def test_reconnection_options_dataclass():
+    """Test StreamableHTTPReconnectionOptions defaults."""
+    from mcp.client.streamable_http import StreamableHTTPReconnectionOptions
+
+    options = StreamableHTTPReconnectionOptions()
+
+    assert options.initial_reconnection_delay == 1.0
+    assert options.max_reconnection_delay == 30.0
+    assert options.reconnection_delay_grow_factor == 1.5
+    assert options.max_retries == 2
+
+
+@pytest.mark.anyio
+async def test_streamablehttp_client_with_reconnection_options(
+    basic_server: None, basic_server_url: str
+):
+    """Test streamablehttp_client accepts reconnection_options parameter."""
+    from mcp.client.streamable_http import StreamableHTTPReconnectionOptions
+
+    options = StreamableHTTPReconnectionOptions(
+        initial_reconnection_delay=0.5,
+        max_reconnection_delay=10.0,
+        reconnection_delay_grow_factor=1.2,
+        max_retries=3,
+    )
+
+    async with streamablehttp_client(
+        f"{basic_server_url}/mcp",
+        reconnection_options=options,
+    ) as (
+        read_stream,
+        write_stream,
+        _,
+    ):
+        async with ClientSession(read_stream, write_stream) as session:
+            result = await session.initialize()
+            assert isinstance(result, InitializeResult)