Clean up namespaces (#410)

No meaningful code changes, just lots of namespace shuffling to have fewer namespaces and to put things in better namespaces.

src/ModelContextProtocol/AIContentExtensions.cs

@@ -1,8 +1,8 @@
 using Microsoft.Extensions.AI;
-using ModelContextProtocol.Protocol.Types;
-using ModelContextProtocol.Utils;
-using ModelContextProtocol.Utils.Json;
+using ModelContextProtocol.Protocol;
+#if !NET
 using System.Runtime.InteropServices;
+#endif
 using System.Text.Json;
 
 namespace ModelContextProtocol;

src/ModelContextProtocol/Client/IClientTransport.cs

@@ -0,0 +1,47 @@
+using ModelContextProtocol.Protocol;
+
+namespace ModelContextProtocol.Client;
+
+/// <summary>
+/// Represents a transport mechanism for Model Context Protocol (MCP) client-to-server communication.
+/// </summary>
+/// <remarks>
+/// <para>
+/// The <see cref="IClientTransport"/> interface abstracts the communication layer between MCP clients
+/// and servers, allowing different transport protocols to be used interchangeably.
+/// </para>
+/// <para>
+/// When creating an <see cref="IMcpClient"/>, <see cref="McpClientFactory"/> is typically used, and is
+/// provided with the <see cref="IClientTransport"/> based on expected server configuration.
+/// </para>
+/// </remarks>
+public interface IClientTransport
+{
+    /// <summary>
+    /// Gets a transport identifier, used for logging purposes.
+    /// </summary>
+    string Name { get; }
+
+    /// <summary>
+    /// Asynchronously establishes a transport session with an MCP server and returns a transport for the duplex message stream.
+    /// </summary>
+    /// <param name="cancellationToken">The <see cref="CancellationToken"/> to monitor for cancellation requests. The default is <see cref="CancellationToken.None"/>.</param>
+    /// <returns>Returns an interface for the duplex message stream.</returns>
+    /// <remarks>
+    /// <para>
+    /// This method is responsible for initializing the connection to the server using the specific transport 
+    /// mechanism implemented by the derived class. The returned <see cref="ITransport"/> interface 
+    /// provides methods to send and receive messages over the established connection.
+    /// </para>
+    /// <para>
+    /// The lifetime of the returned <see cref="ITransport"/> instance is typically managed by the 
+    /// <see cref="McpClient"/> that uses this transport. When the client is disposed, it will dispose
+    /// the transport session as well.
+    /// </para>
+    /// <para>
+    /// This method is used by <see cref="McpClientFactory"/> to initialize the connection.
+    /// </para>
+    /// </remarks>
+    /// <exception cref="InvalidOperationException">The transport connection could not be established.</exception>
+    Task<ITransport> ConnectAsync(CancellationToken cancellationToken = default);
+}

src/ModelContextProtocol/Client/IMcpClient.cs

@@ -1,4 +1,4 @@
-using ModelContextProtocol.Protocol.Types;
+using ModelContextProtocol.Protocol;
 
 namespace ModelContextProtocol.Client;
 

src/ModelContextProtocol/Client/McpClient.cs

@@ -1,9 +1,5 @@
 using Microsoft.Extensions.Logging;
-using ModelContextProtocol.Protocol.Messages;
-using ModelContextProtocol.Protocol.Transport;
-using ModelContextProtocol.Protocol.Types;
-using ModelContextProtocol.Shared;
-using ModelContextProtocol.Utils.Json;
+using ModelContextProtocol.Protocol;
 using System.Text.Json;
 
 namespace ModelContextProtocol.Client;

src/ModelContextProtocol/Client/McpClientExtensions.cs

@@ -1,10 +1,7 @@
 using Microsoft.Extensions.AI;
 using Microsoft.Extensions.Logging;
-using ModelContextProtocol.Protocol.Messages;
-using ModelContextProtocol.Protocol.Types;
+using ModelContextProtocol.Protocol;
 using ModelContextProtocol.Server;
-using ModelContextProtocol.Utils;
-using ModelContextProtocol.Utils.Json;
 using System.Runtime.CompilerServices;
 using System.Text.Json;
 

src/ModelContextProtocol/Client/McpClientFactory.cs

@@ -1,5 +1,3 @@
-using ModelContextProtocol.Protocol.Transport;
-using ModelContextProtocol.Utils;
 using Microsoft.Extensions.Logging;
 
 namespace ModelContextProtocol.Client;

src/ModelContextProtocol/Client/McpClientOptions.cs

@@ -1,4 +1,4 @@
-using ModelContextProtocol.Protocol.Types;
+using ModelContextProtocol.Protocol;
 
 namespace ModelContextProtocol.Client;
 

src/ModelContextProtocol/Client/McpClientPrompt.cs

@@ -1,4 +1,4 @@
-using ModelContextProtocol.Protocol.Types;
+using ModelContextProtocol.Protocol;
 using System.Text.Json;
 
 namespace ModelContextProtocol.Client;

src/ModelContextProtocol/Client/McpClientResource.cs

@@ -1,4 +1,4 @@
-using ModelContextProtocol.Protocol.Types;
+using ModelContextProtocol.Protocol;
 
 namespace ModelContextProtocol.Client;
 

src/ModelContextProtocol/Client/McpClientResourceTemplate.cs

@@ -1,4 +1,4 @@
-using ModelContextProtocol.Protocol.Types;
+using ModelContextProtocol.Protocol;
 
 namespace ModelContextProtocol.Client;
 

src/ModelContextProtocol/Client/McpClientTool.cs

@@ -1,7 +1,5 @@
 using Microsoft.Extensions.AI;
-using ModelContextProtocol.Protocol.Types;
-using ModelContextProtocol.Utils;
-using ModelContextProtocol.Utils.Json;
+using ModelContextProtocol.Protocol;
 using System.Collections.ObjectModel;
 using System.Text.Json;
 

src/ModelContextProtocol/Client/SseClientSessionTransport.cs

@@ -0,0 +1,264 @@
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Logging.Abstractions;
+using ModelContextProtocol.Protocol;
+using System.Diagnostics;
+using System.Net.Http.Headers;
+using System.Net.ServerSentEvents;
+using System.Text;
+using System.Text.Json;
+
+namespace ModelContextProtocol.Client;
+
+/// <summary>
+/// The ServerSideEvents client transport implementation
+/// </summary>
+internal sealed partial class SseClientSessionTransport : TransportBase
+{
+    private readonly HttpClient _httpClient;
+    private readonly SseClientTransportOptions _options;
+    private readonly Uri _sseEndpoint;
+    private Uri? _messageEndpoint;
+    private readonly CancellationTokenSource _connectionCts;
+    private Task? _receiveTask;
+    private readonly ILogger _logger;
+    private readonly TaskCompletionSource<bool> _connectionEstablished;
+
+    /// <summary>
+    /// SSE transport for client endpoints. Unlike stdio it does not launch a process, but connects to an existing server.
+    /// The HTTP server can be local or remote, and must support the SSE protocol.
+    /// </summary>
+    /// <param name="transportOptions">Configuration options for the transport.</param>
+    /// <param name="httpClient">The HTTP client instance used for requests.</param>
+    /// <param name="loggerFactory">Logger factory for creating loggers.</param>
+    /// <param name="endpointName">The endpoint name used for logging purposes.</param>
+    public SseClientSessionTransport(SseClientTransportOptions transportOptions, HttpClient httpClient, ILoggerFactory? loggerFactory, string endpointName)
+        : base(endpointName, loggerFactory)
+    {
+        Throw.IfNull(transportOptions);
+        Throw.IfNull(httpClient);
+
+        _options = transportOptions;
+        _sseEndpoint = transportOptions.Endpoint;
+        _httpClient = httpClient;
+        _connectionCts = new CancellationTokenSource();
+        _logger = (ILogger?)loggerFactory?.CreateLogger<SseClientTransport>() ?? NullLogger.Instance;
+        _connectionEstablished = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+    }
+
+    /// <inheritdoc/>
+    public async Task ConnectAsync(CancellationToken cancellationToken = default)
+    {
+        Debug.Assert(!IsConnected);
+        try
+        {
+            // Start message receiving loop
+            _receiveTask = ReceiveMessagesAsync(_connectionCts.Token);
+
+            await _connectionEstablished.Task.WaitAsync(_options.ConnectionTimeout, cancellationToken).ConfigureAwait(false);
+        }
+        catch (Exception ex)
+        {
+            LogTransportConnectFailed(Name, ex);
+            await CloseAsync().ConfigureAwait(false);
+            throw new InvalidOperationException("Failed to connect transport", ex);
+        }
+    }
+
+    /// <inheritdoc/>
+    public override async Task SendMessageAsync(
+        JsonRpcMessage message,
+        CancellationToken cancellationToken = default)
+    {
+        if (_messageEndpoint == null)
+            throw new InvalidOperationException("Transport not connected");
+
+        using var content = new StringContent(
+            JsonSerializer.Serialize(message, McpJsonUtilities.JsonContext.Default.JsonRpcMessage),
+            Encoding.UTF8,
+            "application/json"
+        );
+
+        string messageId = "(no id)";
+
+        if (message is JsonRpcMessageWithId messageWithId)
+        {
+            messageId = messageWithId.Id.ToString();
+        }
+
+        using var httpRequestMessage = new HttpRequestMessage(HttpMethod.Post, _messageEndpoint)
+        {
+            Content = content,
+        };
+        StreamableHttpClientSessionTransport.CopyAdditionalHeaders(httpRequestMessage.Headers, _options.AdditionalHeaders);
+        var response = await _httpClient.SendAsync(httpRequestMessage, cancellationToken).ConfigureAwait(false);
+
+        response.EnsureSuccessStatusCode();
+
+        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))
+            {
+                LogRejectedPostSensitive(Name, messageId, responseContent);
+            }
+            else
+            {
+                LogRejectedPost(Name, messageId);
+            }
+
+            throw new InvalidOperationException("Failed to send message");
+        }
+    }
+
+    private async Task CloseAsync()
+    {
+        try
+        {
+            await _connectionCts.CancelAsync().ConfigureAwait(false);
+
+            try
+            {
+                if (_receiveTask != null)
+                {
+                    await _receiveTask.ConfigureAwait(false);
+                }
+            }
+            finally
+            {
+                _connectionCts.Dispose();
+            }
+        }
+        finally
+        {
+            SetDisconnected();
+        }
+    }
+
+    /// <inheritdoc/>
+    public override async ValueTask DisposeAsync()
+    {
+        try
+        {
+            await CloseAsync().ConfigureAwait(false);
+        }
+        catch (Exception)
+        {
+            // Ignore exceptions on close
+        }
+    }
+
+    private async Task ReceiveMessagesAsync(CancellationToken cancellationToken)
+    {
+        try
+        {
+            using var request = new HttpRequestMessage(HttpMethod.Get, _sseEndpoint);
+            request.Headers.Accept.Add(new MediaTypeWithQualityHeaderValue("text/event-stream"));
+            StreamableHttpClientSessionTransport.CopyAdditionalHeaders(request.Headers, _options.AdditionalHeaders);
+
+            using var response = await _httpClient.SendAsync(
+                request,
+                HttpCompletionOption.ResponseHeadersRead,
+                cancellationToken
+            ).ConfigureAwait(false);
+
+            response.EnsureSuccessStatusCode();
+
+            using var stream = await response.Content.ReadAsStreamAsync(cancellationToken).ConfigureAwait(false);
+
+            await foreach (SseItem<string> sseEvent in SseParser.Create(stream).EnumerateAsync(cancellationToken).ConfigureAwait(false))
+            {
+                switch (sseEvent.EventType)
+                {
+                    case "endpoint":
+                        HandleEndpointEvent(sseEvent.Data);
+                        break;
+
+                    case "message":
+                        await ProcessSseMessage(sseEvent.Data, cancellationToken).ConfigureAwait(false);
+                        break;
+                }
+            }
+        }
+        catch (Exception ex)
+        {
+            if (cancellationToken.IsCancellationRequested)
+            {
+                // Normal shutdown
+                LogTransportReadMessagesCancelled(Name);
+                _connectionEstablished.TrySetCanceled(cancellationToken);
+            }
+            else
+            {
+                LogTransportReadMessagesFailed(Name, ex);
+                _connectionEstablished.TrySetException(ex);
+                throw;
+            }
+        }
+        finally
+        {
+            SetDisconnected();
+        }
+    }
+
+    private async Task ProcessSseMessage(string data, CancellationToken cancellationToken)
+    {
+        if (!IsConnected)
+        {
+            LogTransportMessageReceivedBeforeConnected(Name);
+            return;
+        }
+
+        try
+        {
+            var message = JsonSerializer.Deserialize(data, McpJsonUtilities.JsonContext.Default.JsonRpcMessage);
+            if (message == null)
+            {
+                LogTransportMessageParseUnexpectedTypeSensitive(Name, data);
+                return;
+            }
+
+            await WriteMessageAsync(message, cancellationToken).ConfigureAwait(false);
+        }
+        catch (JsonException ex)
+        {
+            if (_logger.IsEnabled(LogLevel.Trace))
+            {
+                LogTransportMessageParseFailedSensitive(Name, data, ex);
+            }
+            else
+            {
+                LogTransportMessageParseFailed(Name, ex);
+            }
+        }
+    }
+
+    private void HandleEndpointEvent(string data)
+    {
+        if (string.IsNullOrEmpty(data))
+        {
+            LogTransportEndpointEventInvalid(Name);
+            return;
+        }
+
+        // If data is an absolute URL, the Uri will be constructed entirely from it and not the _sseEndpoint.
+        _messageEndpoint = new Uri(_sseEndpoint, data);
+
+        // Set connected state
+        SetConnected();
+        _connectionEstablished.TrySetResult(true);
+    }
+
+    [LoggerMessage(Level = LogLevel.Information, Message = "{EndpointName} accepted SSE transport POST for message ID '{MessageId}'.")]
+    private partial void LogAcceptedPost(string endpointName, string messageId);
+
+    [LoggerMessage(Level = LogLevel.Information, Message = "{EndpointName} rejected SSE transport POST for message ID '{MessageId}'.")]
+    private partial void LogRejectedPost(string endpointName, string messageId);
+
+    [LoggerMessage(Level = LogLevel.Trace, Message = "{EndpointName} rejected SSE transport POST for message ID '{MessageId}'. Server response: '{responseContent}'.")]
+    private partial void LogRejectedPostSensitive(string endpointName, string messageId, string responseContent);
+}

src/ModelContextProtocol/Client/SseClientTransport.cs

@@ -0,0 +1,89 @@
+using Microsoft.Extensions.Logging;
+using ModelContextProtocol.Protocol;
+
+namespace ModelContextProtocol.Client;
+
+/// <summary>
+/// Provides an <see cref="IClientTransport"/> over HTTP using the Server-Sent Events (SSE) protocol.
+/// </summary>
+/// <remarks>
+/// This transport connects to an MCP server over HTTP using SSE,
+/// allowing for real-time server-to-client communication with a standard HTTP request.
+/// Unlike the <see cref="StdioClientTransport"/>, this transport connects to an existing server
+/// rather than launching a new process.
+/// </remarks>
+public sealed class SseClientTransport : IClientTransport, IAsyncDisposable
+{
+    private readonly SseClientTransportOptions _options;
+    private readonly HttpClient _httpClient;
+    private readonly ILoggerFactory? _loggerFactory;
+    private readonly bool _ownsHttpClient;
+
+    /// <summary>
+    /// Initializes a new instance of the <see cref="SseClientTransport"/> class.
+    /// </summary>
+    /// <param name="transportOptions">Configuration options for the transport.</param>
+    /// <param name="loggerFactory">Logger factory for creating loggers used for diagnostic output during transport operations.</param>
+    public SseClientTransport(SseClientTransportOptions transportOptions, ILoggerFactory? loggerFactory = null)
+        : this(transportOptions, new HttpClient(), loggerFactory, ownsHttpClient: true)
+    {
+    }
+
+    /// <summary>
+    /// Initializes a new instance of the <see cref="SseClientTransport"/> class with a provided HTTP client.
+    /// </summary>
+    /// <param name="transportOptions">Configuration options for the transport.</param>
+    /// <param name="httpClient">The HTTP client instance used for requests.</param>
+    /// <param name="loggerFactory">Logger factory for creating loggers used for diagnostic output during transport operations.</param>
+    /// <param name="ownsHttpClient">
+    /// <see langword="true"/> to dispose of <paramref name="httpClient"/> when the transport is disposed; 
+    /// <see langword="false"/> if the caller is retaining ownership of the <paramref name="httpClient"/>'s lifetime.
+    /// </param>
+    public SseClientTransport(SseClientTransportOptions transportOptions, HttpClient httpClient, ILoggerFactory? loggerFactory = null, bool ownsHttpClient = false)
+    {
+        Throw.IfNull(transportOptions);
+        Throw.IfNull(httpClient);
+
+        _options = transportOptions;
+        _httpClient = httpClient;
+        _loggerFactory = loggerFactory;
+        _ownsHttpClient = ownsHttpClient;
+        Name = transportOptions.Name ?? transportOptions.Endpoint.ToString();
+    }
+
+    /// <inheritdoc />
+    public string Name { get; }
+
+    /// <inheritdoc />
+    public async Task<ITransport> ConnectAsync(CancellationToken cancellationToken = default)
+    {
+        if (_options.UseStreamableHttp)
+        {
+            return new StreamableHttpClientSessionTransport(_options, _httpClient, _loggerFactory, Name);
+        }
+
+        var sessionTransport = new SseClientSessionTransport(_options, _httpClient, _loggerFactory, Name);
+
+        try
+        {
+            await sessionTransport.ConnectAsync(cancellationToken).ConfigureAwait(false);
+            return sessionTransport;
+        }
+        catch
+        {
+            await sessionTransport.DisposeAsync().ConfigureAwait(false);
+            throw;
+        }
+    }
+
+    /// <inheritdoc />
+    public ValueTask DisposeAsync()
+    {
+        if (_ownsHttpClient)
+        {
+            _httpClient.Dispose();
+        }
+
+        return default;
+    }
+}

src/ModelContextProtocol/Client/SseClientTransportOptions.cs

@@ -0,0 +1,65 @@
+namespace ModelContextProtocol.Client;
+
+/// <summary>
+/// Provides options for configuring <see cref="SseClientTransport"/> instances.
+/// </summary>
+public record SseClientTransportOptions
+{
+    /// <summary>
+    /// Gets or sets the base address of the server for SSE connections.
+    /// </summary>
+    public required Uri Endpoint
+    {
+        get;
+        init
+        {
+            if (value is null)
+            {
+                throw new ArgumentNullException(nameof(value), "Endpoint cannot be null.");
+            }
+            if (!value.IsAbsoluteUri)
+            {
+                throw new ArgumentException("Endpoint must be an absolute URI.", nameof(value));
+            }
+            if (value.Scheme != Uri.UriSchemeHttp && value.Scheme != Uri.UriSchemeHttps)
+            {
+                throw new ArgumentException("Endpoint must use HTTP or HTTPS scheme.", nameof(value));
+            }
+
+            field = value;
+        }
+    }
+
+    /// <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>.
+    /// </summary>
+    public bool UseStreamableHttp { get; init; }
+
+    /// <summary>
+    /// Gets a transport identifier used for logging purposes.
+    /// </summary>
+    public string? Name { get; init; }
+
+    /// <summary>
+    /// Gets or sets a timeout used to establish the initial connection to the SSE server. Defaults to 30 seconds.
+    /// </summary>
+    /// <remarks>
+    /// This timeout controls how long the client waits for:
+    /// <list type="bullet">
+    ///   <item><description>The initial HTTP connection to be established with the SSE server</description></item>
+    ///   <item><description>The endpoint event to be received, which indicates the message endpoint URL</description></item>
+    /// </list>
+    /// If the timeout expires before the connection is established, a <see cref="TimeoutException"/> will be thrown.
+    /// </remarks>
+    public TimeSpan ConnectionTimeout { get; init; } = TimeSpan.FromSeconds(30);
+
+    /// <summary>
+    /// Gets custom HTTP headers to include in requests to the SSE server.
+    /// </summary>
+    /// <remarks>
+    /// Use this property to specify custom HTTP headers that should be sent with each request to the server.
+    /// </remarks>
+    public Dictionary<string, string>? AdditionalHeaders { get; init; }
+}

src/ModelContextProtocol/Client/StdioClientSessionTransport.cs

@@ -0,0 +1,111 @@
+using Microsoft.Extensions.Logging;
+using ModelContextProtocol.Protocol;
+using System.Diagnostics;
+
+namespace ModelContextProtocol.Client;
+
+/// <summary>Provides the client side of a stdio-based session transport.</summary>
+internal sealed class StdioClientSessionTransport : StreamClientSessionTransport
+{
+    private readonly StdioClientTransportOptions _options;
+    private readonly Process _process;
+    private readonly Queue<string> _stderrRollingLog;
+    private int _cleanedUp = 0;
+
+    public StdioClientSessionTransport(StdioClientTransportOptions options, Process process, string endpointName, Queue<string> stderrRollingLog, ILoggerFactory? loggerFactory)
+        : base(process.StandardInput, process.StandardOutput, endpointName, loggerFactory)
+    {
+        _process = process;
+        _options = options;
+        _stderrRollingLog = stderrRollingLog;
+    }
+
+    /// <inheritdoc/>
+    public override async Task SendMessageAsync(JsonRpcMessage message, CancellationToken cancellationToken = default)
+    {
+        try
+        {
+            await base.SendMessageAsync(message, cancellationToken);
+        }
+        catch (IOException)
+        {
+            // We failed to send due to an I/O error. If the server process has exited, which is then very likely the cause
+            // for the I/O error, we should throw an exception for that instead.
+            if (await GetUnexpectedExitExceptionAsync(cancellationToken).ConfigureAwait(false) is Exception processExitException)
+            {
+                throw processExitException;
+            }
+
+            throw;
+        }
+    }
+
+    /// <inheritdoc/>
+    protected override async ValueTask CleanupAsync(Exception? error = null, CancellationToken cancellationToken = default)
+    {
+        // Only clean up once.
+        if (Interlocked.Exchange(ref _cleanedUp, 1) != 0)
+        {
+            return;
+        }
+
+        // We've not yet forcefully terminated the server. If it's already shut down, something went wrong,
+        // so create an exception with details about that.
+        error ??= await GetUnexpectedExitExceptionAsync(cancellationToken).ConfigureAwait(false);
+
+        // Now terminate the server process.
+        try
+        {
+            StdioClientTransport.DisposeProcess(_process, processRunning: true, _options.ShutdownTimeout, Name);
+        }
+        catch (Exception ex)
+        {
+            LogTransportShutdownFailed(Name, ex);
+        }
+
+        // And handle cleanup in the base type.
+        await base.CleanupAsync(error, cancellationToken);
+    }
+
+    private async ValueTask<Exception?> GetUnexpectedExitExceptionAsync(CancellationToken cancellationToken)
+    {
+        if (!StdioClientTransport.HasExited(_process))
+        {
+            return null;
+        }
+
+        Debug.Assert(StdioClientTransport.HasExited(_process));
+        try
+        {
+            // The process has exited, but we still need to ensure stderr has been flushed.
+#if NET
+            await _process.WaitForExitAsync(cancellationToken).ConfigureAwait(false);
+#else
+            _process.WaitForExit();
+#endif
+        }
+        catch { }
+
+        string errorMessage = "MCP server process exited unexpectedly";
+
+        string? exitCode = null;
+        try
+        {
+            exitCode = $" (exit code: {(uint)_process.ExitCode})";
+        }
+        catch { }
+
+        lock (_stderrRollingLog)
+        {
+            if (_stderrRollingLog.Count > 0)
+            {
+                errorMessage =
+                    $"{errorMessage}{exitCode}{Environment.NewLine}" +
+                    $"Server's stderr tail:{Environment.NewLine}" +
+                    $"{string.Join(Environment.NewLine, _stderrRollingLog)}";
+            }
+        }
+
+        return new IOException(errorMessage);
+    }
+}

src/ModelContextProtocol/Client/StdioClientTransport.cs

@@ -0,0 +1,262 @@
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Logging.Abstractions;
+using ModelContextProtocol.Protocol;
+using System.Diagnostics;
+using System.Runtime.InteropServices;
+using System.Text;
+using System.Text.RegularExpressions;
+
+#pragma warning disable CS1998 // Async method lacks 'await' operators and will run synchronously
+
+namespace ModelContextProtocol.Client;
+
+/// <summary>
+/// Provides a <see cref="IClientTransport"/> implemented via "stdio" (standard input/output).
+/// </summary>
+/// <remarks>
+/// <para>
+/// This transport launches an external process and communicates with it through standard input and output streams.
+/// It's used to connect to MCP servers launched and hosted in child processes.
+/// </para>
+/// <para>
+/// The transport manages the entire lifecycle of the process: starting it with specified command-line arguments
+/// and environment variables, handling output, and properly terminating the process when the transport is closed.
+/// </para>
+/// </remarks>
+public sealed partial class StdioClientTransport : IClientTransport
+{
+    private readonly StdioClientTransportOptions _options;
+    private readonly ILoggerFactory? _loggerFactory;
+
+    /// <summary>
+    /// Initializes a new instance of the <see cref="StdioClientTransport"/> class.
+    /// </summary>
+    /// <param name="options">Configuration options for the transport, including the command to execute, arguments, working directory, and environment variables.</param>
+    /// <param name="loggerFactory">Logger factory for creating loggers used for diagnostic output during transport operations.</param>
+    public StdioClientTransport(StdioClientTransportOptions options, ILoggerFactory? loggerFactory = null)
+    {
+        Throw.IfNull(options);
+
+        _options = options;
+        _loggerFactory = loggerFactory;
+        Name = options.Name ?? $"stdio-{Regex.Replace(Path.GetFileName(options.Command), @"[\s\.]+", "-")}";
+    }
+
+    /// <inheritdoc />
+    public string Name { get; }
+
+    /// <inheritdoc />
+    public async Task<ITransport> ConnectAsync(CancellationToken cancellationToken = default)
+    {
+        string endpointName = Name;
+
+        Process? process = null;
+        bool processStarted = false;
+
+        string command = _options.Command;
+        IList<string>? arguments = _options.Arguments;
+        if (RuntimeInformation.IsOSPlatform(OSPlatform.Windows) &&
+            !string.Equals(Path.GetFileName(command), "cmd.exe", StringComparison.OrdinalIgnoreCase))
+        {
+            // On Windows, for stdio, we need to wrap non-shell commands with cmd.exe /c {command} (usually npx or uvicorn).
+            // The stdio transport will not work correctly if the command is not run in a shell.
+            arguments = arguments is null or [] ? ["/c", command] : ["/c", command, ..arguments];
+            command = "cmd.exe";
+        }
+
+        ILogger logger = (ILogger?)_loggerFactory?.CreateLogger<StdioClientTransport>() ?? NullLogger.Instance;
+        try
+        {
+            LogTransportConnecting(logger, endpointName);
+
+            UTF8Encoding noBomUTF8 = new(encoderShouldEmitUTF8Identifier: false);
+
+            ProcessStartInfo startInfo = new()
+            {
+                FileName = command,
+                RedirectStandardInput = true,
+                RedirectStandardOutput = true,
+                RedirectStandardError = true,
+                UseShellExecute = false,
+                CreateNoWindow = true,
+                WorkingDirectory = _options.WorkingDirectory ?? Environment.CurrentDirectory,
+                StandardOutputEncoding = noBomUTF8,
+                StandardErrorEncoding = noBomUTF8,
+#if NET
+                StandardInputEncoding = noBomUTF8,
+#endif
+            };
+
+            if (arguments is not null) 
+            {
+#if NET
+                foreach (string arg in arguments)
+                {
+                    startInfo.ArgumentList.Add(arg);
+                }
+#else
+                StringBuilder argsBuilder = new();
+                foreach (string arg in arguments)
+                {
+                    PasteArguments.AppendArgument(argsBuilder, arg);
+                }
+
+                startInfo.Arguments = argsBuilder.ToString();
+#endif
+            }
+
+            if (_options.EnvironmentVariables != null)
+            {
+                foreach (var entry in _options.EnvironmentVariables)
+                {
+                    startInfo.Environment[entry.Key] = entry.Value;
+                }
+            }
+
+            if (logger.IsEnabled(LogLevel.Trace))
+            {
+                LogCreateProcessForTransportSensitive(logger, endpointName, _options.Command,
+                    startInfo.Arguments,
+                    string.Join(", ", startInfo.Environment.Select(kvp => $"{kvp.Key}={kvp.Value}")),
+                    startInfo.WorkingDirectory);
+            }
+            else
+            {
+                LogCreateProcessForTransport(logger, endpointName, _options.Command);
+            }
+
+            process = new() { StartInfo = startInfo };
+
+            // Set up stderr handling. Log all stderr output, and keep the last
+            // few lines in a rolling log for use in exceptions.
+            const int MaxStderrLength = 10; // keep the last 10 lines of stderr
+            Queue<string> stderrRollingLog = new(MaxStderrLength);
+            process.ErrorDataReceived += (sender, args) =>
+            {
+                string? data = args.Data;
+                if (data is not null)
+                {
+                    lock (stderrRollingLog)
+                    {
+                        if (stderrRollingLog.Count >= MaxStderrLength)
+                        {
+                            stderrRollingLog.Dequeue();
+                        }
+
+                        stderrRollingLog.Enqueue(data);
+                    }
+
+                    LogReadStderr(logger, endpointName, data);
+                }
+            };
+
+            // We need both stdin and stdout to use a no-BOM UTF-8 encoding. On .NET Core,
+            // we can use ProcessStartInfo.StandardOutputEncoding/StandardInputEncoding, but
+            // StandardInputEncoding doesn't exist on .NET Framework; instead, it always picks
+            // up the encoding from Console.InputEncoding. As such, when not targeting .NET Core,
+            // we temporarily change Console.InputEncoding to no-BOM UTF-8 around the Process.Start
+            // call, to ensure it picks up the correct encoding.
+#if NET
+            processStarted = process.Start();
+#else
+            Encoding originalInputEncoding = Console.InputEncoding;
+            try
+            {
+                Console.InputEncoding = noBomUTF8;
+                processStarted = process.Start();
+            }
+            finally
+            {
+                Console.InputEncoding = originalInputEncoding;
+            }
+#endif
+
+            if (!processStarted)
+            {
+                LogTransportProcessStartFailed(logger, endpointName);
+                throw new IOException("Failed to start MCP server process.");
+            }
+
+            LogTransportProcessStarted(logger, endpointName, process.Id);
+
+            process.BeginErrorReadLine();
+
+            return new StdioClientSessionTransport(_options, process, endpointName, stderrRollingLog, _loggerFactory);
+        }
+        catch (Exception ex)
+        {
+            LogTransportConnectFailed(logger, endpointName, ex);
+
+            try
+            {
+                DisposeProcess(process, processStarted, _options.ShutdownTimeout, endpointName);
+            }
+            catch (Exception ex2)
+            {
+                LogTransportShutdownFailed(logger, endpointName, ex2);
+            }
+
+            throw new IOException("Failed to connect transport.", ex);
+        }
+    }
+
+    internal static void DisposeProcess(
+        Process? process, bool processRunning, TimeSpan shutdownTimeout, string endpointName)
+    {
+        if (process is not null)
+        {
+            try
+            {
+                processRunning = processRunning && !HasExited(process);
+                if (processRunning)
+                {
+                    // Wait for the process to exit.
+                    // Kill the while process tree because the process may spawn child processes
+                    // and Node.js does not kill its children when it exits properly.
+                    process.KillTree(shutdownTimeout);
+                }
+            }
+            finally
+            {
+                process.Dispose();
+            }
+        }
+    }
+
+    /// <summary>Gets whether <paramref name="process"/> has exited.</summary>
+    internal static bool HasExited(Process process)
+    {
+        try
+        {
+            return process.HasExited;
+        }
+        catch
+        {
+            return true;
+        }
+    }
+
+    [LoggerMessage(Level = LogLevel.Information, Message = "{EndpointName} connecting.")]
+    private static partial void LogTransportConnecting(ILogger logger, string endpointName);
+
+    [LoggerMessage(Level = LogLevel.Information, Message = "{EndpointName} starting server process. Command: '{Command}'.")]
+    private static partial void LogCreateProcessForTransport(ILogger logger, string endpointName, string command);
+
+    [LoggerMessage(Level = LogLevel.Trace, Message = "{EndpointName} starting server process. Command: '{Command}', Arguments: {Arguments}, Environment: {Environment}, Working directory: {WorkingDirectory}.")]
+    private static partial void LogCreateProcessForTransportSensitive(ILogger logger, string endpointName, string command, string? arguments, string environment, string workingDirectory);
+
+    [LoggerMessage(Level = LogLevel.Warning, Message = "{EndpointName} failed to start server process.")]
+    private static partial void LogTransportProcessStartFailed(ILogger logger, string endpointName);
+
+    [LoggerMessage(Level = LogLevel.Information, Message = "{EndpointName} received stderr log: '{Data}'.")]
+    private static partial void LogReadStderr(ILogger logger, string endpointName, string data);
+
+    [LoggerMessage(Level = LogLevel.Information, Message = "{EndpointName} started server process with PID {ProcessId}.")]
+    private static partial void LogTransportProcessStarted(ILogger logger, string endpointName, int processId);
+
+    [LoggerMessage(Level = LogLevel.Warning, Message = "{EndpointName} connect failed.")]
+    private static partial void LogTransportConnectFailed(ILogger logger, string endpointName, Exception exception);
+
+    [LoggerMessage(Level = LogLevel.Warning, Message = "{EndpointName} shutdown failed.")]
+    private static partial void LogTransportShutdownFailed(ILogger logger, string endpointName, Exception exception);
+}

src/ModelContextProtocol/Client/StdioClientTransportOptions.cs

@@ -0,0 +1,64 @@
+namespace ModelContextProtocol.Client;
+
+/// <summary>
+/// Provides options for configuring <see cref="StdioClientTransport"/> instances.
+/// </summary>
+public record StdioClientTransportOptions
+{
+    /// <summary>
+    /// Gets or sets the command to execute to start the server process.
+    /// </summary>
+    public required string Command
+    {
+        get;
+        set
+        {
+            if (string.IsNullOrWhiteSpace(value))
+            {
+                throw new ArgumentException("Command cannot be null or empty.", nameof(value));
+            }
+
+            field = value;
+        }
+    }
+
+    /// <summary>
+    /// Gets or sets the arguments to pass to the server process when it is started.
+    /// </summary>
+    public IList<string>? Arguments { get; set; }
+
+    /// <summary>
+    /// Gets or sets a transport identifier used for logging purposes.
+    /// </summary>
+    public string? Name { get; set; }
+
+    /// <summary>
+    /// Gets or sets the working directory for the server process.
+    /// </summary>
+    public string? WorkingDirectory { get; set; }
+
+    /// <summary>
+    /// Gets or sets environment variables to set for the server process.
+    /// </summary>
+    /// <remarks>
+    /// This property allows you to specify environment variables that will be set in the server process's
+    /// environment. This is useful for passing configuration, authentication information, or runtime flags
+    /// to the server without modifying its code.
+    /// </remarks>
+    public Dictionary<string, string>? EnvironmentVariables { get; set; }
+
+    /// <summary>
+    /// Gets or sets the timeout to wait for the server to shut down gracefully.
+    /// </summary>
+    /// <remarks>
+    /// <para>
+    /// This property dictates how long the client should wait for the server process to exit cleanly during shutdown
+    /// before forcibly terminating it. This balances between giving the server enough time to clean up 
+    /// resources and not hanging indefinitely if a server process becomes unresponsive.
+    /// </para>
+    /// <para>
+    /// The default is five seconds.
+    /// </para>
+    /// </remarks>
+    public TimeSpan ShutdownTimeout { get; set; } = TimeSpan.FromSeconds(5);
+}

src/ModelContextProtocol/Client/StreamClientSessionTransport.cs

@@ -0,0 +1,182 @@
+using Microsoft.Extensions.Logging;
+using ModelContextProtocol.Protocol;
+using System.Text.Json;
+
+namespace ModelContextProtocol.Client;
+
+/// <summary>Provides the client side of a stream-based session transport.</summary>
+internal class StreamClientSessionTransport : TransportBase
+{
+    private readonly TextReader _serverOutput;
+    private readonly TextWriter _serverInput;
+    private readonly SemaphoreSlim _sendLock = new(1, 1);
+    private CancellationTokenSource? _shutdownCts = new();
+    private Task? _readTask;
+
+    /// <summary>
+    /// Initializes a new instance of the <see cref="StreamClientSessionTransport"/> class.
+    /// </summary>
+    /// <param name="serverInput">
+    /// The text writer connected to the server's input stream. 
+    /// Messages written to this writer will be sent to the server.
+    /// </param>
+    /// <param name="serverOutput">
+    /// The text reader connected to the server's output stream.
+    /// Messages read from this reader will be received from the server.
+    /// </param>
+    /// <param name="endpointName">
+    /// A name that identifies this transport endpoint in logs.
+    /// </param>
+    /// <param name="loggerFactory">
+    /// Optional factory for creating loggers. If null, a NullLogger will be used.
+    /// </param>
+    /// <remarks>
+    /// This constructor starts a background task to read messages from the server output stream.
+    /// The transport will be marked as connected once initialized.
+    /// </remarks>
+    public StreamClientSessionTransport(
+        TextWriter serverInput, TextReader serverOutput, string endpointName, ILoggerFactory? loggerFactory)
+        : base(endpointName, loggerFactory)
+    {
+        _serverOutput = serverOutput;
+        _serverInput = serverInput;
+
+        // Start reading messages in the background. We use the rarer pattern of new Task + Start
+        // in order to ensure that the body of the task will always see _readTask initialized.
+        // It is then able to reliably null it out on completion.
+        var readTask = new Task<Task>(
+            thisRef => ((StreamClientSessionTransport)thisRef!).ReadMessagesAsync(_shutdownCts.Token), 
+            this,
+            TaskCreationOptions.DenyChildAttach);
+        _readTask = readTask.Unwrap();
+        readTask.Start();
+
+        SetConnected();
+    }
+
+    /// <inheritdoc/>
+    public override async Task SendMessageAsync(JsonRpcMessage message, CancellationToken cancellationToken = default)
+    {
+        string id = "(no id)";
+        if (message is JsonRpcMessageWithId messageWithId)
+        {
+            id = messageWithId.Id.ToString();
+        }
+
+        var json = JsonSerializer.Serialize(message, McpJsonUtilities.JsonContext.Default.JsonRpcMessage);
+
+        using var _ = await _sendLock.LockAsync(cancellationToken).ConfigureAwait(false);
+        try
+        {
+            // Write the message followed by a newline using our UTF-8 writer
+            await _serverInput.WriteLineAsync(json).ConfigureAwait(false);
+            await _serverInput.FlushAsync(cancellationToken).ConfigureAwait(false);
+        }
+        catch (Exception ex)
+        {
+            LogTransportSendFailed(Name, id, ex);
+            throw new IOException("Failed to send message.", ex);
+        }
+    }
+
+    /// <inheritdoc/>
+    public override ValueTask DisposeAsync() => 
+        CleanupAsync(cancellationToken: CancellationToken.None);
+
+    private async Task ReadMessagesAsync(CancellationToken cancellationToken)
+    {
+        Exception? error = null;
+        try
+        {
+            LogTransportEnteringReadMessagesLoop(Name);
+
+            while (true)
+            {
+                if (await _serverOutput.ReadLineAsync(cancellationToken).ConfigureAwait(false) is not string line)
+                {
+                    LogTransportEndOfStream(Name);
+                    break;
+                }
+
+                if (string.IsNullOrWhiteSpace(line))
+                {
+                    continue;
+                }
+
+                LogTransportReceivedMessageSensitive(Name, line);
+
+                await ProcessMessageAsync(line, cancellationToken).ConfigureAwait(false);
+            }
+        }
+        catch (OperationCanceledException)
+        {
+            LogTransportReadMessagesCancelled(Name);
+        }
+        catch (Exception ex)
+        {
+            error = ex;
+            LogTransportReadMessagesFailed(Name, ex);
+        }
+        finally
+        {
+            _readTask = null;
+            await CleanupAsync(error, cancellationToken).ConfigureAwait(false);
+        }
+    }
+
+    private async Task ProcessMessageAsync(string line, CancellationToken cancellationToken)
+    {
+        try
+        {
+            var message = (JsonRpcMessage?)JsonSerializer.Deserialize(line.AsSpan().Trim(), McpJsonUtilities.DefaultOptions.GetTypeInfo(typeof(JsonRpcMessage)));
+            if (message != null)
+            {
+                await WriteMessageAsync(message, cancellationToken).ConfigureAwait(false);
+            }
+            else
+            {
+                LogTransportMessageParseUnexpectedTypeSensitive(Name, line);
+            }
+        }
+        catch (JsonException ex)
+        {
+            if (Logger.IsEnabled(LogLevel.Trace))
+            {
+                LogTransportMessageParseFailedSensitive(Name, line, ex);
+            }
+            else
+            {
+                LogTransportMessageParseFailed(Name, ex);
+            }
+        }
+    }
+
+    protected virtual async ValueTask CleanupAsync(Exception? error = null, CancellationToken cancellationToken = default)
+    {
+        LogTransportShuttingDown(Name);
+
+        if (Interlocked.Exchange(ref _shutdownCts, null) is { } shutdownCts)
+        {
+            await shutdownCts.CancelAsync().ConfigureAwait(false);
+            shutdownCts.Dispose();
+        }
+
+        if (Interlocked.Exchange(ref _readTask, null) is Task readTask)
+        {
+            try
+            {
+                await readTask.WaitAsync(TimeSpan.FromSeconds(5), cancellationToken).ConfigureAwait(false);
+            }
+            catch (OperationCanceledException)
+            {
+            }
+            catch (Exception ex)
+            {
+                LogTransportCleanupReadTaskFailed(Name, ex);
+            }
+        }
+
+        SetDisconnected(error);
+        LogTransportShutDown(Name);
+    }
+}

src/ModelContextProtocol/Client/StreamClientTransport.cs

@@ -0,0 +1,55 @@
+using Microsoft.Extensions.Logging;
+using ModelContextProtocol.Client;
+
+namespace ModelContextProtocol.Protocol;
+
+/// <summary>
+/// Provides an <see cref="IClientTransport"/> implemented around a pair of input/output streams.
+/// </summary>
+/// <remarks>
+/// This transport is useful for scenarios where you already have established streams for communication,
+/// such as custom network protocols, pipe connections, or for testing purposes. It works with any
+/// readable and writable streams.
+/// </remarks>
+public sealed class StreamClientTransport : IClientTransport
+{
+    private readonly Stream _serverInput;
+    private readonly Stream _serverOutput;
+    private readonly ILoggerFactory? _loggerFactory;
+
+    /// <summary>
+    /// Initializes a new instance of the <see cref="StreamClientTransport"/> class.
+    /// </summary>
+    /// <param name="serverInput">
+    /// The stream representing the connected server's input. 
+    /// Writes to this stream will be sent to the server.
+    /// </param>
+    /// <param name="serverOutput">
+    /// The stream representing the connected server's output.
+    /// Reads from this stream will receive messages from the server.
+    /// </param>
+    /// <param name="loggerFactory">A logger factory for creating loggers.</param>
+    public StreamClientTransport(
+        Stream serverInput, Stream serverOutput, ILoggerFactory? loggerFactory = null)
+    {
+        Throw.IfNull(serverInput);
+        Throw.IfNull(serverOutput);
+
+        _serverInput = serverInput;
+        _serverOutput = serverOutput;
+        _loggerFactory = loggerFactory;
+    }
+
+    /// <inheritdoc />
+    public string Name => "in-memory-stream";
+
+    /// <inheritdoc />
+    public Task<ITransport> ConnectAsync(CancellationToken cancellationToken = default)
+    {
+        return Task.FromResult<ITransport>(new StreamClientSessionTransport(
+            new StreamWriter(_serverInput),
+            new StreamReader(_serverOutput),
+            "Client (stream)",
+            _loggerFactory));
+    }
+}

src/ModelContextProtocol/Client/StreamableHttpClientSessionTransport.cs

@@ -0,0 +1,238 @@
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Logging.Abstractions;
+using System.Net.Http.Headers;
+using System.Net.ServerSentEvents;
+using System.Text.Json;
+using ModelContextProtocol.Protocol;
+#if NET
+using System.Net.Http.Json;
+#else
+using System.Text;
+#endif
+
+namespace ModelContextProtocol.Client;
+
+/// <summary>
+/// The Streamable HTTP client transport implementation
+/// </summary>
+internal sealed partial class StreamableHttpClientSessionTransport : TransportBase
+{
+    private static readonly MediaTypeWithQualityHeaderValue s_applicationJsonMediaType = new("application/json");
+    private static readonly MediaTypeWithQualityHeaderValue s_textEventStreamMediaType = new("text/event-stream");
+
+    private readonly HttpClient _httpClient;
+    private readonly SseClientTransportOptions _options;
+    private readonly CancellationTokenSource _connectionCts;
+    private readonly ILogger _logger;
+
+    private string? _mcpSessionId;
+    private Task? _getReceiveTask;
+
+    public StreamableHttpClientSessionTransport(SseClientTransportOptions transportOptions, HttpClient httpClient, ILoggerFactory? loggerFactory, string endpointName)
+        : base(endpointName, loggerFactory)
+    {
+        Throw.IfNull(transportOptions);
+        Throw.IfNull(httpClient);
+
+        _options = transportOptions;
+        _httpClient = httpClient;
+        _connectionCts = new CancellationTokenSource();
+        _logger = (ILogger?)loggerFactory?.CreateLogger<SseClientTransport>() ?? NullLogger.Instance;
+
+        // We connect with the initialization request with the MCP transport. This means that any errors won't be observed
+        // until the first call to SendMessageAsync. Fortunately, that happens internally in McpClientFactory.ConnectAsync
+        // so we still throw any connection-related Exceptions from there and never expose a pre-connected client to the user.
+        SetConnected();
+    }
+
+    /// <inheritdoc/>
+    public override async Task SendMessageAsync(
+        JsonRpcMessage message,
+        CancellationToken cancellationToken = default)
+    {
+        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"
+        );
+#endif
+
+        using var httpRequestMessage = new HttpRequestMessage(HttpMethod.Post, _options.Endpoint)
+        {
+            Content = content,
+            Headers =
+            {
+                Accept = { s_applicationJsonMediaType, s_textEventStreamMediaType },
+            },
+        };
+
+        CopyAdditionalHeaders(httpRequestMessage.Headers, _options.AdditionalHeaders, _mcpSessionId);
+        using var response = await _httpClient.SendAsync(httpRequestMessage, HttpCompletionOption.ResponseHeadersRead, cancellationToken).ConfigureAwait(false);
+
+        response.EnsureSuccessStatusCode();
+
+        var rpcRequest = message as JsonRpcRequest;
+        JsonRpcMessage? rpcResponseCandidate = null;
+
+        if (response.Content.Headers.ContentType?.MediaType == "application/json")
+        {
+            var responseContent = await response.Content.ReadAsStringAsync(cancellationToken).ConfigureAwait(false);
+            rpcResponseCandidate = await ProcessMessageAsync(responseContent, cancellationToken).ConfigureAwait(false);
+        }
+        else if (response.Content.Headers.ContentType?.MediaType == "text/event-stream")
+        {
+            using var responseBodyStream = await response.Content.ReadAsStreamAsync(cancellationToken);
+            rpcResponseCandidate = await ProcessSseResponseAsync(responseBodyStream, rpcRequest, cancellationToken).ConfigureAwait(false);
+        }
+
+        if (rpcRequest is null)
+        {
+            return;
+        }
+
+        if (rpcResponseCandidate is not JsonRpcMessageWithId messageWithId || messageWithId.Id != rpcRequest.Id)
+        {
+            throw new McpException($"Streamable HTTP POST response completed without a reply to request with ID: {rpcRequest.Id}");
+        }
+
+        if (rpcRequest.Method == RequestMethods.Initialize && rpcResponseCandidate is JsonRpcResponse)
+        {
+            // We've successfully initialized! Copy session-id and start GET request if any.
+            if (response.Headers.TryGetValues("mcp-session-id", out var sessionIdValues))
+            {
+                _mcpSessionId = sessionIdValues.FirstOrDefault();
+            }
+
+            _getReceiveTask = ReceiveUnsolicitedMessagesAsync();
+        }
+    }
+
+    public override async ValueTask DisposeAsync()
+    {
+        try
+        {
+            await _connectionCts.CancelAsync().ConfigureAwait(false);
+
+            try
+            {
+                if (_getReceiveTask != null)
+                {
+                    await _getReceiveTask.ConfigureAwait(false);
+                }
+            }
+            catch (OperationCanceledException)
+            {
+            }
+            finally
+            {
+                _connectionCts.Dispose();
+            }
+        }
+        finally
+        {
+            SetDisconnected();
+        }
+    }
+
+    private async Task ReceiveUnsolicitedMessagesAsync()
+    {
+        // Send a GET request to handle any unsolicited messages not sent over a POST response.
+        using var request = new HttpRequestMessage(HttpMethod.Get, _options.Endpoint);
+        request.Headers.Accept.Add(s_textEventStreamMediaType);
+        CopyAdditionalHeaders(request.Headers, _options.AdditionalHeaders, _mcpSessionId);
+
+        using var response = await _httpClient.SendAsync(request, HttpCompletionOption.ResponseHeadersRead, _connectionCts.Token).ConfigureAwait(false);
+
+        if (!response.IsSuccessStatusCode)
+        {
+            // Server support for the GET request is optional. If it fails, we don't care. It just means we won't receive unsolicited messages.
+            return;
+        }
+
+        using var responseStream = await response.Content.ReadAsStreamAsync(_connectionCts.Token).ConfigureAwait(false);
+        await ProcessSseResponseAsync(responseStream, relatedRpcRequest: null, _connectionCts.Token).ConfigureAwait(false);
+    }
+
+    private async Task<JsonRpcMessageWithId?> ProcessSseResponseAsync(Stream responseStream, JsonRpcRequest? relatedRpcRequest, CancellationToken cancellationToken)
+    {
+        await foreach (SseItem<string> sseEvent in SseParser.Create(responseStream).EnumerateAsync(cancellationToken).ConfigureAwait(false))
+        {
+            if (sseEvent.EventType != "message")
+            {
+                continue;
+            }
+
+            var message = await ProcessMessageAsync(sseEvent.Data, cancellationToken).ConfigureAwait(false);
+
+            // The server SHOULD end the response here anyway, but we won't leave it to chance. This transport makes
+            // a GET request for any notifications that might need to be sent after the completion of each POST.
+            if (message is JsonRpcMessageWithId messageWithId && relatedRpcRequest?.Id == messageWithId.Id)
+            {
+                return messageWithId;
+            }
+        }
+
+        return null;
+    }
+
+    private async Task<JsonRpcMessage?> ProcessMessageAsync(string data, CancellationToken cancellationToken)
+    {
+        try
+        {
+            var message = JsonSerializer.Deserialize(data, McpJsonUtilities.JsonContext.Default.JsonRpcMessage);
+            if (message is null)
+            {
+                LogTransportMessageParseUnexpectedTypeSensitive(Name, data);
+                return null;
+            }
+
+            await WriteMessageAsync(message, cancellationToken).ConfigureAwait(false);
+            return message;
+        }
+        catch (JsonException ex)
+        {
+            LogJsonException(ex, data);
+        }
+
+        return null;
+    }
+
+    private void LogJsonException(JsonException ex, string data)
+    {
+        if (_logger.IsEnabled(LogLevel.Trace))
+        {
+            LogTransportMessageParseFailedSensitive(Name, data, ex);
+        }
+        else
+        {
+            LogTransportMessageParseFailed(Name, ex);
+        }
+    }
+
+    internal static void CopyAdditionalHeaders(HttpRequestHeaders headers, Dictionary<string, string>? additionalHeaders, string? sessionId = null)
+    {
+        if (sessionId is not null)
+        {
+            headers.Add("mcp-session-id", sessionId);
+        }
+
+        if (additionalHeaders is null)
+        {
+            return;
+        }
+
+        foreach (var header in additionalHeaders)
+        {
+            if (!headers.TryAddWithoutValidation(header.Key, header.Value))
+            {
+                throw new InvalidOperationException($"Failed to add header '{header.Key}' with value '{header.Value}' from {nameof(SseClientTransportOptions.AdditionalHeaders)}.");
+            }
+        }
+    }
+}