modelcontextprotocol/csharp-sdk
Watch
1
0
Fork
You've already forked csharp-sdk
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

Remove batching support from StreamableHttpServerTransport (#372)

Browse source
This commit is contained in:
Stephen Halter 2025-05-01 15:48:53 -07:00 • committed by GitHub
commit bac97efec6
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 5 additions and 77 deletions
Download patch file Download diff file Expand all files Collapse all files

2
src/ModelContextProtocol/Protocol/Transport/StreamClientSessionTransport.cs
View file

@ -70,7 +70,7 @@ internal class StreamClientSessionTransport : TransportBase
id = messageWithId.Id.ToString();
}
var json = JsonSerializer.Serialize(message, McpJsonUtilities.DefaultOptions.GetTypeInfo(typeof(JsonRpcMessage)));
var json = JsonSerializer.Serialize(message, McpJsonUtilities.JsonContext.Default.JsonRpcMessage);
using var _ = await _sendLock.LockAsync(cancellationToken).ConfigureAwait(false);
try

45
src/ModelContextProtocol/Protocol/Transport/StreamableHttpPostTransport.cs
View file

@ -32,7 +32,9 @@ internal sealed class StreamableHttpPostTransport(ChannelWriter<JsonRpcMessage>?
// The incomingChannel is null to handle the potential client GET request to handle unsolicited JsonRpcMessages.
if (incomingChannel is not null)
{
await OnPostBodyReceivedAsync(httpBodies.Input, cancellationToken).ConfigureAwait(false);
var message = await JsonSerializer.DeserializeAsync(httpBodies.Input.AsStream(),
McpJsonUtilities.JsonContext.Default.JsonRpcMessage, cancellationToken).ConfigureAwait(false);
await OnMessageReceivedAsync(message, cancellationToken).ConfigureAwait(false);
}
if (_pendingRequests.Count == 0)
@ -72,24 +74,6 @@ internal sealed class StreamableHttpPostTransport(ChannelWriter<JsonRpcMessage>?
}
}
private async ValueTask OnPostBodyReceivedAsync(PipeReader streamableHttpRequestBody, CancellationToken cancellationToken)
{
if (!await IsJsonArrayAsync(streamableHttpRequestBody, cancellationToken).ConfigureAwait(false))
{
var message = await JsonSerializer.DeserializeAsync(streamableHttpRequestBody.AsStream(), McpJsonUtilities.JsonContext.Default.JsonRpcMessage, cancellationToken).ConfigureAwait(false);
await OnMessageReceivedAsync(message, cancellationToken).ConfigureAwait(false);
}
else
{
// Batched JSON-RPC message
var messages = JsonSerializer.DeserializeAsyncEnumerable(streamableHttpRequestBody.AsStream(), McpJsonUtilities.JsonContext.Default.JsonRpcMessage, cancellationToken).ConfigureAwait(false);
await foreach (var message in messages.WithCancellation(cancellationToken))
{
await OnMessageReceivedAsync(message, cancellationToken).ConfigureAwait(false);
}
}
}
private async ValueTask OnMessageReceivedAsync(JsonRpcMessage? message, CancellationToken cancellationToken)
{
if (message is null)
@ -108,27 +92,4 @@ internal sealed class StreamableHttpPostTransport(ChannelWriter<JsonRpcMessage>?
Throw.IfNull(incomingChannel);
await incomingChannel.WriteAsync(message, cancellationToken).ConfigureAwait(false);
}
private async ValueTask<bool> IsJsonArrayAsync(PipeReader requestBody, CancellationToken cancellationToken)
{
// REVIEW: Should we bother trimming whitespace before checking for '['?
var firstCharacterResult = await requestBody.ReadAtLeastAsync(1, cancellationToken).ConfigureAwait(false);
try
{
if (firstCharacterResult.Buffer.Length == 0)
{
return false;
}
Span<byte> firstCharBuffer = stackalloc byte[1];
firstCharacterResult.Buffer.Slice(0, 1).CopyTo(firstCharBuffer);
return firstCharBuffer[0] == (byte)'[';
}
finally
{
// Never consume data when checking for '['. System.Text.Json still needs to consume it.
requestBody.AdvanceTo(firstCharacterResult.Buffer.Start);
}
}
}

2
src/ModelContextProtocol/Shared/McpSession.cs
View file

@ -75,7 +75,7 @@ internal sealed partial class McpSession : IDisposable
StdioClientSessionTransport or StdioServerTransport => "stdio",
StreamClientSessionTransport or StreamServerTransport => "stream",
SseClientSessionTransport or SseResponseStreamTransport => "sse",
StreamableHttpServerTransport or StreamableHttpPostTransport => "http",
StreamableHttpClientSessionTransport or StreamableHttpServerTransport or StreamableHttpPostTransport => "http",
_ => "unknownTransport"
};

33
tests/ModelContextProtocol.AspNetCore.Tests/StreamableHttpServerConformanceTests.cs
View file

@ -182,39 +182,6 @@ public class StreamableHttpServerConformanceTests(ITestOutputHelper outputHelper
await CallInitializeAndValidateAsync();
}
[Fact]
public async Task BatchedJsonRpcRequests_IsHandled_WithCompleteSseResponse()
{
await StartAsync();
using var response = await HttpClient.PostAsync("", JsonContent($"[{InitializeRequest},{EchoRequest}]"), TestContext.Current.CancellationToken);
Assert.Equal(HttpStatusCode.OK, response.StatusCode);
var eventCount = 0;
await foreach (var sseEvent in ReadSseAsync(response.Content))
{
var jsonRpcResponse = JsonSerializer.Deserialize(sseEvent, GetJsonTypeInfo<JsonRpcResponse>());
Assert.NotNull(jsonRpcResponse);
var responseId = Assert.IsType<long>(jsonRpcResponse.Id.Id);
switch (responseId)
{
case 1:
AssertServerInfo(jsonRpcResponse);
break;
case 2:
AssertEchoResponse(jsonRpcResponse);
break;
default:
throw new Exception($"Unexpected response ID: {jsonRpcResponse.Id}");
}
eventCount++;
}
Assert.Equal(2, eventCount);
}
[Fact]
public async Task SingleJsonRpcRequest_ThatThrowsIsHandled_WithCompleteSseResponse()
{