Add automatic transport fallback for SSE client connections

global.json

@@ -1,6 +1,6 @@
 {
   "sdk": {
-    "version": "9.0.100",
+    "version": "8.0.115",
     "rollForward": "minor"
   }
 }

src/ModelContextProtocol/Client/AutoDetectingClientSessionTransport.cs

@@ -0,0 +1,200 @@
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Logging.Abstractions;
+using ModelContextProtocol.Protocol;
+using System.Diagnostics;
+using System.Threading.Channels;
+
+namespace ModelContextProtocol.Client;
+
+/// <summary>
+/// A transport that automatically detects whether to use Streamable HTTP or SSE transport
+/// by trying Streamable HTTP first and falling back to SSE if that fails.
+/// </summary>
+internal sealed class AutoDetectingClientSessionTransport : ITransport
+{
+    private readonly SseClientTransportOptions _options;
+    private readonly HttpClient _httpClient;
+    private readonly ILoggerFactory? _loggerFactory;
+    private readonly ILogger _logger;
+    private readonly string _name;
+
+    private StreamableHttpClientSessionTransport? _streamableHttpTransport;
+    private SseClientSessionTransport? _sseTransport;
+    private readonly Channel<JsonRpcMessage> _messageChannel;
+
+    public AutoDetectingClientSessionTransport(SseClientTransportOptions transportOptions, HttpClient httpClient, ILoggerFactory? loggerFactory, string endpointName)
+    {
+        Throw.IfNull(transportOptions);
+        Throw.IfNull(httpClient);
+
+        _options = transportOptions;
+        _httpClient = httpClient;
+        _loggerFactory = loggerFactory;
+        _logger = (ILogger?)loggerFactory?.CreateLogger<AutoDetectingClientSessionTransport>() ?? NullLogger.Instance;
+        _name = endpointName;
+
+        // Unbounded channel to prevent blocking on writes
+        _messageChannel = Channel.CreateUnbounded<JsonRpcMessage>(new UnboundedChannelOptions
+        {
+            SingleReader = true,
+            SingleWriter = false,
+        });
+    }
+
+    public ChannelReader<JsonRpcMessage> MessageReader => _messageChannel.Reader;
+
+    /// <inheritdoc/>
+    public async Task SendMessageAsync(JsonRpcMessage message, CancellationToken cancellationToken = default)
+    {
+        if (_streamableHttpTransport == null && _sseTransport == null)
+        {
+            var rpcRequest = message as JsonRpcRequest;
+
+            // The first message must be an initialize request
+            Debug.Assert(rpcRequest != null && rpcRequest.Method == RequestMethods.Initialize, 
+                "First message must be an initialize request");
+
+            // Try StreamableHttp first
+            _streamableHttpTransport = new StreamableHttpClientSessionTransport(_options, _httpClient, _loggerFactory, _name);
+
+            try
+            {
+                var response = await _streamableHttpTransport.SendInitialRequestAsync(message, cancellationToken).ConfigureAwait(false);
+
+                // If the status code is not success, fall back to SSE
+                if (!response.IsSuccessStatusCode)
+                {
+                    _logger.LogDebug("Streamable HTTP transport failed for {EndpointName} with status code {StatusCode}, falling back to SSE transport",
+                        _name, response.StatusCode);
+
+                    await _streamableHttpTransport.DisposeAsync().ConfigureAwait(false);
+                    _streamableHttpTransport = null;
+
+                    await InitializeSseTransportAsync(message, cancellationToken).ConfigureAwait(false);
+                    return;
+                }
+
+                // Process the response
+                var responseContent = await response.Content.ReadAsStringAsync(cancellationToken).ConfigureAwait(false);
+                if (response.Content.Headers.ContentType?.MediaType == "application/json")
+                {
+                    await ProcessMessageFromStreamableHttpAsync(responseContent, cancellationToken).ConfigureAwait(false);
+                }
+                else if (response.Content.Headers.ContentType?.MediaType == "text/event-stream")
+                {
+                    using var responseBodyStream = await response.Content.ReadAsStreamAsync(cancellationToken);
+                    await ProcessSseResponseFromStreamableHttpAsync(responseBodyStream, rpcRequest, cancellationToken).ConfigureAwait(false);
+                }
+
+                // Start forwarding messages
+                _ = ForwardMessagesAsync(_streamableHttpTransport.MessageReader, cancellationToken);
+            }
+            catch (Exception ex)
+            {
+                _logger.LogDebug(ex, "Streamable HTTP transport failed for {EndpointName}, falling back to SSE transport", _name);
+
+                if (_streamableHttpTransport != null)
+                {
+                    await _streamableHttpTransport.DisposeAsync().ConfigureAwait(false);
+                    _streamableHttpTransport = null;
+                }
+
+                await InitializeSseTransportAsync(message, cancellationToken).ConfigureAwait(false);
+            }
+        }
+        else if (_streamableHttpTransport != null)
+        {
+            await _streamableHttpTransport.SendMessageAsync(message, cancellationToken).ConfigureAwait(false);
+        }
+        else if (_sseTransport != null)
+        {
+            await _sseTransport.SendMessageAsync(message, cancellationToken).ConfigureAwait(false);
+        }
+    }
+
+    private async Task InitializeSseTransportAsync(JsonRpcMessage message, CancellationToken cancellationToken)
+    {
+        _sseTransport = new SseClientSessionTransport(_options, _httpClient, _loggerFactory, _name);
+        await _sseTransport.ConnectAsync(cancellationToken).ConfigureAwait(false);
+        await _sseTransport.SendMessageAsync(message, cancellationToken).ConfigureAwait(false);
+
+        // Start forwarding messages
+        _ = ForwardMessagesAsync(_sseTransport.MessageReader, cancellationToken);
+    }
+
+    private async Task ProcessMessageFromStreamableHttpAsync(string data, CancellationToken cancellationToken)
+    {
+        try
+        {
+            var message = System.Text.Json.JsonSerializer.Deserialize(data, McpJsonUtilities.JsonContext.Default.JsonRpcMessage);
+            if (message is null)
+            {
+                _logger.LogWarning("Failed to parse message from Streamable HTTP response for {EndpointName}", _name);
+                return;
+            }
+
+            bool wrote = _messageChannel.Writer.TryWrite(message);
+            Debug.Assert(wrote, "Failed to write message to channel");
+        }
+        catch (System.Text.Json.JsonException ex)
+        {
+            _logger.LogError(ex, "Failed to parse JSON message from Streamable HTTP response for {EndpointName}", _name);
+        }
+    }
+
+    private async Task ProcessSseResponseFromStreamableHttpAsync(Stream responseStream, JsonRpcRequest relatedRpcRequest, CancellationToken cancellationToken)
+    {
+        await foreach (System.Net.ServerSentEvents.SseItem<string> sseEvent in System.Net.ServerSentEvents.SseParser.Create(responseStream)
+                       .EnumerateAsync(cancellationToken).ConfigureAwait(false))
+        {
+            if (sseEvent.EventType != "message")
+            {
+                continue;
+            }
+
+            await ProcessMessageFromStreamableHttpAsync(sseEvent.Data, cancellationToken).ConfigureAwait(false);
+        }
+    }
+
+    private async Task ForwardMessagesAsync(ChannelReader<JsonRpcMessage> reader, CancellationToken cancellationToken)
+    {
+        try
+        {
+            await foreach (var message in reader.ReadAllAsync(cancellationToken).ConfigureAwait(false))
+            {
+                bool wrote = _messageChannel.Writer.TryWrite(message);
+                Debug.Assert(wrote, "Failed to write message to channel");
+            }
+        }
+        catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
+        {
+            // Expected when cancelling
+        }
+        catch (Exception ex)
+        {
+            _logger.LogError(ex, "Error forwarding messages from active transport for {EndpointName}", _name);
+        }
+    }
+
+    public async ValueTask DisposeAsync()
+    {
+        try
+        {
+            if (_streamableHttpTransport != null)
+            {
+                await _streamableHttpTransport.DisposeAsync().ConfigureAwait(false);
+                _streamableHttpTransport = null;
+            }
+
+            if (_sseTransport != null)
+            {
+                await _sseTransport.DisposeAsync().ConfigureAwait(false);
+                _sseTransport = null;
+            }
+        }
+        finally
+        {
+            _messageChannel.Writer.Complete();
+        }
+    }
+}

src/ModelContextProtocol/Client/HttpTransportMode.cs

@@ -0,0 +1,23 @@
+namespace ModelContextProtocol.Client;
+
+/// <summary>
+/// Specifies the transport mode for HTTP client connections.
+/// </summary>
+public enum HttpTransportMode
+{
+    /// <summary>
+    /// Automatically detect the appropriate transport by trying Streamable HTTP first, then falling back to SSE if that fails.
+    /// This is the recommended mode for maximum compatibility.
+    /// </summary>
+    AutoDetect,
+
+    /// <summary>
+    /// Use only the Streamable HTTP transport.
+    /// </summary>
+    StreamableHttp,
+
+    /// <summary>
+    /// Use only the HTTP with SSE transport.
+    /// </summary>
+    Sse
+}

src/ModelContextProtocol/Client/SseClientTransport.cs

@@ -57,11 +57,24 @@ public SseClientTransport(SseClientTransportOptions transportOptions, HttpClient
     /// <inheritdoc />
     public async Task<ITransport> ConnectAsync(CancellationToken cancellationToken = default)
     {
-        if (_options.UseStreamableHttp)
+        switch (_options.TransportMode)
         {
-            return new StreamableHttpClientSessionTransport(_options, _httpClient, _loggerFactory, Name);
+            case HttpTransportMode.StreamableHttp:
+                return new StreamableHttpClientSessionTransport(_options, _httpClient, _loggerFactory, Name);
+
+            case HttpTransportMode.Sse:
+                return await ConnectSseTransportAsync(cancellationToken).ConfigureAwait(false);
+
+            case HttpTransportMode.AutoDetect:
+                return new AutoDetectingClientSessionTransport(_options, _httpClient, _loggerFactory, Name);
+
+            default:
+                throw new ArgumentOutOfRangeException(nameof(_options.TransportMode), _options.TransportMode, "Unsupported transport mode");
         }
+    }
 
+    private async Task<ITransport> ConnectSseTransportAsync(CancellationToken cancellationToken)
+    {
         var sessionTransport = new SseClientSessionTransport(_options, _httpClient, _loggerFactory, Name);
 
         try

src/ModelContextProtocol/Client/SseClientTransportOptions.cs

@@ -30,12 +30,19 @@ public required Uri Endpoint
         }
     }
 
+
+
     /// <summary>
-    /// Gets or sets a value indicating whether to use "Streamable HTTP" for the transport rather than "HTTP with SSE". Defaults to false.
-    /// <see href="https://modelcontextprotocol.io/specification/2025-03-26/basic/transports#streamable-http">Streamable HTTP transport specification</see>.
-    /// <see href="https://modelcontextprotocol.io/specification/2024-11-05/basic/transports#http-with-sse">HTTP with SSE transport specification</see>.
+    /// Gets or sets the transport mode to use for the connection. Defaults to <see cref="HttpTransportMode.AutoDetect"/>.
     /// </summary>
-    public bool UseStreamableHttp { get; init; }
+    /// <remarks>
+    /// <para>
+    /// When set to <see cref="HttpTransportMode.AutoDetect"/> (the default), the client will first attempt to use
+    /// Streamable HTTP transport and automatically fall back to SSE transport if the server doesn't support it.
+    /// This provides the best compatibility and matches the behavior of VS Code.
+    /// </para>
+    /// </remarks>
+    public HttpTransportMode TransportMode { get; init; } = HttpTransportMode.AutoDetect;
 
     /// <summary>
     /// Gets a transport identifier used for logging purposes.

src/ModelContextProtocol/Client/StreamableHttpClientSessionTransport.cs

@@ -203,16 +203,40 @@ private async Task ReceiveUnsolicitedMessagesAsync()
         return null;
     }
 
-    private void LogJsonException(JsonException ex, string data)
+    /// <summary>
+    /// Sends the initial initialization request and returns the HTTP response so the status code can be checked.
+    /// </summary>
+    /// <param name="message">The initialize message to send, which must be a JsonRpcRequest with the method "initialize".</param>
+    /// <param name="cancellationToken">The cancellation token.</param>
+    /// <returns>The HTTP response message for the initialization request.</returns>
+    internal async Task<HttpResponseMessage> SendInitialRequestAsync(
+        JsonRpcMessage message,
+        CancellationToken cancellationToken = default)
     {
-        if (_logger.IsEnabled(LogLevel.Trace))
-        {
-            LogTransportMessageParseFailedSensitive(Name, data, ex);
-        }
-        else
+        using var sendCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, _connectionCts.Token);
+        cancellationToken = sendCts.Token;
+
+#if NET
+        using var content = JsonContent.Create(message, McpJsonUtilities.JsonContext.Default.JsonRpcMessage);
+#else
+        using var content = new StringContent(
+            JsonSerializer.Serialize(message, McpJsonUtilities.JsonContext.Default.JsonRpcMessage),
+            Encoding.UTF8,
+            "application/json; charset=utf-8"
+        );
+#endif
+
+        using var httpRequestMessage = new HttpRequestMessage(HttpMethod.Post, _options.Endpoint)
         {
-            LogTransportMessageParseFailed(Name, ex);
-        }
+            Content = content,
+            Headers =
+            {
+                Accept = { s_applicationJsonMediaType, s_textEventStreamMediaType },
+            },
+        };
+
+        CopyAdditionalHeaders(httpRequestMessage.Headers, _options.AdditionalHeaders, _mcpSessionId);
+        return await _httpClient.SendAsync(httpRequestMessage, HttpCompletionOption.ResponseHeadersRead, cancellationToken).ConfigureAwait(false);
     }
 
     internal static void CopyAdditionalHeaders(HttpRequestHeaders headers, Dictionary<string, string>? additionalHeaders, string? sessionId = null)