Add ModelContextProtocol.AspNetCore (#160)
* Add ModelContextProtocol.AspNetCore - Remove IServerTransport and HttpListenerSseServerTransport - Replace HttpListenerSseServerTransport usage in tests with ASP.NET Core. Co-authored-by: Eirik Tsarpalis <eirik.tsarpalis@gmail.com> Co-authored-by: Jeff Handley <jeffhandley@users.noreply.github.com>
This commit is contained in:
Stephen Halter
GitHub
parent
330e526446
commit
973c7c093f
12 changed files with 332 additions and 831 deletions
|
|
@ -79,10 +79,9 @@ internal sealed class McpClient : McpJsonRpcEndpoint, IMcpClient
|
|||
{
|
||||
// Connect transport
|
||||
_sessionTransport = await _clientTransport.ConnectAsync(cancellationToken).ConfigureAwait(false);
|
||||
InitializeSession(_sessionTransport);
|
||||
// We don't want the ConnectAsync token to cancel the session after we've successfully connected.
|
||||
// The base class handles cleaning up the session in DisposeAsync without our help.
|
||||
StartSession(fullSessionCancellationToken: CancellationToken.None);
|
||||
StartSession(_sessionTransport, fullSessionCancellationToken: CancellationToken.None);
|
||||
|
||||
// Perform initialization sequence
|
||||
using var initializationCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
|
||||
|
|
@ -142,13 +141,14 @@ internal sealed class McpClient : McpJsonRpcEndpoint, IMcpClient
|
|||
/// <inheritdoc/>
|
||||
public override async ValueTask DisposeUnsynchronizedAsync()
|
||||
{
|
||||
if (_connectCts is not null)
|
||||
{
|
||||
await _connectCts.CancelAsync().ConfigureAwait(false);
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
if (_connectCts is not null)
|
||||
{
|
||||
await _connectCts.CancelAsync().ConfigureAwait(false);
|
||||
_connectCts.Dispose();
|
||||
}
|
||||
|
||||
await base.DisposeUnsynchronizedAsync().ConfigureAwait(false);
|
||||
}
|
||||
finally
|
||||
|
|
@ -157,8 +157,6 @@ internal sealed class McpClient : McpJsonRpcEndpoint, IMcpClient
|
|||
{
|
||||
await _sessionTransport.DisposeAsync().ConfigureAwait(false);
|
||||
}
|
||||
|
||||
_connectCts?.Dispose();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -337,7 +337,7 @@ public static partial class McpServerBuilderExtensions
|
|||
Throw.IfNull(builder);
|
||||
|
||||
builder.Services.AddSingleton<ITransport, StdioServerTransport>();
|
||||
builder.Services.AddHostedService<McpServerSingleSessionHostedService>();
|
||||
builder.Services.AddHostedService<StdioMcpServerHostedService>();
|
||||
|
||||
builder.Services.AddSingleton(services =>
|
||||
{
|
||||
|
|
@ -350,18 +350,5 @@ public static partial class McpServerBuilderExtensions
|
|||
|
||||
return builder;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Adds a server transport that uses SSE via a HttpListener for communication.
|
||||
/// </summary>
|
||||
/// <param name="builder">The builder instance.</param>
|
||||
public static IMcpServerBuilder WithHttpListenerSseServerTransport(this IMcpServerBuilder builder)
|
||||
{
|
||||
Throw.IfNull(builder);
|
||||
|
||||
builder.Services.AddSingleton<IServerTransport, HttpListenerSseServerTransport>();
|
||||
builder.Services.AddHostedService<McpServerMultiSessionHostedService>();
|
||||
return builder;
|
||||
}
|
||||
#endregion
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,43 +0,0 @@
|
|||
using Microsoft.Extensions.Hosting;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Microsoft.Extensions.Options;
|
||||
using ModelContextProtocol.Protocol.Transport;
|
||||
using ModelContextProtocol.Server;
|
||||
|
||||
namespace ModelContextProtocol.Hosting;
|
||||
|
||||
/// <summary>
|
||||
/// Hosted service for a multi-session (i.e. HTTP) MCP server.
|
||||
/// </summary>
|
||||
internal sealed class McpServerMultiSessionHostedService : BackgroundService
|
||||
{
|
||||
private readonly IServerTransport _serverTransport;
|
||||
private readonly McpServerOptions _serverOptions;
|
||||
private readonly ILoggerFactory _loggerFactory;
|
||||
private readonly IServiceProvider _serviceProvider;
|
||||
|
||||
public McpServerMultiSessionHostedService(
|
||||
IServerTransport serverTransport,
|
||||
IOptions<McpServerOptions> serverOptions,
|
||||
ILoggerFactory loggerFactory,
|
||||
IServiceProvider serviceProvider)
|
||||
{
|
||||
_serverTransport = serverTransport;
|
||||
_serverOptions = serverOptions.Value;
|
||||
_loggerFactory = loggerFactory;
|
||||
_serviceProvider = serviceProvider;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
||||
{
|
||||
while (await AcceptSessionAsync(stoppingToken).ConfigureAwait(false) is { } server)
|
||||
{
|
||||
// TODO: Track all running sessions and wait for all sessions to complete for graceful shutdown.
|
||||
_ = server.RunAsync(stoppingToken);
|
||||
}
|
||||
}
|
||||
|
||||
private Task<IMcpServer> AcceptSessionAsync(CancellationToken cancellationToken)
|
||||
=> McpServerFactory.AcceptAsync(_serverTransport, _serverOptions, _loggerFactory, _serviceProvider, cancellationToken);
|
||||
}
|
||||
|
|
@ -0,0 +1,13 @@
|
|||
using Microsoft.Extensions.Hosting;
|
||||
using ModelContextProtocol.Server;
|
||||
|
||||
namespace ModelContextProtocol.Hosting;
|
||||
|
||||
/// <summary>
|
||||
/// Hosted service for a single-session (i.e stdio) MCP server.
|
||||
/// </summary>
|
||||
internal sealed class StdioMcpServerHostedService(IMcpServer session) : BackgroundService
|
||||
{
|
||||
/// <inheritdoc />
|
||||
protected override Task ExecuteAsync(CancellationToken stoppingToken) => session.RunAsync(stoppingToken);
|
||||
}
|
||||
|
|
@ -1,210 +0,0 @@
|
|||
using System.Net;
|
||||
|
||||
namespace ModelContextProtocol.Protocol.Transport;
|
||||
|
||||
/// <summary>
|
||||
/// HTTP server provider using HttpListener.
|
||||
/// </summary>
|
||||
internal sealed class HttpListenerServerProvider : IAsyncDisposable
|
||||
{
|
||||
private static readonly byte[] s_accepted = "Accepted"u8.ToArray();
|
||||
|
||||
private const string SseEndpoint = "/sse";
|
||||
private const string MessageEndpoint = "/message";
|
||||
|
||||
private readonly HttpListener _listener;
|
||||
private readonly CancellationTokenSource _shutdownTokenSource = new();
|
||||
private Task _listeningTask = Task.CompletedTask;
|
||||
|
||||
private readonly TaskCompletionSource<bool> _completed = new();
|
||||
private int _outstandingOperations;
|
||||
|
||||
private int _state;
|
||||
private const int StateNotStarted = 0;
|
||||
private const int StateRunning = 1;
|
||||
private const int StateStopped = 2;
|
||||
|
||||
/// <summary>
|
||||
/// Creates a new instance of the HTTP server provider.
|
||||
/// </summary>
|
||||
/// <param name="port">The port to listen on</param>
|
||||
public HttpListenerServerProvider(int port)
|
||||
{
|
||||
if (port < 0)
|
||||
{
|
||||
throw new ArgumentOutOfRangeException(nameof(port));
|
||||
}
|
||||
|
||||
_listener = new();
|
||||
_listener.Prefixes.Add($"http://localhost:{port}/");
|
||||
}
|
||||
|
||||
public required Func<Stream, CancellationToken, Task> OnSseConnectionAsync { get; set; }
|
||||
public required Func<Stream, CancellationToken, Task<bool>> OnMessageAsync { get; set; }
|
||||
|
||||
public void Start()
|
||||
{
|
||||
if (Interlocked.CompareExchange(ref _state, StateRunning, StateNotStarted) != StateNotStarted)
|
||||
{
|
||||
throw new ObjectDisposedException("Server may not be started twice.");
|
||||
}
|
||||
|
||||
// Start listening for connections
|
||||
_listener.Start();
|
||||
|
||||
OperationAdded(); // for the listening task
|
||||
_listeningTask = Task.Run(async () =>
|
||||
{
|
||||
try
|
||||
{
|
||||
using var cts = CancellationTokenSource.CreateLinkedTokenSource(_shutdownTokenSource.Token);
|
||||
cts.Token.Register(_listener.Stop);
|
||||
while (!cts.IsCancellationRequested)
|
||||
{
|
||||
try
|
||||
{
|
||||
var context = await _listener.GetContextAsync().ConfigureAwait(false);
|
||||
|
||||
// Process the request in a separate task
|
||||
OperationAdded(); // for the processing task; decremented in ProcessRequestAsync
|
||||
_ = Task.Run(() => ProcessRequestAsync(context, cts.Token), CancellationToken.None);
|
||||
}
|
||||
catch (Exception)
|
||||
{
|
||||
if (cts.IsCancellationRequested)
|
||||
{
|
||||
// Shutdown requested, exit gracefully
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
finally
|
||||
{
|
||||
OperationCompleted(); // for the listening task
|
||||
}
|
||||
}, CancellationToken.None);
|
||||
}
|
||||
|
||||
/// <inheritdoc/>
|
||||
public async ValueTask DisposeAsync()
|
||||
{
|
||||
if (Interlocked.CompareExchange(ref _state, StateStopped, StateRunning) != StateRunning)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
await _shutdownTokenSource.CancelAsync().ConfigureAwait(false);
|
||||
_listener.Stop();
|
||||
await _listeningTask.ConfigureAwait(false);
|
||||
await _completed.Task.ConfigureAwait(false);
|
||||
}
|
||||
|
||||
/// <summary>Gets a <see cref="Task"/> that completes when the server has finished its work.</summary>
|
||||
public Task Completed => _completed.Task;
|
||||
|
||||
private void OperationAdded() => Interlocked.Increment(ref _outstandingOperations);
|
||||
|
||||
private void OperationCompleted()
|
||||
{
|
||||
if (Interlocked.Decrement(ref _outstandingOperations) == 0)
|
||||
{
|
||||
// All operations completed
|
||||
_completed.TrySetResult(true);
|
||||
}
|
||||
}
|
||||
|
||||
private async Task ProcessRequestAsync(HttpListenerContext context, CancellationToken cancellationToken)
|
||||
{
|
||||
var request = context.Request;
|
||||
var response = context.Response;
|
||||
try
|
||||
{
|
||||
if (request is null || response is null)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
// Handle SSE connection
|
||||
if (request.HttpMethod == "GET" && request.Url?.LocalPath == SseEndpoint)
|
||||
{
|
||||
await HandleSseConnectionAsync(context, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
// Handle message POST
|
||||
else if (request.HttpMethod == "POST" && request.Url?.LocalPath == MessageEndpoint)
|
||||
{
|
||||
await HandleMessageAsync(context, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
else
|
||||
{
|
||||
// Not found
|
||||
response.StatusCode = 404;
|
||||
response.Close();
|
||||
}
|
||||
}
|
||||
catch (Exception)
|
||||
{
|
||||
try
|
||||
{
|
||||
response.StatusCode = 500;
|
||||
response.Close();
|
||||
}
|
||||
catch { /* Ignore errors during error handling */ }
|
||||
}
|
||||
finally
|
||||
{
|
||||
OperationCompleted();
|
||||
}
|
||||
}
|
||||
|
||||
private async Task HandleSseConnectionAsync(HttpListenerContext context, CancellationToken cancellationToken)
|
||||
{
|
||||
var response = context.Response;
|
||||
|
||||
// Set SSE headers
|
||||
response.ContentType = "text/event-stream";
|
||||
response.Headers.Add("Cache-Control", "no-cache");
|
||||
response.Headers.Add("Connection", "keep-alive");
|
||||
|
||||
// Keep the connection open until cancelled
|
||||
try
|
||||
{
|
||||
await OnSseConnectionAsync(response.OutputStream, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
catch (Exception)
|
||||
{
|
||||
}
|
||||
finally
|
||||
{
|
||||
// Remove client on disconnect
|
||||
try
|
||||
{
|
||||
response.Close();
|
||||
}
|
||||
catch { /* Ignore errors during cleanup */ }
|
||||
}
|
||||
}
|
||||
|
||||
private async Task HandleMessageAsync(HttpListenerContext context, CancellationToken cancellationToken)
|
||||
{
|
||||
var request = context.Request;
|
||||
var response = context.Response;
|
||||
|
||||
// Process the message asynchronously
|
||||
if (await OnMessageAsync(request.InputStream, cancellationToken))
|
||||
{
|
||||
// Return 202 Accepted
|
||||
response.StatusCode = 202;
|
||||
|
||||
// Write "accepted" response
|
||||
await response.OutputStream.WriteAsync(s_accepted, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
else
|
||||
{
|
||||
// Return 400 Bad Request
|
||||
response.StatusCode = 400;
|
||||
}
|
||||
|
||||
response.Close();
|
||||
}
|
||||
}
|
||||
|
|
@ -1,76 +0,0 @@
|
|||
using Microsoft.Extensions.Logging;
|
||||
using ModelContextProtocol.Logging;
|
||||
using ModelContextProtocol.Protocol.Messages;
|
||||
using ModelContextProtocol.Utils;
|
||||
using ModelContextProtocol.Utils.Json;
|
||||
using System.Net;
|
||||
using System.Text.Json;
|
||||
|
||||
namespace ModelContextProtocol.Protocol.Transport;
|
||||
|
||||
/// <summary>
|
||||
/// Implements the MCP transport protocol using <see cref="HttpListener"/>.
|
||||
/// </summary>
|
||||
internal sealed class HttpListenerSseServerSessionTransport : TransportBase
|
||||
{
|
||||
private readonly string _serverName;
|
||||
private readonly ILogger<HttpListenerSseServerTransport> _logger;
|
||||
private SseResponseStreamTransport _responseStreamTransport;
|
||||
|
||||
private string EndpointName => $"Server (SSE) ({_serverName})";
|
||||
|
||||
public HttpListenerSseServerSessionTransport(string serverName, SseResponseStreamTransport responseStreamTransport, ILoggerFactory loggerFactory)
|
||||
: base(loggerFactory)
|
||||
{
|
||||
Throw.IfNull(serverName);
|
||||
|
||||
_serverName = serverName;
|
||||
_responseStreamTransport = responseStreamTransport;
|
||||
_logger = loggerFactory.CreateLogger<HttpListenerSseServerTransport>();
|
||||
SetConnected(true);
|
||||
}
|
||||
|
||||
/// <inheritdoc/>
|
||||
public override async Task SendMessageAsync(IJsonRpcMessage message, CancellationToken cancellationToken = default)
|
||||
{
|
||||
if (!IsConnected)
|
||||
{
|
||||
_logger.TransportNotConnected(EndpointName);
|
||||
throw new McpTransportException("Transport is not connected");
|
||||
}
|
||||
|
||||
string id = "(no id)";
|
||||
if (message is IJsonRpcMessageWithId messageWithId)
|
||||
{
|
||||
id = messageWithId.Id.ToString();
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
if (_logger.IsEnabled(LogLevel.Debug))
|
||||
{
|
||||
var json = JsonSerializer.Serialize(message, McpJsonUtilities.DefaultOptions.GetTypeInfo<IJsonRpcMessage>());
|
||||
_logger.TransportSendingMessage(EndpointName, id, json);
|
||||
}
|
||||
|
||||
await _responseStreamTransport.SendMessageAsync(message, cancellationToken).ConfigureAwait(false);
|
||||
|
||||
_logger.TransportSentMessage(EndpointName, id);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.TransportSendFailed(EndpointName, id, ex);
|
||||
throw new McpTransportException("Failed to send message", ex);
|
||||
}
|
||||
}
|
||||
|
||||
public Task OnMessageReceivedAsync(IJsonRpcMessage message, CancellationToken cancellationToken)
|
||||
=> WriteMessageAsync(message, cancellationToken);
|
||||
|
||||
/// <inheritdoc/>
|
||||
public override ValueTask DisposeAsync()
|
||||
{
|
||||
SetConnected(false);
|
||||
return default;
|
||||
}
|
||||
}
|
||||
|
|
@ -1,180 +0,0 @@
|
|||
using System.Net;
|
||||
using System.Text.Json;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using ModelContextProtocol.Protocol.Messages;
|
||||
using ModelContextProtocol.Utils.Json;
|
||||
using ModelContextProtocol.Logging;
|
||||
using ModelContextProtocol.Server;
|
||||
using ModelContextProtocol.Utils;
|
||||
using System.Threading.Channels;
|
||||
|
||||
namespace ModelContextProtocol.Protocol.Transport;
|
||||
|
||||
/// <summary>
|
||||
/// Implements the MCP transport protocol using <see cref="HttpListener"/>.
|
||||
/// </summary>
|
||||
public sealed class HttpListenerSseServerTransport : IServerTransport, IAsyncDisposable
|
||||
{
|
||||
private readonly string _serverName;
|
||||
private readonly HttpListenerServerProvider _httpServerProvider;
|
||||
private readonly ILoggerFactory _loggerFactory;
|
||||
private readonly ILogger<HttpListenerSseServerTransport> _logger;
|
||||
|
||||
private readonly Channel<ITransport> _incomingSessions;
|
||||
|
||||
private HttpListenerSseServerSessionTransport? _sessionTransport;
|
||||
|
||||
private string EndpointName => $"Server (SSE) ({_serverName})";
|
||||
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the SseServerTransport class.
|
||||
/// </summary>
|
||||
/// <param name="serverOptions">The server options.</param>
|
||||
/// <param name="port">The port to listen on.</param>
|
||||
/// <param name="loggerFactory">A logger factory for creating loggers.</param>
|
||||
public HttpListenerSseServerTransport(McpServerOptions serverOptions, int port, ILoggerFactory loggerFactory)
|
||||
: this(GetServerName(serverOptions), port, loggerFactory)
|
||||
{
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the SseServerTransport class.
|
||||
/// </summary>
|
||||
/// <param name="serverName">The name of the server.</param>
|
||||
/// <param name="port">The port to listen on.</param>
|
||||
/// <param name="loggerFactory">A logger factory for creating loggers.</param>
|
||||
public HttpListenerSseServerTransport(string serverName, int port, ILoggerFactory loggerFactory)
|
||||
{
|
||||
Throw.IfNull(serverName);
|
||||
|
||||
_serverName = serverName;
|
||||
_loggerFactory = loggerFactory;
|
||||
_logger = loggerFactory.CreateLogger<HttpListenerSseServerTransport>();
|
||||
_httpServerProvider = new HttpListenerServerProvider(port)
|
||||
{
|
||||
OnSseConnectionAsync = OnSseConnectionAsync,
|
||||
OnMessageAsync = OnMessageAsync,
|
||||
};
|
||||
|
||||
// Until we support session IDs, there's no way to support more than one concurrent session.
|
||||
// Any new SSE connection overwrites the old session and any new /messages go to the new session.
|
||||
_incomingSessions = Channel.CreateBounded<ITransport>(new BoundedChannelOptions(1)
|
||||
{
|
||||
FullMode = BoundedChannelFullMode.DropOldest,
|
||||
});
|
||||
|
||||
// REVIEW: We could add another layer of async for binding similar to Kestrel's IConnectionListenerFactory,
|
||||
// but this wouldn't play well with a static factory method to accept new sessions. Ultimately,
|
||||
// ASP.NET Core is not going to hand over binding to the MCP SDK, so I decided to just bind in the transport
|
||||
// constructor for now.
|
||||
_httpServerProvider.Start();
|
||||
}
|
||||
|
||||
/// <inheritdoc/>
|
||||
public async Task<ITransport?> AcceptAsync(CancellationToken cancellationToken = default)
|
||||
{
|
||||
while (await _incomingSessions.Reader.WaitToReadAsync(cancellationToken).ConfigureAwait(false))
|
||||
{
|
||||
if (_incomingSessions.Reader.TryRead(out var session))
|
||||
{
|
||||
return session;
|
||||
}
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
/// <inheritdoc/>
|
||||
public async ValueTask DisposeAsync()
|
||||
{
|
||||
_logger.TransportCleaningUp(EndpointName);
|
||||
|
||||
await _httpServerProvider.DisposeAsync().ConfigureAwait(false);
|
||||
_incomingSessions.Writer.TryComplete();
|
||||
|
||||
_logger.TransportCleanedUp(EndpointName);
|
||||
}
|
||||
|
||||
private async Task OnSseConnectionAsync(Stream responseStream, CancellationToken cancellationToken)
|
||||
{
|
||||
var sseResponseStreamTransport = new SseResponseStreamTransport(responseStream);
|
||||
var sessionTransport = new HttpListenerSseServerSessionTransport(_serverName, sseResponseStreamTransport, _loggerFactory);
|
||||
|
||||
await using (sseResponseStreamTransport.ConfigureAwait(false))
|
||||
await using (sseResponseStreamTransport.ConfigureAwait(false))
|
||||
{
|
||||
_sessionTransport = sessionTransport;
|
||||
await _incomingSessions.Writer.WriteAsync(sessionTransport).ConfigureAwait(false);
|
||||
await sseResponseStreamTransport.RunAsync(cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Handles HTTP messages received by the HTTP server provider.
|
||||
/// </summary>
|
||||
/// <returns>true if the message was accepted (return 202), false otherwise (return 400)</returns>
|
||||
private async Task<bool> OnMessageAsync(Stream requestStream, CancellationToken cancellationToken)
|
||||
{
|
||||
string request;
|
||||
IJsonRpcMessage? message = null;
|
||||
|
||||
if (_logger.IsEnabled(LogLevel.Information))
|
||||
{
|
||||
using var reader = new StreamReader(requestStream);
|
||||
request = await reader.ReadToEndAsync(cancellationToken).ConfigureAwait(false);
|
||||
message = JsonSerializer.Deserialize(request, McpJsonUtilities.DefaultOptions.GetTypeInfo<IJsonRpcMessage>());
|
||||
|
||||
_logger.TransportReceivedMessage(EndpointName, request);
|
||||
}
|
||||
else
|
||||
{
|
||||
request = "(Enable information-level logs to see the request)";
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
message ??= await JsonSerializer.DeserializeAsync(requestStream, McpJsonUtilities.DefaultOptions.GetTypeInfo<IJsonRpcMessage>()).ConfigureAwait(false);
|
||||
if (message != null)
|
||||
{
|
||||
string messageId = "(no id)";
|
||||
if (message is IJsonRpcMessageWithId messageWithId)
|
||||
{
|
||||
messageId = messageWithId.Id.ToString();
|
||||
}
|
||||
|
||||
_logger.TransportReceivedMessageParsed(EndpointName, messageId);
|
||||
|
||||
if (_sessionTransport is null)
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
await _sessionTransport.OnMessageReceivedAsync(message, cancellationToken).ConfigureAwait(false);
|
||||
|
||||
_logger.TransportMessageWritten(EndpointName, messageId);
|
||||
|
||||
return true;
|
||||
}
|
||||
else
|
||||
{
|
||||
_logger.TransportMessageParseUnexpectedType(EndpointName, request);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
catch (JsonException ex)
|
||||
{
|
||||
_logger.TransportMessageParseFailed(EndpointName, request, ex);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Validates the <paramref name="serverOptions"/> and extracts from it the server name to use.</summary>
|
||||
private static string GetServerName(McpServerOptions serverOptions)
|
||||
{
|
||||
Throw.IfNull(serverOptions);
|
||||
Throw.IfNull(serverOptions.ServerInfo);
|
||||
Throw.IfNull(serverOptions.ServerInfo.Name);
|
||||
|
||||
return serverOptions.ServerInfo.Name;
|
||||
}
|
||||
}
|
||||
|
|
@ -1,14 +0,0 @@
|
|||
namespace ModelContextProtocol.Protocol.Transport;
|
||||
|
||||
/// <summary>
|
||||
/// Represents a transport mechanism for MCP communication (from the server).
|
||||
/// </summary>
|
||||
public interface IServerTransport
|
||||
{
|
||||
/// <summary>
|
||||
/// Asynchronously accepts a transport session initiated by an MCP client and returns an interface for the duplex JSON-RPC message stream.
|
||||
/// </summary>
|
||||
/// <param name="cancellationToken">Used to signal the cancellation of the asynchronous operation.</param>
|
||||
/// <returns>Returns an interface for the duplex JSON-RPC message stream.</returns>
|
||||
Task<ITransport?> AcceptAsync(CancellationToken cancellationToken = default);
|
||||
}
|
||||
|
|
@ -15,8 +15,8 @@ namespace ModelContextProtocol.Protocol.Transport;
|
|||
/// <param name="messageEndpoint">The endpoint to send JSON-RPC messages to. Defaults to "/message".</param>
|
||||
public sealed class SseResponseStreamTransport(Stream sseResponseStream, string messageEndpoint = "/message") : ITransport
|
||||
{
|
||||
private readonly Channel<IJsonRpcMessage> _incomingChannel = CreateSingleItemChannel<IJsonRpcMessage>();
|
||||
private readonly Channel<SseItem<IJsonRpcMessage?>> _outgoingSseChannel = CreateSingleItemChannel<SseItem<IJsonRpcMessage?>>();
|
||||
private readonly Channel<IJsonRpcMessage> _incomingChannel = CreateBoundedChannel<IJsonRpcMessage>();
|
||||
private readonly Channel<SseItem<IJsonRpcMessage?>> _outgoingSseChannel = CreateBoundedChannel<SseItem<IJsonRpcMessage?>>();
|
||||
|
||||
private Task? _sseWriteTask;
|
||||
private Utf8JsonWriter? _jsonWriter;
|
||||
|
|
@ -47,7 +47,10 @@ public sealed class SseResponseStreamTransport(Stream sseResponseStream, string
|
|||
|
||||
// The very first SSE event isn't really an IJsonRpcMessage, but there's no API to write a single item of a different type,
|
||||
// so we fib and special-case the "endpoint" event type in the formatter.
|
||||
_outgoingSseChannel.Writer.TryWrite(new SseItem<IJsonRpcMessage?>(null, "endpoint"));
|
||||
if (!_outgoingSseChannel.Writer.TryWrite(new SseItem<IJsonRpcMessage?>(null, "endpoint")))
|
||||
{
|
||||
throw new InvalidOperationException($"You must call ${nameof(RunAsync)} before calling ${nameof(SendMessageAsync)}.");
|
||||
}
|
||||
|
||||
var sseItems = _outgoingSseChannel.Reader.ReadAllAsync(cancellationToken);
|
||||
return _sseWriteTask = SseFormatter.WriteAsync(sseItems, sseResponseStream, WriteJsonRpcMessageToBuffer, cancellationToken);
|
||||
|
|
@ -73,7 +76,7 @@ public sealed class SseResponseStreamTransport(Stream sseResponseStream, string
|
|||
throw new InvalidOperationException($"Transport is not connected. Make sure to call {nameof(RunAsync)} first.");
|
||||
}
|
||||
|
||||
await _outgoingSseChannel.Writer.WriteAsync(new SseItem<IJsonRpcMessage?>(message), cancellationToken).AsTask();
|
||||
await _outgoingSseChannel.Writer.WriteAsync(new SseItem<IJsonRpcMessage?>(message), cancellationToken);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
|
|
@ -90,11 +93,11 @@ public sealed class SseResponseStreamTransport(Stream sseResponseStream, string
|
|||
throw new InvalidOperationException($"Transport is not connected. Make sure to call {nameof(RunAsync)} first.");
|
||||
}
|
||||
|
||||
await _incomingChannel.Writer.WriteAsync(message, cancellationToken).AsTask();
|
||||
await _incomingChannel.Writer.WriteAsync(message, cancellationToken);
|
||||
}
|
||||
|
||||
private static Channel<T> CreateSingleItemChannel<T>() =>
|
||||
Channel.CreateBounded<T>(new BoundedChannelOptions(1)
|
||||
private static Channel<T> CreateBoundedChannel<T>(int capacity = 1) =>
|
||||
Channel.CreateBounded<T>(new BoundedChannelOptions(capacity)
|
||||
{
|
||||
SingleReader = true,
|
||||
SingleWriter = false,
|
||||
|
|
|
|||
|
|
@ -12,30 +12,12 @@ namespace ModelContextProtocol.Server;
|
|||
/// <inheritdoc />
|
||||
internal sealed class McpServer : McpJsonRpcEndpoint, IMcpServer
|
||||
{
|
||||
private readonly IServerTransport? _serverTransport;
|
||||
private readonly EventHandler? _toolsChangedDelegate;
|
||||
private readonly EventHandler? _promptsChangedDelegate;
|
||||
|
||||
private ITransport? _sessionTransport;
|
||||
private ITransport _sessionTransport;
|
||||
private string _endpointName;
|
||||
|
||||
/// <summary>
|
||||
/// Creates a new instance of <see cref="McpServer"/>.
|
||||
/// </summary>
|
||||
/// <param name="serverTransport">Transport to use for the server that is ready to accept new sessions asynchronously.</param>
|
||||
/// <param name="options">Configuration options for this server, including capabilities.
|
||||
/// Make sure to accurately reflect exactly what capabilities the server supports and does not support.</param>
|
||||
/// <param name="loggerFactory">Logger factory to use for logging</param>
|
||||
/// <param name="serviceProvider">Optional service provider to use for dependency injection</param>
|
||||
/// <exception cref="McpServerException"></exception>
|
||||
public McpServer(IServerTransport serverTransport, McpServerOptions options, ILoggerFactory? loggerFactory, IServiceProvider? serviceProvider)
|
||||
: this(options, loggerFactory, serviceProvider)
|
||||
{
|
||||
Throw.IfNull(serverTransport);
|
||||
|
||||
_serverTransport = serverTransport;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Creates a new instance of <see cref="McpServer"/>.
|
||||
/// </summary>
|
||||
|
|
@ -46,27 +28,12 @@ internal sealed class McpServer : McpJsonRpcEndpoint, IMcpServer
|
|||
/// <param name="serviceProvider">Optional service provider to use for dependency injection</param>
|
||||
/// <exception cref="McpServerException"></exception>
|
||||
public McpServer(ITransport transport, McpServerOptions options, ILoggerFactory? loggerFactory, IServiceProvider? serviceProvider)
|
||||
: this(options, loggerFactory, serviceProvider)
|
||||
{
|
||||
Throw.IfNull(transport);
|
||||
|
||||
_sessionTransport = transport;
|
||||
InitializeSession(transport);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Creates a new instance of <see cref="McpServer"/>.
|
||||
/// </summary>
|
||||
/// <param name="options">Configuration options for this server, including capabilities.
|
||||
/// Make sure to accurately reflect exactly what capabilities the server supports and does not support.</param>
|
||||
/// <param name="loggerFactory">Logger factory to use for logging</param>
|
||||
/// <param name="serviceProvider">Optional service provider to use for dependency injection</param>
|
||||
/// <exception cref="McpServerException"></exception>
|
||||
private McpServer(McpServerOptions options, ILoggerFactory? loggerFactory, IServiceProvider? serviceProvider)
|
||||
: base(loggerFactory)
|
||||
{
|
||||
Throw.IfNull(transport);
|
||||
Throw.IfNull(options);
|
||||
|
||||
_sessionTransport = transport;
|
||||
ServerOptions = options;
|
||||
Services = serviceProvider;
|
||||
_endpointName = $"Server ({options.ServerInfo.Name} {options.ServerInfo.Version})";
|
||||
|
|
@ -128,36 +95,13 @@ internal sealed class McpServer : McpJsonRpcEndpoint, IMcpServer
|
|||
/// <inheritdoc />
|
||||
public override string EndpointName => _endpointName;
|
||||
|
||||
public async Task AcceptSessionAsync(CancellationToken cancellationToken = default)
|
||||
{
|
||||
// Below is effectively an assertion. The McpServerFactory should only use this with the IServerTransport constructor.
|
||||
Throw.IfNull(_serverTransport);
|
||||
|
||||
try
|
||||
{
|
||||
_sessionTransport = await _serverTransport.AcceptAsync(cancellationToken).ConfigureAwait(false);
|
||||
|
||||
if (_sessionTransport is null)
|
||||
{
|
||||
throw new McpServerException("The server transport closed before a client started a new session.");
|
||||
}
|
||||
|
||||
InitializeSession(_sessionTransport);
|
||||
}
|
||||
catch (Exception e)
|
||||
{
|
||||
_logger.ServerInitializationError(EndpointName, e);
|
||||
throw;
|
||||
}
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async Task RunAsync(CancellationToken cancellationToken = default)
|
||||
{
|
||||
try
|
||||
{
|
||||
// Start processing messages
|
||||
StartSession(fullSessionCancellationToken: cancellationToken);
|
||||
StartSession(_sessionTransport, fullSessionCancellationToken: cancellationToken);
|
||||
await MessageProcessingTask.ConfigureAwait(false);
|
||||
}
|
||||
finally
|
||||
|
|
@ -178,18 +122,7 @@ internal sealed class McpServer : McpJsonRpcEndpoint, IMcpServer
|
|||
prompts.Changed -= _promptsChangedDelegate;
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
await base.DisposeUnsynchronizedAsync().ConfigureAwait(false);
|
||||
}
|
||||
finally
|
||||
{
|
||||
if (_serverTransport is not null && _sessionTransport is not null)
|
||||
{
|
||||
// We created the _sessionTransport from the _serverTransport, so we own it.
|
||||
await _sessionTransport.DisposeAsync().ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
await base.DisposeUnsynchronizedAsync().ConfigureAwait(false);
|
||||
}
|
||||
|
||||
private void SetPingHandler()
|
||||
|
|
@ -208,7 +141,7 @@ internal sealed class McpServer : McpJsonRpcEndpoint, IMcpServer
|
|||
|
||||
// Use the ClientInfo to update the session EndpointName for logging.
|
||||
_endpointName = $"{_endpointName}, Client ({ClientInfo?.Name} {ClientInfo?.Version})";
|
||||
GetSessionOrThrow().EndpointName = EndpointName;
|
||||
GetSessionOrThrow().EndpointName = _endpointName;
|
||||
|
||||
return Task.FromResult(new InitializeResult()
|
||||
{
|
||||
|
|
|
|||
|
|
@ -33,42 +33,4 @@ public static class McpServerFactory
|
|||
|
||||
return new McpServer(transport, serverOptions, loggerFactory, serviceProvider);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Waits for the client to establish a new MCP session, then initializes a new instance of the <see cref="IMcpServer"/> class.
|
||||
/// </summary>
|
||||
/// <param name="serverTransport">Transport to use for the server that is ready to accept new MCP sessions asynchronously.</param>
|
||||
/// <param name="serverOptions">
|
||||
/// Configuration options for this server, including capabilities.
|
||||
/// Make sure to accurately reflect exactly what capabilities the server supports and does not support.
|
||||
/// </param>
|
||||
/// <param name="loggerFactory">Logger factory to use for logging</param>
|
||||
/// <param name="serviceProvider">Optional service provider to create new instances.</param>
|
||||
/// <param name="cancellationToken">Cancel waiting for a client to establish a new MCP session.</param>
|
||||
/// <returns>An <see cref="IMcpServer"/>.</returns>
|
||||
/// <exception cref="ArgumentNullException"><paramref name="serverTransport"/> is <see langword="null"/>.</exception>
|
||||
/// <exception cref="ArgumentNullException"><paramref name="serverOptions"/> is <see langword="null"/>.</exception>
|
||||
public static async Task<IMcpServer> AcceptAsync(
|
||||
IServerTransport serverTransport,
|
||||
McpServerOptions serverOptions,
|
||||
ILoggerFactory? loggerFactory = null,
|
||||
IServiceProvider? serviceProvider = null,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
Throw.IfNull(serverTransport);
|
||||
Throw.IfNull(serverOptions);
|
||||
|
||||
var mcpServer = new McpServer(serverTransport, serverOptions, loggerFactory, serviceProvider);
|
||||
|
||||
try
|
||||
{
|
||||
await mcpServer.AcceptSessionAsync(cancellationToken).ConfigureAwait(false);
|
||||
return mcpServer;
|
||||
}
|
||||
catch
|
||||
{
|
||||
await mcpServer.DisposeAsync().ConfigureAwait(false);
|
||||
throw;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -60,13 +60,8 @@ internal abstract class McpJsonRpcEndpoint : IAsyncDisposable
|
|||
/// </summary>
|
||||
protected Task? MessageProcessingTask { get; set; }
|
||||
|
||||
protected void InitializeSession(ITransport sessionTransport)
|
||||
{
|
||||
_session = new McpSession(sessionTransport, EndpointName, _requestHandlers, _notificationHandlers, _logger);
|
||||
}
|
||||
|
||||
[MemberNotNull(nameof(MessageProcessingTask))]
|
||||
protected void StartSession(CancellationToken fullSessionCancellationToken = default)
|
||||
protected void StartSession(ITransport sessionTransport, CancellationToken fullSessionCancellationToken = default)
|
||||
{
|
||||
if (Interlocked.Exchange(ref _started, 1) != 0)
|
||||
{
|
||||
|
|
@ -74,7 +69,8 @@ internal abstract class McpJsonRpcEndpoint : IAsyncDisposable
|
|||
}
|
||||
|
||||
_sessionCts = CancellationTokenSource.CreateLinkedTokenSource(fullSessionCancellationToken);
|
||||
MessageProcessingTask = GetSessionOrThrow().ProcessMessagesAsync(_sessionCts.Token);
|
||||
_session = new McpSession(sessionTransport, EndpointName, _requestHandlers, _notificationHandlers, _logger);
|
||||
MessageProcessingTask = _session.ProcessMessagesAsync(_sessionCts.Token);
|
||||
}
|
||||
|
||||
public async ValueTask DisposeAsync()
|
||||
|
|
@ -122,5 +118,5 @@ internal abstract class McpJsonRpcEndpoint : IAsyncDisposable
|
|||
}
|
||||
|
||||
protected McpSession GetSessionOrThrow()
|
||||
=> _session ?? throw new InvalidOperationException($"This should be unreachable from public API! Call {nameof(InitializeSession)} before sending messages.");
|
||||
=> _session ?? throw new InvalidOperationException($"This should be unreachable from public API! Call {nameof(StartSession)} before sending messages.");
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue