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

SSE client transport: tolerate return of RPC responses in POST requests. (#445)

Browse source 
* SSE client transport: handle return of RPC responses in POST requests.

* Remove accidental change.

* Address feedback

* Remove unused namespace.
This commit is contained in:
Eirik Tsarpalis 2025-05-23 21:39:54 +03:00 • committed by GitHub
commit 77f9e64e04
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 30 additions and 14 deletions
Download patch file Download diff file Expand all files Collapse all files

14
src/ModelContextProtocol/Client/SseClientSessionTransport.cs
View file

@ -92,26 +92,18 @@ internal sealed partial class SseClientSessionTransport : TransportBase
StreamableHttpClientSessionTransport.CopyAdditionalHeaders(httpRequestMessage.Headers, _options.AdditionalHeaders); StreamableHttpClientSessionTransport.CopyAdditionalHeaders(httpRequestMessage.Headers, _options.AdditionalHeaders);
var response = await _httpClient.SendAsync(httpRequestMessage, cancellationToken).ConfigureAwait(false); var response = await _httpClient.SendAsync(httpRequestMessage, cancellationToken).ConfigureAwait(false);
response.EnsureSuccessStatusCode(); if (!response.IsSuccessStatusCode)
var responseContent = await response.Content.ReadAsStringAsync(cancellationToken).ConfigureAwait(false);
if (string.IsNullOrEmpty(responseContent) || responseContent.Equals("accepted", StringComparison.OrdinalIgnoreCase))
{
LogAcceptedPost(Name, messageId);
}
else
{ {
if (_logger.IsEnabled(LogLevel.Trace)) if (_logger.IsEnabled(LogLevel.Trace))
{ {
LogRejectedPostSensitive(Name, messageId, responseContent); LogRejectedPostSensitive(Name, messageId, await response.Content.ReadAsStringAsync(cancellationToken).ConfigureAwait(false));
} }
else else
{ {
LogRejectedPost(Name, messageId); LogRejectedPost(Name, messageId);
} }
throw new InvalidOperationException("Failed to send message"); response.EnsureSuccessStatusCode();
} }
} }

30
tests/ModelContextProtocol.AspNetCore.Tests/SseIntegrationTests.cs
View file

@ -58,6 +58,22 @@ public partial class SseIntegrationTests(ITestOutputHelper outputHelper) : Kestr
Assert.True(true); Assert.True(true);
} }
[Fact]
public async Task ConnectAndReceiveMessage_ServerReturningJsonInPostRequest()
{
await using var app = Builder.Build();
MapAbsoluteEndpointUriMcp(app, respondInJson: true);
await app.StartAsync(TestContext.Current.CancellationToken);
await using var mcpClient = await ConnectMcpClientAsync();
// Send a test message through POST endpoint
await mcpClient.SendNotificationAsync("test/message", new Envelope { Message = "Hello, SSE!" }, serializerOptions: JsonContext.Default.Options, cancellationToken: TestContext.Current.CancellationToken);
Assert.True(true);
}
[Fact] [Fact]
public async Task ConnectAndReceiveNotification_InMemoryServer() public async Task ConnectAndReceiveNotification_InMemoryServer()
{ {
@ -220,7 +236,7 @@ public partial class SseIntegrationTests(ITestOutputHelper outputHelper) : Kestr
Assert.Equal("Failed to add header '' with value '' from AdditionalHeaders.", ex.Message); Assert.Equal("Failed to add header '' with value '' from AdditionalHeaders.", ex.Message);
} }
private static void MapAbsoluteEndpointUriMcp(IEndpointRouteBuilder endpoints) private static void MapAbsoluteEndpointUriMcp(IEndpointRouteBuilder endpoints, bool respondInJson = false)
{ {
var loggerFactory = endpoints.ServiceProvider.GetRequiredService<ILoggerFactory>(); var loggerFactory = endpoints.ServiceProvider.GetRequiredService<ILoggerFactory>();
var optionsSnapshot = endpoints.ServiceProvider.GetRequiredService<IOptions<McpServerOptions>>(); var optionsSnapshot = endpoints.ServiceProvider.GetRequiredService<IOptions<McpServerOptions>>();
@ -267,7 +283,7 @@ public partial class SseIntegrationTests(ITestOutputHelper outputHelper) : Kestr
await Results.BadRequest("Session not started.").ExecuteAsync(context); await Results.BadRequest("Session not started.").ExecuteAsync(context);
return; return;
} }
var message = (JsonRpcMessage?)await context.Request.ReadFromJsonAsync(McpJsonUtilities.DefaultOptions.GetTypeInfo(typeof(JsonRpcMessage)), context.RequestAborted); var message = await context.Request.ReadFromJsonAsync<JsonRpcMessage>(McpJsonUtilities.DefaultOptions, context.RequestAborted);
if (message is null) if (message is null)
{ {
await Results.BadRequest("No message in request body.").ExecuteAsync(context); await Results.BadRequest("No message in request body.").ExecuteAsync(context);
@ -276,7 +292,15 @@ public partial class SseIntegrationTests(ITestOutputHelper outputHelper) : Kestr
await session.OnMessageReceivedAsync(message, context.RequestAborted); await session.OnMessageReceivedAsync(message, context.RequestAborted);
context.Response.StatusCode = StatusCodes.Status202Accepted; context.Response.StatusCode = StatusCodes.Status202Accepted;
await context.Response.WriteAsync("Accepted");
if (respondInJson)
{
await context.Response.WriteAsJsonAsync(message, McpJsonUtilities.DefaultOptions, cancellationToken: context.RequestAborted);
}
else
{
await context.Response.WriteAsync("Accepted");
}
}); });
} }