Skip to content

Commit 5c6b17d

Browse files
halter73halter73
committed
Move the SseResponseStreamTransport out of sample and into the main project
1 parent c8c974a commit 5c6b17d

File tree

5 files changed

+43
-18
lines changed

5 files changed

+43
-18
lines changed

‎Directory.Packages.props‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,7 @@
1313
<PackageVersion Include="Microsoft.Extensions.AI.Abstractions" Version="$(MicrosoftExtensionsAIVersion)" />
1414
<PackageVersion Include="Microsoft.Extensions.Hosting.Abstractions" Version="$(MicrosoftExtensionsVersion)" />
1515
<PackageVersion Include="Microsoft.Extensions.Logging.Abstractions" Version="$(MicrosoftExtensionsVersion)" />
16-
<PackageVersion Include="System.Net.ServerSentEvents" Version="$(SystemVersion)" />
16+
<PackageVersion Include="System.Net.ServerSentEvents" Version="$(System10Version)" />
1717
<PackageVersion Include="System.Text.Json" Version="$(SystemVersion)" />
1818
<PackageVersion Include="System.Threading.Channels" Version="$(SystemVersion)" />
1919

‎samples/AspNetCoreSseServer/AspNetCoreSseServer.csproj‎

Lines changed: 0 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -10,8 +10,4 @@
1010
<ProjectReference Include="..\..\src\ModelContextProtocol\ModelContextProtocol.csproj" />
1111
</ItemGroup>
1212

13-
<ItemGroup>
14-
<PackageReference Include="System.Net.ServerSentEvents" VersionOverride="10.0.0-preview.1.25080.5" />
15-
</ItemGroup>
16-
1713
</Project>

‎samples/AspNetCoreSseServer/McpEndpointRouteBuilderExtensions.cs‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@
22
using ModelContextProtocol.Server;
33
using ModelContextProtocol.Utils.Json;
44
using Microsoft.Extensions.Options;
5+
using ModelContextProtocol.Protocol.Transport;
56

67
namespace AspNetCoreSseServer;
78

@@ -10,15 +11,15 @@ public static class McpEndpointRouteBuilderExtensions
1011
public static IEndpointConventionBuilder MapMcpSse(this IEndpointRouteBuilder endpoints)
1112
{
1213
IMcpServer? server = null;
13-
SseServerStreamTransport? transport = null;
14+
SseResponseStreamTransport? transport = null;
1415
var loggerFactory = endpoints.ServiceProvider.GetRequiredService<ILoggerFactory>();
1516
var mcpServerOptions = endpoints.ServiceProvider.GetRequiredService<IOptions<McpServerOptions>>();
1617

1718
var routeGroup = endpoints.MapGroup("");
1819

1920
routeGroup.MapGet("/sse", async (HttpResponse response, CancellationToken requestAborted) =>
2021
{
21-
await using var localTransport = transport = new SseServerStreamTransport(response.Body);
22+
await using var localTransport = transport = new SseResponseStreamTransport(response.Body);
2223
await using var localServer = server = McpServerFactory.Create(transport, mcpServerOptions.Value, loggerFactory, endpoints.ServiceProvider);
2324

2425
await localServer.StartAsync(requestAborted);
@@ -37,7 +38,7 @@ public static IEndpointConventionBuilder MapMcpSse(this IEndpointRouteBuilder en
3738
}
3839
});
3940

40-
routeGroup.MapPost("/message", async (HttpContext context) =>
41+
routeGroup.MapPost("/message", async context =>
4142
{
4243
if (transport is null)
4344
{

‎src/ModelContextProtocol/Protocol/Transport/HttpListenerSseServerTransport.cs‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
1-
using System.Text.Json;
1+
using System.Net;
2+
using System.Text.Json;
23
using Microsoft.Extensions.Logging;
34
using ModelContextProtocol.Protocol.Messages;
45
using ModelContextProtocol.Utils.Json;
@@ -9,7 +10,7 @@
910
namespace ModelContextProtocol.Protocol.Transport;
1011

1112
/// <summary>
12-
/// Implements the MCP transport protocol over standard input/output streams.
13+
/// Implements the MCP transport protocol using <see cref="HttpListener"/>.
1314
/// </summary>
1415
public sealed class HttpListenerSseServerTransport : TransportBase, IServerTransport
1516
{

‎samples/AspNetCoreSseServer/SseServerStreamTransport.cs‎ renamed to ‎src/ModelContextProtocol/Protocol/Transport/SseResponseStreamTransport.cs‎

Lines changed: 35 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -3,21 +3,31 @@
33
using System.Text.Json;
44
using System.Threading.Channels;
55
using ModelContextProtocol.Protocol.Messages;
6-
using ModelContextProtocol.Protocol.Transport;
76
using ModelContextProtocol.Utils.Json;
87

9-
namespace AspNetCoreSseServer;
8+
namespace ModelContextProtocol.Protocol.Transport;
109

11-
public class SseServerStreamTransport(Stream sseResponseStream) : ITransport
10+
/// <summary>
11+
/// Implements the MCP SSE server transport protocol using the SSE response <see cref="Stream"/>.
12+
/// </summary>
13+
/// <param name="sseResponseStream">The stream to write the SSE response body to.</param>
14+
public sealed class SseResponseStreamTransport(Stream sseResponseStream) : ITransport
1215
{
1316
private readonly Channel<IJsonRpcMessage> _incomingChannel = CreateSingleItemChannel<IJsonRpcMessage>();
1417
private readonly Channel<SseItem<IJsonRpcMessage?>> _outgoingSseChannel = CreateSingleItemChannel<SseItem<IJsonRpcMessage?>>();
1518

1619
private Task? _sseWriteTask;
1720
private Utf8JsonWriter? _jsonWriter;
1821

22+
/// <inherityydoc/>
1923
public bool IsConnected => _sseWriteTask?.IsCompleted == false;
2024

25+
/// <summary>
26+
/// Starts the transport and writes the JSON-RPC messages sent via <see cref="SendMessageAsync(IJsonRpcMessage, CancellationToken)"/>
27+
/// to the SSE response stream until cancelled or disposed.
28+
/// </summary>
29+
/// <param name="cancellationToken">A token to cancel writing to the SSE response stream.</param>
30+
/// <returns>A task representing the send loop that writes JSON-RPC messages to the SSE response stream.</returns>
2131
public Task RunAsync(CancellationToken cancellationToken)
2232
{
2333
void WriteJsonRpcMessageToBuffer(SseItem<IJsonRpcMessage?> item, IBufferWriter<byte> writer)
@@ -28,7 +38,7 @@ void WriteJsonRpcMessageToBuffer(SseItem<IJsonRpcMessage?> item, IBufferWriter<b
2838
return;
2939
}
3040

31-
JsonSerializer.Serialize(GetUtf8JsonWriter(writer), item.Data, McpJsonUtilities.DefaultOptions);
41+
JsonSerializer.Serialize(GetUtf8JsonWriter(writer), item.Data, McpJsonUtilities.DefaultOptions.GetTypeInfo<IJsonRpcMessage?>());
3242
}
3343

3444
// The very first SSE event isn't really an IJsonRpcMessage, but there's no API to write a single item of a different type,
@@ -39,23 +49,40 @@ void WriteJsonRpcMessageToBuffer(SseItem<IJsonRpcMessage?> item, IBufferWriter<b
3949
return _sseWriteTask = SseFormatter.WriteAsync(sseItems, sseResponseStream, WriteJsonRpcMessageToBuffer, cancellationToken);
4050
}
4151

52+
/// <inheritdoc/>
4253
public ChannelReader<IJsonRpcMessage> MessageReader => _incomingChannel.Reader;
4354

55+
/// <inheritdoc/>
4456
public ValueTask DisposeAsync()
4557
{
4658
_incomingChannel.Writer.TryComplete();
4759
_outgoingSseChannel.Writer.TryComplete();
4860
return new ValueTask(_sseWriteTask ?? Task.CompletedTask);
4961
}
5062

51-
public Task SendMessageAsync(IJsonRpcMessage message, CancellationToken cancellationToken = default) =>
52-
_outgoingSseChannel.Writer.WriteAsync(new SseItem<IJsonRpcMessage?>(message), cancellationToken).AsTask();
63+
/// <inheritdoc/>
64+
public Task SendMessageAsync(IJsonRpcMessage message, CancellationToken cancellationToken = default)
65+
{
66+
if (_sseWriteTask is null)
67+
{
68+
throw new InvalidOperationException($"Transport is not connected. Make sure to call {nameof(RunAsync)} first.");
69+
}
70+
71+
return _outgoingSseChannel.Writer.WriteAsync(new SseItem<IJsonRpcMessage?>(message), cancellationToken).AsTask();
72+
}
5373

74+
/// <summary>
75+
/// Handles incoming JSON-RPC messages received on the /message endpoint.
76+
/// </summary>
77+
/// <param name="message">The JSON-RPC message received.</param>
78+
/// <param name="cancellationToken">A token to cancel the operation.</param>
79+
/// <returns>A task representing the potentially asynchronous operation to buffer or process the JSON-RPC message.</returns>
80+
/// <exception cref="InvalidOperationException">Thrown when there is an attempt to process a message before calling <see cref="RunAsync(CancellationToken)"/>.</exception>
5481
public Task OnMessageReceivedAsync(IJsonRpcMessage message, CancellationToken cancellationToken)
5582
{
56-
if (!IsConnected)
83+
if (_sseWriteTask is null)
5784
{
58-
throw new McpTransportException("Transport is not connected");
85+
throw new InvalidOperationException($"Transport is not connected. Make sure to call {nameof(RunAsync)} first.");
5986
}
6087

6188
return _incomingChannel.Writer.WriteAsync(message, cancellationToken).AsTask();

0 commit comments

Comments
 (0)