From 30f3f192cacfeaf8e2be97af351da4db07c876fd Mon Sep 17 00:00:00 2001 From: Peder Holdgaard Pedersen <127606677+PederHP@users.noreply.github.com> Date: Thu, 27 Feb 2025 21:14:20 +0100 Subject: [PATCH] SSE ClientTransport fixes and tests (#27) SSE fixes Additional SSE integration tests with everything server --- .../McpDotNet.Extensions.AI.csproj | 10 +- .../McpSessionScope.cs | 4 +- src/mcpdotnet/Logging/Log.cs | 21 +++ .../Protocol/Transport/SseClientTransport.cs | 103 ++++++----- src/mcpdotnet/mcpdotnet.csproj | 2 +- .../EverythingSseServerFixture.cs | 63 +++++++ tests/mcpdotnet.Tests/SseIntegrationTests.cs | 168 +++++++++++++++++- tests/mcpdotnet.Tests/Utils/TestSseServer.cs | 89 +++++++--- 8 files changed, 380 insertions(+), 80 deletions(-) create mode 100644 tests/mcpdotnet.Tests/EverythingSseServerFixture.cs diff --git a/src/McpDotNet.Extensions.AI/McpDotNet.Extensions.AI.csproj b/src/McpDotNet.Extensions.AI/McpDotNet.Extensions.AI.csproj index 17a901d..49671fd 100644 --- a/src/McpDotNet.Extensions.AI/McpDotNet.Extensions.AI.csproj +++ b/src/McpDotNet.Extensions.AI/McpDotNet.Extensions.AI.csproj @@ -9,10 +9,10 @@ McpDotNet.Extensions.AI - 1.0.1.1 + 1.0.1.3 PederHP Microsoft.Extensions.AI integration for the Model Context Protocol (MCP). Enables seamless use of MCP tools as AI functions in any IChatClient implementation. - https://github.com/PederHP/mcpdotnet/src/McpDotNet.Extensions.AI + https://github.com/PederHP/mcpdotnet/tree/main/src/McpDotNet.Extensions.AI https://github.com/PederHP/mcpdotnet git mcp;ai;microsoft-extensions-ai;chatclient;llm @@ -26,6 +26,10 @@ snupkg + + + + @@ -33,7 +37,7 @@ - + diff --git a/src/McpDotNet.Extensions.AI/McpSessionScope.cs b/src/McpDotNet.Extensions.AI/McpSessionScope.cs index 47a03bc..872c44f 100644 --- a/src/McpDotNet.Extensions.AI/McpSessionScope.cs +++ b/src/McpDotNet.Extensions.AI/McpSessionScope.cs @@ -20,7 +20,7 @@ public class McpSessionScope : IAsyncDisposable /// /// The list of tools exposed by the servers in this scope. /// - public IList Tools { get; private set; } + public IList Tools { get; private set; } = []; /// /// The server instructions provided by the servers in this scope. @@ -84,7 +84,7 @@ public class McpSessionScope : IAsyncDisposable return client; } - /// + /// public async ValueTask DisposeAsync() { if (_disposed) return; diff --git a/src/mcpdotnet/Logging/Log.cs b/src/mcpdotnet/Logging/Log.cs index c0cf0c2..ed5e286 100644 --- a/src/mcpdotnet/Logging/Log.cs +++ b/src/mcpdotnet/Logging/Log.cs @@ -480,4 +480,25 @@ internal static partial class Log this ILogger logger, string endpointName, Exception e); + + [LoggerMessage( + EventId = 7018, + Level = LogLevel.Debug, + Message = "SSE transport POST accepted for {endpointName} with message ID {messageId}" + )] + public static partial void SSETransportPostAccepted( + this ILogger logger, + string endpointName, + string messageId); + + [LoggerMessage( + EventId = 7019, + Level = LogLevel.Error, + Message = "SSE transport POST not accepted for {endpointName} with message ID {messageId} and server response {responseContent}" + )] + public static partial void SSETransportPostNotAccepted( + this ILogger logger, + string endpointName, + string messageId, + string responseContent); } \ No newline at end of file diff --git a/src/mcpdotnet/Protocol/Transport/SseClientTransport.cs b/src/mcpdotnet/Protocol/Transport/SseClientTransport.cs index f50ca83..6c96188 100644 --- a/src/mcpdotnet/Protocol/Transport/SseClientTransport.cs +++ b/src/mcpdotnet/Protocol/Transport/SseClientTransport.cs @@ -99,6 +99,13 @@ public sealed class SseClientTransport : TransportBase, IClientTransport "application/json" ); + string messageId = "(no id)"; + RequestId? requestId = null; + if (message is IJsonRpcMessageWithId messageWithId) + { + messageId = messageWithId.Id.ToString(); + } + var response = await _httpClient.PostAsync( _messageEndpoint, content, @@ -107,31 +114,40 @@ public sealed class SseClientTransport : TransportBase, IClientTransport response.EnsureSuccessStatusCode(); - var responseContent = await response.Content.ReadAsStringAsync(); - // Handle notifications, which don't have responses - if (message is JsonRpcNotification) + // Check if the message was an initialize request + if (message is JsonRpcRequest request && request.Method == "initialize") { - await HandleNotificationResponseAsync(message, responseContent, cancellationToken); + // If the response is not a JSON-RPC response, it is an SSE message + if (responseContent.ToLowerInvariant() == "accepted") + { + _logger.SSETransportPostAccepted(EndpointName, messageId); + // The response will arrive as an SSE message + } + else + { + JsonRpcResponse initializeResponse = JsonSerializer.Deserialize(responseContent, _jsonOptions); + if (initializeResponse == null) + { + throw new McpTransportException("Failed to initialize client"); + } + _logger.TransportReceivedMessageParsed(EndpointName, messageId); + await WriteMessageAsync(initializeResponse, cancellationToken); + _logger.TransportMessageWritten(EndpointName, messageId); + } return; } - var responseMessage = JsonSerializer.Deserialize(responseContent, _jsonOptions); - if (responseMessage != null) + // Otherwise, check if the response was accepted (the response will come as an SSE message) + if (responseContent.ToLowerInvariant() == "accepted") { - string messageId = "(no id)"; - if (responseMessage is IJsonRpcMessageWithId messageWithId) - { - messageId = messageWithId.Id.ToString(); - } - _logger.TransportReceivedMessageParsed(EndpointName, messageId); - await WriteMessageAsync(responseMessage, cancellationToken); - _logger.TransportMessageWritten(EndpointName, messageId); + _logger.SSETransportPostAccepted(EndpointName, messageId); } else { - _logger.TransportMessageParseUnexpectedType(EndpointName, responseContent); + _logger.SSETransportPostNotAccepted(EndpointName, messageId, responseContent); + throw new McpTransportException("Failed to send message"); } } @@ -154,28 +170,6 @@ public sealed class SseClientTransport : TransportBase, IClientTransport GC.SuppressFinalize(this); } - private async Task HandleNotificationResponseAsync( - IJsonRpcMessage message, - string responseContent, - CancellationToken cancellationToken) - { - // Check for error response, but don't require one - try - { - var errorResponse = JsonSerializer.Deserialize(responseContent, _jsonOptions); - if (errorResponse is not null) - { - await WriteMessageAsync(errorResponse, cancellationToken); - return; - } - } - catch (JsonException) - { - // Ignore deserialization errors for notifications - // This is expected as most notifications won't have responses - } - } - internal Uri? MessageEndpoint => _messageEndpoint; internal SseClientTransportOptions Options => _options; @@ -207,7 +201,7 @@ public sealed class SseClientTransport : TransportBase, IClientTransport string? currentEvent = null; string? line; - while ((line = await reader.ReadLineAsync()) != null) + while ((line = await reader.ReadLineAsync(cancellationToken)) != null) { if (line.StartsWith("event: ")) { @@ -227,6 +221,16 @@ public sealed class SseClientTransport : TransportBase, IClientTransport } } } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + _logger.TransportReadMessagesCancelled(EndpointName); + // Normal shutdown + } + catch (IOException) when (cancellationToken.IsCancellationRequested) + { + _logger.TransportReadMessagesCancelled(EndpointName); + // Normal shutdown + } catch (Exception ex) when (!cancellationToken.IsCancellationRequested) { _logger.TransportConnectionError(EndpointName, ex); @@ -279,14 +283,31 @@ public sealed class SseClientTransport : TransportBase, IClientTransport { try { - var endpointData = JsonSerializer.Deserialize(data, _jsonOptions); - if (endpointData?.Uri == null) + if (string.IsNullOrEmpty(data)) { _logger.TransportEndpointEventInvalid(EndpointName, data); return; } - _messageEndpoint = new Uri(endpointData.Uri); + // Check if data is absolute URI + if (data.ToLowerInvariant().StartsWith("http://") || data.ToLowerInvariant().StartsWith("https://")) + { + // Since the endpoint is an absolute URI, we can use it directly + _messageEndpoint = new Uri(data); + } + else + { + // If the endpoint is a relative URI, we need to combine it with the relative path of the SSE endpoint + var hostUrl = _sseEndpoint.AbsoluteUri; + if (hostUrl.EndsWith("/sse")) + hostUrl = hostUrl[..^4]; + + var endpointUri = hostUrl + data; + + _messageEndpoint = new Uri(endpointUri); + } + + // Set connected state SetConnected(true); _connectionEstablished.TrySetResult(true); } diff --git a/src/mcpdotnet/mcpdotnet.csproj b/src/mcpdotnet/mcpdotnet.csproj index bd46eb5..f0cfde9 100644 --- a/src/mcpdotnet/mcpdotnet.csproj +++ b/src/mcpdotnet/mcpdotnet.csproj @@ -9,7 +9,7 @@ mcpdotnet - 1.0.1.1 + 1.0.1.3 PederHP .NET client library for the Model Context Protocol (MCP) https://github.com/PederHP/mcpdotnet diff --git a/tests/mcpdotnet.Tests/EverythingSseServerFixture.cs b/tests/mcpdotnet.Tests/EverythingSseServerFixture.cs new file mode 100644 index 0000000..dcffe04 --- /dev/null +++ b/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}"); + } + } +} \ No newline at end of file diff --git a/tests/mcpdotnet.Tests/SseIntegrationTests.cs b/tests/mcpdotnet.Tests/SseIntegrationTests.cs index 5ec5f34..2b4e135 100644 --- a/tests/mcpdotnet.Tests/SseIntegrationTests.cs +++ b/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(), + 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(), + 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 + { + ["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()); + 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(), + 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); } diff --git a/tests/mcpdotnet.Tests/Utils/TestSseServer.cs b/tests/mcpdotnet.Tests/Utils/TestSseServer.cs index c419b5b..1bee18e 100644 --- a/tests/mcpdotnet.Tests/Utils/TestSseServer.cs +++ b/tests/mcpdotnet.Tests/Utils/TestSseServer.cs @@ -34,6 +34,11 @@ public sealed class TestSseServer : IAsyncDisposable _messagePath = "/message"; } + /// + /// This is to be able to use the full URL for the endpoint event. + /// + public bool UseFullUrlForEndpointEvent { get; set; } + /// /// Full URL for the SSE endpoint, e.g. "http://localhost:5000/sse" /// @@ -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(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(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(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 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