diff --git a/src/ModelContextProtocol.AspNetCore/StreamableHttpHandler.cs b/src/ModelContextProtocol.AspNetCore/StreamableHttpHandler.cs index f7db4630e..81fb8c5fa 100644 --- a/src/ModelContextProtocol.AspNetCore/StreamableHttpHandler.cs +++ b/src/ModelContextProtocol.AspNetCore/StreamableHttpHandler.cs @@ -243,6 +243,10 @@ await WriteJsonRpcErrorAsync(context, // RequestAborted always triggers when the client disconnects before a complete response body is written, // but this is how SSE connections are typically closed. } + finally + { + session.EndGetRequest(); + } } #pragma warning disable MCP9006 // Stateful Streamable HTTP resumability types are obsolete but still wired up internally. diff --git a/src/ModelContextProtocol.AspNetCore/StreamableHttpSession.cs b/src/ModelContextProtocol.AspNetCore/StreamableHttpSession.cs index 5065ddcfb..786d0644a 100644 --- a/src/ModelContextProtocol.AspNetCore/StreamableHttpSession.cs +++ b/src/ModelContextProtocol.AspNetCore/StreamableHttpSession.cs @@ -100,6 +100,7 @@ public async ValueTask EnsureStartedAsync(CancellationToken cancellationToken) } public bool TryStartGetRequest() => Interlocked.Exchange(ref _getRequestStarted, 1) == 0; + public void EndGetRequest() => Volatile.Write(ref _getRequestStarted, 0); public bool HasSameUserId(ClaimsPrincipal user) => userId == StreamableHttpHandler.GetUserIdClaim(user); public async ValueTask DisposeAsync() diff --git a/src/ModelContextProtocol.Core/Server/StreamableHttpServerTransport.cs b/src/ModelContextProtocol.Core/Server/StreamableHttpServerTransport.cs index 131c836dc..534da1653 100644 --- a/src/ModelContextProtocol.Core/Server/StreamableHttpServerTransport.cs +++ b/src/ModelContextProtocol.Core/Server/StreamableHttpServerTransport.cs @@ -185,6 +185,14 @@ public async Task HandleGetRequestAsync(Stream sseResponseStream, CancellationTo _httpSseWriter = null; writer.Dispose(); } + + // Event-store sessions retain this state so notifications can be persisted + // for Last-Event-ID replay while the client is disconnected. + if (_storeSseWriter is null) + { + _httpResponseTcs = null; + _getHttpRequestStarted = false; + } } } } diff --git a/tests/ModelContextProtocol.AspNetCore.Tests/MapMcpStreamableHttpTests.cs b/tests/ModelContextProtocol.AspNetCore.Tests/MapMcpStreamableHttpTests.cs index 889a7daab..104c0b85f 100644 --- a/tests/ModelContextProtocol.AspNetCore.Tests/MapMcpStreamableHttpTests.cs +++ b/tests/ModelContextProtocol.AspNetCore.Tests/MapMcpStreamableHttpTests.cs @@ -72,6 +72,82 @@ public async Task StreamableHttpMode_Works_WithRootEndpoint() Assert.Equal("StreamableHttpTestServer", mcpClient.ServerInfo.Name); } + [Fact] + public async Task GetStream_CanReconnectAfterPreviousRequestEnds() + { + Assert.SkipWhen(Stateless, "Unsolicited GET streams are not supported in stateless mode."); + + Builder.Services.AddMcpServer().WithHttpTransport(ConfigureStateless); + await using var app = Builder.Build(); + + var firstGetCompleted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var secondGetCompleted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var completedGetCount = 0; + app.Use(async (context, next) => + { + try + { + await next(context); + } + finally + { + if (context.Request.Method == HttpMethods.Get) + { + if (Interlocked.Increment(ref completedGetCount) == 1) + { + firstGetCompleted.TrySetResult(); + } + else + { + secondGetCompleted.TrySetResult(); + } + } + } + }); + app.MapMcp(); + await app.StartAsync(TestContext.Current.CancellationToken); + + const string initializeRequest = """ + {"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-11-25","capabilities":{},"clientInfo":{"name":"test-client","version":"1.0.0"}}} + """; + using var initRequest = new HttpRequestMessage(HttpMethod.Post, "/") + { + Content = new StringContent(initializeRequest, System.Text.Encoding.UTF8, "application/json"), + }; + initRequest.Headers.Accept.ParseAdd("application/json"); + initRequest.Headers.Accept.ParseAdd("text/event-stream"); + using var initResponse = await HttpClient.SendAsync(initRequest, TestContext.Current.CancellationToken); + Assert.True(initResponse.IsSuccessStatusCode); + var sessionId = Assert.Single(initResponse.Headers.GetValues("Mcp-Session-Id")); + + using (var firstResponse = await HttpClient.SendAsync( + CreateGetRequest(sessionId), + HttpCompletionOption.ResponseHeadersRead, + TestContext.Current.CancellationToken)) + { + Assert.Equal(HttpStatusCode.OK, firstResponse.StatusCode); + } + await firstGetCompleted.Task.WaitAsync(TestConstants.DefaultTimeout, TestContext.Current.CancellationToken); + + using (var secondResponse = await HttpClient.SendAsync( + CreateGetRequest(sessionId), + HttpCompletionOption.ResponseHeadersRead, + TestContext.Current.CancellationToken)) + { + Assert.Equal(HttpStatusCode.OK, secondResponse.StatusCode); + } + await secondGetCompleted.Task.WaitAsync(TestConstants.DefaultTimeout, TestContext.Current.CancellationToken); + + static HttpRequestMessage CreateGetRequest(string sessionId) + { + var request = new HttpRequestMessage(HttpMethod.Get, "/"); + request.Headers.Accept.ParseAdd("text/event-stream"); + request.Headers.Add("Mcp-Session-Id", sessionId); + request.Headers.Add("MCP-Protocol-Version", "2025-11-25"); + return request; + } + } + [Fact] public async Task AutoDetectMode_Works_WithRootEndpoint() { diff --git a/tests/ModelContextProtocol.Tests/Transport/StreamableHttpServerTransportTests.cs b/tests/ModelContextProtocol.Tests/Transport/StreamableHttpServerTransportTests.cs index ce2147e27..cfd846508 100644 --- a/tests/ModelContextProtocol.Tests/Transport/StreamableHttpServerTransportTests.cs +++ b/tests/ModelContextProtocol.Tests/Transport/StreamableHttpServerTransportTests.cs @@ -6,6 +6,18 @@ namespace ModelContextProtocol.Tests.Transport; public class StreamableHttpServerTransportTests(ITestOutputHelper testOutputHelper) : LoggedTest(testOutputHelper) { + [Fact] + public async Task HandleGetRequestAsync_CanStartAgainAfterPreviousRequestEnds() + { + await using var transport = new StreamableHttpServerTransport() + { + SessionId = "test-session", + }; + + await RunGetRequestAndCancelAsync(transport); + await RunGetRequestAndCancelAsync(transport); + } + [Fact] public async Task SendMessageAsync_AfterGetRequestEnds_DoesNotWriteToResponseStream() { @@ -43,6 +55,18 @@ await transport.SendMessageAsync( Assert.Equal(writeCountBeforeCancel, responseStream.WriteCount); } + private static async Task RunGetRequestAndCancelAsync(StreamableHttpServerTransport transport) + { + var responseStream = new RecordingStream(); + using var cts = new CancellationTokenSource(); + var getTask = transport.HandleGetRequestAsync(responseStream, cts.Token); + + await responseStream.FirstActivity.WaitAsync(TestConstants.DefaultTimeout, TestContext.Current.CancellationToken); + + cts.Cancel(); + await Assert.ThrowsAnyAsync(() => getTask); + } + private sealed class RecordingStream : Stream { private readonly TaskCompletionSource _firstActivity = new(TaskCreationOptions.RunContinuationsAsynchronously);