[WIP] PR feedback + cleanups

Author: MackinnonBuck

src/ModelContextProtocol.AspNetCore/HttpServerTransportOptions.cs

@@ -56,20 +56,20 @@ public class HttpServerTransportOptions
     /// <item><description>Send priming events to establish resumability before any actual messages</description></item>
     /// </list>
     /// </remarks>
-    public IEventStore? EventStore { get; set; }
+    public ISseEventStore? EventStore { get; set; }
 
     /// <summary>
     /// Gets or sets the retry interval to suggest to clients in SSE retry field.
     /// </summary>
     /// <value>
-    /// The retry interval. The default is <see langword="null"/>, meaning no retry field is sent.
+    /// The retry interval. The default is 5 seconds.
     /// </value>
     /// <remarks>
-    /// When set along with <see cref="EventStore"/>, the server will include a retry field in priming events.
-    /// This suggests to clients how long to wait before attempting to reconnect after a connection is lost.
+    /// When <see cref="EventStore"/> is set, the server will include a retry field in priming events.
+    /// This value suggests to clients how long to wait before attempting to reconnect after a connection is lost.
     /// Clients may use this value to implement polling behavior during long-running operations.
     /// </remarks>
-    public TimeSpan? RetryInterval { get; set; }
+    public TimeSpan RetryInterval { get; set; } = TimeSpan.FromSeconds(5);
 
     /// <summary>
     /// Gets or sets a value that indicates whether the server uses a single execution context for the entire session.

src/ModelContextProtocol.Core/McpSessionHandler.cs

@@ -41,15 +41,15 @@ internal sealed partial class McpSessionHandler : IAsyncDisposable
     ];
 
     /// <summary>
-    /// Checks if the given protocol version supports resumability (priming events).
+    /// Checks if the given protocol version supports priming events.
     /// </summary>
     /// <param name="protocolVersion">The protocol version to check.</param>
     /// <returns>True if the protocol version supports resumability.</returns>
     /// <remarks>
     /// Priming events are only supported in protocol version &gt;= 2025-11-25.
     /// Older clients may crash when receiving SSE events with empty data.
     /// </remarks>
-    internal static bool SupportsResumability(string? protocolVersion)
+    internal static bool SupportsPrimingEvent(string? protocolVersion)
     {
         const string MinResumabilityProtocolVersion = "2025-11-25";
         return string.Compare(protocolVersion, MinResumabilityProtocolVersion, StringComparison.Ordinal) >= 0;

src/ModelContextProtocol.Core/Protocol/JsonRpcMessageContext.cs

@@ -1,6 +1,5 @@
 using ModelContextProtocol.Server;
 using System.Security.Claims;
-using System.Text.Json.Serialization;
 
 namespace ModelContextProtocol.Protocol;
 
@@ -58,45 +57,4 @@ public class JsonRpcMessageContext
     /// </para>
     /// </remarks>
     public ClaimsPrincipal? User { get; set; }
-
-    /// <summary>
-    /// Gets or sets a callback that closes the SSE stream associated with the current JSON-RPC request.
-    /// </summary>
-    /// <remarks>
-    /// <para>
-    /// This callback implements the SSE polling pattern from SEP-1699. When invoked, it gracefully closes
-    /// the SSE stream for the current request, signaling to the client that it should reconnect to receive
-    /// the response. The server must have sent a priming event with an event ID before this callback is invoked.
-    /// </para>
-    /// <para>
-    /// This is useful for long-running operations where the server wants to free resources while the operation
-    /// is in progress. The client will reconnect with the Last-Event-ID header, and the server will replay
-    /// any events that were sent after that ID.
-    /// </para>
-    /// <para>
-    /// This callback is only available when using the Streamable HTTP transport with resumability enabled.
-    /// For other transports, this property will be <see langword="null"/>.
-    /// </para>
-    /// </remarks>
-    public Action? CloseSseStream { get; set; }
-
-    /// <summary>
-    /// Gets or sets a callback that closes the standalone SSE stream for the current session.
-    /// </summary>
-    /// <remarks>
-    /// <para>
-    /// This callback closes the standalone SSE stream that is used for server-initiated messages
-    /// (notifications and requests from server to client). Unlike <see cref="CloseSseStream"/>,
-    /// this affects the session-level SSE stream rather than a request-specific stream.
-    /// </para>
-    /// <para>
-    /// This is useful when the server needs to signal the client to reconnect its standalone SSE stream,
-    /// for example during server restarts or resource cleanup.
-    /// </para>
-    /// <para>
-    /// This callback is only available when using the Streamable HTTP transport.
-    /// For other transports, this property will be <see langword="null"/>.
-    /// </para>
-    /// </remarks>
-    public Action? CloseStandaloneSseStream { get; set; }
 }

…ntextProtocol.Core/Server/IEventStore.cs …xtProtocol.Core/Server/ISseEventStore.cssrc/ModelContextProtocol.Core/Server/IEventStore.cs renamed to src/ModelContextProtocol.Core/Server/ISseEventStore.cs src/ModelContextProtocol.Core/Server/IEventStore.cs renamed to src/ModelContextProtocol.Core/Server/ISseEventStore.cs

@@ -19,11 +19,14 @@ namespace ModelContextProtocol.Server;
 /// Implementations should be thread-safe, as events may be stored and replayed concurrently.
 /// </para>
 /// </remarks>
-public interface IEventStore
+public interface ISseEventStore
 {
     /// <summary>
     /// Stores an event for later retrieval.
     /// </summary>
+    /// <param name="sessionId">
+    /// The ID of the session, or <c>null</c>.
+    /// </param>
     /// <param name="streamId">
     /// The ID of the stream the event belongs to. This is typically the JSON-RPC request ID
     /// for POST SSE responses, or a special identifier for the standalone GET SSE stream.
@@ -34,23 +37,18 @@ public interface IEventStore
     /// </param>
     /// <param name="cancellationToken">A token to cancel the operation.</param>
     /// <returns>The generated event ID for the stored event.</returns>
-    ValueTask<string> StoreEventAsync(string streamId, JsonRpcMessage? message, CancellationToken cancellationToken = default);
+    ValueTask<string> StoreEventAsync(string sessionId, string streamId, JsonRpcMessage? message, CancellationToken cancellationToken = default);
 
     /// <summary>
     /// Replays events that occurred after the specified event ID.
     /// </summary>
     /// <param name="lastEventId">The ID of the last event the client received.</param>
-    /// <param name="sendCallback">
-    /// A callback function to send each replayed event to the client.
-    /// The callback receives the message and its event ID.
-    /// </param>
     /// <param name="cancellationToken">A token to cancel the operation.</param>
     /// <returns>
-    /// The stream ID of the replayed events if the event ID was found and events were replayed;
+    /// An <see cref="SseReplayResult"/> containing the events to replay if the event ID was found;
     /// <see langword="null"/> if the event ID was not found in the store.
     /// </returns>
-    ValueTask<string?> ReplayEventsAfterAsync(
+    ValueTask<SseReplayResult?> GetEventsAfterAsync(
         string lastEventId,
-        Func<JsonRpcMessage, string, CancellationToken, ValueTask> sendCallback,
         CancellationToken cancellationToken = default);
 }

src/ModelContextProtocol.Core/Server/RequestContext.cs

@@ -81,52 +81,4 @@ public McpServer Server
     /// including the method name, parameters, request ID, and associated transport and user information.
     /// </remarks>
     public JsonRpcRequest JsonRpcRequest { get; }
-
-    /// <summary>
-    /// Closes the SSE stream for the current request, signaling to the client that it should reconnect.
-    /// </summary>
-    /// <remarks>
-    /// <para>
-    /// This method implements the SSE polling pattern from SEP-1699. When called, it gracefully closes
-    /// the SSE stream for the current request. For this to work correctly, the server must have sent
-    /// a priming event with an event ID before this method is called (which happens automatically when
-    /// resumability is enabled via <see cref="StreamableHttpServerTransport.EventStore"/>).
-    /// </para>
-    /// <para>
-    /// After calling this method, the client will receive a stream end and should reconnect with the
-    /// Last-Event-ID header. The server will then replay any events that were sent after that ID
-    /// and continue streaming new events.
-    /// </para>
-    /// <para>
-    /// This method only has an effect when using the Streamable HTTP transport with resumability enabled.
-    /// For other transports or when resumability is not configured, this method does nothing.
-    /// </para>
-    /// </remarks>
-    public void CloseSseStream()
-    {
-        JsonRpcRequest.Context?.CloseSseStream?.Invoke();
-    }
-
-    /// <summary>
-    /// Closes the standalone SSE stream for server-initiated messages.
-    /// </summary>
-    /// <remarks>
-    /// <para>
-    /// This method closes the standalone SSE stream that is used for server-initiated messages
-    /// (notifications and requests from server to client). Unlike <see cref="CloseSseStream"/>,
-    /// this affects the session-level SSE stream (from GET requests) rather than a request-specific stream.
-    /// </para>
-    /// <para>
-    /// This is useful when the server needs to signal the client to reconnect its standalone SSE stream,
-    /// for example during server restarts or resource cleanup.
-    /// </para>
-    /// <para>
-    /// This method only has an effect when using the Streamable HTTP transport.
-    /// For other transports, this method does nothing.
-    /// </para>
-    /// </remarks>
-    public void CloseStandaloneSseStream()
-    {
-        JsonRpcRequest.Context?.CloseStandaloneSseStream?.Invoke();
-    }
 }

src/ModelContextProtocol.Core/Server/SseReplayResult.cs

@@ -0,0 +1,33 @@
+using ModelContextProtocol.Protocol;
+
+namespace ModelContextProtocol.Server;
+
+/// <summary>
+/// Represents the result of replaying SSE events to a client during resumption.
+/// </summary>
+/// <remarks>
+/// This class is returned by <see cref="ISseEventStore.GetEventsAfterAsync"/> when a client
+/// reconnects with a <c>Last-Event-ID</c> header. It contains the stream and session identifiers
+/// along with an async enumerable of events to replay.
+/// </remarks>
+public sealed class SseReplayResult
+{
+    /// <summary>
+    /// Gets the session ID that the events belong to.
+    /// </summary>
+    public required string SessionId { get; init; }
+
+    /// <summary>
+    /// Gets the stream ID that the events belong to.
+    /// </summary>
+    /// <remarks>
+    /// This is typically the JSON-RPC request ID for POST SSE responses,
+    /// or a special identifier for the standalone GET SSE stream.
+    /// </remarks>
+    public required string StreamId { get; init; }
+
+    /// <summary>
+    /// Gets the async enumerable of events to replay to the client.
+    /// </summary>
+    public required IAsyncEnumerable<StoredSseEvent> Events { get; init; }
+}

src/ModelContextProtocol.Core/Server/SseStreamEventStore.cs

@@ -0,0 +1,47 @@
+using ModelContextProtocol.Protocol;
+
+namespace ModelContextProtocol.Server;
+
+/// <summary>
+/// Wraps an <see cref="ISseEventStore"/> with session and stream context for a specific SSE stream.
+/// </summary>
+/// <remarks>
+/// This class simplifies event storage by binding the session ID, stream ID, and retry interval
+/// so that callers only need to provide the message when storing events.
+/// </remarks>
+internal sealed class SseStreamEventStore
+{
+    private readonly ISseEventStore _eventStore;
+    private readonly string _sessionId;
+    private readonly string _streamId;
+    private readonly TimeSpan _retryInterval;
+
+    /// <summary>
+    /// Gets the retry interval to suggest to clients in SSE retry field.
+    /// </summary>
+    public TimeSpan RetryInterval => _retryInterval;
+
+    /// <summary>
+    /// Initializes a new instance of the <see cref="SseStreamEventStore"/> class.
+    /// </summary>
+    /// <param name="eventStore">The underlying event store to use for storage.</param>
+    /// <param name="sessionId">The session ID, or <see langword="null"/> to generate a new one.</param>
+    /// <param name="streamId">The stream ID for this SSE stream.</param>
+    /// <param name="retryInterval">The retry interval to suggest to clients.</param>
+    public SseStreamEventStore(ISseEventStore eventStore, string? sessionId, string streamId, TimeSpan retryInterval)
+    {
+        _eventStore = eventStore;
+        _sessionId = sessionId ?? Guid.NewGuid().ToString("N");
+        _streamId = streamId;
+        _retryInterval = retryInterval;
+    }
+
+    /// <summary>
+    /// Stores an event in the underlying event store with the bound session and stream context.
+    /// </summary>
+    /// <param name="message">The JSON-RPC message to store, or <see langword="null"/> for priming events.</param>
+    /// <param name="cancellationToken">A token to cancel the operation.</param>
+    /// <returns>The generated event ID for the stored event.</returns>
+    public ValueTask<string> StoreEventAsync(JsonRpcMessage? message, CancellationToken cancellationToken = default)
+        => _eventStore.StoreEventAsync(_sessionId, _streamId, message, cancellationToken);
+}