SSE ClientTransport fixes and tests (#27)

SSE fixes
Additional SSE integration tests with everything server

tests/mcpdotnet.Tests/EverythingSseServerFixture.cs

@@ -0,0 +1,63 @@
+using System.Diagnostics;
+
+namespace McpDotNet.Tests;
+
+public class EverythingSseServerFixture : IAsyncDisposable
+{
+    private Process _process;
+
+    public async Task StartAsync()
+    {
+        var processStartInfo = new ProcessStartInfo
+        {
+            FileName = "docker",
+            Arguments = "run -p 3001:3001 --rm tzolov/mcp-everything-server:v1",
+            RedirectStandardInput = true,
+            RedirectStandardOutput = true,
+            RedirectStandardError = true,
+            UseShellExecute = false,
+        };
+
+        _process = Process.Start(processStartInfo);
+
+        // Wait for the server to start
+        await Task.Delay(10000);
+    }
+    public async ValueTask DisposeAsync()
+    {
+        try
+        {
+            // Find the container ID
+            var psInfo = new ProcessStartInfo
+            {
+                FileName = "docker",
+                Arguments = "ps -q --filter ancestor=tzolov/mcp-everything-server:v1",
+                RedirectStandardOutput = true,
+                UseShellExecute = false
+            };
+
+            using var psProcess = Process.Start(psInfo);
+            string containerId = await psProcess.StandardOutput.ReadToEndAsync();
+            containerId = containerId.Trim();
+
+            if (!string.IsNullOrEmpty(containerId))
+            {
+                // Stop the container
+                var stopInfo = new ProcessStartInfo
+                {
+                    FileName = "docker",
+                    Arguments = $"stop {containerId}",
+                    UseShellExecute = false
+                };
+
+                using var stopProcess = Process.Start(stopInfo);
+                await stopProcess.WaitForExitAsync();
+            }
+        }
+        catch (Exception ex)
+        {
+            // Log the exception but don't throw
+            Console.Error.WriteLine($"Error stopping Docker container: {ex.Message}");
+        }
+    }
+}

tests/mcpdotnet.Tests/SseIntegrationTests.cs

@@ -2,6 +2,7 @@
 using McpDotNet.Configuration;
 using McpDotNet.Protocol.Messages;
 using McpDotNet.Protocol.Transport;
+using McpDotNet.Protocol.Types;
 using McpDotNet.Tests.Utils;
 using Microsoft.Extensions.Logging;
 using System.Text.Json;
@@ -49,18 +50,169 @@ public class SseIntegrationTests
         await server.WaitForConnectionAsync(TimeSpan.FromSeconds(10));
 
         // Send a test message through POST endpoint
-        var testMessage = new JsonRpcRequest()
+        await client.SendNotificationAsync("test/message", new { message = "Hello, SSE!" });
+
+        // Assert
+        Assert.True(true);
+    }
+
+    [Fact]
+    public async Task ConnectAndReceiveMessage_EverythingServerWithSse()
+    {
+        using var loggerFactory = Microsoft.Extensions.Logging.LoggerFactory.Create(builder =>
+            builder.AddConsole()
+            .SetMinimumLevel(LogLevel.Debug));
+
+        await using var fixture = new EverythingSseServerFixture();
+        await fixture.StartAsync();
+
+        var defaultOptions = new McpClientOptions
         {
-            Id = RequestId.FromNumber(1),
-            Method = "test/message",
-            Params = new { message = "Hello, SSE!" }
+            ClientInfo = new() { Name = "IntegrationTestClient", Version = "1.0.0" }
         };
-        using var httpClient = new HttpClient();
-        await httpClient.PostAsync(
-            "http://localhost:5000/message",
-            new StringContent(JsonSerializer.Serialize(testMessage))
+
+        var defaultConfig = new McpServerConfig
+        {
+            Id = "everything",
+            Name = "Everything",
+            TransportType = "sse",
+            TransportOptions = new Dictionary<string, string>(),
+            Location = "http://localhost:3001/sse"
+        };
+
+        var factory = new McpClientFactory(
+            [defaultConfig],
+            defaultOptions,
+            loggerFactory
         );
 
+        // Create client and run tests
+        var client = await factory.GetClientAsync("everything");
+        var tools = await client.ListToolsAsync();
+
+        // assert
+        Assert.NotNull(tools);
+        Assert.NotEmpty(tools.Tools);
+    }
+
+    [Fact]
+    public async Task Sampling_Sse_EverythingServer()
+    {
+        // arrange
+        using var loggerFactory = Microsoft.Extensions.Logging.LoggerFactory.Create(builder =>
+            builder.AddConsole()
+            .SetMinimumLevel(LogLevel.Debug));
+
+        await using var fixture = new EverythingSseServerFixture();
+        await fixture.StartAsync();
+
+        var defaultOptions = new McpClientOptions
+        {
+            ClientInfo = new()
+            {
+                Name = "IntegrationTestClient",
+                Version = "1.0.0"                
+            },
+            Capabilities = new()
+            {
+                Sampling = new()
+            }
+        };
+
+        var defaultConfig = new McpServerConfig
+        {
+            Id = "everything",
+            Name = "Everything",
+            TransportType = "sse",
+            TransportOptions = new Dictionary<string, string>(),
+            Location = "http://localhost:3001/sse"
+        };
+
+        var factory = new McpClientFactory(
+            [defaultConfig],
+            defaultOptions,
+            loggerFactory
+        );
+        var client = await factory.GetClientAsync("everything");
+
+        // Set up the sampling handler
+        int samplingHandlerCalls = 0;
+#pragma warning disable CS1998 // Async method lacks 'await' operators and will run synchronously
+        client.SamplingHandler = async (_, _) =>
+        {
+            samplingHandlerCalls++;
+            return new CreateMessageResult
+            {
+                Model = "test-model",
+                Role = "assistant",
+                Content = new Content
+                {
+                    Type = "text",
+                    Text = "Test response"
+                }
+            };
+        };
+#pragma warning restore CS1998 // Async method lacks 'await' operators and will run synchronously
+
+        // Call the server's sampleLLM tool which should trigger our sampling handler
+        var result = await client.CallToolAsync(
+            "sampleLLM",
+            new Dictionary<string, object>
+            {
+                ["prompt"] = "Test prompt",
+                ["maxTokens"] = 100
+            }
+        );
+
+        // assert
+        Assert.NotNull(result);
+        var textContent = Assert.Single(result.Content);
+        Assert.Equal("text", textContent.Type);
+        Assert.False(string.IsNullOrEmpty(textContent.Text));
+    }
+
+    [Fact]
+    public async Task ConnectAndReceiveMessage_InMemoryServer_WithFullEndpointEventUri()
+    {
+        // Arrange
+        using var loggerFactory = Microsoft.Extensions.Logging.LoggerFactory.Create(builder =>
+            builder.AddConsole()
+            .SetMinimumLevel(LogLevel.Debug));
+
+        await using TestSseServer server = new(logger: loggerFactory.CreateLogger<TestSseServer>());
+        server.UseFullUrlForEndpointEvent = true;
+        await server.StartAsync();
+
+
+        var defaultOptions = new McpClientOptions
+        {
+            ClientInfo = new() { Name = "IntegrationTestClient", Version = "1.0.0" }
+        };
+
+        var defaultConfig = new McpServerConfig
+        {
+            Id = "test_server",
+            Name = "In-memory Test Server",
+            TransportType = "sse",
+            TransportOptions = new Dictionary<string, string>(),
+            Location = "http://localhost:5000/sse"
+        };
+
+        var factory = new McpClientFactory(
+            [defaultConfig],
+            defaultOptions,
+            loggerFactory
+        );
+
+        // Act
+        var client = await factory.GetClientAsync("test_server");
+
+        // Wait for SSE connection to be established
+        await server.WaitForConnectionAsync(TimeSpan.FromSeconds(10));
+
+        // Send a test message through POST endpoint
+        await client.SendNotificationAsync("test/message", new { message = "Hello, SSE!" });
+
         // Assert
         Assert.True(true);
     }

tests/mcpdotnet.Tests/Utils/TestSseServer.cs

@@ -34,6 +34,11 @@ public sealed class TestSseServer : IAsyncDisposable
         _messagePath = "/message";
     }
 
+    /// <summary>
+    /// This is to be able to use the full URL for the endpoint event.
+    /// </summary>
+    public bool UseFullUrlForEndpointEvent { get; set; }
+
     /// <summary>
     /// Full URL for the SSE endpoint, e.g. "http://localhost:5000/sse"
     /// </summary>
@@ -120,14 +125,15 @@ public sealed class TestSseServer : IAsyncDisposable
         _sseClients.Add(writer);
 
         // Immediately send the "endpoint" event with the POST URL
-        var endpointData = new
-        {
-            uri = MessageEndpoint
-        };
-        var json = JsonSerializer.Serialize(endpointData);
-
         await writer.WriteLineAsync("event: endpoint");
-        await writer.WriteLineAsync($"data: {json}");
+        if (UseFullUrlForEndpointEvent)
+        {
+            await writer.WriteLineAsync($"data: {MessageEndpoint}");
+        }
+        else
+        {
+            await writer.WriteLineAsync($"data: {_messagePath}");
+        }
         await writer.WriteLineAsync(); // blank line to end an SSE message
         await writer.FlushAsync();
 
@@ -183,33 +189,50 @@ public sealed class TestSseServer : IAsyncDisposable
             using var reader = new StreamReader(request.InputStream);
             string content = await reader.ReadToEndAsync();
 
+            var jsonRpcNotification = JsonSerializer.Deserialize<JsonRpcNotification>(content);
+            if (jsonRpcNotification != null && jsonRpcNotification.Method != "initialize")
+            {
+                // Test server so just ignore notifications
+
+                // Set status code to 202 Accepted
+                response.StatusCode = 202;
+
+                // Write "accepted" as the response body
+                using var writer = new StreamWriter(response.OutputStream);
+                await writer.WriteAsync("accepted");
+                await writer.FlushAsync();
+                return;
+            }
+
             var jsonRpcRequest = JsonSerializer.Deserialize<JsonRpcRequest>(content);
 
-            if (jsonRpcRequest != null && !jsonRpcRequest.Method.StartsWith("notification"))
+            if (jsonRpcRequest != null)
             {
                 if (jsonRpcRequest.Method == "initialize")
                 {
                     await HandleInitializationRequest(response, jsonRpcRequest);
-                    return;
                 }
+                else
+                {
+                    // Set status code to 202 Accepted
+                    response.StatusCode = 202;
 
-                // Method not found - proper JSON-RPC error response
-                await SendJsonRpcErrorAsync(response, jsonRpcRequest.Id, -32601, $"Method '{jsonRpcRequest.Method}' not found");
-                return;
-            }
- 
-            var notification = JsonSerializer.Deserialize<JsonRpcNotification>(content);
-            if (notification != null)
-            {
-                response.StatusCode = 200;
-                response.ContentType = "application/json";
-                await using var writer = new StreamWriter(response.OutputStream);
-                await writer.WriteAsync(JsonSerializer.Serialize(new { jsonrpc = "2.0" }));
-                return;
-            }
+                    // Write "accepted" as the response body
+                    using var writer = new StreamWriter(response.OutputStream);
+                    await writer.WriteAsync("accepted");
+                    await writer.FlushAsync();
 
-            // Invalid JSON-RPC message - proper JSON-RPC error response
-            await SendJsonRpcErrorAsync(response, null, -32700, "Parse error");
+                    // Then process the message and send the response via SSE
+                    if (jsonRpcRequest != null)
+                    {
+                        // Process the request and generate a response
+                        var responseMessage = await ProcessRequestAsync(jsonRpcRequest);
+
+                        // Send the response via the SSE connection instead of HTTP response
+                        await SendSseMessageAsync(responseMessage);
+                    }
+                }
+            }
         }
         catch (Exception ex)
         {
@@ -222,6 +245,22 @@ public sealed class TestSseServer : IAsyncDisposable
         }
     }
 
+    private async Task<JsonRpcResponse> ProcessRequestAsync(JsonRpcRequest jsonRpcRequest)
+    {
+        // This is a test server so we just echo back the request
+        return new JsonRpcResponse
+        {
+            Id = jsonRpcRequest.Id,
+            Result = jsonRpcRequest.Params
+        };
+    }
+
+    private async Task SendSseMessageAsync(JsonRpcResponse jsonRpcResponse)
+    {
+        // This is a test server so we just send to all connected clients
+        await BroadcastMessageAsync(JsonSerializer.Serialize(jsonRpcResponse));
+    }
+
     private async Task SendJsonRpcErrorAsync(HttpListenerResponse response, RequestId? id, int code, string message, string? data = null)
     {
         var errorResponse = new JsonRpcError