Skip to content

Convert EverythingServer to use Streamable HTTP #709

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
wants to merge 4 commits into
base: main
Choose a base branch
from
+93 −41
Draft

Convert EverythingServer to use Streamable HTTP #709

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
64da739
Convert EverythingServer to use Streamable HTTP
mikekistler Aug 14, 2025
10c96f2
Fix resource subscriptions for HTTP
mikekistler Aug 15, 2025
e96c833
Fix SetLoggingLevel handler / log message generator
mikekistler Aug 15, 2025
5159f94
Fix resource notification generator
mikekistler Aug 15, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 2 additions & 3 deletions samples/EverythingServer/EverythingServer.csproj
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
<Project Sdk="Microsoft.NET.Sdk">
<Project Sdk="Microsoft.NET.Sdk.Web">

<PropertyGroup>
<TargetFramework>net9.0</TargetFramework>
Expand All @@ -8,14 +8,13 @@
</PropertyGroup>

<ItemGroup>
<PackageReference Include="Microsoft.Extensions.Hosting" />
<PackageReference Include="OpenTelemetry.Exporter.OpenTelemetryProtocol" />
<PackageReference Include="OpenTelemetry.Extensions.Hosting" />
<PackageReference Include="OpenTelemetry.Instrumentation.Http" />
</ItemGroup>

<ItemGroup>
<ProjectReference Include="..\..\src\ModelContextProtocol\ModelContextProtocol.csproj" />
<ProjectReference Include="..\..\src\ModelContextProtocol.AspNetCore\ModelContextProtocol.AspNetCore.csproj" />
</ItemGroup>

</Project>
37 changes: 26 additions & 11 deletions samples/EverythingServer/LoggingUpdateMessageSender.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,11 +1,10 @@
using Microsoft.Extensions.Hosting;
using ModelContextProtocol;
using ModelContextProtocol;
using ModelContextProtocol.Protocol;
using ModelContextProtocol.Server;

namespace EverythingServer;

public class LoggingUpdateMessageSender(IMcpServer server, Func<LoggingLevel> getMinLevel) : BackgroundService
public class LoggingUpdateMessageSender(IServiceProvider serviceProvider) : BackgroundService
{
readonly Dictionary<LoggingLevel, string> _loggingLevelMap = new()
{
Expand All @@ -21,19 +20,35 @@ public class LoggingUpdateMessageSender(IMcpServer server, Func<LoggingLevel> ge

protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
// Wait for the application to fully start before trying to access the MCP server
await Task.Delay(2000, stoppingToken);

while (!stoppingToken.IsCancellationRequested)
{
var newLevel = (LoggingLevel)Random.Shared.Next(_loggingLevelMap.Count);

var message = new
try
{
// Try to get the server from the service provider
var server = serviceProvider.GetService<IMcpServer>();
if (server != null)
{
Level = newLevel.ToString().ToLower(),
Data = _loggingLevelMap[newLevel],
};
var newLevel = (LoggingLevel)Random.Shared.Next(_loggingLevelMap.Count);

if (newLevel > getMinLevel())
var message = new
{
Level = newLevel.ToString().ToLower(),
Data = _loggingLevelMap[newLevel],
};

if (newLevel > server.LoggingLevel)
{
await server.SendNotificationAsync("notifications/message", message, cancellationToken: stoppingToken);
}
}
}
catch (Exception ex)
{
await server.SendNotificationAsync("notifications/message", message, cancellationToken: stoppingToken);
// Log the exception but don't crash the service
Console.WriteLine($"Error in LoggingUpdateMessageSender: {ex.Message}");
}

await Task.Delay(15000, stoppingToken);
Expand Down
42 changes: 23 additions & 19 deletions samples/EverythingServer/Program.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,9 +3,6 @@
using EverythingServer.Resources;
using EverythingServer.Tools;
using Microsoft.Extensions.AI;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using ModelContextProtocol;
using ModelContextProtocol.Protocol;
using ModelContextProtocol.Server;
Expand All @@ -15,19 +12,14 @@
using OpenTelemetry.Resources;
using OpenTelemetry.Trace;

var builder = Host.CreateApplicationBuilder(args);
builder.Logging.AddConsole(consoleLogOptions =>
{
// Configure all logs to go to stderr
consoleLogOptions.LogToStandardErrorThreshold = LogLevel.Trace;
});
var builder = WebApplication.CreateBuilder(args);

HashSet<string> subscriptions = [];
var _minimumLoggingLevel = LoggingLevel.Debug;
// Subscriptions tracks resource URIs to McpServer instances
Dictionary<string, List<IMcpServer>> subscriptions = new();
Copy link
Contributor

@halter73 halter73 Aug 18, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We'll need to use thread-safe data structures like ConcurrentDictionary and ConcurrentQueue to manage the subscriptions since handlers can run in parallel even in the context of a single session. And the sessions can arrive in parallel as well. We'll also want to remove entries from any singleton data structures like this when the session ends.

You can override RunSessionHandler to run custom logic before the session begins and after it ends. We'll probably need to track all the resource URIs each session is subscribed to in order to remove them after IMcpServer.RunAsync() completes.

However, given that we fire-and-forget message handlers like the SubscribeToResourcesHandler, that means that IMcpServer.RunAsync() can complete before all the handlers do, which could make cleanup racy if you're not aware of that.

csharp-sdk/src/ModelContextProtocol.Core/McpSession.cs

Lines 118 to 122 in b067261

// Fire and forget the message handling to avoid blocking the transport.
if (message.ExecutionContext is null)
{
_ = ProcessMessageAsync();
}

I don't think this race would come up super often, but I still hope this scenario is motivation to fix that. I think IMcpServer.RunAsync() should wait for all handlers to complete similar to how WebApplication.RunAsync() will wait for Kestrel to complete all its RequestDelegates before completing unless the host shutdown timeout fires first.


builder.Services
.AddMcpServer()
.WithStdioServerTransport()
.WithHttpTransport()
.WithTools<AddTool>()
.WithTools<AnnotatedMessageTool>()
.WithTools<EchoTool>()
Expand All @@ -44,7 +36,11 @@

if (uri is not null)
{
subscriptions.Add(uri);
if (!subscriptions.ContainsKey(uri))
{
subscriptions[uri] = new List<IMcpServer>();
}
subscriptions[uri].Add(ctx.Server);

await ctx.Server.SampleAsync([
new ChatMessage(ChatRole.System, "You are a helpful test server"),
Expand All @@ -65,7 +61,11 @@ await ctx.Server.SampleAsync([
var uri = ctx.Params?.Uri;
if (uri is not null)
{
subscriptions.Remove(uri);
if (subscriptions.ContainsKey(uri))
{
// Remove ctx.Server from the subscription list
subscriptions[uri].Remove(ctx.Server);
}
}
return new EmptyResult();
})
Expand Down Expand Up @@ -126,13 +126,13 @@ await ctx.Server.SampleAsync([
throw new McpException("Missing required argument 'level'", McpErrorCode.InvalidParams);
}

_minimumLoggingLevel = ctx.Params.Level;
// The SDK updates the LoggingLevel field of the IMcpServer

await ctx.Server.SendNotificationAsync("notifications/message", new
{
Level = "debug",
Logger = "test-server",
Data = $"Logging level set to {_minimumLoggingLevel}",
Data = $"Logging level set to {ctx.Params.Level}",
}, cancellationToken: ct);

return new EmptyResult();
Expand All @@ -145,10 +145,14 @@ await ctx.Server.SampleAsync([
.WithLogging(b => b.SetResourceBuilder(resource))
.UseOtlpExporter();

builder.Services.AddSingleton(subscriptions);
builder.Services.AddSingleton<IDictionary<string, List<IMcpServer>>>(subscriptions);
builder.Services.AddHostedService<SubscriptionMessageSender>();
builder.Services.AddHostedService<LoggingUpdateMessageSender>();

builder.Services.AddSingleton<Func<LoggingLevel>>(_ => () => _minimumLoggingLevel);
var app = builder.Build();

app.UseHttpsRedirection();

app.MapMcp();

await builder.Build().RunAsync();
app.Run();
21 changes: 21 additions & 0 deletions samples/EverythingServer/Properties/launchSettings.json
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,21 @@
{
"$schema": "https://json.schemastore.org/launchsettings.json",
"profiles": {
"http": {
"commandName": "Project",
"dotnetRunMessages": true,
"applicationUrl": "http://localhost:3001",
"environmentVariables": {
"ASPNETCORE_ENVIRONMENT": "Development",
}
},
"https": {
"commandName": "Project",
"dotnetRunMessages": true,
"applicationUrl": "https://localhost:7133;http://localhost:3001",
"environmentVariables": {
"ASPNETCORE_ENVIRONMENT": "Development",
}
}
}
}
29 changes: 21 additions & 8 deletions samples/EverythingServer/SubscriptionMessageSender.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,20 +1,33 @@
using Microsoft.Extensions.Hosting;
using ModelContextProtocol;
using ModelContextProtocol;
using ModelContextProtocol.Server;

internal class SubscriptionMessageSender(IMcpServer server, HashSet<string> subscriptions) : BackgroundService
internal class SubscriptionMessageSender(IDictionary<string, List<IMcpServer>> subscriptions) : BackgroundService
{
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
// Wait for the application to fully start before trying to access the MCP server
await Task.Delay(2000, stoppingToken);

while (!stoppingToken.IsCancellationRequested)
{
foreach (var uri in subscriptions)
try
{
await server.SendNotificationAsync("notifications/resource/updated",
new
foreach (var (uri, servers) in subscriptions)
{
foreach (var server in servers)
{
Uri = uri,
}, cancellationToken: stoppingToken);
await server.SendNotificationAsync("notifications/resource/updated",
new
{
Uri = uri,
}, cancellationToken: stoppingToken);
}
}
}
catch (Exception ex)
{
// Log the exception but don't crash the service
Console.WriteLine($"Error in SubscriptionMessageSender: {ex.Message}");
}

await Task.Delay(5000, stoppingToken);
Expand Down
Loading