Make ClientTransportClosedException public and unify transport exception handling (#1467)

Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>
Co-authored-by: stephentoub <2642209+stephentoub@users.noreply.github.com>
Co-authored-by: halter73 <54385+halter73@users.noreply.github.com>
Co-authored-by: Stephen Toub <stoub@microsoft.com>
This commit is contained in:
Copilot 2026-03-29 20:21:03 -04:00 • committed by GitHub
commit 7662ce6b70
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
11 changed files with 253 additions and 38 deletions

View file

@ -0,0 +1,40 @@
using ModelContextProtocol.Protocol;
using System.Threading.Channels;
namespace ModelContextProtocol.Client;
/// <summary>
/// An <see cref="IOException"/> that indicates the client transport was closed, carrying
/// structured <see cref="ClientCompletionDetails"/> about why the closure occurred.
/// </summary>
/// <remarks>
/// <para>
/// This exception is thrown when an MCP transport closes, either during initialization
/// (e.g., from <see cref="McpClient.CreateAsync"/>) or during an active session.
/// Callers can catch this exception to access the <see cref="Details"/> property
/// for structured information about the closure.
/// </para>
/// <para>
/// For stdio-based transports, the <see cref="Details"/> will be a
/// <see cref="StdioClientCompletionDetails"/> instance providing access to the
/// server process exit code, process ID, and standard error output.
/// </para>
/// <para>
/// Custom <see cref="ITransport"/> implementations can provide their own
/// <see cref="ClientCompletionDetails"/>-derived types by completing their
/// <see cref="ChannelWriter{T}"/> with this exception.
/// </para>
/// </remarks>
public sealed class ClientTransportClosedException(ClientCompletionDetails details) :
IOException(details.Exception?.Message ?? "The transport was closed.", details.Exception)
{
/// <summary>
/// Gets the structured details about why the transport was closed.
/// </summary>
/// <remarks>
/// The concrete type of the returned <see cref="ClientCompletionDetails"/> depends on
/// the transport that was used. For example, <see cref="StdioClientCompletionDetails"/>
/// for stdio-based transports and <see cref="HttpClientCompletionDetails"/> for HTTP-based transports.
/// </remarks>
public ClientCompletionDetails Details { get; } = details;
}

View file

@ -1,6 +1,7 @@
using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging;
using ModelContextProtocol.Protocol; using ModelContextProtocol.Protocol;
using ModelContextProtocol.Server; using ModelContextProtocol.Server;
using System.Diagnostics;
using System.Diagnostics.CodeAnalysis; using System.Diagnostics.CodeAnalysis;
using System.Text.Json; using System.Text.Json;
using System.Text.Json.Nodes; using System.Text.Json.Nodes;
@ -52,13 +53,25 @@ public abstract partial class McpClient : McpSession
{ {
await clientSession.ConnectAsync(cancellationToken).ConfigureAwait(false); await clientSession.ConnectAsync(cancellationToken).ConfigureAwait(false);
} }
catch catch (Exception ex) when (ex is not OperationCanceledException and not ClientTransportClosedException)
{ {
try // ConnectAsync already disposed the session (which includes awaiting Completion).
// Check if the transport provided structured completion details indicating
// why the transport closed that aren't already in the original exception chain.
Debug.Assert(clientSession.Completion.IsCompleted, "Completion should already be finished after ConnectAsync's DisposeAsync.");
var completionDetails = await clientSession.Completion.ConfigureAwait(false);
// If the transport closed with a non-graceful error (e.g., server process exited)
// and the completion details carry an exception that's NOT already in the original
// exception chain, throw a ClientTransportClosedException with the structured details so
// callers can programmatically inspect the closure reason (exit code, stderr, etc.).
// When the same exception is already in the chain (e.g., HttpRequestException from
// an HTTP transport), the original exception is more appropriate to re-throw.
if (completionDetails.Exception is { } detailsException &&
!ExceptionChainContains(ex, detailsException))
{ {
await clientSession.DisposeAsync().ConfigureAwait(false); throw new ClientTransportClosedException(completionDetails);
} }
catch { } // allow the original exception to propagate
throw; throw;
} }
@ -66,6 +79,23 @@ public abstract partial class McpClient : McpSession
return clientSession; return clientSession;
} }
/// <summary>
/// Returns <see langword="true"/> if <paramref name="target"/> is the same object as
/// <paramref name="exception"/> or any exception in its <see cref="Exception.InnerException"/> chain.
/// </summary>
private static bool ExceptionChainContains(Exception exception, Exception target)
{
for (Exception? current = exception; current is not null; current = current.InnerException)
{
if (ReferenceEquals(current, target))
{
return true;
}
}
return false;
}
/// <summary> /// <summary>
/// Recreates an <see cref="McpClient"/> using an existing transport session without sending a new initialize request. /// Recreates an <see cref="McpClient"/> using an existing transport session without sending a new initialize request.
/// </summary> /// </summary>

View file

@ -58,11 +58,15 @@ internal sealed partial class SseClientSessionTransport : TransportBase
await _connectionEstablished.Task.WaitAsync(_options.ConnectionTimeout, cancellationToken).ConfigureAwait(false); await _connectionEstablished.Task.WaitAsync(_options.ConnectionTimeout, cancellationToken).ConfigureAwait(false);
} }
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
{
throw;
}
catch (Exception ex) catch (Exception ex)
{ {
LogTransportConnectFailed(Name, ex); LogTransportConnectFailed(Name, ex);
await CloseAsync().ConfigureAwait(false); await CloseAsync().ConfigureAwait(false);
throw new InvalidOperationException("Failed to connect transport", ex); throw new IOException("Failed to connect transport.", ex);
} }
} }
@ -125,7 +129,7 @@ internal sealed partial class SseClientSessionTransport : TransportBase
} }
finally finally
{ {
SetDisconnected(new TransportClosedException(new HttpClientCompletionDetails())); SetDisconnected(new ClientTransportClosedException(new HttpClientCompletionDetails()));
} }
} }
@ -186,7 +190,7 @@ internal sealed partial class SseClientSessionTransport : TransportBase
} }
else else
{ {
SetDisconnected(new TransportClosedException(new HttpClientCompletionDetails SetDisconnected(new ClientTransportClosedException(new HttpClientCompletionDetails
{ {
HttpStatusCode = failureStatusCode, HttpStatusCode = failureStatusCode,
Exception = ex, Exception = ex,
@ -199,7 +203,7 @@ internal sealed partial class SseClientSessionTransport : TransportBase
} }
finally finally
{ {
SetDisconnected(new TransportClosedException(new HttpClientCompletionDetails())); SetDisconnected(new ClientTransportClosedException(new HttpClientCompletionDetails()));
} }
} }

View file

@ -64,12 +64,12 @@ internal sealed class StdioClientSessionTransport : StreamClientSessionTransport
_process, _process,
processRunning: true, processRunning: true,
_options.ShutdownTimeout, _options.ShutdownTimeout,
beforeDispose: () => SetDisconnected(new TransportClosedException(BuildCompletionDetails(error)))); beforeDispose: () => SetDisconnected(new ClientTransportClosedException(BuildCompletionDetails(error))));
} }
catch (Exception ex) catch (Exception ex)
{ {
LogTransportShutdownFailed(Name, ex); LogTransportShutdownFailed(Name, ex);
SetDisconnected(new TransportClosedException(BuildCompletionDetails(error))); SetDisconnected(new ClientTransportClosedException(BuildCompletionDetails(error)));
} }
// And handle cleanup in the base type. SetDisconnected has already been // And handle cleanup in the base type. SetDisconnected has already been

View file

@ -85,6 +85,10 @@ internal class StreamClientSessionTransport : TransportBase
await _serverInputStream.WriteAsync(s_newlineBytes, cancellationToken).ConfigureAwait(false); await _serverInputStream.WriteAsync(s_newlineBytes, cancellationToken).ConfigureAwait(false);
await _serverInputStream.FlushAsync(cancellationToken).ConfigureAwait(false); await _serverInputStream.FlushAsync(cancellationToken).ConfigureAwait(false);
} }
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
{
throw;
}
catch (Exception ex) catch (Exception ex)
{ {
LogTransportSendFailed(Name, id, ex); LogTransportSendFailed(Name, id, ex);

View file

@ -25,7 +25,7 @@ internal sealed partial class StreamableHttpClientSessionTransport : TransportBa
private string? _negotiatedProtocolVersion; private string? _negotiatedProtocolVersion;
private Task? _getReceiveTask; private Task? _getReceiveTask;
private volatile TransportClosedException? _disconnectError; private volatile ClientTransportClosedException? _disconnectError;
private readonly SemaphoreSlim _disposeLock = new(1, 1); private readonly SemaphoreSlim _disposeLock = new(1, 1);
private bool _disposed; private bool _disposed;
@ -200,7 +200,7 @@ internal sealed partial class StreamableHttpClientSessionTransport : TransportBa
{ {
// _disconnectError is set when the server returns 404 indicating session expiry. // _disconnectError is set when the server returns 404 indicating session expiry.
// When null, this is a graceful client-initiated closure (no error). // When null, this is a graceful client-initiated closure (no error).
SetDisconnected(_disconnectError ?? new TransportClosedException(new HttpClientCompletionDetails())); SetDisconnected(_disconnectError ?? new ClientTransportClosedException(new HttpClientCompletionDetails()));
} }
} }
} }
@ -491,7 +491,7 @@ internal sealed partial class StreamableHttpClientSessionTransport : TransportBa
{ {
// Store the error before canceling so DisposeAsync can use it if it races us, especially // Store the error before canceling so DisposeAsync can use it if it races us, especially
// after the call to Cancel below, to invoke SetDisconnected. // after the call to Cancel below, to invoke SetDisconnected.
_disconnectError = new TransportClosedException(new HttpClientCompletionDetails _disconnectError = new ClientTransportClosedException(new HttpClientCompletionDetails
{ {
HttpStatusCode = HttpStatusCode.NotFound, HttpStatusCode = HttpStatusCode.NotFound,
Exception = new McpException( Exception = new McpException(

View file

@ -1,19 +0,0 @@
using ModelContextProtocol.Protocol;
using System.Threading.Channels;
namespace ModelContextProtocol.Client;
/// <summary>
/// <see cref="IOException"/> used to smuggle <see cref="ClientCompletionDetails"/> through
/// the <see cref="ChannelWriter{T}.TryComplete(Exception?)"/> mechanism.
/// </summary>
/// <remarks>
/// This could be made public in the future to allow custom <see cref="ITransport"/>
/// implementations to provide their own <see cref="ClientCompletionDetails"/>-derived types
/// by completing their channel with this exception.
/// </remarks>
internal sealed class TransportClosedException(ClientCompletionDetails details) :
IOException(details.Exception?.Message, details.Exception)
{
public ClientCompletionDetails Details { get; } = details;
}

View file

@ -156,7 +156,7 @@ internal sealed partial class McpSessionHandler : IAsyncDisposable
/// <summary> /// <summary>
/// Gets a task that completes when the client session has completed, providing details about the closure. /// Gets a task that completes when the client session has completed, providing details about the closure.
/// Completion details are resolved from the transport's channel completion exception: if a transport /// Completion details are resolved from the transport's channel completion exception: if a transport
/// completes its channel with a <see cref="TransportClosedException"/>, the wrapped /// completes its channel with a <see cref="ClientTransportClosedException"/>, the wrapped
/// <see cref="ClientCompletionDetails"/> is unwrapped. Otherwise, a default instance is returned. /// <see cref="ClientCompletionDetails"/> is unwrapped. Otherwise, a default instance is returned.
/// </summary> /// </summary>
internal Task<ClientCompletionDetails> CompletionTask => internal Task<ClientCompletionDetails> CompletionTask =>
@ -325,16 +325,23 @@ internal sealed partial class McpSessionHandler : IAsyncDisposable
} }
// Fail any pending requests, as they'll never be satisfied. // Fail any pending requests, as they'll never be satisfied.
// If the transport's channel was completed with a ClientTransportClosedException,
// propagate it so callers can access the structured completion details.
Exception pendingException =
_transport.MessageReader.Completion is { IsCompleted: true, IsFaulted: true } completion &&
completion.Exception?.InnerException is { } innerException
? innerException
: new IOException("The server shut down unexpectedly.");
foreach (var entry in _pendingRequests) foreach (var entry in _pendingRequests)
{ {
entry.Value.TrySetException(new IOException("The server shut down unexpectedly.")); entry.Value.TrySetException(pendingException);
} }
} }
} }
/// <summary> /// <summary>
/// Resolves <see cref="ClientCompletionDetails"/> from the transport's channel completion. /// Resolves <see cref="ClientCompletionDetails"/> from the transport's channel completion.
/// If the channel was completed with a <see cref="TransportClosedException"/>, the wrapped /// If the channel was completed with a <see cref="ClientTransportClosedException"/>, the wrapped
/// details are returned. Otherwise a default instance is created from the completion state. /// details are returned. Otherwise a default instance is created from the completion state.
/// </summary> /// </summary>
private static async Task<ClientCompletionDetails> GetCompletionDetailsAsync(Task channelCompletion) private static async Task<ClientCompletionDetails> GetCompletionDetailsAsync(Task channelCompletion)
@ -344,7 +351,7 @@ internal sealed partial class McpSessionHandler : IAsyncDisposable
await channelCompletion.ConfigureAwait(false); await channelCompletion.ConfigureAwait(false);
return new ClientCompletionDetails(); return new ClientCompletionDetails();
} }
catch (TransportClosedException tce) catch (ClientTransportClosedException tce)
{ {
return tce.Details; return tce.Details;
} }
@ -942,7 +949,7 @@ internal sealed partial class McpSessionHandler : IAsyncDisposable
catch catch
{ {
// Ignore exceptions from the message processing loop. It may fault with // Ignore exceptions from the message processing loop. It may fault with
// OperationCanceledException on normal shutdown or TransportClosedException // OperationCanceledException on normal shutdown or ClientTransportClosedException
// when the transport's channel completes with an error. // when the transport's channel completes with an error.
} }
} }

View file

@ -80,6 +80,10 @@ public class StreamServerTransport : TransportBase
await _outputStream.WriteAsync(s_newlineBytes, cancellationToken).ConfigureAwait(false); await _outputStream.WriteAsync(s_newlineBytes, cancellationToken).ConfigureAwait(false);
await _outputStream.FlushAsync(cancellationToken).ConfigureAwait(false); await _outputStream.FlushAsync(cancellationToken).ConfigureAwait(false);
} }
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
{
throw;
}
catch (Exception ex) catch (Exception ex)
{ {
LogTransportSendFailed(Name, id, ex); LogTransportSendFailed(Name, id, ex);

View file

@ -4,6 +4,48 @@ namespace ModelContextProtocol.Tests.Client;
public class ClientCompletionDetailsTests public class ClientCompletionDetailsTests
{ {
[Fact]
public void ClientTransportClosedException_ExposesDetails()
{
var details = new StdioClientCompletionDetails
{
ExitCode = 42,
ProcessId = 12345,
StandardErrorTail = ["error line"],
Exception = new IOException("process exited"),
};
var exception = new ClientTransportClosedException(details);
Assert.IsType<StdioClientCompletionDetails>(exception.Details);
var stdioDetails = (StdioClientCompletionDetails)exception.Details;
Assert.Equal(42, stdioDetails.ExitCode);
Assert.Equal(12345, stdioDetails.ProcessId);
Assert.Equal(["error line"], stdioDetails.StandardErrorTail);
Assert.Equal("process exited", exception.Message);
Assert.IsType<IOException>(exception.InnerException);
}
[Fact]
public void ClientTransportClosedException_WithNullException_HasDefaultMessage()
{
var details = new ClientCompletionDetails();
var exception = new ClientTransportClosedException(details);
Assert.Equal("The transport was closed.", exception.Message);
Assert.Null(exception.InnerException);
Assert.Same(details, exception.Details);
}
[Fact]
public void ClientTransportClosedException_IsIOException()
{
var details = new ClientCompletionDetails();
IOException exception = new ClientTransportClosedException(details);
Assert.IsType<ClientTransportClosedException>(exception);
}
[Fact] [Fact]
public void ClientCompletionDetails_PropertiesRoundtrip() public void ClientCompletionDetails_PropertiesRoundtrip()
{ {

View file

@ -1,12 +1,14 @@
using Microsoft.Extensions.Logging;
using ModelContextProtocol.Client; using ModelContextProtocol.Client;
using ModelContextProtocol.Protocol; using ModelContextProtocol.Protocol;
using ModelContextProtocol.Tests.Utils;
using System.IO.Pipelines; using System.IO.Pipelines;
using System.Text.Json; using System.Text.Json;
using System.Threading.Channels; using System.Threading.Channels;
namespace ModelContextProtocol.Tests.Client; namespace ModelContextProtocol.Tests.Client;
public class McpClientCreationTests public class McpClientCreationTests(ITestOutputHelper testOutputHelper) : LoggedTest(testOutputHelper)
{ {
[Fact] [Fact]
public async Task CreateAsync_WithInvalidArgs_Throws() public async Task CreateAsync_WithInvalidArgs_Throws()
@ -101,6 +103,51 @@ public class McpClientCreationTests
} }
} }
[Fact]
public async Task CreateAsync_TransportChannelClosed_ThrowsClientTransportClosedException()
{
// Arrange - transport completes its read channel with ClientTransportClosedException
// when the client tries to send the initialize request (simulating a server process
// exit detected by the reader loop). SendMessageAsync returns successfully —
// only the read side fails.
var transport = new ChannelClosedDuringInitTransport();
// Act & Assert
var ex = await Assert.ThrowsAsync<ClientTransportClosedException>(
() => McpClient.CreateAsync(transport, loggerFactory: LoggerFactory, cancellationToken: TestContext.Current.CancellationToken));
var details = Assert.IsType<StdioClientCompletionDetails>(ex.Details);
Assert.Equal(42, details.ExitCode);
Assert.Equal(9999, details.ProcessId);
Assert.NotNull(details.StandardErrorTail);
Assert.Equal("Feature disabled", details.StandardErrorTail![0]);
// Verify initialization error was logged
Assert.Contains(MockLoggerProvider.LogMessages, log =>
log.LogLevel == LogLevel.Error &&
log.Message.Contains("client initialization error"));
}
[Fact]
public async Task CreateAsync_SendFails_PropagatesOriginalIOException()
{
// Arrange - transport throws IOException from SendMessageAsync, but the channel
// is not completed with ClientTransportClosedException. The original IOException should
// propagate without being wrapped in ClientTransportClosedException.
var transport = new SendFailsDuringInitTransport();
// Act & Assert
var ex = await Assert.ThrowsAsync<IOException>(
() => McpClient.CreateAsync(transport, loggerFactory: LoggerFactory, cancellationToken: TestContext.Current.CancellationToken));
Assert.Equal(SendFailsDuringInitTransport.ExpectedMessage, ex.Message);
// Verify initialization error was logged
Assert.Contains(MockLoggerProvider.LogMessages, log =>
log.LogLevel == LogLevel.Error &&
log.Message.Contains("client initialization error"));
}
private class NopTransport : ITransport, IClientTransport private class NopTransport : ITransport, IClientTransport
{ {
private readonly Channel<JsonRpcMessage> _channel = Channel.CreateUnbounded<JsonRpcMessage>(); private readonly Channel<JsonRpcMessage> _channel = Channel.CreateUnbounded<JsonRpcMessage>();
@ -155,4 +202,60 @@ public class McpClientCreationTests
throw new InvalidOperationException(ExpectedMessage); throw new InvalidOperationException(ExpectedMessage);
} }
} }
/// <summary>
/// Simulates a transport where the read channel closes with structured completion details during
/// initialization, as happens when a stdio server process exits before completing the handshake.
/// The send succeeds — only the read side carries the failure.
/// </summary>
private sealed class ChannelClosedDuringInitTransport : ITransport, IClientTransport
{
private readonly Channel<JsonRpcMessage> _channel = Channel.CreateUnbounded<JsonRpcMessage>();
public bool IsConnected => true;
public string? SessionId => null;
public ChannelReader<JsonRpcMessage> MessageReader => _channel.Reader;
public Task<ITransport> ConnectAsync(CancellationToken cancellationToken = default) => Task.FromResult<ITransport>(this);
public ValueTask DisposeAsync()
{
_channel.Writer.TryComplete();
return default;
}
public string Name => "Test ChannelClosed Transport";
public Task SendMessageAsync(JsonRpcMessage message, CancellationToken cancellationToken = default)
{
// Simulate the server process exiting: complete the channel with a
// ClientTransportClosedException carrying structured completion details.
// The send itself succeeds — the failure comes from the read side.
var details = new StdioClientCompletionDetails
{
ExitCode = 42,
ProcessId = 9999,
StandardErrorTail = ["Feature disabled"],
Exception = new IOException("MCP server process exited unexpectedly (exit code: 42)"),
};
_channel.Writer.TryComplete(new ClientTransportClosedException(details));
return Task.CompletedTask;
}
}
/// <summary>
/// Simulates a transport where SendMessageAsync throws IOException but the channel
/// doesn't carry a ClientTransportClosedException (e.g., a write pipe break without structured details).
/// </summary>
private sealed class SendFailsDuringInitTransport : NopTransport
{
public const string ExpectedMessage = "Failed to write to transport";
public override Task SendMessageAsync(JsonRpcMessage message, CancellationToken cancellationToken = default)
{
throw new IOException(ExpectedMessage);
}
}
} }