Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 17 additions & 3 deletions src/ModelContextProtocol.Core/Client/SseClientSessionTransport.cs
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ internal sealed partial class SseClientSessionTransport : TransportBase
private Task? _receiveTask;
private readonly ILogger _logger;
private readonly TaskCompletionSource<bool> _connectionEstablished;
private volatile bool _sseAdopted;

/// <summary>
/// SSE transport for a single session. Unlike stdio it does not launch a process, but connects to an existing server.
Expand Down Expand Up @@ -107,6 +108,8 @@ public override async Task SendMessageAsync(

throw HttpResponseMessageExtensions.CreateHttpRequestException(response, responseBody);
}

_sseAdopted = true;
}

private async Task CloseAsync()
Expand All @@ -129,7 +132,7 @@ private async Task CloseAsync()
}
finally
{
SetDisconnected(new ClientTransportClosedException(new HttpClientCompletionDetails()));
SetSseDisconnected(new ClientTransportClosedException(new HttpClientCompletionDetails()));
}
}

Expand Down Expand Up @@ -190,7 +193,7 @@ private async Task ReceiveMessagesAsync(CancellationToken cancellationToken)
}
else
{
SetDisconnected(new ClientTransportClosedException(new HttpClientCompletionDetails
SetSseDisconnected(new ClientTransportClosedException(new HttpClientCompletionDetails
{
HttpStatusCode = failureStatusCode,
Exception = ex,
Expand All @@ -202,7 +205,18 @@ private async Task ReceiveMessagesAsync(CancellationToken cancellationToken)
}
finally
{
SetDisconnected(new ClientTransportClosedException(new HttpClientCompletionDetails()));
SetSseDisconnected(new ClientTransportClosedException(new HttpClientCompletionDetails()));
}
}

private void SetSseDisconnected(Exception error)
{
// If AutoDetect is still probing SSE, leave its shared message channel open so it can
// retry with another transport. A successful POST means SSE was selected and owns the
// channel from that point on, matching Streamable HTTP's adoption behavior.
if (_options.TransportMode is not HttpTransportMode.AutoDetect || _sseAdopted)
Comment thread
PranavSenthilnathan marked this conversation as resolved.
{
SetDisconnected(error);
}
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@
using ModelContextProtocol.Protocol;
using ModelContextProtocol.Tests.Utils;
using Microsoft.Extensions.Logging;
using System.IO.Pipelines;
using System.Net;

namespace ModelContextProtocol.Tests.Transport;
Expand Down Expand Up @@ -163,6 +164,115 @@ public async Task AutoDetectMode_FallsBackToSse_WhenStreamableHttpFails()
Assert.NotNull(session);
}

[Fact]
public async Task AutoDetectMode_WhenProvisionalSseFails_LeavesSharedMessageChannelOpen()
{
var options = new HttpClientTransportOptions
{
Endpoint = new Uri("http://localhost"),
TransportMode = HttpTransportMode.AutoDetect,
Name = "AutoDetect shared channel test client"
};

using var mockHttpHandler = new MockHttpHandler();
using var httpClient = new HttpClient(mockHttpHandler);
await using var transport = new HttpClientTransport(options, httpClient, LoggerFactory);
var streamableHttpPostCount = 0;

mockHttpHandler.RequestHandler = request =>
{
if (request.Method == HttpMethod.Get)
{
return Task.FromResult(new HttpResponseMessage(HttpStatusCode.MethodNotAllowed));
}

if (request.Method == HttpMethod.Post && ++streamableHttpPostCount == 1)
{
return Task.FromResult(new HttpResponseMessage(HttpStatusCode.NotFound)
{
Content = new StringContent("Invalid session ID"),
});
}

return Task.FromResult(new HttpResponseMessage(HttpStatusCode.OK)
{
Content = new StringContent(
"{\"jsonrpc\":\"2.0\",\"id\":2,\"result\":{\"protocolVersion\":\"2025-11-25\",\"capabilities\":{},\"serverInfo\":{\"name\":\"test\",\"version\":\"1.0\"}}}",
System.Text.Encoding.UTF8,
"application/json"),
});
};

await using var session = await transport.ConnectAsync(TestContext.Current.CancellationToken);

await Assert.ThrowsAsync<HttpRequestException>(() =>
session.SendMessageAsync(
new JsonRpcRequest { Method = RequestMethods.ServerDiscover, Id = new RequestId(1) },
TestContext.Current.CancellationToken));

await session.SendMessageAsync(
new JsonRpcRequest { Method = RequestMethods.Initialize, Id = new RequestId(2) },
TestContext.Current.CancellationToken);

var response = await session.MessageReader.ReadAsync(TestContext.Current.CancellationToken);
Assert.Equal(new RequestId(2), Assert.IsType<JsonRpcResponse>(response).Id);
}

[Fact]
public async Task AutoDetectMode_WhenAdoptedSseDisconnects_CompletesSharedMessageChannel()
{
var options = new HttpClientTransportOptions
{
Endpoint = new Uri("http://localhost"),
TransportMode = HttpTransportMode.AutoDetect,
Name = "AutoDetect adopted SSE test client"
};

using var mockHttpHandler = new MockHttpHandler();
using var httpClient = new HttpClient(mockHttpHandler);
await using var transport = new HttpClientTransport(options, httpClient, LoggerFactory);
var ssePipe = new Pipe();
var postCount = 0;

await ssePipe.Writer.WriteAsync(
System.Text.Encoding.UTF8.GetBytes("event: endpoint\r\ndata: /sse-endpoint\r\n\r\n"),
TestContext.Current.CancellationToken);

mockHttpHandler.RequestHandler = request =>
{
if (request.Method == HttpMethod.Get)
{
var content = new StreamContent(ssePipe.Reader.AsStream());
content.Headers.ContentType = new("text/event-stream");
return Task.FromResult(new HttpResponseMessage(HttpStatusCode.OK) { Content = content });
}

if (request.Method == HttpMethod.Post && ++postCount == 1)
{
return Task.FromResult(new HttpResponseMessage(HttpStatusCode.NotFound)
{
Content = new StringContent("Streamable HTTP not supported"),
});
}

return Task.FromResult(new HttpResponseMessage(HttpStatusCode.OK));
};

await using var session = await transport.ConnectAsync(TestContext.Current.CancellationToken);

await session.SendMessageAsync(
new JsonRpcRequest { Method = RequestMethods.Initialize, Id = new RequestId(1) },
TestContext.Current.CancellationToken);

await ssePipe.Writer.CompleteAsync();

var exception = await Assert.ThrowsAsync<ClientTransportClosedException>(
async () => await session.MessageReader.Completion.WaitAsync(
TestConstants.DefaultTimeout,
TestContext.Current.CancellationToken));
Assert.IsType<HttpClientCompletionDetails>(exception.Details);
}

// Regression test for https://github.com/modelcontextprotocol/csharp-sdk/issues/1526
// When Streamable HTTP returns 415 (e.g. wrong Content-Type) and the SSE fallback also fails
// (e.g. a Streamable-HTTP-only server returns 405 to the GET), the surfaced exception must
Expand Down