From 755707aa33d4ce3cc67bee4f4f8bcb611aa6a7bd Mon Sep 17 00:00:00 2001 From: Loi Nguyen Date: Mon, 20 Jul 2026 04:26:31 +0700 Subject: [PATCH] feat: add JSON response mode for Streamable HTTP --- docs/concepts/stateless/stateless.md | 1 + docs/concepts/transports/transports.md | 17 +- .../HttpServerTransportOptions.cs | 15 ++ .../StreamableHttpHandler.cs | 13 +- .../Server/StreamableHttpPostTransport.cs | 72 +++++-- .../Server/StreamableHttpServerTransport.cs | 14 ++ .../JsonResponseModeTests.cs | 201 ++++++++++++++++++ 7 files changed, 316 insertions(+), 17 deletions(-) create mode 100644 tests/ModelContextProtocol.AspNetCore.Tests/JsonResponseModeTests.cs diff --git a/docs/concepts/stateless/stateless.md b/docs/concepts/stateless/stateless.md index 69cf767b3..f48a04b5d 100644 --- a/docs/concepts/stateless/stateless.md +++ b/docs/concepts/stateless/stateless.md @@ -418,6 +418,7 @@ builder.Services.AddMcpServer() | Property | Type | Default | Description | |----------|------|---------|-------------| | | `bool` | `true` | Enables stateless mode. No sessions, no `Mcp-Session-Id` header, no server-to-client requests on the legacy protocol. Required by the `2026-07-28` protocol revision. | +| | `bool` | `false` | Returns the final JSON-RPC response as `application/json` instead of an SSE stream. Intermediate request-related messages, POST resumability, and polling are unavailable in this mode. | | | `TimeSpan` | 2 hours | _Stateful only (`MCP9006`)._ Duration of inactivity before a session is closed. Checked every 5 seconds. | | | `int` | 10,000 | _Stateful only (`MCP9006`)._ Maximum idle sessions before the oldest are forcibly terminated. | | | `Func?` | `null` | Per-session callback to customize `McpServerOptions` with access to `HttpContext`. In stateless mode (including all `2026-07-28` requests), this runs on every HTTP request. | diff --git a/docs/concepts/transports/transports.md b/docs/concepts/transports/transports.md index 8fb2af0c7..9aaa9b61d 100644 --- a/docs/concepts/transports/transports.md +++ b/docs/concepts/transports/transports.md @@ -185,6 +185,21 @@ app.Run(); By default, the HTTP transport runs **statelessly** — the server does not assign an `Mcp-Session-Id` or track transport session state in memory. This simplifies deployment, enables horizontal scaling without session affinity, and matches the `2026-07-28` Streamable HTTP wire format. Set `Stateless = false` explicitly when your server needs stateful sessions for unsolicited notifications, resource subscriptions, or per-client isolation. See [Sessions](xref:stateless) for a detailed guide on when to use stateless vs. stateful mode, configure session options, and understand [cancellation and disposal](xref:stateless#cancellation-and-disposal) behavior during shutdown. +#### JSON response mode + +By default, each Streamable HTTP POST returns an SSE stream so the server can send progress notifications and other intermediate messages before the final JSON-RPC response. Set to `true` when an intermediary, such as a web application firewall or reverse proxy, cannot pass SSE responses: + +```csharp +builder.Services.AddMcpServer() + .WithHttpTransport(options => + { + options.EnableJsonResponse = true; + }) + .WithTools(); +``` + +In this mode, POST requests return the final JSON-RPC response directly with an `application/json` content type. Request-related progress notifications and other intermediate messages are omitted, event-store resumability and polling are unavailable for POST responses, and notification-only POST requests still return an empty `202 Accepted` response. Stateful servers can still use the standalone GET SSE stream for unsolicited messages. + #### Host name validation For local HTTP servers, keep the set of accepted host names limited to loopback values. This helps protect against DNS rebinding, where a browser reaches a local server through an attacker-controlled DNS name while sending that DNS name in the HTTP `Host` header. ASP.NET Core's Kestrel server doesn't validate `Host` headers by default, so configure `AllowedHosts` with known host names rather than `"*"`. This also avoids reflecting untrusted host names through ASP.NET Core features such as absolute URL generation. See [Host filtering with ASP.NET Core Kestrel web server | Microsoft Learn](https://learn.microsoft.com/aspnet/core/fundamentals/servers/kestrel/host-filtering) and [URL generation concepts | Microsoft Learn](https://learn.microsoft.com/aspnet/core/fundamentals/routing#url-generation-concepts). @@ -245,7 +260,7 @@ app.MapMcp("/mcp").RequireCors("McpBrowserClient"); #### How messages flow -In Streamable HTTP, client requests arrive as HTTP POST requests. The server holds each POST response body open as an SSE stream and writes the JSON-RPC response — plus any intermediate messages like progress notifications or server-to-client requests — back through it. This provides natural HTTP-level backpressure: each POST holds its connection until the handler completes. +In Streamable HTTP, client requests arrive as HTTP POST requests. By default, the server holds each POST response body open as an SSE stream and writes the JSON-RPC response — plus any intermediate messages like progress notifications or server-to-client requests — back through it. JSON response mode instead returns only the final response as `application/json`. Both modes provide natural HTTP-level backpressure because each POST remains open until the handler completes. In stateful mode, the client can also open a long-lived GET request to receive **unsolicited** messages — notifications or server-to-client requests that the server initiates outside any active request handler (e.g., resource-changed notifications from a background watcher). In stateless mode, the GET endpoint is not mapped, so every message must be part of a POST response. See [How Streamable HTTP delivers messages](xref:stateless#how-streamable-http-delivers-messages) for a detailed breakdown. diff --git a/src/ModelContextProtocol.AspNetCore/HttpServerTransportOptions.cs b/src/ModelContextProtocol.AspNetCore/HttpServerTransportOptions.cs index 024772240..108df6ab3 100644 --- a/src/ModelContextProtocol.AspNetCore/HttpServerTransportOptions.cs +++ b/src/ModelContextProtocol.AspNetCore/HttpServerTransportOptions.cs @@ -72,6 +72,21 @@ public class HttpServerTransportOptions /// public bool Stateless { get; set; } = true; + /// + /// Gets or sets a value that indicates whether Streamable HTTP POST requests return a single JSON response + /// instead of an SSE stream. + /// + /// + /// to return the final JSON-RPC response as application/json; + /// to use an SSE response stream. The default is . + /// + /// + /// JSON response mode is intended for simple request/response scenarios where intermediaries do not support SSE. + /// Request-related notifications and other intermediate messages are not included in the response. Standalone GET + /// requests continue to use SSE when stateful mode is enabled. + /// + public bool EnableJsonResponse { get; set; } + /// /// Gets or sets a value that indicates whether the server maps legacy SSE endpoints (/sse and /message) /// for backward compatibility with clients that do not support the Streamable HTTP transport. diff --git a/src/ModelContextProtocol.AspNetCore/StreamableHttpHandler.cs b/src/ModelContextProtocol.AspNetCore/StreamableHttpHandler.cs index f7db4630e..810fe2f67 100644 --- a/src/ModelContextProtocol.AspNetCore/StreamableHttpHandler.cs +++ b/src/ModelContextProtocol.AspNetCore/StreamableHttpHandler.cs @@ -124,7 +124,15 @@ await WriteJsonRpcErrorAsync(context, await using var _ = await session.AcquireReferenceAsync(context.RequestAborted); - InitializeSseResponse(context); + if (session.Transport.EnableJsonResponse) + { + context.Response.ContentType = "application/json"; + } + else + { + InitializeSseResponse(context); + } + var wroteResponse = await session.Transport.HandlePostRequestAsync(message, context.Response.Body, context.RequestAborted); if (!wroteResponse) { @@ -449,6 +457,7 @@ private async ValueTask StartNewSessionAsync(HttpContext transport = new(loggerFactory) { SessionId = sessionId, + EnableJsonResponse = HttpServerTransportOptions.EnableJsonResponse, FlowExecutionContextFromRequests = !HttpServerTransportOptions.PerSessionExecutionContext, EventStreamStore = HttpServerTransportOptions.EventStreamStore, OnSessionInitialized = HttpServerTransportOptions.SessionMigrationHandler is { } handler @@ -468,6 +477,7 @@ private async ValueTask StartNewSessionAsync(HttpContext transport = new(loggerFactory) { Stateless = true, + EnableJsonResponse = HttpServerTransportOptions.EnableJsonResponse, }; } @@ -524,6 +534,7 @@ private async ValueTask MigrateSessionAsync( var transport = new StreamableHttpServerTransport(loggerFactory) { SessionId = sessionId, + EnableJsonResponse = HttpServerTransportOptions.EnableJsonResponse, #pragma warning disable MCP9006 // Stateful Streamable HTTP options are obsolete but still wired up internally. FlowExecutionContextFromRequests = !HttpServerTransportOptions.PerSessionExecutionContext, EventStreamStore = HttpServerTransportOptions.EventStreamStore, diff --git a/src/ModelContextProtocol.Core/Server/StreamableHttpPostTransport.cs b/src/ModelContextProtocol.Core/Server/StreamableHttpPostTransport.cs index ed8c3293a..de9c16f6f 100644 --- a/src/ModelContextProtocol.Core/Server/StreamableHttpPostTransport.cs +++ b/src/ModelContextProtocol.Core/Server/StreamableHttpPostTransport.cs @@ -19,7 +19,7 @@ internal sealed partial class StreamableHttpPostTransport( { private readonly SemaphoreSlim _messageLock = new(1, 1); private readonly TaskCompletionSource _httpResponseTcs = new(TaskCreationOptions.RunContinuationsAsynchronously); - private readonly SseEventWriter _httpSseWriter = new(responseStream); + private readonly SseEventWriter? _httpSseWriter = parentTransport.EnableJsonResponse ? null : new(responseStream); private TaskCompletionSource? _storeStreamTcs; #pragma warning disable MCP9006 // Stateful Streamable HTTP resumability types are obsolete but still wired up internally. @@ -71,20 +71,23 @@ public async ValueTask HandlePostAsync(JsonRpcMessage message, Cancellatio using (await _messageLock.LockAsync(cancellationToken).ConfigureAwait(false)) { - var primingItem = await TryStartSseEventStreamAsync(_pendingRequest).ConfigureAwait(false); - if (primingItem.HasValue) + if (!parentTransport.EnableJsonResponse) { - await _httpSseWriter.WriteAsync(primingItem.Value, cancellationToken).ConfigureAwait(false); - } - else - { - // If there's no priming write, flush the stream to ensure HTTP response headers are - // sent to the client now that the server is ready to process the request. - // This prevents HttpClient timeout for long-running requests. - await responseStream.FlushAsync(cancellationToken).ConfigureAwait(false); + var primingItem = await TryStartSseEventStreamAsync(_pendingRequest).ConfigureAwait(false); + if (primingItem.HasValue) + { + await _httpSseWriter!.WriteAsync(primingItem.Value, cancellationToken).ConfigureAwait(false); + } + else + { + // If there's no priming write, flush the stream to ensure HTTP response headers are + // sent to the client now that the server is ready to process the request. + // This prevents HttpClient timeout for long-running requests. + await responseStream.FlushAsync(cancellationToken).ConfigureAwait(false); + } } - // Ensure that we've sent the priming event before processing the incoming request. + // In SSE mode, ensure that we've sent the priming event before processing the incoming request. await parentTransport.MessageWriter.WriteAsync(message, cancellationToken).ConfigureAwait(false); } @@ -108,6 +111,40 @@ public async Task SendMessageAsync(JsonRpcMessage message, CancellationToken can try { + if (parentTransport.EnableJsonResponse) + { + if (_finalResponseMessageSent) + { + return; + } + + if ((message is JsonRpcResponse or JsonRpcError) && ((JsonRpcMessageWithId)message).Id == _pendingRequest) + { + try + { + if (!_httpResponseCompleted) + { + await JsonSerializer.SerializeAsync( + responseStream, + message, + McpJsonUtilities.JsonContext.Default.JsonRpcMessage, + cancellationToken).ConfigureAwait(false); + } + } + catch (Exception ex) when (!cancellationToken.IsCancellationRequested) + { + _httpResponseTcs.TrySetException(ex); + } + finally + { + _finalResponseMessageSent = true; + _httpResponseTcs.TrySetResult(true); + } + } + + // JSON mode only returns the final correlated response. Intermediate messages are omitted. + return; + } if (_finalResponseMessageSent) { @@ -130,7 +167,7 @@ public async Task SendMessageAsync(JsonRpcMessage message, CancellationToken can try { - await _httpSseWriter.WriteAsync(item, cancellationToken).ConfigureAwait(false); + await _httpSseWriter!.WriteAsync(item, cancellationToken).ConfigureAwait(false); } catch (Exception ex) when (!cancellationToken.IsCancellationRequested) { @@ -152,6 +189,11 @@ public async Task SendMessageAsync(JsonRpcMessage message, CancellationToken can public async ValueTask EnablePollingAsync(TimeSpan retryInterval, CancellationToken cancellationToken) { + if (parentTransport.EnableJsonResponse) + { + throw new InvalidOperationException("Polling is not supported when JSON responses are enabled."); + } + if (parentTransport.Stateless) { throw new InvalidOperationException("Polling is not supported in stateless mode."); @@ -173,7 +215,7 @@ public async ValueTask EnablePollingAsync(TimeSpan retryInterval, CancellationTo // Write to the response stream if it still exists. if (!_httpResponseCompleted) { - await _httpSseWriter.WriteAsync(primingItem, cancellationToken).ConfigureAwait(false); + await _httpSseWriter!.WriteAsync(primingItem, cancellationToken).ConfigureAwait(false); } // Set the mode to 'Polling' so that the replay stream ends as soon as all available messages have been sent. @@ -240,7 +282,7 @@ public async ValueTask DisposeAsync() _httpResponseTcs.TrySetResult(true); - _httpSseWriter.Dispose(); + _httpSseWriter?.Dispose(); // Don't dispose the event stream writer here, as we may continue to write to the event store // after disposal if there are pending messages. diff --git a/src/ModelContextProtocol.Core/Server/StreamableHttpServerTransport.cs b/src/ModelContextProtocol.Core/Server/StreamableHttpServerTransport.cs index 131c836dc..b403c57fd 100644 --- a/src/ModelContextProtocol.Core/Server/StreamableHttpServerTransport.cs +++ b/src/ModelContextProtocol.Core/Server/StreamableHttpServerTransport.cs @@ -69,6 +69,20 @@ public StreamableHttpServerTransport(ILoggerFactory? loggerFactory = null) /// public bool Stateless { get; init; } + /// + /// Gets or initializes a value that indicates whether POST requests return the final JSON-RPC response as JSON + /// instead of streaming messages as SSE events. + /// + /// + /// to write a single JSON response; to use SSE. The default is + /// . + /// + /// + /// When enabled, request-related notifications and other intermediate messages are omitted from POST responses. + /// Standalone GET requests are unaffected and continue to use SSE. + /// + public bool EnableJsonResponse { get; init; } + /// /// Gets or initializes a value indicating whether the execution context should flow from the calls to /// to the corresponding property contained in the instances returned by the . diff --git a/tests/ModelContextProtocol.AspNetCore.Tests/JsonResponseModeTests.cs b/tests/ModelContextProtocol.AspNetCore.Tests/JsonResponseModeTests.cs new file mode 100644 index 000000000..775c6b7a8 --- /dev/null +++ b/tests/ModelContextProtocol.AspNetCore.Tests/JsonResponseModeTests.cs @@ -0,0 +1,201 @@ +using Microsoft.AspNetCore.Builder; +using Microsoft.Extensions.DependencyInjection; +using ModelContextProtocol.AspNetCore.Tests.Utils; +using ModelContextProtocol.Protocol; +using ModelContextProtocol.Server; +using System.Net; +using System.Text; +using System.Text.Json; +using System.Text.Json.Serialization.Metadata; + +namespace ModelContextProtocol.AspNetCore.Tests; + +public class JsonResponseModeTests(ITestOutputHelper outputHelper) : KestrelInMemoryTest(outputHelper), IAsyncDisposable +{ + private WebApplication? _app; + + public async ValueTask DisposeAsync() + { + if (_app is not null) + { + await _app.DisposeAsync(); + } + + base.Dispose(); + } + + [Fact] + public async Task JsonResponseMode_ReturnsRawJsonRpcResponse() + { + await StartAsync(enableJsonResponse: true); + + using var response = await SendAsync(InitializeRequest); + + Assert.Equal(HttpStatusCode.OK, response.StatusCode); + Assert.Equal("application/json", response.Content.Headers.ContentType?.MediaType); + Assert.Null(response.Headers.CacheControl); + Assert.False(response.Headers.Contains("X-Accel-Buffering")); + + var responseBody = await response.Content.ReadAsStringAsync(TestContext.Current.CancellationToken); + Assert.DoesNotContain("event:", responseBody, StringComparison.Ordinal); + + var rpcResponse = JsonSerializer.Deserialize(responseBody, GetJsonTypeInfo()); + Assert.IsType(rpcResponse); + Assert.Equal(new RequestId(1), ((JsonRpcResponse)rpcResponse).Id); + } + + [Fact] + public async Task JsonResponseMode_SuppressesProgressAndReturnsToolResult() + { + await StartAsync(enableJsonResponse: true); + var sessionId = await InitializeSessionAsync(); + + using var response = await SendAsync(CallProgressToolRequest, sessionId, "2025-03-26"); + + Assert.Equal(HttpStatusCode.OK, response.StatusCode); + Assert.Equal("application/json", response.Content.Headers.ContentType?.MediaType); + + var responseBody = await response.Content.ReadAsStringAsync(TestContext.Current.CancellationToken); + Assert.DoesNotContain("notifications/progress", responseBody, StringComparison.Ordinal); + + var rpcMessage = JsonSerializer.Deserialize(responseBody, GetJsonTypeInfo()); + var rpcResponse = Assert.IsType(rpcMessage); + var result = JsonSerializer.Deserialize(rpcResponse.Result, GetJsonTypeInfo()); + Assert.NotNull(result); + var content = Assert.Single(result.Content); + Assert.Equal("done", Assert.IsType(content).Text); + } + + [Fact] + public async Task JsonResponseMode_NotificationReturnsEmptyAcceptedResponse() + { + await StartAsync(enableJsonResponse: true); + var sessionId = await InitializeSessionAsync(); + + using var response = await SendAsync(InitializedNotification, sessionId, "2025-03-26"); + + Assert.Equal(HttpStatusCode.Accepted, response.StatusCode); + Assert.Null(response.Content.Headers.ContentType); + Assert.Empty(await response.Content.ReadAsByteArrayAsync(TestContext.Current.CancellationToken)); + } + + [Fact] + public async Task JsonResponseMode_RejectsPolling() + { + InvalidOperationException? pollingException = null; + var pollingTool = McpServerTool.Create(async (RequestContext context) => + { + try + { + await context.EnablePollingAsync(TimeSpan.FromSeconds(1)); + } + catch (InvalidOperationException ex) + { + pollingException = ex; + } + + return "done"; + }, new() { Name = "polling" }); + + await StartAsync(enableJsonResponse: true, tools: [pollingTool]); + var sessionId = await InitializeSessionAsync(); + + using var response = await SendAsync(CallPollingToolRequest, sessionId, "2025-03-26"); + response.EnsureSuccessStatusCode(); + + Assert.NotNull(pollingException); + Assert.Equal("Polling is not supported when JSON responses are enabled.", pollingException.Message); + } + + [Fact] + public async Task DefaultMode_StillReturnsSse() + { + await StartAsync(enableJsonResponse: false); + + using var response = await SendAsync(InitializeRequest); + + Assert.Equal(HttpStatusCode.OK, response.StatusCode); + Assert.Equal("text/event-stream", response.Content.Headers.ContentType?.MediaType); + Assert.Equal("no", Assert.Single(response.Headers.GetValues("X-Accel-Buffering"))); + } + + private async Task StartAsync(bool enableJsonResponse, McpServerTool[]? tools = null) + { + tools ??= + [ + McpServerTool.Create((IProgress progress) => + { + progress.Report(new() { Progress = 1, Total = 1, Message = "working" }); + return "done"; + }, new() { Name = "progress" }), + ]; + + Builder.Services.AddMcpServer(options => + { + options.ServerInfo = new Implementation + { + Name = nameof(JsonResponseModeTests), + Version = "1.0.0", + }; + }) + .WithHttpTransport(options => + { + options.Stateless = false; + options.EnableJsonResponse = enableJsonResponse; + }) + .WithTools(tools); + + _app = Builder.Build(); + _app.MapMcp(); + await _app.StartAsync(TestContext.Current.CancellationToken); + + HttpClient.DefaultRequestHeaders.Accept.Add(new("application/json")); + HttpClient.DefaultRequestHeaders.Accept.Add(new("text/event-stream")); + } + + private async Task InitializeSessionAsync() + { + using var response = await SendAsync(InitializeRequest); + response.EnsureSuccessStatusCode(); + return Assert.Single(response.Headers.GetValues("mcp-session-id")); + } + + private async Task SendAsync(string json, string? sessionId = null, string? protocolVersion = null) + { + using var request = new HttpRequestMessage(HttpMethod.Post, "") + { + Content = new StringContent(json, Encoding.UTF8, "application/json"), + }; + + if (sessionId is not null) + { + request.Headers.Add("mcp-session-id", sessionId); + } + + if (protocolVersion is not null) + { + request.Headers.Add("mcp-protocol-version", protocolVersion); + } + + return await HttpClient.SendAsync(request, TestContext.Current.CancellationToken); + } + + private static JsonTypeInfo GetJsonTypeInfo() => + (JsonTypeInfo)McpJsonUtilities.DefaultOptions.GetTypeInfo(typeof(T)); + + private const string InitializeRequest = """ + {"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-03-26","capabilities":{},"clientInfo":{"name":"JsonResponseModeTests","version":"1.0.0"}}} + """; + + private const string InitializedNotification = """ + {"jsonrpc":"2.0","method":"notifications/initialized"} + """; + + private const string CallProgressToolRequest = """ + {"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"progress","arguments":{},"_meta":{"progressToken":"progress-token"}}} + """; + + private const string CallPollingToolRequest = """ + {"jsonrpc":"2.0","id":3,"method":"tools/call","params":{"name":"polling","arguments":{}}} + """; +}