Write "event: message" to SSE response (#192)

* Write "event: message" to SSE response
* Rename McpJsonRpcEndpoint to McpEndpoint
* Remove unused McpServerConfig.Arguments property

src/ModelContextProtocol/Client/McpClient.cs

@@ -10,7 +10,7 @@ using System.Text.Json;
 namespace ModelContextProtocol.Client;
 
 /// <inheritdoc/>
-internal sealed class McpClient : McpJsonRpcEndpoint, IMcpClient
+internal sealed class McpClient : McpEndpoint, IMcpClient
 {
     private readonly IClientTransport _clientTransport;
     private readonly McpClientOptions _options;

src/ModelContextProtocol/Configuration/McpServerConfig.cs

@@ -27,11 +27,6 @@ public record McpServerConfig
     /// </summary>
     public string? Location { get; set; }
 
-    /// <summary>
-    /// Arguments (if any) to pass to the executable.
-    /// </summary>
-    public string[]? Arguments { get; init; }
-
     /// <summary>
     /// Additional transport-specific configuration.
     /// </summary>

src/ModelContextProtocol/Protocol/Transport/SseResponseStreamTransport.cs

@@ -76,7 +76,8 @@ 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);
+        // Emit redundant "event: message" lines for better compatibility with other SDKs.
+        await _outgoingSseChannel.Writer.WriteAsync(new SseItem<IJsonRpcMessage?>(message, SseParser.EventTypeDefault), cancellationToken);
     }
 
     /// <summary>

src/ModelContextProtocol/Server/McpServer.cs

@@ -9,7 +9,7 @@ using ModelContextProtocol.Utils.Json;
 namespace ModelContextProtocol.Server;
 
 /// <inheritdoc />
-internal sealed class McpServer : McpJsonRpcEndpoint, IMcpServer
+internal sealed class McpServer : McpEndpoint, IMcpServer
 {
     private readonly EventHandler? _toolsChangedDelegate;
     private readonly EventHandler? _promptsChangedDelegate;

src/ModelContextProtocol/Shared/McpEndpoint.cs

@@ -0,0 +1,130 @@
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Logging.Abstractions;
+using ModelContextProtocol.Logging;
+using ModelContextProtocol.Protocol.Messages;
+using ModelContextProtocol.Protocol.Transport;
+using ModelContextProtocol.Server;
+using ModelContextProtocol.Utils;
+using System.Diagnostics.CodeAnalysis;
+using System.Text.Json.Serialization.Metadata;
+
+namespace ModelContextProtocol.Shared;
+
+/// <summary>
+/// Base class for an MCP JSON-RPC endpoint. This covers both MCP clients and servers.
+/// It is not supported, nor necessary, to implement both client and server functionality in the same class.
+/// If an application needs to act as both a client and a server, it should use separate objects for each.
+/// This is especially true as a client represents a connection to one and only one server, and vice versa.
+/// Any multi-client or multi-server functionality should be implemented at a higher level of abstraction.
+/// </summary>
+internal abstract class McpEndpoint : IAsyncDisposable
+{
+    private readonly RequestHandlers _requestHandlers = [];
+    private readonly NotificationHandlers _notificationHandlers = [];
+
+    private McpSession? _session;
+    private CancellationTokenSource? _sessionCts;
+
+    private readonly SemaphoreSlim _disposeLock = new(1, 1);
+    private bool _disposed;
+
+    protected readonly ILogger _logger;
+
+    /// <summary>
+    /// Initializes a new instance of the <see cref="McpEndpoint"/> class.
+    /// </summary>
+    /// <param name="loggerFactory">The logger factory.</param>
+    protected McpEndpoint(ILoggerFactory? loggerFactory = null)
+    {
+        _logger = loggerFactory?.CreateLogger(GetType()) ?? NullLogger.Instance;
+    }
+
+    protected void SetRequestHandler<TRequest, TResponse>(
+        string method,
+        Func<TRequest?, CancellationToken, Task<TResponse>> handler,
+        JsonTypeInfo<TRequest> requestTypeInfo,
+        JsonTypeInfo<TResponse> responseTypeInfo)
+
+        => _requestHandlers.Set(method, handler, requestTypeInfo, responseTypeInfo);
+
+    public void AddNotificationHandler(string method, Func<JsonRpcNotification, Task> handler)
+        => _notificationHandlers.Add(method, handler);
+
+    public Task<JsonRpcResponse> SendRequestAsync(JsonRpcRequest request, CancellationToken cancellationToken = default)
+        => GetSessionOrThrow().SendRequestAsync(request, cancellationToken);
+
+    public Task SendMessageAsync(IJsonRpcMessage message, CancellationToken cancellationToken = default)
+        => GetSessionOrThrow().SendMessageAsync(message, cancellationToken);
+
+    /// <summary>
+    /// Gets the name of the endpoint for logging and debug purposes.
+    /// </summary>
+    public abstract string EndpointName { get; }
+
+    /// <summary>
+    /// Task that processes incoming messages from the transport.
+    /// </summary>
+    protected Task? MessageProcessingTask { get; private set; }
+
+    [MemberNotNull(nameof(MessageProcessingTask))]
+    protected void StartSession(ITransport sessionTransport)
+    {
+        _sessionCts = new CancellationTokenSource();
+        _session = new McpSession(this is IMcpServer, sessionTransport, EndpointName, _requestHandlers, _notificationHandlers, _logger);
+        MessageProcessingTask = _session.ProcessMessagesAsync(_sessionCts.Token);
+    }
+
+    protected void CancelSession() => _sessionCts?.Cancel();
+
+    public async ValueTask DisposeAsync()
+    {
+        using var _ = await _disposeLock.LockAsync().ConfigureAwait(false);
+
+        if (_disposed)
+        {
+            return;
+        }
+        _disposed = true;
+
+        await DisposeUnsynchronizedAsync().ConfigureAwait(false);
+    }
+
+    /// <summary>
+    /// Cleans up the endpoint and releases resources.
+    /// </summary>
+    /// <returns></returns>
+    public virtual async ValueTask DisposeUnsynchronizedAsync()
+    {
+        _logger.CleaningUpEndpoint(EndpointName);
+
+        try
+        {
+            if (_sessionCts is not null)
+            {
+                await _sessionCts.CancelAsync().ConfigureAwait(false);
+            }
+
+            if (MessageProcessingTask is not null)
+            {
+                try
+                {
+                    await MessageProcessingTask.ConfigureAwait(false);
+                }
+                catch (OperationCanceledException)
+                {
+                    // Ignore cancellation
+                }
+            }
+        }
+        finally
+        {
+            _session?.Dispose();
+            _sessionCts?.Dispose();
+        }
+
+        _logger.EndpointCleanedUp(EndpointName);
+    }
+
+    protected McpSession GetSessionOrThrow()
+        => _session ?? throw new InvalidOperationException($"This should be unreachable from public API! Call {nameof(StartSession)} before sending messages.");
+}

src/ModelContextProtocol/Shared/McpSession.cs

@@ -45,7 +45,7 @@ internal sealed class McpSession : IDisposable
     /// </summary>
     private readonly ConcurrentDictionary<RequestId, CancellationTokenSource> _handlingRequests = new();
     private readonly ILogger _logger;
-    
+
     private readonly string _id = Guid.NewGuid().ToString("N");
     private long _nextRequestId;
 

tests/ModelContextProtocol.Tests/SseServerIntegrationTests.cs

@@ -1,6 +1,8 @@
 using ModelContextProtocol.Client;
 using ModelContextProtocol.Protocol.Types;
 using ModelContextProtocol.Tests.Utils;
+using System.Net;
+using System.Text;
 
 namespace ModelContextProtocol.Tests;
 
@@ -30,6 +32,12 @@ public class SseServerIntegrationTests : LoggedTest, IClassFixture<SseServerInte
             cancellationToken: TestContext.Current.CancellationToken);
     }
 
+    private HttpClient GetHttpClient() =>
+        new()
+        {
+            BaseAddress = new(_fixture.DefaultConfig.Location!),
+        };
+
     [Fact]
     public async Task ConnectAndPing_Sse_TestServer()
     {
@@ -271,4 +279,44 @@ public class SseServerIntegrationTests : LoggedTest, IClassFixture<SseServerInte
             Assert.Equal($"Echo: Hello MCP! {i}", textContent.Text);
         }
     }
+
+    [Fact]
+    public async Task EventSourceStream_Includes_MessageEventType()
+    {
+        // Simulate our own MCP client handshake using a plain HttpClient so we can look for "event: message"
+        // in the raw SSE response stream which is not exposed by the real MCP client.
+        using var httpClient = GetHttpClient();
+        await using var sseResponse = await httpClient.GetStreamAsync("", TestContext.Current.CancellationToken);
+        using var streamReader = new StreamReader(sseResponse);
+
+        var endpointEvent = await streamReader.ReadLineAsync(TestContext.Current.CancellationToken);
+        Assert.Equal("event: endpoint", endpointEvent);
+
+        var endpointData = await streamReader.ReadLineAsync(TestContext.Current.CancellationToken);
+        Assert.NotNull(endpointData);
+        Assert.StartsWith("data: ", endpointData);
+        var messageEndpoint = endpointData["data: ".Length..];
+
+        const string initializeRequest = """
+            {"jsonrpc":"2.0","id":"1","method":"initialize","params":{"protocolVersion":"2024-11-05","capabilities":{},"clientInfo":{"name":"IntegrationTestClient","version":"1.0.0"}}}
+            """;
+        using (var initializeRequestBody = new StringContent(initializeRequest, Encoding.UTF8, "application/json"))
+        {
+            var response = await httpClient.PostAsync(messageEndpoint, initializeRequestBody, TestContext.Current.CancellationToken);
+            Assert.Equal(HttpStatusCode.Accepted, response.StatusCode);
+        }
+
+        const string initializedNotification = """
+            {"jsonrpc":"2.0","method":"notifications/initialized"}
+            """;
+        using (var initializedNotificationBody = new StringContent(initializedNotification, Encoding.UTF8, "application/json"))
+        {
+            var response = await httpClient.PostAsync(messageEndpoint, initializedNotificationBody, TestContext.Current.CancellationToken);
+            Assert.Equal(HttpStatusCode.Accepted, response.StatusCode);
+        }
+
+        Assert.Equal("", await streamReader.ReadLineAsync(TestContext.Current.CancellationToken));
+        var messageEvent = await streamReader.ReadLineAsync(TestContext.Current.CancellationToken);
+        Assert.Equal("event: message", messageEvent);
+    }
 }