Clean up logging (#286)
This commit is contained in:
Stephen Toub
GitHub
parent
dc544f929b
commit
312aede7d6
6 changed files with 401 additions and 570 deletions
|
|
@ -1,9 +1,9 @@
|
|||
using Microsoft.Extensions.Logging;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using ModelContextProtocol.Logging;
|
||||
using ModelContextProtocol.Protocol.Messages;
|
||||
using ModelContextProtocol.Utils;
|
||||
using ModelContextProtocol.Utils.Json;
|
||||
using System.Diagnostics;
|
||||
using System.Net.Http.Headers;
|
||||
using System.Net.ServerSentEvents;
|
||||
using System.Text;
|
||||
|
|
@ -14,9 +14,8 @@ namespace ModelContextProtocol.Protocol.Transport;
|
|||
/// <summary>
|
||||
/// The ServerSideEvents client transport implementation
|
||||
/// </summary>
|
||||
internal sealed class SseClientSessionTransport : TransportBase
|
||||
internal sealed partial class SseClientSessionTransport : TransportBase
|
||||
{
|
||||
private readonly string _endpointName;
|
||||
private readonly HttpClient _httpClient;
|
||||
private readonly SseClientTransportOptions _options;
|
||||
private readonly Uri _sseEndpoint;
|
||||
|
|
@ -35,7 +34,7 @@ internal sealed class SseClientSessionTransport : TransportBase
|
|||
/// <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(loggerFactory)
|
||||
: base(endpointName, loggerFactory)
|
||||
{
|
||||
Throw.IfNull(transportOptions);
|
||||
Throw.IfNull(httpClient);
|
||||
|
|
@ -45,36 +44,23 @@ internal sealed class SseClientSessionTransport : TransportBase
|
|||
_httpClient = httpClient;
|
||||
_connectionCts = new CancellationTokenSource();
|
||||
_logger = (ILogger?)loggerFactory?.CreateLogger<SseClientTransport>() ?? NullLogger.Instance;
|
||||
_connectionEstablished = new TaskCompletionSource<bool>();
|
||||
_endpointName = endpointName;
|
||||
_connectionEstablished = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
}
|
||||
|
||||
/// <inheritdoc/>
|
||||
public async Task ConnectAsync(CancellationToken cancellationToken = default)
|
||||
{
|
||||
Debug.Assert(!IsConnected);
|
||||
try
|
||||
{
|
||||
if (IsConnected)
|
||||
{
|
||||
_logger.TransportAlreadyConnected(_endpointName);
|
||||
throw new McpTransportException("Transport is already connected");
|
||||
}
|
||||
|
||||
// Start message receiving loop
|
||||
_receiveTask = ReceiveMessagesAsync(_connectionCts.Token);
|
||||
|
||||
_logger.TransportReadingMessages(_endpointName);
|
||||
|
||||
await _connectionEstablished.Task.WaitAsync(_options.ConnectionTimeout, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
catch (McpTransportException)
|
||||
catch (Exception ex) when (ex is not McpTransportException) // propagate transport exceptions
|
||||
{
|
||||
// Rethrow transport exceptions
|
||||
throw;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.TransportConnectFailed(_endpointName, ex);
|
||||
LogTransportConnectFailed(Name, ex);
|
||||
await CloseAsync().ConfigureAwait(false);
|
||||
throw new McpTransportException("Failed to connect transport", ex);
|
||||
}
|
||||
|
|
@ -118,7 +104,7 @@ internal sealed class SseClientSessionTransport : TransportBase
|
|||
// If the response is not a JSON-RPC response, it is an SSE message
|
||||
if (string.IsNullOrEmpty(responseContent) || responseContent.Equals("accepted", StringComparison.OrdinalIgnoreCase))
|
||||
{
|
||||
_logger.SSETransportPostAccepted(_endpointName, messageId);
|
||||
LogAcceptedPost(Name, messageId);
|
||||
// The response will arrive as an SSE message
|
||||
}
|
||||
else
|
||||
|
|
@ -126,21 +112,30 @@ internal sealed class SseClientSessionTransport : TransportBase
|
|||
JsonRpcResponse initializeResponse = JsonSerializer.Deserialize(responseContent, McpJsonUtilities.JsonContext.Default.JsonRpcResponse) ??
|
||||
throw new McpTransportException("Failed to initialize client");
|
||||
|
||||
_logger.TransportReceivedMessageParsed(_endpointName, messageId);
|
||||
LogTransportReceivedMessage(Name, messageId);
|
||||
await WriteMessageAsync(initializeResponse, cancellationToken).ConfigureAwait(false);
|
||||
_logger.TransportMessageWritten(_endpointName, messageId);
|
||||
LogTransportMessageWritten(Name, messageId);
|
||||
}
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
// Otherwise, check if the response was accepted (the response will come as an SSE message)
|
||||
if (string.IsNullOrEmpty(responseContent) || responseContent.Equals("accepted", StringComparison.OrdinalIgnoreCase))
|
||||
{
|
||||
_logger.SSETransportPostAccepted(_endpointName, messageId);
|
||||
LogAcceptedPost(Name, messageId);
|
||||
}
|
||||
else
|
||||
{
|
||||
_logger.SSETransportPostNotAccepted(_endpointName, messageId, responseContent);
|
||||
if (_logger.IsEnabled(LogLevel.Trace))
|
||||
{
|
||||
LogRejectedPostSensitive(Name, messageId, responseContent);
|
||||
}
|
||||
else
|
||||
{
|
||||
LogRejectedPost(Name, messageId);
|
||||
}
|
||||
|
||||
throw new McpTransportException("Failed to send message");
|
||||
}
|
||||
}
|
||||
|
|
@ -151,12 +146,17 @@ internal sealed class SseClientSessionTransport : TransportBase
|
|||
{
|
||||
await _connectionCts.CancelAsync().ConfigureAwait(false);
|
||||
|
||||
if (_receiveTask != null)
|
||||
try
|
||||
{
|
||||
await _receiveTask.ConfigureAwait(false);
|
||||
if (_receiveTask != null)
|
||||
{
|
||||
await _receiveTask.ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
finally
|
||||
{
|
||||
_connectionCts.Dispose();
|
||||
}
|
||||
|
||||
_connectionCts.Dispose();
|
||||
}
|
||||
finally
|
||||
{
|
||||
|
|
@ -216,13 +216,13 @@ internal sealed class SseClientSessionTransport : TransportBase
|
|||
catch when (cancellationToken.IsCancellationRequested)
|
||||
{
|
||||
// Normal shutdown
|
||||
LogTransportReadMessagesCancelled(Name);
|
||||
_connectionEstablished.TrySetCanceled(cancellationToken);
|
||||
_logger.TransportReadMessagesCancelled(_endpointName);
|
||||
}
|
||||
catch (Exception ex) when (!cancellationToken.IsCancellationRequested)
|
||||
catch (Exception ex)
|
||||
{
|
||||
LogTransportReadMessagesFailed(Name, ex);
|
||||
_connectionEstablished.TrySetException(ex);
|
||||
_logger.TransportConnectionError(_endpointName, ex);
|
||||
throw;
|
||||
}
|
||||
finally
|
||||
|
|
@ -235,7 +235,7 @@ internal sealed class SseClientSessionTransport : TransportBase
|
|||
{
|
||||
if (!IsConnected)
|
||||
{
|
||||
_logger.TransportMessageReceivedBeforeConnected(_endpointName, data);
|
||||
LogTransportMessageReceivedBeforeConnected(Name);
|
||||
return;
|
||||
}
|
||||
|
||||
|
|
@ -244,7 +244,7 @@ internal sealed class SseClientSessionTransport : TransportBase
|
|||
var message = JsonSerializer.Deserialize(data, McpJsonUtilities.JsonContext.Default.IJsonRpcMessage);
|
||||
if (message == null)
|
||||
{
|
||||
_logger.TransportMessageParseUnexpectedType(_endpointName, data);
|
||||
LogTransportMessageParseUnexpectedTypeSensitive(Name, data);
|
||||
return;
|
||||
}
|
||||
|
||||
|
|
@ -254,13 +254,20 @@ internal sealed class SseClientSessionTransport : TransportBase
|
|||
messageId = messageWithId.Id.ToString();
|
||||
}
|
||||
|
||||
_logger.TransportReceivedMessageParsed(_endpointName, messageId);
|
||||
LogTransportReceivedMessage(Name, messageId);
|
||||
await WriteMessageAsync(message, cancellationToken).ConfigureAwait(false);
|
||||
_logger.TransportMessageWritten(_endpointName, messageId);
|
||||
LogTransportMessageWritten(Name, messageId);
|
||||
}
|
||||
catch (JsonException ex)
|
||||
{
|
||||
_logger.TransportMessageParseFailed(_endpointName, data, ex);
|
||||
if (_logger.IsEnabled(LogLevel.Trace))
|
||||
{
|
||||
LogTransportMessageParseFailedSensitive(Name, data, ex);
|
||||
}
|
||||
else
|
||||
{
|
||||
LogTransportMessageParseFailed(Name, ex);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -270,7 +277,7 @@ internal sealed class SseClientSessionTransport : TransportBase
|
|||
{
|
||||
if (string.IsNullOrEmpty(data))
|
||||
{
|
||||
_logger.TransportEndpointEventInvalid(_endpointName, data);
|
||||
LogTransportEndpointEventInvalid(Name);
|
||||
return;
|
||||
}
|
||||
|
||||
|
|
@ -283,7 +290,15 @@ internal sealed class SseClientSessionTransport : TransportBase
|
|||
}
|
||||
catch (JsonException ex)
|
||||
{
|
||||
_logger.TransportEndpointEventParseFailed(_endpointName, data, ex);
|
||||
if (_logger.IsEnabled(LogLevel.Trace))
|
||||
{
|
||||
LogTransportEndpointEventParseFailedSensitive(Name, data, ex);
|
||||
}
|
||||
else
|
||||
{
|
||||
LogTransportEndpointEventParseFailed(Name, ex);
|
||||
}
|
||||
|
||||
throw new McpTransportException("Failed to parse endpoint event", ex);
|
||||
}
|
||||
}
|
||||
|
|
@ -301,4 +316,13 @@ internal sealed class SseClientSessionTransport : TransportBase
|
|||
}
|
||||
}
|
||||
}
|
||||
|
||||
[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);
|
||||
}
|
||||
|
|
@ -1,6 +1,6 @@
|
|||
using Microsoft.Extensions.Logging;
|
||||
using ModelContextProtocol.Logging;
|
||||
using ModelContextProtocol.Protocol.Messages;
|
||||
using System;
|
||||
using System.Diagnostics;
|
||||
|
||||
namespace ModelContextProtocol.Protocol.Transport;
|
||||
|
|
@ -49,7 +49,6 @@ internal sealed class StdioClientSessionTransport : StreamClientSessionTransport
|
|||
|
||||
if (hasExited)
|
||||
{
|
||||
Logger.TransportNotConnected(EndpointName);
|
||||
throw new McpTransportException("Transport is not connected", processException);
|
||||
}
|
||||
|
||||
|
|
@ -59,7 +58,14 @@ internal sealed class StdioClientSessionTransport : StreamClientSessionTransport
|
|||
/// <inheritdoc/>
|
||||
protected override ValueTask CleanupAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
StdioClientTransport.DisposeProcess(_process, processRunning: true, Logger, _options.ShutdownTimeout, EndpointName);
|
||||
try
|
||||
{
|
||||
StdioClientTransport.DisposeProcess(_process, processRunning: true, _options.ShutdownTimeout, Name);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
LogTransportShutdownFailed(Name, ex);
|
||||
}
|
||||
|
||||
return base.CleanupAsync(cancellationToken);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,6 +1,5 @@
|
|||
using Microsoft.Extensions.Logging;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using ModelContextProtocol.Logging;
|
||||
using ModelContextProtocol.Utils;
|
||||
using System.Diagnostics;
|
||||
using System.Runtime.InteropServices;
|
||||
|
|
@ -24,7 +23,7 @@ namespace ModelContextProtocol.Protocol.Transport;
|
|||
/// and environment variables, handling output, and properly terminating the process when the transport is closed.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public sealed class StdioClientTransport : IClientTransport
|
||||
public sealed partial class StdioClientTransport : IClientTransport
|
||||
{
|
||||
private readonly StdioClientTransportOptions _options;
|
||||
private readonly ILoggerFactory? _loggerFactory;
|
||||
|
|
@ -68,7 +67,7 @@ public sealed class StdioClientTransport : IClientTransport
|
|||
ILogger logger = (ILogger?)_loggerFactory?.CreateLogger<StdioClientTransport>() ?? NullLogger.Instance;
|
||||
try
|
||||
{
|
||||
logger.TransportConnecting(endpointName);
|
||||
LogTransportConnecting(logger, endpointName);
|
||||
|
||||
UTF8Encoding noBomUTF8 = new(encoderShouldEmitUTF8Identifier: false);
|
||||
|
||||
|
|
@ -114,14 +113,22 @@ public sealed class StdioClientTransport : IClientTransport
|
|||
}
|
||||
}
|
||||
|
||||
logger.CreateProcessForTransport(endpointName, _options.Command,
|
||||
startInfo.Arguments, string.Join(", ", startInfo.Environment.Select(kvp => kvp.Key + "=" + kvp.Value)),
|
||||
startInfo.WorkingDirectory, _options.ShutdownTimeout.ToString());
|
||||
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 error logging
|
||||
process.ErrorDataReceived += (sender, args) => logger.ReadStderr(endpointName, args.Data ?? "(no data)");
|
||||
process.ErrorDataReceived += (sender, args) => LogReadStderr(logger, endpointName, args.Data ?? "(no 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
|
||||
|
|
@ -146,11 +153,11 @@ public sealed class StdioClientTransport : IClientTransport
|
|||
|
||||
if (!processStarted)
|
||||
{
|
||||
logger.TransportProcessStartFailed(endpointName);
|
||||
LogTransportProcessStartFailed(logger, endpointName);
|
||||
throw new McpTransportException("Failed to start MCP server process");
|
||||
}
|
||||
|
||||
logger.TransportProcessStarted(endpointName, process.Id);
|
||||
LogTransportProcessStarted(logger, endpointName, process.Id);
|
||||
|
||||
process.BeginErrorReadLine();
|
||||
|
||||
|
|
@ -158,14 +165,23 @@ public sealed class StdioClientTransport : IClientTransport
|
|||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
logger.TransportConnectFailed(endpointName, ex);
|
||||
DisposeProcess(process, processStarted, logger, _options.ShutdownTimeout, endpointName);
|
||||
LogTransportConnectFailed(logger, endpointName, ex);
|
||||
|
||||
try
|
||||
{
|
||||
DisposeProcess(process, processStarted, _options.ShutdownTimeout, endpointName);
|
||||
}
|
||||
catch (Exception ex2)
|
||||
{
|
||||
LogTransportShutdownFailed(logger, endpointName, ex2);
|
||||
}
|
||||
|
||||
throw new McpTransportException("Failed to connect transport", ex);
|
||||
}
|
||||
}
|
||||
|
||||
internal static void DisposeProcess(
|
||||
Process? process, bool processRunning, ILogger logger, TimeSpan shutdownTimeout, string endpointName)
|
||||
Process? process, bool processRunning, TimeSpan shutdownTimeout, string endpointName)
|
||||
{
|
||||
if (process is not null)
|
||||
{
|
||||
|
|
@ -188,18 +204,37 @@ public sealed class StdioClientTransport : IClientTransport
|
|||
// 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.
|
||||
logger.TransportWaitingForShutdown(endpointName);
|
||||
process.KillTree(shutdownTimeout);
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
logger.TransportShutdownFailed(endpointName, ex);
|
||||
}
|
||||
finally
|
||||
{
|
||||
process.Dispose();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
[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);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,6 +1,4 @@
|
|||
using Microsoft.Extensions.Logging;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using ModelContextProtocol.Logging;
|
||||
using ModelContextProtocol.Protocol.Messages;
|
||||
using ModelContextProtocol.Utils;
|
||||
using ModelContextProtocol.Utils.Json;
|
||||
|
|
@ -40,17 +38,14 @@ internal class StreamClientSessionTransport : TransportBase
|
|||
/// </remarks>
|
||||
public StreamClientSessionTransport(
|
||||
TextWriter serverInput, TextReader serverOutput, string endpointName, ILoggerFactory? loggerFactory)
|
||||
: base(loggerFactory)
|
||||
: base(endpointName, loggerFactory)
|
||||
{
|
||||
Logger = (ILogger?)loggerFactory?.CreateLogger<StdioClientTransport>() ?? NullLogger.Instance;
|
||||
_serverOutput = serverOutput;
|
||||
_serverInput = serverInput;
|
||||
EndpointName = endpointName;
|
||||
|
||||
// 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.
|
||||
Logger.TransportReadingMessages(endpointName);
|
||||
var readTask = new Task<Task>(
|
||||
thisRef => ((StreamClientSessionTransport)thisRef!).ReadMessagesAsync(_shutdownCts.Token),
|
||||
this,
|
||||
|
|
@ -61,25 +56,6 @@ internal class StreamClientSessionTransport : TransportBase
|
|||
SetConnected(true);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Gets the logger instance used by this transport for diagnostic output.
|
||||
/// If no logger factory was provided in the constructor, this will be a NullLogger.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Derived classes can use this logger to emit additional diagnostic information
|
||||
/// specific to their implementation.
|
||||
/// </remarks>
|
||||
protected ILogger Logger { get; private set; }
|
||||
|
||||
/// <summary>
|
||||
/// Gets the name that identifies this transport endpoint in logs.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// This name is provided during construction and is used in log messages to identify
|
||||
/// the source of transport-related events.
|
||||
/// </remarks>
|
||||
protected string EndpointName { get; }
|
||||
|
||||
/// <inheritdoc/>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
|
|
@ -102,7 +78,6 @@ internal class StreamClientSessionTransport : TransportBase
|
|||
{
|
||||
if (!IsConnected)
|
||||
{
|
||||
Logger.TransportNotConnected(EndpointName);
|
||||
throw new McpTransportException("Transport is not connected");
|
||||
}
|
||||
|
||||
|
|
@ -117,18 +92,13 @@ internal class StreamClientSessionTransport : TransportBase
|
|||
using var _ = await _sendLock.LockAsync(cancellationToken).ConfigureAwait(false);
|
||||
try
|
||||
{
|
||||
Logger.TransportSendingMessage(EndpointName, id, json);
|
||||
Logger.TransportMessageBytesUtf8(EndpointName, json);
|
||||
|
||||
// Write the message followed by a newline using our UTF-8 writer
|
||||
await _serverInput.WriteLineAsync(json).ConfigureAwait(false);
|
||||
await _serverInput.FlushAsync(cancellationToken).ConfigureAwait(false);
|
||||
|
||||
Logger.TransportSentMessage(EndpointName, id);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
Logger.TransportSendFailed(EndpointName, id, ex);
|
||||
LogTransportSendFailed(Name, id, ex);
|
||||
throw new McpTransportException("Failed to send message", ex);
|
||||
}
|
||||
}
|
||||
|
|
@ -151,14 +121,13 @@ internal class StreamClientSessionTransport : TransportBase
|
|||
{
|
||||
try
|
||||
{
|
||||
Logger.TransportEnteringReadMessagesLoop(EndpointName);
|
||||
LogTransportEnteringReadMessagesLoop(Name);
|
||||
|
||||
while (!cancellationToken.IsCancellationRequested)
|
||||
{
|
||||
Logger.TransportWaitingForMessage(EndpointName);
|
||||
if (await _serverOutput.ReadLineAsync(cancellationToken).ConfigureAwait(false) is not string line)
|
||||
{
|
||||
Logger.TransportEndOfStream(EndpointName);
|
||||
LogTransportEndOfStream(Name);
|
||||
break;
|
||||
}
|
||||
|
||||
|
|
@ -167,20 +136,18 @@ internal class StreamClientSessionTransport : TransportBase
|
|||
continue;
|
||||
}
|
||||
|
||||
Logger.TransportReceivedMessage(EndpointName, line);
|
||||
Logger.TransportMessageBytesUtf8(EndpointName, line);
|
||||
LogTransportReceivedMessageSensitive(Name, line);
|
||||
|
||||
await ProcessMessageAsync(line, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
Logger.TransportExitingReadMessagesLoop(EndpointName);
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
Logger.TransportReadMessagesCancelled(EndpointName);
|
||||
LogTransportReadMessagesCancelled(Name);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
Logger.TransportReadMessagesFailed(EndpointName, ex);
|
||||
LogTransportReadMessagesFailed(Name, ex);
|
||||
}
|
||||
finally
|
||||
{
|
||||
|
|
@ -202,24 +169,31 @@ internal class StreamClientSessionTransport : TransportBase
|
|||
messageId = messageWithId.Id.ToString();
|
||||
}
|
||||
|
||||
Logger.TransportReceivedMessageParsed(EndpointName, messageId);
|
||||
LogTransportReceivedMessage(Name, messageId);
|
||||
await WriteMessageAsync(message, cancellationToken).ConfigureAwait(false);
|
||||
Logger.TransportMessageWritten(EndpointName, messageId);
|
||||
LogTransportMessageWritten(Name, messageId);
|
||||
}
|
||||
else
|
||||
{
|
||||
Logger.TransportMessageParseUnexpectedType(EndpointName, line);
|
||||
LogTransportMessageParseUnexpectedTypeSensitive(Name, line);
|
||||
}
|
||||
}
|
||||
catch (JsonException ex)
|
||||
{
|
||||
Logger.TransportMessageParseFailed(EndpointName, line, ex);
|
||||
if (Logger.IsEnabled(LogLevel.Trace))
|
||||
{
|
||||
LogTransportMessageParseFailedSensitive(Name, line, ex);
|
||||
}
|
||||
else
|
||||
{
|
||||
LogTransportMessageParseFailed(Name, ex);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
protected virtual async ValueTask CleanupAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
Logger.TransportCleaningUp(EndpointName);
|
||||
LogTransportShuttingDown(Name);
|
||||
|
||||
if (Interlocked.Exchange(ref _shutdownCts, null) is { } shutdownCts)
|
||||
{
|
||||
|
|
@ -231,25 +205,18 @@ internal class StreamClientSessionTransport : TransportBase
|
|||
{
|
||||
try
|
||||
{
|
||||
Logger.TransportWaitingForReadTask(EndpointName);
|
||||
await readTask.WaitAsync(TimeSpan.FromSeconds(5), cancellationToken).ConfigureAwait(false);
|
||||
Logger.TransportReadTaskCleanedUp(EndpointName);
|
||||
}
|
||||
catch (TimeoutException)
|
||||
{
|
||||
Logger.TransportCleanupReadTaskTimeout(EndpointName);
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
Logger.TransportCleanupReadTaskCancelled(EndpointName);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
Logger.TransportCleanupReadTaskFailed(EndpointName, ex);
|
||||
LogTransportCleanupReadTaskFailed(Name, ex);
|
||||
}
|
||||
}
|
||||
|
||||
SetConnected(false);
|
||||
Logger.TransportCleanedUp(EndpointName);
|
||||
LogTransportShutDown(Name);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,6 +1,5 @@
|
|||
using Microsoft.Extensions.Logging;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using ModelContextProtocol.Logging;
|
||||
using ModelContextProtocol.Protocol.Messages;
|
||||
using ModelContextProtocol.Utils;
|
||||
using ModelContextProtocol.Utils.Json;
|
||||
|
|
@ -25,7 +24,6 @@ public class StreamServerTransport : TransportBase
|
|||
|
||||
private readonly TextReader _inputReader;
|
||||
private readonly Stream _outputStream;
|
||||
private readonly string _endpointName;
|
||||
|
||||
private readonly SemaphoreSlim _sendLock = new(1, 1);
|
||||
private readonly CancellationTokenSource _shutdownCts = new();
|
||||
|
|
@ -43,7 +41,7 @@ public class StreamServerTransport : TransportBase
|
|||
/// <exception cref="ArgumentNullException"><paramref name="inputStream"/> is <see langword="null"/>.</exception>
|
||||
/// <exception cref="ArgumentNullException"><paramref name="outputStream"/> is <see langword="null"/>.</exception>
|
||||
public StreamServerTransport(Stream inputStream, Stream outputStream, string? serverName = null, ILoggerFactory? loggerFactory = null)
|
||||
: base(loggerFactory)
|
||||
: base(serverName is not null ? $"Server (stream) ({serverName})" : "Server (stream)", loggerFactory)
|
||||
{
|
||||
Throw.IfNull(inputStream);
|
||||
Throw.IfNull(outputStream);
|
||||
|
|
@ -55,8 +53,6 @@ public class StreamServerTransport : TransportBase
|
|||
|
||||
SetConnected(true);
|
||||
_readLoopCompleted = Task.Run(ReadMessagesAsync, _shutdownCts.Token);
|
||||
|
||||
_endpointName = serverName is not null ? $"Server (stream) ({serverName})" : "Server (stream)";
|
||||
}
|
||||
|
||||
/// <inheritdoc/>
|
||||
|
|
@ -64,7 +60,6 @@ public class StreamServerTransport : TransportBase
|
|||
{
|
||||
if (!IsConnected)
|
||||
{
|
||||
_logger.TransportNotConnected(_endpointName);
|
||||
throw new McpTransportException("Transport is not connected");
|
||||
}
|
||||
|
||||
|
|
@ -78,17 +73,13 @@ public class StreamServerTransport : TransportBase
|
|||
|
||||
try
|
||||
{
|
||||
_logger.TransportSendingMessage(_endpointName, id);
|
||||
|
||||
await JsonSerializer.SerializeAsync(_outputStream, message, McpJsonUtilities.DefaultOptions.GetTypeInfo(typeof(IJsonRpcMessage)), cancellationToken).ConfigureAwait(false);
|
||||
await _outputStream.WriteAsync(s_newlineBytes, cancellationToken).ConfigureAwait(false);
|
||||
await _outputStream.FlushAsync(cancellationToken).ConfigureAwait(false);
|
||||
|
||||
_logger.TransportSentMessage(_endpointName, id);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.TransportSendFailed(_endpointName, id, ex);
|
||||
LogTransportSendFailed(Name, id, ex);
|
||||
throw new McpTransportException("Failed to send message", ex);
|
||||
}
|
||||
}
|
||||
|
|
@ -98,26 +89,23 @@ public class StreamServerTransport : TransportBase
|
|||
CancellationToken shutdownToken = _shutdownCts.Token;
|
||||
try
|
||||
{
|
||||
_logger.TransportEnteringReadMessagesLoop(_endpointName);
|
||||
LogTransportEnteringReadMessagesLoop(Name);
|
||||
|
||||
while (!shutdownToken.IsCancellationRequested)
|
||||
{
|
||||
_logger.TransportWaitingForMessage(_endpointName);
|
||||
|
||||
var line = await _inputReader.ReadLineAsync(shutdownToken).ConfigureAwait(false);
|
||||
if (string.IsNullOrWhiteSpace(line))
|
||||
{
|
||||
if (line is null)
|
||||
{
|
||||
_logger.TransportEndOfStream(_endpointName);
|
||||
LogTransportEndOfStream(Name);
|
||||
break;
|
||||
}
|
||||
|
||||
continue;
|
||||
}
|
||||
|
||||
_logger.TransportReceivedMessage(_endpointName, line);
|
||||
_logger.TransportMessageBytesUtf8(_endpointName, line);
|
||||
LogTransportReceivedMessageSensitive(Name, line);
|
||||
|
||||
try
|
||||
{
|
||||
|
|
@ -128,32 +116,38 @@ public class StreamServerTransport : TransportBase
|
|||
{
|
||||
messageId = messageWithId.Id.ToString();
|
||||
}
|
||||
_logger.TransportReceivedMessageParsed(_endpointName, messageId);
|
||||
|
||||
LogTransportReceivedMessage(Name, messageId);
|
||||
await WriteMessageAsync(message, shutdownToken).ConfigureAwait(false);
|
||||
_logger.TransportMessageWritten(_endpointName, messageId);
|
||||
LogTransportMessageWritten(Name, messageId);
|
||||
}
|
||||
else
|
||||
{
|
||||
_logger.TransportMessageParseUnexpectedType(_endpointName, line);
|
||||
LogTransportMessageParseUnexpectedTypeSensitive(Name, line);
|
||||
}
|
||||
}
|
||||
catch (JsonException ex)
|
||||
{
|
||||
_logger.TransportMessageParseFailed(_endpointName, line, ex);
|
||||
if (Logger.IsEnabled(LogLevel.Trace))
|
||||
{
|
||||
LogTransportMessageParseFailedSensitive(Name, line, ex);
|
||||
}
|
||||
else
|
||||
{
|
||||
LogTransportMessageParseFailed(Name, ex);
|
||||
}
|
||||
|
||||
// Continue reading even if we fail to parse a message
|
||||
}
|
||||
}
|
||||
|
||||
_logger.TransportExitingReadMessagesLoop(_endpointName);
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
_logger.TransportReadMessagesCancelled(_endpointName);
|
||||
LogTransportReadMessagesCancelled(Name);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.TransportReadMessagesFailed(_endpointName, ex);
|
||||
LogTransportReadMessagesFailed(Name, ex);
|
||||
}
|
||||
finally
|
||||
{
|
||||
|
|
@ -171,7 +165,7 @@ public class StreamServerTransport : TransportBase
|
|||
|
||||
try
|
||||
{
|
||||
_logger.TransportCleaningUp(_endpointName);
|
||||
LogTransportShuttingDown(Name);
|
||||
|
||||
// Signal to the stdin reading loop to stop.
|
||||
await _shutdownCts.CancelAsync().ConfigureAwait(false);
|
||||
|
|
@ -185,27 +179,20 @@ public class StreamServerTransport : TransportBase
|
|||
// Make sure the work has quiesced.
|
||||
try
|
||||
{
|
||||
_logger.TransportWaitingForReadTask(_endpointName);
|
||||
await _readLoopCompleted.ConfigureAwait(false);
|
||||
_logger.TransportReadTaskCleanedUp(_endpointName);
|
||||
}
|
||||
catch (TimeoutException)
|
||||
{
|
||||
_logger.TransportCleanupReadTaskTimeout(_endpointName);
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
_logger.TransportCleanupReadTaskCancelled(_endpointName);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.TransportCleanupReadTaskFailed(_endpointName, ex);
|
||||
LogTransportCleanupReadTaskFailed(Name, ex);
|
||||
}
|
||||
}
|
||||
finally
|
||||
{
|
||||
SetConnected(false);
|
||||
_logger.TransportCleanedUp(_endpointName);
|
||||
LogTransportShutDown(Name);
|
||||
}
|
||||
|
||||
GC.SuppressFinalize(this);
|
||||
|
|
|
|||
|
|
@ -1,6 +1,5 @@
|
|||
using System.Threading.Channels;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using ModelContextProtocol.Logging;
|
||||
using ModelContextProtocol.Protocol.Messages;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
|
||||
|
|
@ -20,7 +19,7 @@ namespace ModelContextProtocol.Protocol.Transport;
|
|||
/// to handle the specific transport mechanism being used.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public abstract class TransportBase : ITransport
|
||||
public abstract partial class TransportBase : ITransport
|
||||
{
|
||||
private readonly Channel<IJsonRpcMessage> _messageChannel;
|
||||
private readonly ILogger _logger;
|
||||
|
|
@ -29,17 +28,30 @@ public abstract class TransportBase : ITransport
|
|||
/// <summary>
|
||||
/// Initializes a new instance of the <see cref="TransportBase"/> class.
|
||||
/// </summary>
|
||||
protected TransportBase(ILoggerFactory? loggerFactory)
|
||||
protected TransportBase(string name, ILoggerFactory? loggerFactory)
|
||||
{
|
||||
Name = name;
|
||||
_logger = loggerFactory?.CreateLogger(GetType()) ?? NullLogger.Instance;
|
||||
|
||||
// Unbounded channel to prevent blocking on writes
|
||||
_messageChannel = Channel.CreateUnbounded<IJsonRpcMessage>(new UnboundedChannelOptions
|
||||
{
|
||||
SingleReader = true,
|
||||
SingleWriter = true,
|
||||
});
|
||||
_logger = loggerFactory?.CreateLogger(GetType()) ?? NullLogger.Instance;
|
||||
}
|
||||
|
||||
/// <summary>Gets the logger used by this transport.</summary>
|
||||
private protected ILogger Logger => _logger;
|
||||
|
||||
/// <summary>
|
||||
/// Gets the name that identifies this transport endpoint in logs.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// This name is used in log messages to identify the source of transport-related events.
|
||||
/// </remarks>
|
||||
protected string Name { get; }
|
||||
|
||||
/// <inheritdoc/>
|
||||
public bool IsConnected => _isConnected == 1;
|
||||
|
||||
|
|
@ -64,9 +76,7 @@ public abstract class TransportBase : ITransport
|
|||
throw new McpTransportException("Transport is not connected");
|
||||
}
|
||||
|
||||
_logger.TransportWritingMessageToChannel(message);
|
||||
await _messageChannel.Writer.WriteAsync(message, cancellationToken).ConfigureAwait(false);
|
||||
_logger.TransportMessageWrittenToChannel();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
|
|
@ -86,4 +96,64 @@ public abstract class TransportBase : ITransport
|
|||
_messageChannel.Writer.Complete();
|
||||
}
|
||||
}
|
||||
|
||||
[LoggerMessage(Level = LogLevel.Error, Message = "{EndpointName} transport connect failed.")]
|
||||
private protected partial void LogTransportConnectFailed(string endpointName, Exception exception);
|
||||
|
||||
[LoggerMessage(Level = LogLevel.Error, Message = "{EndpointName} transport send failed for message ID '{MessageId}'.")]
|
||||
private protected partial void LogTransportSendFailed(string endpointName, string messageId, Exception exception);
|
||||
|
||||
[LoggerMessage(Level = LogLevel.Information, Message = "{EndpointName} transport reading messages.")]
|
||||
private protected partial void LogTransportEnteringReadMessagesLoop(string endpointName);
|
||||
|
||||
[LoggerMessage(Level = LogLevel.Information, Message = "{EndpointName} transport completed reading messages.")]
|
||||
private protected partial void LogTransportEndOfStream(string endpointName);
|
||||
|
||||
[LoggerMessage(Level = LogLevel.Trace, Message = "{EndpointName} transport received message. Message: '{Message}'.")]
|
||||
private protected partial void LogTransportReceivedMessageSensitive(string endpointName, string message);
|
||||
|
||||
[LoggerMessage(Level = LogLevel.Debug, Message = "{EndpointName} transport received message with ID '{MessageId}'.")]
|
||||
private protected partial void LogTransportReceivedMessage(string endpointName, string messageId);
|
||||
|
||||
[LoggerMessage(Level = LogLevel.Debug, Message = "{EndpointName} transport sent message with ID '{MessageId}'.")]
|
||||
private protected partial void LogTransportMessageWritten(string endpointName, string messageId);
|
||||
|
||||
[LoggerMessage(Level = LogLevel.Trace, Message = "{EndpointName} transport received unexpected message. Message: '{Message}'.")]
|
||||
private protected partial void LogTransportMessageParseUnexpectedTypeSensitive(string endpointName, string message);
|
||||
|
||||
[LoggerMessage(Level = LogLevel.Information, Message = "{EndpointName} transport message parsing failed.")]
|
||||
private protected partial void LogTransportMessageParseFailed(string endpointName, Exception exception);
|
||||
|
||||
[LoggerMessage(Level = LogLevel.Trace, Message = "{EndpointName} transport message parsing failed. Message: '{Message}'.")]
|
||||
private protected partial void LogTransportMessageParseFailedSensitive(string endpointName, string message, Exception exception);
|
||||
|
||||
[LoggerMessage(Level = LogLevel.Information, Message = "{EndpointName} transport message reading canceled.")]
|
||||
private protected partial void LogTransportReadMessagesCancelled(string endpointName);
|
||||
|
||||
[LoggerMessage(Level = LogLevel.Warning, Message = "{EndpointName} transport message reading failed.")]
|
||||
private protected partial void LogTransportReadMessagesFailed(string endpointName, Exception exception);
|
||||
|
||||
[LoggerMessage(Level = LogLevel.Information, Message = "{EndpointName} shutting down.")]
|
||||
private protected partial void LogTransportShuttingDown(string endpointName);
|
||||
|
||||
[LoggerMessage(Level = LogLevel.Warning, Message = "{EndpointName} shutdown failed.")]
|
||||
private protected partial void LogTransportShutdownFailed(string endpointName, Exception exception);
|
||||
|
||||
[LoggerMessage(Level = LogLevel.Warning, Message = "{EndpointName} shutdown failed waiting for message reading completion.")]
|
||||
private protected partial void LogTransportCleanupReadTaskFailed(string endpointName, Exception exception);
|
||||
|
||||
[LoggerMessage(Level = LogLevel.Information, Message = "{EndpointName} shut down.")]
|
||||
private protected partial void LogTransportShutDown(string endpointName);
|
||||
|
||||
[LoggerMessage(Level = LogLevel.Warning, Message = "{EndpointName} received message before connected.")]
|
||||
private protected partial void LogTransportMessageReceivedBeforeConnected(string endpointName);
|
||||
|
||||
[LoggerMessage(Level = LogLevel.Warning, Message = "{EndpointName} endpoint event received out of order.")]
|
||||
private protected partial void LogTransportEndpointEventInvalid(string endpointName);
|
||||
|
||||
[LoggerMessage(Level = LogLevel.Warning, Message = "{EndpointName} failed to parse event.")]
|
||||
private protected partial void LogTransportEndpointEventParseFailed(string endpointName, Exception exception);
|
||||
|
||||
[LoggerMessage(Level = LogLevel.Warning, Message = "{EndpointName} failed to parse event. Message: '{Message}'.")]
|
||||
private protected partial void LogTransportEndpointEventParseFailedSensitive(string endpointName, string message, Exception exception);
|
||||
}
|
||||
Loading…
Reference in a new issue