HttpListener based SSE Server Support (#73)

This commit is contained in:
Peder Holdgaard Pedersen 2025-03-17 22:56:03 +01:00 • committed by GitHub
commit b5a26e4c33
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 1197 additions and 11 deletions

View file

@ -20,4 +20,18 @@ public class McpServerBuilderExtensionsTransportsTests
Assert.NotNull(transportType);
Assert.Equal(typeof(StdioServerTransport), transportType.ImplementationType);
}
[Fact]
public void WithHttpListenerSseServerTransport_Sets_Transport()
{
var services = new ServiceCollection();
var builder = new Mock<IMcpServerBuilder>();
builder.SetupGet(b => b.Services).Returns(services);
builder.Object.WithHttpListenerSseServerTransport();
var transportType = services.FirstOrDefault(s => s.ServiceType == typeof(IServerTransport));
Assert.NotNull(transportType);
Assert.Equal(typeof(HttpListenerSseServerTransport), transportType.ImplementationType);
}
}

View file

@ -18,7 +18,7 @@ public class SseIntegrationTests
builder.AddConsole()
.SetMinimumLevel(LogLevel.Debug));
await using TestSseServer server = new(logger: loggerFactory.CreateLogger<TestSseServer>());
await using InMemoryTestSseServer server = new(logger: loggerFactory.CreateLogger<InMemoryTestSseServer>());
await server.StartAsync();
@ -164,7 +164,7 @@ public class SseIntegrationTests
builder.AddConsole()
.SetMinimumLevel(LogLevel.Debug));
await using TestSseServer server = new(logger: loggerFactory.CreateLogger<TestSseServer>());
await using InMemoryTestSseServer server = new(logger: loggerFactory.CreateLogger<InMemoryTestSseServer>());
server.UseFullUrlForEndpointEvent = true;
await server.StartAsync();
@ -204,7 +204,7 @@ public class SseIntegrationTests
builder.AddConsole()
.SetMinimumLevel(LogLevel.Debug));
await using TestSseServer server = new(logger: loggerFactory.CreateLogger<TestSseServer>());
await using InMemoryTestSseServer server = new(logger: loggerFactory.CreateLogger<InMemoryTestSseServer>());
await server.StartAsync();
@ -253,7 +253,7 @@ public class SseIntegrationTests
builder.AddConsole()
.SetMinimumLevel(LogLevel.Debug));
await using TestSseServer server = new(logger: loggerFactory.CreateLogger<TestSseServer>());
await using InMemoryTestSseServer server = new(logger: loggerFactory.CreateLogger<InMemoryTestSseServer>());
await server.StartAsync();

View file

@ -0,0 +1,83 @@
using McpDotNet.Client;
using McpDotNet.Configuration;
using McpDotNet.Protocol.Transport;
using Microsoft.Extensions.Logging;
using System.Diagnostics;
namespace McpDotNet.Tests;
public class SseServerIntegrationTestFixture : IDisposable
{
private Process _process;
public ILoggerFactory LoggerFactory { get; }
public McpClientFactory Factory { get; }
public McpClientOptions DefaultOptions { get; }
public McpServerConfig DefaultConfig { get; }
public SseServerIntegrationTestFixture()
{
LoggerFactory = Microsoft.Extensions.Logging.LoggerFactory.Create(builder =>
builder.AddConsole()
.SetMinimumLevel(LogLevel.Debug));
DefaultOptions = new()
{
ClientInfo = new() { Name = "IntegrationTestClient", Version = "1.0.0" },
Capabilities = new() { Sampling = new(), Roots = new() }
};
DefaultConfig = new McpServerConfig
{
Id = "test_server",
Name = "TestServer",
TransportType = TransportTypes.Sse,
TransportOptions = [],
Location = "http://localhost:3001/sse"
};
// Inject the mock transport into the factory
Factory = new McpClientFactory(
[DefaultConfig],
DefaultOptions,
LoggerFactory
);
Start();
}
public void Start()
{
// Start the server (which is at TestSseServer.exe on windows and "dotnet TestSseServer.dll" on linux)
var processStartInfo = new ProcessStartInfo
{
FileName = OperatingSystem.IsWindows() ? "TestSseServer.exe" : "dotnet",
Arguments = "TestSseServer.dll",
RedirectStandardInput = true,
RedirectStandardOutput = true,
RedirectStandardError = true,
UseShellExecute = false,
};
_process = Process.Start(processStartInfo)
?? throw new InvalidOperationException($"Could not start process for {processStartInfo.FileName} with '{processStartInfo.Arguments}'.");
// Wait 1 second
Thread.Sleep(1000);
}
public void Dispose()
{
try
{
var client = Factory.GetClientAsync("test_server").Result;
client.DisposeAsync().AsTask().Wait();
LoggerFactory?.Dispose();
}
finally
{
// Kill the server process
_process.Kill();
}
}
}

View file

@ -0,0 +1,235 @@
using McpDotNet.Client;
using McpDotNet.Protocol.Types;
namespace McpDotNet.Tests;
public class SseServerIntegrationTests : IClassFixture<SseServerIntegrationTestFixture>
{
private readonly SseServerIntegrationTestFixture _fixture;
public SseServerIntegrationTests(SseServerIntegrationTestFixture fixture)
{
_fixture = fixture;
}
[Fact]
public async Task ConnectAndPing_Sse_TestServer()
{
// Arrange
// Act
var client = await _fixture.Factory.GetClientAsync("test_server");
await client.PingAsync(CancellationToken.None);
// Assert
Assert.NotNull(client);
}
[Fact]
public async Task Connect_TestServer_ShouldProvideServerFields()
{
// Arrange
// Act
var client = await _fixture.Factory.GetClientAsync("test_server");
// Assert
Assert.NotNull(client.ServerCapabilities);
Assert.NotNull(client.ServerInfo);
}
[Fact]
public async Task ListTools_Sse_TestServer()
{
// arrange
// act
var client = await _fixture.Factory.GetClientAsync("test_server");
var tools = await client.ListToolsAsync().ToListAsync();
// assert
Assert.NotNull(tools);
}
[Fact]
public async Task CallTool_Sse_EchoServer()
{
// arrange
// act
var client = await _fixture.Factory.GetClientAsync("test_server");
var result = await client.CallToolAsync(
"echo",
new Dictionary<string, object>
{
["message"] = "Hello MCP!"
},
CancellationToken.None
);
// assert
Assert.NotNull(result);
Assert.False(result.IsError);
var textContent = Assert.Single(result.Content, c => c.Type == "text");
Assert.Equal("Echo: Hello MCP!", textContent.Text);
}
[Fact]
public async Task ListResources_Sse_TestServer()
{
// arrange
// act
var client = await _fixture.Factory.GetClientAsync("test_server");
List<Resource> allResources = [];
string? cursor = null;
do
{
var resources = await client.ListResourcesAsync(cursor, CancellationToken.None);
allResources.AddRange(resources.Resources);
cursor = resources.NextCursor;
}
while (cursor != null);
// The everything server provides 100 test resources
Assert.Equal(100, allResources.Count);
}
[Fact]
public async Task ReadResource_Sse_TextResource()
{
// arrange
// act
var client = await _fixture.Factory.GetClientAsync("test_server");
// Odd numbered resources are text in the everything server (despite the docs saying otherwise)
// 1 is index 0, which is "even" in the 0-based index
// We copied this oddity to the test server
var result = await client.ReadResourceAsync("test://static/resource/1", CancellationToken.None);
Assert.NotNull(result);
Assert.Single(result.Contents);
Assert.NotNull(result.Contents[0].Text);
}
[Fact]
public async Task ReadResource_Sse_BinaryResource()
{
// arrange
// act
var client = await _fixture.Factory.GetClientAsync("test_server");
// Even numbered resources are binary in the everything server (despite the docs saying otherwise)
// 2 is index 1, which is "odd" in the 0-based index
// We copied this oddity to the test server
var result = await client.ReadResourceAsync("test://static/resource/2", CancellationToken.None);
Assert.NotNull(result);
Assert.Single(result.Contents);
Assert.NotNull(result.Contents[0].Blob);
}
[Fact]
public async Task ListPrompts_Sse_TestServer()
{
// arrange
// act
var client = await _fixture.Factory.GetClientAsync("test_server");
var prompts = await client.ListPromptsAsync().ToListAsync();
// assert
Assert.NotNull(prompts);
Assert.NotEmpty(prompts);
// We could add specific assertions for the known prompts
Assert.Contains(prompts, p => p.Name == "simple_prompt");
Assert.Contains(prompts, p => p.Name == "complex_prompt");
}
[Fact]
public async Task GetPrompt_Sse_SimplePrompt()
{
// arrange
// act
var client = await _fixture.Factory.GetClientAsync("test_server");
var result = await client.GetPromptAsync("simple_prompt", null, CancellationToken.None);
// assert
Assert.NotNull(result);
Assert.NotEmpty(result.Messages);
}
[Fact]
public async Task GetPrompt_Sse_ComplexPrompt()
{
// arrange
// act
var client = await _fixture.Factory.GetClientAsync("test_server");
var arguments = new Dictionary<string, object>
{
{ "temperature", "0.7" },
{ "style", "formal" }
};
var result = await client.GetPromptAsync("complex_prompt", arguments, CancellationToken.None);
// assert
Assert.NotNull(result);
Assert.NotEmpty(result.Messages);
}
[Fact]
public async Task GetPrompt_Sse_NonExistent_ThrowsException()
{
// arrange
// act
var client = await _fixture.Factory.GetClientAsync("test_server");
await Assert.ThrowsAsync<McpClientException>(() =>
client.GetPromptAsync("non_existent_prompt", null, CancellationToken.None));
}
[Fact]
public async Task Sampling_Sse_TestServer()
{
// arrange
var client = await _fixture.Factory.GetClientAsync("test_server");
// Set up the sampling handler
int samplingHandlerCalls = 0;
#pragma warning disable CS1998 // Async method lacks 'await' operators and will run synchronously
client.SetSamplingHandler(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));
}
}

View file

@ -0,0 +1,382 @@
using System.Collections.Concurrent;
using System.Net;
using System.Text.Json;
using McpDotNet.Protocol.Messages;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
namespace McpDotNet.Tests.Utils;
public sealed class InMemoryTestSseServer : IAsyncDisposable
{
private readonly HttpListener _listener;
private readonly CancellationTokenSource _cts;
private readonly ILogger<InMemoryTestSseServer> _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<StreamWriter> _sseClients = [];
public InMemoryTestSseServer(int port = 5000, ILogger<InMemoryTestSseServer>? logger = null)
{
_listener = new HttpListener();
_listener.Prefixes.Add($"http://localhost:{port}/");
_cts = new CancellationTokenSource();
_logger = logger ?? NullLogger<InMemoryTestSseServer>.Instance;
_endpointPath = "/sse";
_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>
public string SseEndpoint
=> $"http://localhost:{_listener.Prefixes.First().Split(':')[2].TrimEnd('/')}{_endpointPath}";
/// <summary>
/// Full URL for the message endpoint, e.g. "http://localhost:5000/message"
/// </summary>
public string MessageEndpoint
=> $"http://localhost:{_listener.Prefixes.First().Split(':')[2].TrimEnd('/')}{_messagePath}";
/// <summary>
/// Starts the server so it can accept incoming connections and POST requests.
/// </summary>
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) == true)
{
await HandleSseConnectionAsync(context, ct);
}
// Handle POST /message
else if (request.HttpMethod.Equals("POST", StringComparison.OrdinalIgnoreCase)
&& request.Url?.AbsolutePath.Equals(_messagePath, StringComparison.OrdinalIgnoreCase) == true)
{
await HandlePostMessageAsync(context, ct);
}
else
{
response.StatusCode = 404;
response.Close();
}
}
/// <summary>
/// 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.
/// </summary>
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
await writer.WriteLineAsync("event: endpoint");
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(ct);
_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(ct);
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);
/// <summary>
/// Handle POST /message endpoint.
/// Echo the content back to the caller and broadcast it over SSE as well.
/// </summary>
private async Task HandlePostMessageAsync(HttpListenerContext context, CancellationToken cancellationToken)
{
var request = context.Request;
var response = context.Response;
try
{
using var reader = new StreamReader(request.InputStream);
string content = await reader.ReadToEndAsync(cancellationToken);
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(cancellationToken);
return;
}
var jsonRpcRequest = JsonSerializer.Deserialize<JsonRpcRequest>(content);
if (jsonRpcRequest != null)
{
if (jsonRpcRequest.Method == "initialize")
{
await HandleInitializationRequest(response, jsonRpcRequest);
}
else
{
// 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(cancellationToken);
// 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)
{
await SendJsonRpcErrorAsync(response, null, -32603, "Internal error", ex.Message);
response.StatusCode = 500;
}
finally
{
response.Close();
}
}
private static Task<JsonRpcResponse> ProcessRequestAsync(JsonRpcRequest jsonRpcRequest)
{
// This is a test server so we just echo back the request
return Task.FromResult(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 static 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 static 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 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));
}
/// <summary>
/// Broadcast a message to all currently connected SSE clients.
/// </summary>
/// <param name="message">The raw string to send</param>
public async Task BroadcastMessageAsync(string message)
{
foreach (var client in _sseClients.ToArray()) // ToArray to avoid mutation issues
{
try
{
// SSE requires "event: <name>" + "data: <payload>" + 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()
{
await _cts.CancelAsync();
if (_serverTask != null)
{
try
{
await Task.WhenAny(_serverTask, Task.Delay(2000));
}
catch (TaskCanceledException)
{
// ignore
}
}
_listener.Close();
_cts.Dispose();
_logger.LogInformation("Test SSE server stopped");
}
/// <summary>
/// Send a test notification to all connected SSE clients.
/// </summary>
/// <param name="content"></param>
/// <returns></returns>
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);
}
}

View file

@ -32,6 +32,7 @@
<ItemGroup>
<ProjectReference Include="..\..\src\mcpdotnet\mcpdotnet.csproj" />
<ProjectReference Include="..\mcpdotnet.TestServer\mcpdotnet.TestServer.csproj" />
<ProjectReference Include="..\mcpdotnet.TestSseServer\mcpdotnet.TestSseServer.csproj" />
</ItemGroup>
<ItemGroup>
@ -41,6 +42,12 @@
<Content Condition="$([MSBuild]::IsOSPlatform('Linux'))" Include="..\mcpdotnet.TestServer\bin\$(Configuration)\net8.0\TestServer.dll">
<CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
</Content>
<Content Condition="$([MSBuild]::IsOSPlatform('Windows'))" Include="..\mcpdotnet.TestSseServer\bin\$(Configuration)\net8.0\TestSseServer.exe">
<CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
</Content>
<Content Condition="$([MSBuild]::IsOSPlatform('Linux'))" Include="..\mcpdotnet.TestSseServer\bin\$(Configuration)\net8.0\TestSseServer.dll">
<CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
</Content>
</ItemGroup>
</Project>