Server-side resumability via Last-Event-ID

Author: MackinnonBuck

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

@@ -38,12 +38,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/McpSessionHandler.cs

@@ -29,16 +29,32 @@ internal sealed partial class McpSessionHandler : IAsyncDisposable
         "mcp.server.operation.duration", "Measures the duration of inbound message processing.", longBuckets: false);
 
     /// <summary>The latest version of the protocol supported by this implementation.</summary>
-    internal const string LatestProtocolVersion = "2025-06-18";
+    internal const string LatestProtocolVersion = "2025-11-25";
 
     /// <summary>All protocol versions supported by this implementation.</summary>
     internal static readonly string[] SupportedProtocolVersions =
     [
         "2024-11-05",
         "2025-03-26",
+        "2025-06-18",
         LatestProtocolVersion,
     ];
 
+    /// <summary>
+    /// 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 SupportsPrimingEvent(string? protocolVersion)
+    {
+        const string MinResumabilityProtocolVersion = "2025-11-25";
+        return string.Compare(protocolVersion, MinResumabilityProtocolVersion, StringComparison.Ordinal) >= 0;
+    }
+
     private readonly bool _isServer;
     private readonly string _transportKind;
     private readonly ITransport _transport;

src/ModelContextProtocol.Core/Server/SseResponseStreamTransport.cs

@@ -67,7 +67,7 @@ public async Task SendMessageAsync(JsonRpcMessage message, CancellationToken can
     {
         Throw.IfNull(message);
         // If the underlying writer has been disposed, just drop the message.
-        await _sseWriter.SendMessageAsync(message, cancellationToken).ConfigureAwait(false);
+        await _sseWriter.SendMessageAsync(message, eventStreamWriter: null, cancellationToken).ConfigureAwait(false);
     }
 
     /// <summary>

src/ModelContextProtocol.Core/Server/SseWriter.cs

@@ -22,8 +22,6 @@ internal sealed class SseWriter(string? messageEndpoint = null, BoundedChannelOp
     private readonly SemaphoreSlim _disposeLock = new(1, 1);
     private bool _disposed;
 
-    public Func<IAsyncEnumerable<SseItem<JsonRpcMessage?>>, CancellationToken, IAsyncEnumerable<SseItem<JsonRpcMessage?>>>? MessageFilter { get; set; }
-
     public Task WriteAllAsync(Stream sseResponseStream, CancellationToken cancellationToken)
     {
         Throw.IfNull(sseResponseStream);
@@ -38,30 +36,48 @@ public Task WriteAllAsync(Stream sseResponseStream, CancellationToken cancellati
         _writeCancellationToken = cancellationToken;
 
         var messages = _messages.Reader.ReadAllAsync(cancellationToken);
-        if (MessageFilter is not null)
-        {
-            messages = MessageFilter(messages, cancellationToken);
-        }
-
         _writeTask = SseFormatter.WriteAsync(messages, sseResponseStream, WriteJsonRpcMessageToBuffer, cancellationToken);
         return _writeTask;
     }
 
-    public async Task<bool> SendMessageAsync(JsonRpcMessage message, CancellationToken cancellationToken = default)
+    public Task<bool> SendPrimingEventAsync(TimeSpan retryInterval, ISseEventStreamWriter eventStreamWriter, CancellationToken cancellationToken = default)
+    {
+        // Create a priming event: empty data with an event ID
+        var primingItem = new SseItem<JsonRpcMessage?>(null, "prime")
+        {
+            ReconnectionInterval = retryInterval,
+        };
+
+        return SendMessageAsync(primingItem, eventStreamWriter, cancellationToken);
+    }
+
+    public Task<bool> SendMessageAsync(JsonRpcMessage message, ISseEventStreamWriter? eventStreamWriter, CancellationToken cancellationToken = default)
     {
         Throw.IfNull(message);
 
+        // Emit redundant "event: message" lines for better compatibility with other SDKs.
+        return SendMessageAsync(new SseItem<JsonRpcMessage?>(message, SseParser.EventTypeDefault), eventStreamWriter, cancellationToken);
+    }
+
+    private async Task<bool> SendMessageAsync(SseItem<JsonRpcMessage?> item, ISseEventStreamWriter? eventStreamWriter, CancellationToken cancellationToken = default)
+    {
         using var _ = await _disposeLock.LockAsync(cancellationToken).ConfigureAwait(false);
 
+        if (eventStreamWriter is not null && item.EventId is null)
+        {
+            // Store the event first, even if the underlying writer has completed, so that
+            // messages can still be retrieved from the event store.
+            item = await eventStreamWriter.WriteEventAsync(item, cancellationToken: cancellationToken).ConfigureAwait(false);
+        }
+
         if (_disposed)
         {
             // Don't throw ObjectDisposedException here; just return false to indicate the message wasn't sent.
             // The calling transport can determine what to do in this case (drop the message, or fall back to another transport).
             return false;
         }
 
-        // Emit redundant "event: message" lines for better compatibility with other SDKs.
-        await _messages.Writer.WriteAsync(new SseItem<JsonRpcMessage?>(message, SseParser.EventTypeDefault), cancellationToken).ConfigureAwait(false);
+        await _messages.Writer.WriteAsync(item, cancellationToken).ConfigureAwait(false);
         return true;
     }
 
@@ -101,7 +117,10 @@ private void WriteJsonRpcMessageToBuffer(SseItem<JsonRpcMessage?> item, IBufferW
             return;
         }
 
-        JsonSerializer.Serialize(GetUtf8JsonWriter(writer), item.Data, McpJsonUtilities.JsonContext.Default.JsonRpcMessage!);
+        if (item.Data is not null)
+        {
+            JsonSerializer.Serialize(GetUtf8JsonWriter(writer), item.Data, McpJsonUtilities.JsonContext.Default.JsonRpcMessage!);
+        }
     }
 
     private Utf8JsonWriter GetUtf8JsonWriter(IBufferWriter<byte> writer)
@@ -118,3 +137,4 @@ private Utf8JsonWriter GetUtf8JsonWriter(IBufferWriter<byte> writer)
         return _jsonWriter;
     }
 }
+