Streamable HTTP resumability + redelivery + SSE polling via server-side disconnect

src/ModelContextProtocol.AspNetCore/HttpServerTransportOptions.cs

@@ -43,6 +43,34 @@ public class HttpServerTransportOptions
     /// </remarks>
     public bool Stateless { get; set; }
 
+    /// <summary>
+    /// Gets or sets the event store for resumability support.
+    /// When set, events are stored and can be replayed when clients reconnect with a Last-Event-ID header.
+    /// </summary>
+    /// <remarks>
+    /// When configured, the server will:
+    /// <list type="bullet">
+    /// <item><description>Generate unique event IDs for each SSE message</description></item>
+    /// <item><description>Store events for later replay</description></item>
+    /// <item><description>Replay missed events when a client reconnects with a Last-Event-ID header</description></item>
+    /// <item><description>Send priming events to establish resumability before any actual messages</description></item>
+    /// </list>
+    /// </remarks>
+    public ISseEventStreamStore? EventStreamStore { 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 1 second.
+    /// </value>
+    /// <remarks>
+    /// When <see cref="EventStreamStore"/> 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; } = TimeSpan.FromSeconds(1);
+
     /// <summary>
     /// Gets or sets a value that indicates whether the server uses a single execution context for the entire session.
     /// </summary>

src/ModelContextProtocol.AspNetCore/McpEndpointRouteBuilderExtensions.cs

@@ -39,12 +39,12 @@ public static IEndpointConventionBuilder MapMcp(this IEndpointRouteBuilder endpo
             .WithMetadata(new ProducesResponseTypeMetadata(StatusCodes.Status200OK, contentTypes: ["text/event-stream"]))
             .WithMetadata(new ProducesResponseTypeMetadata(StatusCodes.Status202Accepted));
 
+        streamableHttpGroup.MapGet("", streamableHttpHandler.HandleGetRequestAsync)
+            .WithMetadata(new ProducesResponseTypeMetadata(StatusCodes.Status200OK, contentTypes: ["text/event-stream"]));
+
         if (!streamableHttpHandler.HttpServerTransportOptions.Stateless)
         {
-            // The GET and DELETE endpoints are not mapped in Stateless mode since there's no way to send unsolicited messages
-            // for the GET to handle, and there is no server-side state for the DELETE to clean up.
-            streamableHttpGroup.MapGet("", streamableHttpHandler.HandleGetRequestAsync)
-                .WithMetadata(new ProducesResponseTypeMetadata(StatusCodes.Status200OK, contentTypes: ["text/event-stream"]));
+            // The DELETE endpoints are not mapped in Stateless mode since there's no server-side state to clean up.
             streamableHttpGroup.MapDelete("", streamableHttpHandler.HandleDeleteRequestAsync);
 
             // Map legacy HTTP with SSE endpoints only if not in Stateless mode, because we cannot guarantee the /message requests

src/ModelContextProtocol.AspNetCore/StreamableHttpHandler.cs

@@ -23,6 +23,7 @@ internal sealed class StreamableHttpHandler(
     ILoggerFactory loggerFactory)
 {
     private const string McpSessionIdHeaderName = "Mcp-Session-Id";
+    private const string LastEventIdHeaderName = "Last-Event-ID";
 
     private static readonly JsonTypeInfo<JsonRpcMessage> s_messageTypeInfo = GetRequiredJsonTypeInfo<JsonRpcMessage>();
     private static readonly JsonTypeInfo<JsonRpcError> s_errorTypeInfo = GetRequiredJsonTypeInfo<JsonRpcError>();
@@ -81,17 +82,80 @@ await WriteJsonRpcErrorAsync(context,
             return;
         }
 
+        StreamableHttpSession? session = null;
+        ISseEventStreamReader? eventStreamReader = null;
+
         var sessionId = context.Request.Headers[McpSessionIdHeaderName].ToString();
-        var session = await GetSessionAsync(context, sessionId);
+        var lastEventId = context.Request.Headers[LastEventIdHeaderName].ToString();
+
+        if (!string.IsNullOrEmpty(sessionId))
+        {
+            session = await GetSessionAsync(context, sessionId);
+            if (session is null)
+            {
+                // There was an error obtaining the session; consider the request failed.
+                return;
+            }
+        }
+
+        if (!string.IsNullOrEmpty(lastEventId))
+        {
+            if (HttpServerTransportOptions.Stateless)
+            {
+                await WriteJsonRpcErrorAsync(context,
+                    "Bad Request: The Last-Event-ID header is not supported in stateless mode.",
+                    StatusCodes.Status400BadRequest);
+                return;
+            }
+
+            eventStreamReader = await GetEventStreamReaderAsync(context, lastEventId);
+            if (eventStreamReader is null)
+            {
+                // There was an error obtaining the event stream; consider the request failed.
+                return;
+            }
+        }
+
+        if (session is not null && eventStreamReader is not null && !string.Equals(session.Id, eventStreamReader.SessionId, StringComparison.Ordinal))
+        {
+            await WriteJsonRpcErrorAsync(context,
+                "Bad Request: The Last-Event-ID header refers to a session with a different session ID.",
+                StatusCodes.Status400BadRequest);
+            return;
+        }
+
+        if (eventStreamReader is null || string.Equals(eventStreamReader.StreamId, StreamableHttpServerTransport.UnsolicitedMessageStreamId, StringComparison.Ordinal))
+        {
+            await HandleUnsolicitedMessageStreamAsync(context, session, eventStreamReader);
+        }
+        else
+        {
+            await HandleResumePostResponseStreamAsync(context, eventStreamReader);
+        }
+    }
+
+    private async Task HandleUnsolicitedMessageStreamAsync(HttpContext context, StreamableHttpSession? session, ISseEventStreamReader? eventStreamReader)
+    {
+        if (HttpServerTransportOptions.Stateless)
+        {
+            await WriteJsonRpcErrorAsync(context,
+                "Bad Request: Unsolicited messages are not supported in stateless mode.",
+                StatusCodes.Status400BadRequest);
+            return;
+        }
+
         if (session is null)
         {
+            await WriteJsonRpcErrorAsync(context,
+                "Bad Request: Mcp-Session-Id header is required",
+                StatusCodes.Status400BadRequest);
             return;
         }
 
-        if (!session.TryStartGetRequest())
+        if (eventStreamReader is null && !session.TryStartGetRequest())
         {
             await WriteJsonRpcErrorAsync(context,
-                "Bad Request: This server does not support multiple GET requests. Start a new session to get a new GET SSE response.",
+                "Bad Request: This server does not support multiple GET requests. Use Last-Event-ID header to resume or start a new session.",
                 StatusCodes.Status400BadRequest);
             return;
         }
@@ -111,7 +175,7 @@ await WriteJsonRpcErrorAsync(context,
             // will be sent in response to a different POST request. It might be a while before we send a message
             // over this response body.
             await context.Response.Body.FlushAsync(cancellationToken);
-            await session.Transport.HandleGetRequestAsync(context.Response.Body, cancellationToken);
+            await session.Transport.HandleGetRequestAsync(context.Response.Body, eventStreamReader, cancellationToken);
         }
         catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
         {
@@ -120,6 +184,12 @@ await WriteJsonRpcErrorAsync(context,
         }
     }
 
+    private static async Task HandleResumePostResponseStreamAsync(HttpContext context, ISseEventStreamReader eventStreamReader)
+    {
+        InitializeSseResponse(context);
+        await eventStreamReader.CopyToAsync(context.Response.Body, context.RequestAborted);
+    }
+
     public async Task HandleDeleteRequestAsync(HttpContext context)
     {
         var sessionId = context.Request.Headers[McpSessionIdHeaderName].ToString();
@@ -131,14 +201,7 @@ public async Task HandleDeleteRequestAsync(HttpContext context)
 
     private async ValueTask<StreamableHttpSession?> GetSessionAsync(HttpContext context, string sessionId)
     {
-        StreamableHttpSession? session;
-
-        if (string.IsNullOrEmpty(sessionId))
-        {
-            await WriteJsonRpcErrorAsync(context, "Bad Request: Mcp-Session-Id header is required", StatusCodes.Status400BadRequest);
-            return null;
-        }
-        else if (!sessionManager.TryGetValue(sessionId, out session))
+        if (!sessionManager.TryGetValue(sessionId, out var session))
         {
             // -32001 isn't part of the MCP standard, but this is what the typescript-sdk currently does.
             // One of the few other usages I found was from some Ethereum JSON-RPC documentation and this
@@ -194,12 +257,16 @@ private async ValueTask<StreamableHttpSession> StartNewSessionAsync(HttpContext
             {
                 SessionId = sessionId,
                 FlowExecutionContextFromRequests = !HttpServerTransportOptions.PerSessionExecutionContext,
+                EventStreamStore = HttpServerTransportOptions.EventStreamStore,
+                RetryInterval = HttpServerTransportOptions.RetryInterval,
             };
             context.Response.Headers[McpSessionIdHeaderName] = sessionId;
         }
         else
         {
             // In stateless mode, each request is independent. Don't set any session ID on the transport.
+            // If in the future we support resuming stateless requests, we should populate
+            // the event stream store and retry interval here as well.
             sessionId = "";
             transport = new()
             {
@@ -246,6 +313,28 @@ private async ValueTask<StreamableHttpSession> CreateSessionAsync(
         return session;
     }
 
+    private async ValueTask<ISseEventStreamReader?> GetEventStreamReaderAsync(HttpContext context, string lastEventId)
+    {
+        if (HttpServerTransportOptions.EventStreamStore is not { } eventStreamStore)
+        {
+            await WriteJsonRpcErrorAsync(context,
+                "Bad Request: This server does not support resuming streams.",
+                StatusCodes.Status400BadRequest);
+            return null;
+        }
+
+        var eventStreamReader = await eventStreamStore.GetStreamReaderAsync(lastEventId, context.RequestAborted);
+        if (eventStreamReader is null)
+        {
+            await WriteJsonRpcErrorAsync(context,
+                "Bad Request: The specified Last-Event-ID is either invalid or expired.",
+                StatusCodes.Status400BadRequest);
+            return null;
+        }
+
+        return eventStreamReader;
+    }
+
     private static Task WriteJsonRpcErrorAsync(HttpContext context, string errorMessage, int statusCode, int errorCode = -32000)
     {
         var jsonRpcError = new JsonRpcError

src/ModelContextProtocol.Core/Client/HttpClientTransportOptions.cs

@@ -108,4 +108,17 @@ public required Uri Endpoint
     /// Gets sor sets the authorization provider to use for authentication.
     /// </summary>
     public ClientOAuthOptions? OAuth { get; set; }
+
+    /// <summary>
+    /// Gets or sets the maximum number of reconnection attempts when an SSE stream is disconnected.
+    /// </summary>
+    /// <value>
+    /// The maximum number of reconnection attempts. The default is 2.
+    /// </value>
+    /// <remarks>
+    /// When an SSE stream is disconnected (e.g., due to a network issue), the client will attempt to
+    /// reconnect using the Last-Event-ID header to resume from where it left off. This property controls
+    /// how many reconnection attempts are made before giving up.
+    /// </remarks>
+    public int MaxReconnectionAttempts { get; set; } = 2;
 }