diff --git a/README.MD b/README.MD index 6c5e2bd..f4c9a9f 100644 --- a/README.MD +++ b/README.MD @@ -1,5 +1,5 @@ # mcpdotnet -[![NuGet version](https://img.shields.io/nuget/v/mcpdotnet.svg?style=flat)](https://www.nuget.org/packages/mcpdotnet/) +[![NuGet version](https://img.shields.io/nuget/v/mcpdotnet.svg)](https://www.nuget.org/packages/mcpdotnet/) A .NET client implementation of the Model Context Protocol (MCP), enabling .NET applications to connect to and interact with MCP servers. @@ -12,23 +12,15 @@ For more information about MCP: - [Protocol Specification](https://spec.modelcontextprotocol.io/) - [GitHub Organization](https://github.com/modelcontextprotocol) -## Status +## Design Goals -This project is in early development. While functional, it may undergo significant changes as the protocol and implementation evolve. - -### Current Limitations -- Only stdio transport is currently supported (HTTP/SSE transport coming soon) -- Client library only - server implementation support may be added in the future -- Provider-agnostic implementation focused purely on the MCP protocol -- Applications using this library need to implement their own integration with LLM APIs/clients - -The library's current focus is on implementing the core MCP protocol, with the main utility being the ability to launch and connect to MCP servers based on configuration objects. +This library aims to provide a clean, specification-compliant implementation of the MCP protocol, with minimal additional abstraction. While transport implementations necessarily include additional code, they follow patterns established by the official SDKs where possible. ## Features - MCP client implementation for .NET applications -- Support for stdio transport -- Support for all capabilities: Tool, Resource, Prompt, Sampling, Roots +- Support for stdio and SSE transports +- Support for all MCP capabilities: Tool, Resource, Prompt, Sampling, Roots - Support for pagination and notifications - Async/await pattern throughout - Comprehensive logging support @@ -36,7 +28,13 @@ The library's current focus is on implementing the core MCP protocol, with the m ## Getting Started -To create a client you need to instantiate a factory with configuration objects: +To use mcpdotnet, first install it via NuGet: + +```powershell +dotnet add package mcpdotnet +``` + +Then create a client and start using tools, or other capabilities, from the servers you configure: ```csharp var options = new McpClientOptions() { ClientInfo = new() { Name = "TestClient", Version = "1.0.0" } }; @@ -58,14 +56,12 @@ var factory = new McpClientFactory( options, NullLoggerFactory.Instance ); -``` -You can then retrieve an IMcpClient reference using the `id` of a server from the configuration, and use `ListToolsAsync` and `CallToolAsync` to integrate with your LLM-based code. - -```csharp var client = await factory.GetClientAsync("everything"); + // Get the list of tools, for passing to an LLM var tools = await client.ListToolsAsync(); + // Execute a tool, in practice this would normally be driven by LLM tool invocations var result = await client.CallToolAsync( "echo", @@ -75,6 +71,7 @@ var result = await client.CallToolAsync( }, CancellationToken.None ); + // echo always returns one and only one text content object Console.WriteLine(result.Content.FirstOrDefault(c => c.Type == "text").Text); ``` @@ -85,6 +82,16 @@ It is also highly recommended that you pass a proper LoggerFactory instance to t Additional examples and documentation will be added as in the near future. +## Roadmap + +- Expand documentation with detailed guides for: + - Advanced scenarios (Sampling, Resources, Prompts) + - Transport configuration + - Error handling and recovery +- Increase test coverage +- Add example applications +- Performance optimization and profiling + ## License This project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details. \ No newline at end of file diff --git a/TODO.md b/TODO.md index 7aae238..2a5b36c 100644 --- a/TODO.md +++ b/TODO.md @@ -1,13 +1,18 @@ # TODOs for mcpdotnet ## Integration Testing -- [ ] Add comprehensive Roots feature testing once reference servers implement support - - (Contribute PR to everything server adding `listRoots` tool) - - Add integration tests similar to Sampling tests - - Verify roots notification handling +- Add comprehensive Roots feature testing once reference servers implement support + - Consider using everything server adding `listRoots` tool) +- Add integration tests similar to Sampling tests +- Verify roots notification handling +- Expand SSE test server to support all features or use a reference SSE server if one becomes available ## Future Improvements -- [ ] Add HTTPS/SSE transport support +- [X] Add HTTPS/SSE transport support +- [ ] Add more example applications showing different capabilities +- [ ] Add comprehensive documentation for advanced scenarios +- [ ] Profile and optimize performance +- [ ] Linux support in stdio transport ## Code Improvements - [ ] Consolidate notification handling in McpClient to reduce duplication between SendNotificationAsync overloads \ No newline at end of file diff --git a/src/mcpdotnet.Tests/IntegrationTests.cs b/src/mcpdotnet.Tests/IntegrationTests.cs index 080f444..854be6e 100644 --- a/src/mcpdotnet.Tests/IntegrationTests.cs +++ b/src/mcpdotnet.Tests/IntegrationTests.cs @@ -2,7 +2,6 @@ using McpDotNet.Configuration; using McpDotNet.Protocol.Messages; using McpDotNet.Protocol.Types; -using Microsoft.Extensions.Logging; namespace McpDotNet.Tests; diff --git a/src/mcpdotnet.Tests/McpClientFactoryTests.cs b/src/mcpdotnet.Tests/McpClientFactoryTests.cs index 5d8dd3d..1a27adb 100644 --- a/src/mcpdotnet.Tests/McpClientFactoryTests.cs +++ b/src/mcpdotnet.Tests/McpClientFactoryTests.cs @@ -1,4 +1,3 @@ -using Castle.Core.Logging; using McpDotNet.Client; using McpDotNet.Configuration; using McpDotNet.Protocol.Messages; @@ -6,7 +5,6 @@ using McpDotNet.Protocol.Transport; using Microsoft.Extensions.Logging.Abstractions; using Moq; using System.Threading.Channels; -using Xunit; namespace McpDotNet.Tests.Client; @@ -192,23 +190,177 @@ public class McpClientFactoryTests } [Fact] - public async Task GetClientAsync_WithHttpTransport_ThrowsNotImplementedException() + public async Task GetClientAsync_WithSseTransport_CanCreateClient() { // Arrange var config = new McpServerConfig { Id = "test-server", Name = "Test Server", - TransportType = "http", + TransportType = "sse", + Location = "http://localhost:8080" + }; + + // Create a mock transport + var mockTransport = new Mock(); + mockTransport.Setup(t => t.ConnectAsync(It.IsAny())) + .Returns(Task.CompletedTask); + mockTransport.Setup(t => t.IsConnected).Returns(true); + mockTransport.Setup(t => t.MessageReader).Returns(Mock.Of>()); + + // Create a mock client + var mockClient = new Mock(); + mockClient.Setup(c => c.ConnectAsync(It.IsAny())) + .Returns(Task.CompletedTask); + mockClient.Setup(c => c.IsInitialized).Returns(true); + + // Inject the mock transport into the factory + var factory = new McpClientFactory( + [config], + _defaultOptions, + NullLoggerFactory.Instance, + transportFactoryMethod: _ => mockTransport.Object, + clientFactoryMethod: (_, _, _) => mockClient.Object + ); + + // Act + var client = await factory.GetClientAsync("test-server"); + + // Assert + Assert.NotNull(client); + } + + [Fact] + public void McpFactory_WithSse_CreatesCorrectTransportOptions() + { + // Arrange + var config = new McpServerConfig + { + Id = "test-server", + Name = "Test Server", + TransportType = "sse", + Location = "http://localhost:8080", + TransportOptions = new Dictionary + { + ["connectionTimeout"] = "10", + ["maxReconnectAttempts"] = "2", + ["reconnectDelay"] = "5", + ["header.test"] = "the_header_value" + } + }; + + // Act + var factory = new McpClientFactory( + [config], + _defaultOptions, + NullLoggerFactory.Instance + ); + + var transport = factory.TransportFactoryMethod(config) as SseTransport; + + // Assert + Assert.NotNull(transport); + Assert.True(transport.Options.ConnectionTimeout == TimeSpan.FromSeconds(10)); + Assert.True(transport.Options.MaxReconnectAttempts == 2); + Assert.True(transport.Options.ReconnectDelay == TimeSpan.FromSeconds(5)); + Assert.True(transport.Options.AdditionalHeaders["test"] == "the_header_value"); + } + + [Fact] + public void McpFactory_WithSseAndNoOptions_CreatesDefaultTransportOptions() + { + // Arrange + var config = new McpServerConfig + { + Id = "test-server", + Name = "Test Server", + TransportType = "sse", Location = "http://localhost:8080" }; - var factory = new McpClientFactory([config], _defaultOptions, - NullLoggerFactory.Instance); + var defaultOptions = new SseTransportOptions(); - // Act & Assert - await Assert.ThrowsAsync( - () => factory.GetClientAsync("test-server") + // Act + var factory = new McpClientFactory( + [config], + _defaultOptions, + NullLoggerFactory.Instance ); + + var transport = factory.TransportFactoryMethod(config) as SseTransport; + + // Assert + Assert.NotNull(transport); + Assert.True(transport.Options.ConnectionTimeout == defaultOptions.ConnectionTimeout); + Assert.True(transport.Options.MaxReconnectAttempts == defaultOptions.MaxReconnectAttempts); + Assert.True(transport.Options.ReconnectDelay == defaultOptions.ReconnectDelay); + Assert.True(transport.Options.AdditionalHeaders == null && defaultOptions.AdditionalHeaders == null); } -} \ No newline at end of file + + [Fact] + public void McpFactory_WithSseAndMissingOptions_CreatesCorrectTransportOptions() + { + // Arrange + var config = new McpServerConfig + { + Id = "test-server", + Name = "Test Server", + TransportType = "sse", + Location = "http://localhost:8080", + TransportOptions = new Dictionary + { + ["connectionTimeout"] = "10", + ["header.test"] = "the_header_value" + } + }; + + var defaultOptions = new SseTransportOptions(); + + // Act + var factory = new McpClientFactory( + [config], + _defaultOptions, + NullLoggerFactory.Instance + ); + + var transport = factory.TransportFactoryMethod(config) as SseTransport; + + // Assert + Assert.NotNull(transport); + Assert.True(transport.Options.ConnectionTimeout == TimeSpan.FromSeconds(10)); + Assert.True(transport.Options.MaxReconnectAttempts == defaultOptions.MaxReconnectAttempts); + Assert.True(transport.Options.ReconnectDelay == defaultOptions.ReconnectDelay); + Assert.True(transport.Options.AdditionalHeaders["test"] == "the_header_value"); + } + + [Theory] + [InlineData("connectionTimeout", "not_a_number")] + [InlineData("maxReconnectAttempts", "invalid")] + [InlineData("reconnectDelay", "bad_value")] + public void McpFactory_WithInvalidTransportOptions_ThrowsFormatException(string key, string value) + { + // arrange + var config = new McpServerConfig + { + Id = "test-server", + Name = "Test Server", + TransportType = "sse", + Location = "http://localhost:8080", + TransportOptions = new Dictionary + { + [key] = value + } + }; + + // Act + var factory = new McpClientFactory( + [config], + _defaultOptions, + NullLoggerFactory.Instance + ); + + // act & assert + Assert.Throws(() => + factory.TransportFactoryMethod(config)); + } +} diff --git a/src/mcpdotnet.Tests/SseIntegrationTests.cs b/src/mcpdotnet.Tests/SseIntegrationTests.cs new file mode 100644 index 0000000..c429884 --- /dev/null +++ b/src/mcpdotnet.Tests/SseIntegrationTests.cs @@ -0,0 +1,164 @@ +using McpDotNet.Client; +using McpDotNet.Configuration; +using McpDotNet.Protocol.Messages; +using McpDotNet.Protocol.Transport; +using McpDotNet.Tests.Utils; +using Microsoft.Extensions.Logging; +using System.Text.Json; + +namespace McpDotNet.Tests; + +public class SseIntegrationTests +{ + [Fact] + public async Task ConnectAndReceiveMessage_InMemoryServer() + { + // Arrange + using var loggerFactory = Microsoft.Extensions.Logging.LoggerFactory.Create(builder => + builder.AddConsole() + .SetMinimumLevel(LogLevel.Debug)); + + await using TestSseServer server = new(logger:loggerFactory.CreateLogger()); + 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 + var testMessage = new JsonRpcRequest() + { + Id = RequestId.FromNumber(1), + Method = "test/message", + Params = new { message = "Hello, SSE!" } + }; + using var httpClient = new HttpClient(); + await httpClient.PostAsync( + "http://localhost:5000/message", + new StringContent(JsonSerializer.Serialize(testMessage)) + ); + + // Assert + Assert.True(true); + } + + [Fact] + public async Task ConnectAndReceiveNotification_InMemoryServer() + { + // Arrange + using var loggerFactory = Microsoft.Extensions.Logging.LoggerFactory.Create(builder => + builder.AddConsole() + .SetMinimumLevel(LogLevel.Debug)); + + await using TestSseServer server = new(logger: loggerFactory.CreateLogger()); + 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)); + + var receivedNotification = new TaskCompletionSource(); + client.OnNotification("test/notification", async (args) => + { + var msg = ((JsonElement)args.Params).GetProperty("message").GetString(); + receivedNotification.SetResult(msg); + }); + + // Act + await server.SendTestNotificationAsync("Hello from server!"); + + // Assert + var message = await receivedNotification.Task.WaitAsync(TimeSpan.FromSeconds(10)); + Assert.Equal("Hello from server!", message); + } + + [Fact] + public async Task ConnectTwice_Throws() + { + // Arrange + using var loggerFactory = Microsoft.Extensions.Logging.LoggerFactory.Create(builder => + builder.AddConsole() + .SetMinimumLevel(LogLevel.Debug)); + + await using TestSseServer server = new(logger: loggerFactory.CreateLogger()); + 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"); + var mcpClient = (McpClient)client; + var transport = (SseTransport)mcpClient.Transport; + + // Wait for SSE connection to be established + await server.WaitForConnectionAsync(TimeSpan.FromSeconds(10)); + + // Assert + await Assert.ThrowsAsync(async () => await transport.ConnectAsync()); + } +} diff --git a/src/mcpdotnet.Tests/SseTransportTests.cs b/src/mcpdotnet.Tests/SseTransportTests.cs new file mode 100644 index 0000000..d95dfc4 --- /dev/null +++ b/src/mcpdotnet.Tests/SseTransportTests.cs @@ -0,0 +1,74 @@ +using McpDotNet.Configuration; +using McpDotNet.Protocol.Messages; +using McpDotNet.Protocol.Transport; +using Microsoft.Extensions.Logging.Abstractions; + +namespace McpDotNet.Tests.Client; + +public class SseTransportTests +{ + [Fact] + public void SseTransportConstructor_WithValidConfig_CreatesTransport() + { + // Arrange + var config = new McpServerConfig + { + Id = "test-server", + Name = "Test Server", + TransportType = "sse", + Location = "http://localhost:8080" + }; + + var transportOptions = new SseTransportOptions + { + ConnectionTimeout = TimeSpan.FromSeconds(10), + MaxReconnectAttempts = 2, + ReconnectDelay = TimeSpan.FromSeconds(5), + AdditionalHeaders = new Dictionary + { + ["test"] = "header" + } + }; + + // Act + var transport = new SseTransport(transportOptions, config, NullLoggerFactory.Instance); + + // Assert + Assert.NotNull(transport); + Assert.True(transport.Options.ConnectionTimeout == TimeSpan.FromSeconds(10)); + Assert.True(transport.Options.MaxReconnectAttempts == 2); + Assert.True(transport.Options.ReconnectDelay == TimeSpan.FromSeconds(5)); + Assert.True(transport.Options.AdditionalHeaders["test"] == "header"); + } + + [Fact] + public async void SseTransportSendMessageAsync_WithMessageEndpointNotSet_ThrowsException() + { + // Arrange + var config = new McpServerConfig + { + Id = "test-server", + Name = "Test Server", + TransportType = "sse", + Location = "http://localhost:8080" + }; + + var transportOptions = new SseTransportOptions + { + ConnectionTimeout = TimeSpan.FromSeconds(10), + MaxReconnectAttempts = 2, + ReconnectDelay = TimeSpan.FromSeconds(5), + AdditionalHeaders = new Dictionary + { + ["test"] = "header" + } + }; + + // Act + var transport = new SseTransport(transportOptions, config, NullLoggerFactory.Instance); + + // Assert + Assert.True(string.IsNullOrEmpty(transport.MessageEndpoint?.ToString())); + await Assert.ThrowsAsync(() => transport.SendMessageAsync(new JsonRpcRequest() { Method = "test"}, CancellationToken.None)); + } +} \ No newline at end of file diff --git a/src/mcpdotnet.Tests/Utils/TestSseServer.cs b/src/mcpdotnet.Tests/Utils/TestSseServer.cs new file mode 100644 index 0000000..c419b5b --- /dev/null +++ b/src/mcpdotnet.Tests/Utils/TestSseServer.cs @@ -0,0 +1,354 @@ +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Logging; +using System.Collections.Concurrent; +using System.Net; +using System.Text.Json; +using McpDotNet.Protocol.Messages; + +namespace McpDotNet.Tests.Utils; + +public sealed class TestSseServer : IAsyncDisposable +{ + private readonly HttpListener _listener; + private readonly CancellationTokenSource _cts; + private readonly ILogger _logger; + private Task? _serverTask; + private readonly TaskCompletionSource _connectionEstablished = new(); + + // SSE endpoint for GET + private readonly string _endpointPath; + // POST endpoint + private readonly string _messagePath; + + // Keep track of all open SSE connections (StreamWriters). + private readonly ConcurrentBag _sseClients = new(); + + public TestSseServer(int port = 5000, ILogger? logger = null) + { + _listener = new HttpListener(); + _listener.Prefixes.Add($"http://localhost:{port}/"); + _cts = new CancellationTokenSource(); + _logger = logger ?? NullLogger.Instance; + + _endpointPath = "/sse"; + _messagePath = "/message"; + } + + /// + /// Full URL for the SSE endpoint, e.g. "http://localhost:5000/sse" + /// + public string SseEndpoint + => $"http://localhost:{_listener.Prefixes.First().Split(':')[2].TrimEnd('/')}{_endpointPath}"; + + /// + /// Full URL for the message endpoint, e.g. "http://localhost:5000/message" + /// + public string MessageEndpoint + => $"http://localhost:{_listener.Prefixes.First().Split(':')[2].TrimEnd('/')}{_messagePath}"; + + /// + /// Starts the server so it can accept incoming connections and POST requests. + /// + public async Task StartAsync() + { + _listener.Start(); + _serverTask = HandleConnectionsAsync(_cts.Token); + + _logger.LogInformation("Test SSE server started on {Endpoint}", SseEndpoint); + await Task.CompletedTask; + } + + private async Task HandleConnectionsAsync(CancellationToken ct) + { + try + { + while (!ct.IsCancellationRequested) + { + var context = await _listener.GetContextAsync(); + _ = Task.Run(() => HandleRequestAsync(context, ct), ct); + } + } + catch (OperationCanceledException) + { + // Ignore, we are shutting down + } + catch (Exception ex) when (!ct.IsCancellationRequested) + { + _logger.LogError(ex, "Error in SSE server connection handling"); + } + } + + private async Task HandleRequestAsync(HttpListenerContext context, CancellationToken ct) + { + var request = context.Request; + var response = context.Response; + + // Handle SSE endpoint + if (request.HttpMethod.Equals("GET", StringComparison.OrdinalIgnoreCase) + && request.Url.AbsolutePath.Equals(_endpointPath, StringComparison.OrdinalIgnoreCase)) + { + await HandleSseConnectionAsync(context, ct); + } + // Handle POST /message + else if (request.HttpMethod.Equals("POST", StringComparison.OrdinalIgnoreCase) + && request.Url.AbsolutePath.Equals(_messagePath, StringComparison.OrdinalIgnoreCase)) + { + await HandlePostMessageAsync(context, ct); + } + else + { + response.StatusCode = 404; + response.Close(); + } + } + + /// + /// Handle Server-Sent Events (SSE) connection. + /// Send the initial event: endpoint with the full POST URL. + /// Keep the connection open until the server is disposed or the client disconnects. + /// + private async Task HandleSseConnectionAsync(HttpListenerContext context, CancellationToken ct) + { + var response = context.Response; + response.ContentType = "text/event-stream"; + response.Headers["Cache-Control"] = "no-cache"; + // Ensures the response is never chunked away by the framework + response.SendChunked = true; + response.StatusCode = (int)HttpStatusCode.OK; + + using var writer = new StreamWriter(response.OutputStream); + _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}"); + await writer.WriteLineAsync(); // blank line to end an SSE message + await writer.FlushAsync(); + + _logger.LogInformation("New SSE client connected."); + _connectionEstablished.TrySetResult(); // Signal connection is ready + + try + { + // Keep the connection open by "pinging" or just waiting + // until the client disconnects or the server is canceled. + while (!ct.IsCancellationRequested && response.OutputStream.CanWrite) + { + _logger.LogDebug("SSE connection alive check"); + // Optionally do a periodic no-op to keep connection alive: + await writer.WriteLineAsync(": keep-alive"); + await writer.FlushAsync(); + await Task.Delay(TimeSpan.FromSeconds(5), ct); + } + } + catch (TaskCanceledException) + { + // This is expected on shutdown + _logger.LogInformation("SSE connection cancelled (expected on shutdown)"); + } + catch (IOException) + { + // Client likely disconnected + _logger.LogInformation("SSE client disconnected"); + } + finally + { + // Remove this writer from bag (we're disposing it anyway) + _sseClients.TryTake(out _); + _logger.LogInformation("SSE client disconnected."); + } + } + + // Add method to wait for connection + public Task WaitForConnectionAsync(TimeSpan timeout) => + _connectionEstablished.Task.WaitAsync(timeout); + + /// + /// Handle POST /message endpoint. + /// Echo the content back to the caller and broadcast it over SSE as well. + /// + private async Task HandlePostMessageAsync(HttpListenerContext context, CancellationToken ct) + { + var request = context.Request; + var response = context.Response; + + try + { + using var reader = new StreamReader(request.InputStream); + string content = await reader.ReadToEndAsync(); + + var jsonRpcRequest = JsonSerializer.Deserialize(content); + + if (jsonRpcRequest != null && !jsonRpcRequest.Method.StartsWith("notification")) + { + if (jsonRpcRequest.Method == "initialize") + { + await HandleInitializationRequest(response, jsonRpcRequest); + return; + } + + // 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; + } + + // Invalid JSON-RPC message - proper JSON-RPC error response + await SendJsonRpcErrorAsync(response, null, -32700, "Parse error"); + } + catch (Exception ex) + { + await SendJsonRpcErrorAsync(response, null, -32603, "Internal error", ex.Message); + response.StatusCode = 500; + } + finally + { + response.Close(); + } + } + + private async Task SendJsonRpcErrorAsync(HttpListenerResponse response, RequestId? id, int code, string message, string? data = null) + { + var errorResponse = new JsonRpcError + { + Id = id ?? RequestId.FromString("error"), + JsonRpc = "2.0", + Error = new JsonRpcErrorDetail + { + Code = code, + Message = message, + Data = data + } + }; + + response.StatusCode = 200; // Always 200 for JSON-RPC + response.ContentType = "application/json"; + await using var writer = new StreamWriter(response.OutputStream); + await writer.WriteAsync(JsonSerializer.Serialize(errorResponse)); + } + + private async Task HandleInitializationRequest(HttpListenerResponse response, JsonRpcRequest jsonRpcRequest) + { + // We don't need to validate the client's initialization request for the test + // Just send back a valid server initialization response + var jsonResponse = new + { + protocolVersion = "2024-11-05", + capabilities = new + { + experimental = (object?)null, + roots = (object?)null, + sampling = (object?)null + } + }; + + var jsonRpcResponse = new JsonRpcResponse() + { + Id = jsonRpcRequest.Id, + Result = new + { + protocolVersion = "2024-11-05", + capabilities = new + { + experimental = (object?)null, + roots = (object?)null, + sampling = (object?)null + }, + serverInfo = new + { + name = "ExampleServer", + version = "1.0.0" + } + } + }; + + // Echo back to the HTTP response + response.StatusCode = 200; + response.ContentType = "application/json"; + + await using var writer = new StreamWriter(response.OutputStream); + await writer.WriteAsync(JsonSerializer.Serialize(jsonRpcResponse)); + } + + /// + /// Broadcast a message to all currently connected SSE clients. + /// + /// The raw string to send + public async Task BroadcastMessageAsync(string message) + { + foreach (var client in _sseClients.ToArray()) // ToArray to avoid mutation issues + { + try + { + // SSE requires "event: " + "data: " + blank line + await client.WriteLineAsync("event: message"); + await client.WriteLineAsync($"data: {message}"); + await client.WriteLineAsync(); + await client.FlushAsync(); + } + catch (IOException) + { + // Client may have disconnected. We let them get cleaned up on next iteration. + } + catch (ObjectDisposedException) + { + // Stream is disposed, ignore. + } + } + } + + public async ValueTask DisposeAsync() + { + _cts.Cancel(); + + if (_serverTask != null) + { + try + { + await _serverTask; + } + catch (TaskCanceledException) + { + // ignore + } + } + + _listener.Close(); + _cts.Dispose(); + + _logger.LogInformation("Test SSE server stopped"); + } + + /// + /// Send a test notification to all connected SSE clients. + /// + /// + /// + public async Task SendTestNotificationAsync(string content) + { + var notification = new JsonRpcNotification + { + JsonRpc = "2.0", + Method = "test/notification", + Params = new { message = content } + }; + + var serialized = JsonSerializer.Serialize(notification); + await BroadcastMessageAsync(serialized); + } +} diff --git a/src/mcpdotnet/Client/McpClient.cs b/src/mcpdotnet/Client/McpClient.cs index 40af3a3..0cd3abf 100644 --- a/src/mcpdotnet/Client/McpClient.cs +++ b/src/mcpdotnet/Client/McpClient.cs @@ -315,6 +315,8 @@ internal class McpClient : IMcpClient ).ConfigureAwait(false); } + internal IMcpTransport Transport => _transport; + private static Dictionary CreateParametersDictionary(string nameParameter, Dictionary? optionalArguments = null) { var parameters = new Dictionary diff --git a/src/mcpdotnet/Client/McpClientFactory.cs b/src/mcpdotnet/Client/McpClientFactory.cs index 2bb2cd6..5d2c77e 100644 --- a/src/mcpdotnet/Client/McpClientFactory.cs +++ b/src/mcpdotnet/Client/McpClientFactory.cs @@ -89,6 +89,8 @@ public class McpClientFactory return client; } + internal Func TransportFactoryMethod => _transportFactoryMethod; + private IMcpTransport CreateTransport(McpServerConfig config) { var options = string.Join(", ", config.TransportOptions?.Select(kv => $"{kv.Key}={kv.Value}") ?? Enumerable.Empty()); @@ -104,10 +106,31 @@ public class McpClientFactory .Where(kv => kv.Key.StartsWith("env:")) .ToDictionary(kv => kv.Key.Substring(4), kv => kv.Value) }, config, _loggerFactory), + "sse" or "http" => new SseTransport( + new SseTransportOptions + { + ConnectionTimeout = TimeSpan.FromSeconds(ParseOrDefault(config.TransportOptions, "connectionTimeout", 30)), + MaxReconnectAttempts = ParseOrDefault(config.TransportOptions, "maxReconnectAttempts", 3), + ReconnectDelay = TimeSpan.FromSeconds(ParseOrDefault(config.TransportOptions, "reconnectDelay", 5)), + AdditionalHeaders = config.TransportOptions? + .Where(kv => kv.Key.StartsWith("header.")) + .ToDictionary(kv => kv.Key.Substring(7), kv => kv.Value) + }, config, _loggerFactory), _ => throw new ArgumentException($"Unsupported transport type '{config.TransportType}'.", nameof(config)) }; } + private static int ParseOrDefault(IDictionary? options, string key, int defaultValue) + { + if (options?.TryGetValue(key, out var value) ?? false) + { + if (!int.TryParse(value, out var result)) + throw new FormatException($"Invalid value '{value}' for option '{key}'"); + return result; + } + return defaultValue; + } + private string GetCommand(McpServerConfig config) { if (config.TransportOptions == null || diff --git a/src/mcpdotnet/Logging/Log.cs b/src/mcpdotnet/Logging/Log.cs index e307199..9b0e01b 100644 --- a/src/mcpdotnet/Logging/Log.cs +++ b/src/mcpdotnet/Logging/Log.cs @@ -277,4 +277,49 @@ internal static partial class Log [LoggerMessage(Level = LogLevel.Information, Message = "Sending notification for {serverId} ({serverName}): {method}")] internal static partial void SendingNotification(this ILogger logger, string serverId, string serverName, string method); + + [LoggerMessage( + EventId = 7000, + Level = LogLevel.Error, + Message = "Transport connection error for server {serverId} ({serverName})" +)] + public static partial void TransportConnectionError( + this ILogger logger, + string serverId, + string serverName, + Exception exception); + + [LoggerMessage( + EventId = 7001, + Level = LogLevel.Warning, + Message = "Transport message received before connected for server {serverId} ({serverName}): {data}" + )] + public static partial void TransportMessageReceivedBeforeConnected( + this ILogger logger, + string serverId, + string serverName, + string data); + + [LoggerMessage( + EventId = 7002, + Level = LogLevel.Error, + Message = "Transport endpoint event received out of order for server {serverId} ({serverName}): {data}" + )] + public static partial void TransportEndpointEventInvalid( + this ILogger logger, + string serverId, + string serverName, + string data); + + [LoggerMessage( + EventId = 7003, + Level = LogLevel.Error, + Message = "Transport event parse failed for server {serverId} ({serverName}): {data}" + )] + public static partial void TransportEndpointEventParseFailed( + this ILogger logger, + string serverId, + string serverName, + string data, + Exception exception); } \ No newline at end of file diff --git a/src/mcpdotnet/Protocol/Messages/JsonRpcError.cs b/src/mcpdotnet/Protocol/Messages/JsonRpcError.cs index 1ad3414..6b1cf6b 100644 --- a/src/mcpdotnet/Protocol/Messages/JsonRpcError.cs +++ b/src/mcpdotnet/Protocol/Messages/JsonRpcError.cs @@ -5,7 +5,7 @@ using System.Text.Json.Serialization; /// /// An error response message in the JSON-RPC protocol. /// -internal record JsonRpcError : IJsonRpcMessageWithId +public record JsonRpcError : IJsonRpcMessageWithId { /// /// JSON-RPC protocol version. Always "2.0". diff --git a/src/mcpdotnet/Protocol/Messages/JsonRpcErrorDetail.cs b/src/mcpdotnet/Protocol/Messages/JsonRpcErrorDetail.cs index 75f7fae..572797d 100644 --- a/src/mcpdotnet/Protocol/Messages/JsonRpcErrorDetail.cs +++ b/src/mcpdotnet/Protocol/Messages/JsonRpcErrorDetail.cs @@ -5,7 +5,7 @@ using System.Text.Json.Serialization; /// /// Detailed error information for JSON-RPC error responses. /// -internal record JsonRpcErrorDetail +public record JsonRpcErrorDetail { /// /// Integer error code. diff --git a/src/mcpdotnet/Protocol/Messages/JsonRpcResponse.cs b/src/mcpdotnet/Protocol/Messages/JsonRpcResponse.cs index eca1ca7..42f7d95 100644 --- a/src/mcpdotnet/Protocol/Messages/JsonRpcResponse.cs +++ b/src/mcpdotnet/Protocol/Messages/JsonRpcResponse.cs @@ -5,7 +5,7 @@ namespace McpDotNet.Protocol.Messages; /// /// A successful response message in the JSON-RPC protocol. /// -internal record JsonRpcResponse : IJsonRpcMessageWithId +public record JsonRpcResponse : IJsonRpcMessageWithId { /// /// JSON-RPC protocol version. Always "2.0". diff --git a/src/mcpdotnet/Protocol/Transport/IMcpTransport.cs b/src/mcpdotnet/Protocol/Transport/IMcpTransport.cs index 140c61b..befea7a 100644 --- a/src/mcpdotnet/Protocol/Transport/IMcpTransport.cs +++ b/src/mcpdotnet/Protocol/Transport/IMcpTransport.cs @@ -1,9 +1,7 @@ -// Protocol/Transport/IMcpTransport.cs +using System.Threading.Channels; +using McpDotNet.Protocol.Messages; + namespace McpDotNet.Protocol.Transport; - -using System.Threading.Channels; -using global::McpDotNet.Protocol.Messages; - /// /// Represents a transport mechanism for MCP communication. /// diff --git a/src/mcpdotnet/Protocol/Transport/SseTransport.cs b/src/mcpdotnet/Protocol/Transport/SseTransport.cs new file mode 100644 index 0000000..4aabf45 --- /dev/null +++ b/src/mcpdotnet/Protocol/Transport/SseTransport.cs @@ -0,0 +1,303 @@ + +using System.Diagnostics.Tracing; +using System.Text.Json; +using System.Text; +using McpDotNet.Protocol.Messages; +using McpDotNet.Configuration; +using Microsoft.Extensions.Logging; +using McpDotNet.Logging; +using System.Net.Http.Headers; +using System.Text.Json.Serialization; +using McpDotNet.Utils.Json; +using System.Threading; + +namespace McpDotNet.Protocol.Transport; + +/// +/// +/// +public sealed class SseTransport : TransportBase +{ + private readonly HttpClient _httpClient; + private readonly SseTransportOptions _options; + private readonly Uri _sseEndpoint; + private Uri? _messageEndpoint; + private readonly CancellationTokenSource _connectionCts; + private Task? _receiveTask; + private readonly ILogger _logger; + private readonly McpServerConfig _serverConfig; + private readonly JsonSerializerOptions _jsonOptions; + private TaskCompletionSource _connectionEstablished; + + /// + /// + /// + /// + /// + /// + public SseTransport(SseTransportOptions transportOptions, McpServerConfig serverConfig, ILoggerFactory loggerFactory) + { + _options = transportOptions; + _serverConfig = serverConfig; + _sseEndpoint = new Uri(serverConfig.Location!); + _httpClient = new HttpClient(); + _connectionCts = new CancellationTokenSource(); + _logger = loggerFactory.CreateLogger(); + _jsonOptions = new JsonSerializerOptions().ConfigureForMcp(loggerFactory); + _connectionEstablished = new TaskCompletionSource(); + } + + /// + public override async Task ConnectAsync(CancellationToken cancellationToken = default) + { + try + { + if (IsConnected) + { + _logger.TransportAlreadyConnected(_serverConfig.Id, _serverConfig.Name); + throw new McpTransportException("Transport is already connected"); + } + + // Start message receiving loop + _receiveTask = ReceiveMessagesAsync(_connectionCts.Token); + + _logger.TransportReadingMessages(_serverConfig.Id, _serverConfig.Name); + + await Task.WhenAny( + _connectionEstablished.Task, + Task.Delay(_options.ConnectionTimeout, cancellationToken) + ); + + if (!IsConnected) + throw new TimeoutException("Failed to receive endpoint event"); + } + catch (McpTransportException) + { + // Rethrow transport exceptions + throw; + } + catch (Exception ex) + { + _logger.TransportConnectFailed(_serverConfig.Id, _serverConfig.Name, ex); + await CloseAsync(); + throw new McpTransportException("Failed to connect transport", ex); + } + } + + public override async Task SendMessageAsync( + IJsonRpcMessage message, + CancellationToken cancellationToken = default) + { + if (_messageEndpoint == null) + throw new InvalidOperationException("Transport not connected"); + + var content = new StringContent( + JsonSerializer.Serialize(message, _jsonOptions), + Encoding.UTF8, + "application/json" + ); + + var response = await _httpClient.PostAsync( + _messageEndpoint, + content, + cancellationToken + ); + + response.EnsureSuccessStatusCode(); + + + var responseContent = await response.Content.ReadAsStringAsync(); + + // Handle notifications, which don't have responses + if (message is JsonRpcNotification) + { + await HandleNotificationResponseAsync(message, responseContent, cancellationToken); + return; + } + + var responseMessage = JsonSerializer.Deserialize(responseContent, _jsonOptions); + if (responseMessage != null) + { + string messageId = "(no id)"; + if (responseMessage is IJsonRpcMessageWithId messageWithId) + { + messageId = messageWithId.Id.ToString(); + } + _logger.TransportReceivedMessageParsed(_serverConfig.Id, _serverConfig.Name, messageId); + await WriteMessageAsync(responseMessage, cancellationToken); + _logger.TransportMessageWritten(_serverConfig.Id, _serverConfig.Name, messageId); + } + else + { + _logger.TransportMessageParseUnexpectedType(_serverConfig.Id, _serverConfig.Name, responseContent); + } + } + + /// + public async Task CloseAsync() + { + _connectionCts.Cancel(); + if (_receiveTask != null) + await _receiveTask; + + _httpClient.Dispose(); + _connectionCts.Dispose(); + SetConnected(false); + } + + /// + public override async ValueTask DisposeAsync() + { + await CloseAsync(); + 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 SseTransportOptions Options => _options; + + private async Task ReceiveMessagesAsync(CancellationToken cancellationToken) + { + int reconnectAttempts = 0; + + while (!cancellationToken.IsCancellationRequested) + { + try + { + var request = new HttpRequestMessage(HttpMethod.Get, _sseEndpoint); + request.Headers.Accept.Add(new MediaTypeWithQualityHeaderValue("text/event-stream")); + + using var response = await _httpClient.SendAsync( + request, + HttpCompletionOption.ResponseHeadersRead, + cancellationToken + ); + + response.EnsureSuccessStatusCode(); + + using var stream = await response.Content.ReadAsStreamAsync(cancellationToken); + using var reader = new StreamReader(stream); + + // Reset reconnect attempts on successful connection + reconnectAttempts = 0; + + string? currentEvent = null; + string? line; + while ((line = await reader.ReadLineAsync()) != null) + { + if (line.StartsWith("event: ")) + { + currentEvent = line.Substring(7); + } + else if (line.StartsWith("data: ")) + { + var data = line.Substring(6); + if (currentEvent == "endpoint") + { + HandleEndpointEvent(data); + } + else + { + await ProcessSseMessage(data, cancellationToken); + } + } + } + } + catch (Exception ex) when (!cancellationToken.IsCancellationRequested) + { + _logger.TransportConnectionError(_serverConfig.Id, _serverConfig.Name, ex); + + reconnectAttempts++; + if (reconnectAttempts >= _options.MaxReconnectAttempts) + { + throw new McpTransportException("Exceeded reconnect limit", ex); + } + + await Task.Delay(_options.ReconnectDelay, cancellationToken); + } + } + } + + private async Task ProcessSseMessage(string data, CancellationToken cancellationToken) + { + if (!IsConnected) + { + _logger.TransportMessageReceivedBeforeConnected(_serverConfig.Id, _serverConfig.Name, data); + return; + } + + try + { + var message = JsonSerializer.Deserialize(data, _jsonOptions); + if (message == null) + { + _logger.TransportMessageParseUnexpectedType(_serverConfig.Id, _serverConfig.Name, data); + return; + } + + string messageId = "(no id)"; + if (message is IJsonRpcMessageWithId messageWithId) + { + messageId = messageWithId.Id.ToString(); + } + + _logger.TransportReceivedMessageParsed(_serverConfig.Id, _serverConfig.Name, messageId); + await WriteMessageAsync(message, cancellationToken); + _logger.TransportMessageWritten(_serverConfig.Id, _serverConfig.Name, messageId); + } + catch (JsonException ex) + { + _logger.TransportMessageParseFailed(_serverConfig.Id, _serverConfig.Name, data, ex); + } + } + + private void HandleEndpointEvent(string data) + { + try + { + var endpointData = JsonSerializer.Deserialize(data, _jsonOptions); + if (endpointData?.Uri == null) + { + _logger.TransportEndpointEventInvalid(_serverConfig.Id, _serverConfig.Name, data); + return; + } + + _messageEndpoint = new Uri(endpointData.Uri); + SetConnected(true); + _connectionEstablished.TrySetResult(true); + } + catch (JsonException ex) + { + _logger.TransportEndpointEventParseFailed(_serverConfig.Id, _serverConfig.Name, data, ex); + throw new McpTransportException("Failed to parse endpoint event", ex); + } + } + + private record EndpointEventData + { + [JsonPropertyName("uri")] + public string? Uri { get; init; } + } +} \ No newline at end of file diff --git a/src/mcpdotnet/Protocol/Transport/SseTransportOptions.cs b/src/mcpdotnet/Protocol/Transport/SseTransportOptions.cs new file mode 100644 index 0000000..a9810cd --- /dev/null +++ b/src/mcpdotnet/Protocol/Transport/SseTransportOptions.cs @@ -0,0 +1,27 @@ +namespace McpDotNet.Protocol.Transport; + +/// +/// Options for configuring the SSE transport. +/// +public record SseTransportOptions +{ + /// + /// Timeout for initial connection and endpoint event. + /// + public TimeSpan ConnectionTimeout { get; init; } = TimeSpan.FromSeconds(30); + + /// + /// Number of reconnection attempts for SSE connection. + /// + public int MaxReconnectAttempts { get; init; } = 3; + + /// + /// Delay between reconnection attempts. + /// + public TimeSpan ReconnectDelay { get; init; } = TimeSpan.FromSeconds(5); + + /// + /// Headers to include in HTTP requests. + /// + public Dictionary? AdditionalHeaders { get; init; } +} \ No newline at end of file diff --git a/src/mcpdotnet/mcpdotnet.csproj b/src/mcpdotnet/mcpdotnet.csproj index 287459a..1007582 100644 --- a/src/mcpdotnet/mcpdotnet.csproj +++ b/src/mcpdotnet/mcpdotnet.csproj @@ -9,7 +9,7 @@ mcpdotnet - 0.5.0.1 + 0.6.0.1 PederHP .NET client library for the Model Context Protocol (MCP) https://github.com/PederHP/mcpdotnet @@ -29,6 +29,10 @@ + + + +