Convert EverythingServer to use Streamable HTTP (#709)

* Convert EverythingServer to use Streamable HTTP

* Apply suggestion from @halter73

Co-authored-by: Stephen Halter <halter73@gmail.com>

---------

Co-authored-by: Stephen Halter <halter73@gmail.com>
This commit is contained in:
Mike Kistler 2025-10-16 14:39:42 -07:00 • committed by GitHub
commit fb07d666a6
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
8 changed files with 177 additions and 37 deletions

View file

@ -1,4 +1,4 @@
<Project Sdk="Microsoft.NET.Sdk"> <Project Sdk="Microsoft.NET.Sdk.Web">
<PropertyGroup> <PropertyGroup>
<TargetFramework>net9.0</TargetFramework> <TargetFramework>net9.0</TargetFramework>
@ -8,14 +8,13 @@
</PropertyGroup> </PropertyGroup>
<ItemGroup> <ItemGroup>
<PackageReference Include="Microsoft.Extensions.Hosting" />
<PackageReference Include="OpenTelemetry.Exporter.OpenTelemetryProtocol" /> <PackageReference Include="OpenTelemetry.Exporter.OpenTelemetryProtocol" />
<PackageReference Include="OpenTelemetry.Extensions.Hosting" /> <PackageReference Include="OpenTelemetry.Extensions.Hosting" />
<PackageReference Include="OpenTelemetry.Instrumentation.Http" /> <PackageReference Include="OpenTelemetry.Instrumentation.Http" />
</ItemGroup> </ItemGroup>
<ItemGroup> <ItemGroup>
<ProjectReference Include="..\..\src\ModelContextProtocol\ModelContextProtocol.csproj" /> <ProjectReference Include="..\..\src\ModelContextProtocol.AspNetCore\ModelContextProtocol.AspNetCore.csproj" />
</ItemGroup> </ItemGroup>
</Project> </Project>

View file

@ -0,0 +1,77 @@
@HostAddress = http://localhost:3001
POST {{HostAddress}}/
Accept: application/json, text/event-stream
Content-Type: application/json
{
"jsonrpc": "2.0",
"id": 1,
"method": "initialize",
"params": {
"clientInfo": {
"name": "RestClient",
"version": "0.1.0"
},
"capabilities": {},
"protocolVersion": "2025-06-18"
}
}
###
@SessionId = ZwwM0VFEtKNOMBsP8D2VzQ
POST {{HostAddress}}/
Accept: application/json, text/event-stream
Content-Type: application/json
MCP-Protocol-Version: 2025-06-18
Mcp-Session-Id: {{SessionId}}
{
"jsonrpc": "2.0",
"id": 2,
"method": "resources/list"
}
###
@resource_uri = test://direct/text/resource
POST {{HostAddress}}/
Accept: application/json, text/event-stream
Content-Type: application/json
MCP-Protocol-Version: 2025-06-18
Mcp-Session-Id: {{SessionId}}
{
"jsonrpc": "2.0",
"id": 3,
"method": "resources/subscribe",
"params": {
"uri": "{{resource_uri}}"
}
}
###
POST {{HostAddress}}/
Accept: application/json, text/event-stream
Content-Type: application/json
MCP-Protocol-Version: 2025-06-18
Mcp-Session-Id: {{SessionId}}
{
"jsonrpc": "2.0",
"id": 4,
"method": "resources/unsubscribe",
"params": {
"uri": "{{resource_uri}}"
}
}
###
DELETE {{HostAddress}}/
MCP-Protocol-Version: 2025-06-18
Mcp-Session-Id: {{SessionId}}

View file

@ -5,7 +5,7 @@ using ModelContextProtocol.Server;
namespace EverythingServer; namespace EverythingServer;
public class LoggingUpdateMessageSender(McpServer server, Func<LoggingLevel> getMinLevel) : BackgroundService public class LoggingUpdateMessageSender(McpServer server) : BackgroundService
{ {
readonly Dictionary<LoggingLevel, string> _loggingLevelMap = new() readonly Dictionary<LoggingLevel, string> _loggingLevelMap = new()
{ {
@ -23,15 +23,15 @@ public class LoggingUpdateMessageSender(McpServer server, Func<LoggingLevel> get
{ {
while (!stoppingToken.IsCancellationRequested) while (!stoppingToken.IsCancellationRequested)
{ {
var newLevel = (LoggingLevel)Random.Shared.Next(_loggingLevelMap.Count); var msgLevel = (LoggingLevel)Random.Shared.Next(_loggingLevelMap.Count);
var message = new var message = new
{ {
Level = newLevel.ToString().ToLower(), Level = msgLevel.ToString().ToLower(),
Data = _loggingLevelMap[newLevel], Data = _loggingLevelMap[msgLevel],
}; };
if (newLevel > getMinLevel()) if (msgLevel > server.LoggingLevel)
{ {
await server.SendNotificationAsync("notifications/message", message, cancellationToken: stoppingToken); await server.SendNotificationAsync("notifications/message", message, cancellationToken: stoppingToken);
} }

View file

@ -3,9 +3,6 @@ using EverythingServer.Prompts;
using EverythingServer.Resources; using EverythingServer.Resources;
using EverythingServer.Tools; using EverythingServer.Tools;
using Microsoft.Extensions.AI; using Microsoft.Extensions.AI;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using ModelContextProtocol; using ModelContextProtocol;
using ModelContextProtocol.Protocol; using ModelContextProtocol.Protocol;
using ModelContextProtocol.Server; using ModelContextProtocol.Server;
@ -14,20 +11,46 @@ using OpenTelemetry.Logs;
using OpenTelemetry.Metrics; using OpenTelemetry.Metrics;
using OpenTelemetry.Resources; using OpenTelemetry.Resources;
using OpenTelemetry.Trace; using OpenTelemetry.Trace;
using System.Collections.Concurrent;
var builder = Host.CreateApplicationBuilder(args); var builder = WebApplication.CreateBuilder(args);
builder.Logging.AddConsole(consoleLogOptions =>
{
// Configure all logs to go to stderr
consoleLogOptions.LogToStandardErrorThreshold = LogLevel.Trace;
});
HashSet<string> subscriptions = []; // Dictionary of session IDs to a set of resource URIs they are subscribed to
var _minimumLoggingLevel = LoggingLevel.Debug; // The value is a ConcurrentDictionary used as a thread-safe HashSet
// because .NET does not have a built-in concurrent HashSet
ConcurrentDictionary<string, ConcurrentDictionary<string, byte>> subscriptions = new();
builder.Services builder.Services
.AddMcpServer() .AddMcpServer()
.WithStdioServerTransport() .WithHttpTransport(options =>
{
// Add a RunSessionHandler to remove all subscriptions for the session when it ends
options.RunSessionHandler = async (httpContext, mcpServer, token) =>
{
if (mcpServer.SessionId == null)
{
// There is no sessionId if the serverOptions.Stateless is true
await mcpServer.RunAsync(token);
return;
}
try
{
subscriptions[mcpServer.SessionId] = new ConcurrentDictionary<string, byte>();
// Start an instance of SubscriptionMessageSender for this session
using var subscriptionSender = new SubscriptionMessageSender(mcpServer, subscriptions[mcpServer.SessionId]);
await subscriptionSender.StartAsync(token);
// Start an instance of LoggingUpdateMessageSender for this session
using var loggingSender = new LoggingUpdateMessageSender(mcpServer);
await loggingSender.StartAsync(token);
await mcpServer.RunAsync(token);
}
finally
{
// This code runs when the session ends
subscriptions.TryRemove(mcpServer.SessionId, out _);
}
};
})
.WithTools<AddTool>() .WithTools<AddTool>()
.WithTools<AnnotatedMessageTool>() .WithTools<AnnotatedMessageTool>()
.WithTools<EchoTool>() .WithTools<EchoTool>()
@ -40,11 +63,13 @@ builder.Services
.WithResources<SimpleResourceType>() .WithResources<SimpleResourceType>()
.WithSubscribeToResourcesHandler(async (ctx, ct) => .WithSubscribeToResourcesHandler(async (ctx, ct) =>
{ {
var uri = ctx.Params?.Uri; if (ctx.Server.SessionId == null)
if (uri is not null)
{ {
subscriptions.Add(uri); throw new McpException("Cannot add subscription for server with null SessionId");
}
if (ctx.Params?.Uri is { } uri)
{
subscriptions[ctx.Server.SessionId].TryAdd(uri, 0);
await ctx.Server.SampleAsync([ await ctx.Server.SampleAsync([
new ChatMessage(ChatRole.System, "You are a helpful test server"), new ChatMessage(ChatRole.System, "You are a helpful test server"),
@ -62,10 +87,13 @@ builder.Services
}) })
.WithUnsubscribeFromResourcesHandler(async (ctx, ct) => .WithUnsubscribeFromResourcesHandler(async (ctx, ct) =>
{ {
var uri = ctx.Params?.Uri; if (ctx.Server.SessionId == null)
if (uri is not null)
{ {
subscriptions.Remove(uri); throw new McpException("Cannot remove subscription for server with null SessionId");
}
if (ctx.Params?.Uri is { } uri)
{
subscriptions[ctx.Server.SessionId].TryRemove(uri, out _);
} }
return new EmptyResult(); return new EmptyResult();
}) })
@ -126,13 +154,13 @@ builder.Services
throw new McpProtocolException("Missing required argument 'level'", McpErrorCode.InvalidParams); throw new McpProtocolException("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 await ctx.Server.SendNotificationAsync("notifications/message", new
{ {
Level = "debug", Level = "debug",
Logger = "test-server", Logger = "test-server",
Data = $"Logging level set to {_minimumLoggingLevel}", Data = $"Logging level set to {ctx.Params.Level}",
}, cancellationToken: ct); }, cancellationToken: ct);
return new EmptyResult(); return new EmptyResult();
@ -145,10 +173,8 @@ builder.Services.AddOpenTelemetry()
.WithLogging(b => b.SetResourceBuilder(resource)) .WithLogging(b => b.SetResourceBuilder(resource))
.UseOtlpExporter(); .UseOtlpExporter();
builder.Services.AddSingleton(subscriptions); var app = builder.Build();
builder.Services.AddHostedService<SubscriptionMessageSender>();
builder.Services.AddHostedService<LoggingUpdateMessageSender>();
builder.Services.AddSingleton<Func<LoggingLevel>>(_ => () => _minimumLoggingLevel); app.MapMcp();
await builder.Build().RunAsync(); app.Run();

View file

@ -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",
}
}
}
}

View file

@ -1,14 +1,14 @@
using Microsoft.Extensions.Hosting; using System.Collections.Concurrent;
using ModelContextProtocol; using ModelContextProtocol;
using ModelContextProtocol.Server; using ModelContextProtocol.Server;
internal class SubscriptionMessageSender(McpServer server, HashSet<string> subscriptions) : BackgroundService internal class SubscriptionMessageSender(McpServer server, ConcurrentDictionary<string, byte> subscriptions) : BackgroundService
{ {
protected override async Task ExecuteAsync(CancellationToken stoppingToken) protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{ {
while (!stoppingToken.IsCancellationRequested) while (!stoppingToken.IsCancellationRequested)
{ {
foreach (var uri in subscriptions) foreach (var uri in subscriptions.Keys)
{ {
await server.SendNotificationAsync("notifications/resource/updated", await server.SendNotificationAsync("notifications/resource/updated",
new new

View file

@ -0,0 +1,8 @@
{
"Logging": {
"LogLevel": {
"Default": "Information",
"Microsoft.AspNetCore": "Warning"
}
}
}

View file

@ -0,0 +1,9 @@
{
"Logging": {
"LogLevel": {
"Default": "Information",
"Microsoft.AspNetCore": "Warning"
}
},
"AllowedHosts": "*"
}