Stream resumability support

samples/AspNetCoreMcpServer/AspNetCoreMcpServer.csproj

@@ -4,7 +4,6 @@
     <TargetFramework>net9.0</TargetFramework>
     <Nullable>enable</Nullable>
     <ImplicitUsings>enable</ImplicitUsings>
-    <PublishAot>true</PublishAot>
   </PropertyGroup>
 
   <ItemGroup>

samples/AspNetCoreMcpServer/EventStore/EventStoreCleanupService.cs

@@ -0,0 +1,47 @@
+using ModelContextProtocol.Server;
+
+namespace AspNetCoreMcpServer.EventStore;
+
+public class EventStoreCleanupService : BackgroundService
+{
+    private readonly TimeSpan _jobRunFrequencyInMinutes;
+    private readonly ILogger<EventStoreCleanupService> _logger;
+    private readonly IEventStoreCleaner? _eventStoreCleaner;
+
+    public EventStoreCleanupService(ILogger<EventStoreCleanupService> logger, IConfiguration configuration, IEventStoreCleaner? eventStoreCleaner = null)
+    {
+        _logger = logger ?? throw new ArgumentNullException(nameof(logger));
+
+        _eventStoreCleaner = eventStoreCleaner;
+        _jobRunFrequencyInMinutes = TimeSpan.FromMinutes(configuration.GetValue<int>("EventStore:CleanupJobRunFrequencyInMinutes", 30));
+    }
+
+    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
+    {
+
+        if (_eventStoreCleaner is null)
+        {
+            _logger.LogWarning("No event store cleaner implementation provided. Event store cleanup job will not run.");
+            return;
+        }
+
+        _logger.LogInformation("Event store cleanup job started.");
+
+        while (!stoppingToken.IsCancellationRequested)
+        {
+            try
+            {
+                _logger.LogInformation("Running event store cleanup job at {CurrentTimeInUtc}.", DateTime.UtcNow);
+                _eventStoreCleaner.CleanEventStore();
+            }
+            catch (Exception ex)
+            {
+                _logger.LogError(ex, "Error running event store cleanup job.");
+            }
+
+            await Task.Delay(_jobRunFrequencyInMinutes, stoppingToken);
+        }
+
+        _logger.LogInformation("Event store cleanup job stopping.");
+    }
+}

samples/AspNetCoreMcpServer/EventStore/IEventStoreCleaner.cs

@@ -0,0 +1,15 @@
+namespace AspNetCoreMcpServer.EventStore;
+
+/// <summary>
+/// Interface for cleaning up the event store
+/// </summary>
+public interface IEventStoreCleaner
+{
+
+    /// <summary>
+    /// Cleans up the event store by removing outdated or unnecessary events.
+    /// </summary>
+    /// <remarks>This method is typically used to maintain the event store's size and performance by clearing
+    /// events that are no longer needed.</remarks>
+    void CleanEventStore();
+}

samples/AspNetCoreMcpServer/EventStore/InMemoryEventStore.cs

@@ -0,0 +1,97 @@
+using ModelContextProtocol.Protocol;
+using ModelContextProtocol.Server;
+using System.Collections.Concurrent;
+using System.Net.ServerSentEvents;
+
+namespace AspNetCoreMcpServer.EventStore;
+
+/// <summary>
+/// Represents an in-memory implementation of an event store that stores and replays events associated with specific
+/// streams. This class is designed to handle events of type <see cref="SseItem{T}"/> where the data payload is a <see
+/// cref="JsonRpcMessage"/>.
+/// </summary>
+/// <remarks>The <see cref="InMemoryEventStore"/> provides functionality to store events for a given stream and
+/// replay events after a specified event ID. It supports resumability for specific types of requests and ensures events
+/// are replayed in the correct order.</remarks>
+public sealed class InMemoryEventStore : IEventStore, IEventStoreCleaner
+{
+    public const string EventIdDelimiter = "_";
+    private static ConcurrentDictionary<string, List<SseItem<JsonRpcMessage?>>> _eventStore = new();
+
+    private readonly ILogger<InMemoryEventStore> _logger;
+    private readonly TimeSpan _eventsRetentionDurationInMinutes;
+
+    public InMemoryEventStore(IConfiguration configuration, ILogger<InMemoryEventStore> logger)
+    {
+        _eventsRetentionDurationInMinutes = TimeSpan.FromMinutes(configuration.GetValue<int>("EventStore:EventsRetentionDurationInMinutes", 60));
+        _logger = logger ?? throw new ArgumentNullException(nameof(logger));
+    }
+
+    public void StoreEvent(string streamId, SseItem<JsonRpcMessage?> messageItem)
+    {
+        // remove ElicitationCreate method check to support resumability for other type of requests
+        if (messageItem.Data is JsonRpcRequest jsonRpcReq && jsonRpcReq.Method == RequestMethods.ElicitationCreate)
+        {
+            var sseItemList = _eventStore.GetOrAdd(streamId, (key) => new List<SseItem<JsonRpcMessage?>>());
+            sseItemList.Add(messageItem);
+        }
+
+        if (messageItem.Data is JsonRpcResponse jsonRpcResp && 
+            _eventStore.TryGetValue(streamId, out var itemList))
+        {
+            itemList.Add(messageItem);
+        }
+    }
+
+    public async Task ReplayEventsAfter(string lastEventId, Action<IAsyncEnumerable<SseItem<JsonRpcMessage>>> sendEvents)
+    {
+        var streamId = lastEventId.Split(EventIdDelimiter)[0];
+        var events = _eventStore.GetValueOrDefault(streamId, new());
+        var sortedAndFilteredEventsToSend = events
+            .Where(e => e.Data is not null && e.EventId != null)
+            .OrderBy(e => e.EventId)
+            // Sending events with EventId greater than lastEventId.
+            .SkipWhile(e => string.Compare(e.EventId!, lastEventId, StringComparison.Ordinal) <= 0)
+            .Select(e =>
+                new SseItem<JsonRpcMessage>(e.Data!, e.EventType)
+                {
+                    EventId = e.EventId,
+                    ReconnectionInterval = e.ReconnectionInterval
+                });
+        sendEvents(SseItemsAsyncEnumerable(sortedAndFilteredEventsToSend));
+    }
+
+    private static async IAsyncEnumerable<SseItem<JsonRpcMessage>> SseItemsAsyncEnumerable(IEnumerable<SseItem<JsonRpcMessage>> enumerableItems)
+    {
+        foreach (var sseItem in enumerableItems)
+        {
+            yield return sseItem;
+        }
+    }
+
+    public string? GetEventId(string streamId, JsonRpcMessage message)
+    {
+        return $"{streamId}{EventIdDelimiter}{DateTime.UtcNow.Ticks}";
+    }
+
+    public void CleanEventStore()
+    {
+        var cutoffTime = DateTime.UtcNow - _eventsRetentionDurationInMinutes;
+        _logger.LogInformation("Cleaning up events older than {CutoffTime} from event store.",  cutoffTime);
+
+        foreach (var key in _eventStore.Keys)
+        {
+            if (_eventStore.TryGetValue(key, out var itemList))
+            {
+                itemList.RemoveAll(item => item.EventId != null && 
+                    long.TryParse(item.EventId.Split(EventIdDelimiter)[1], out var ticks) && 
+                    new DateTime(ticks) < cutoffTime);
+                if (itemList.Count == 0)
+                {
+                    _logger.LogInformation("Removing empty event stream with key {EventStreamKey} from event store.", key);
+                    _eventStore.TryRemove(key, out _);
+                }
+            }
+        }
+    }
+}

samples/AspNetCoreMcpServer/Program.cs

@@ -1,15 +1,18 @@
+using AspNetCoreMcpServer.EventStore;
+using AspNetCoreMcpServer.Resources;
+using AspNetCoreMcpServer.Tools;
+using Microsoft.Extensions.DependencyInjection.Extensions;
+using ModelContextProtocol.Server;
 using OpenTelemetry;
 using OpenTelemetry.Metrics;
 using OpenTelemetry.Trace;
-using AspNetCoreMcpServer.Tools;
-using AspNetCoreMcpServer.Resources;
 using System.Net.Http.Headers;
 
 var builder = WebApplication.CreateBuilder(args);
 builder.Services.AddMcpServer()
     .WithHttpTransport()
     .WithTools<EchoTool>()
-    .WithTools<SampleLlmTool>()
+    .WithTools<CollectUserInformationTool>() // this tool collect user information through elicitation
     .WithTools<WeatherTools>()
     .WithResources<SimpleResourceType>();
 
@@ -30,6 +33,11 @@
     client.DefaultRequestHeaders.UserAgent.Add(new ProductInfoHeaderValue("weather-tool", "1.0"));
 });
 
+// adding InMemoryEventStore to support stream resumability and background cleanup service
+builder.Services.TryAddSingleton<IEventStore, InMemoryEventStore>();
+builder.Services.TryAddSingleton<IEventStoreCleaner, InMemoryEventStore>();
+builder.Services.AddHostedService<EventStoreCleanupService>();
+
 var app = builder.Build();
 
 app.MapMcp();

samples/AspNetCoreMcpServer/Tools/CollectUserInformationTool.cs

@@ -0,0 +1,143 @@
+using ModelContextProtocol;
+using ModelContextProtocol.Protocol;
+using ModelContextProtocol.Server;
+using System.ComponentModel;
+using System.Text.Json;
+
+namespace AspNetCoreMcpServer.Tools;
+
+[McpServerToolType]
+public sealed class CollectUserInformationTool
+{
+    public enum InfoType
+    {
+        contact,
+        preferences,
+        feedback
+    }
+
+    [McpServerTool(Name = "collect-user-info"), Description("A tool that collects user information through elicitation")]
+    public static async Task<CallToolResult> ElicitationEcho(McpServer thisServer, [Description("Type of information to collect")] InfoType infoType)
+    {
+        ElicitRequestParams elicitRequestParams;
+        switch (infoType)
+        {
+            case InfoType.contact:
+                elicitRequestParams = new ElicitRequestParams()
+                { 
+                    Message = "Please provide your contact information",
+                    RequestedSchema = new ElicitRequestParams.RequestSchema
+                    {
+                        Properties = new Dictionary<string, ElicitRequestParams.PrimitiveSchemaDefinition>()
+                        {
+                            ["name"] = new ElicitRequestParams.StringSchema
+                            {
+                                Title = "Full name",
+                                Description = "Your full name",
+                            },
+                            ["email"] = new ElicitRequestParams.StringSchema
+                            {
+                                Title = "Email address",
+                                Description = "Your email address",
+                                Format = "email",
+                            },
+                            ["phone"] = new ElicitRequestParams.StringSchema
+                            {
+                                Title = "Phone number",
+                                Description = "Your phone number (optional)",
+                            }
+                        },
+                        Required = new List<string> { "name", "email" }
+                    }
+                };
+                break;
+
+            case InfoType.preferences:
+                elicitRequestParams = new ElicitRequestParams()
+                {
+                    Message = "Please set your preferences",
+                    RequestedSchema = new ElicitRequestParams.RequestSchema
+                    {
+                        Properties = new Dictionary<string, ElicitRequestParams.PrimitiveSchemaDefinition>()
+                        {
+                            ["theme"] = new ElicitRequestParams.EnumSchema
+                            {
+                                Title = "Theme",
+                                Description = "Choose your preferred theme",
+                                Enum = new List<string> { "light", "dark", "auto" },
+                                EnumNames = new List<string> { "Light", "Dark", "Auto" }
+                            },
+                            ["notifications"] = new ElicitRequestParams.BooleanSchema
+                            {
+                                Title = "Enable notifications",
+                                Description = "Would you like to receive notifications?",
+                                Default = true,
+                            },
+                            ["frequency"] = new ElicitRequestParams.EnumSchema
+                            {
+                                Title = "Notification frequency",
+                                Description = "How often would you like notifications?",
+                                Enum = new List<string> { "daily", "weekly", "monthly" },
+                                EnumNames = new List<string> { "Daily", "Weekly", "Monthly" }
+                            }
+                        },
+                        Required = new List<string> { "theme" }
+                    }
+                };
+
+                break;
+
+            case InfoType.feedback:
+                elicitRequestParams = new ElicitRequestParams()
+                {
+                    Message = "Please provide your feedback",
+                    RequestedSchema = new ElicitRequestParams.RequestSchema
+                    {
+                        Properties = new Dictionary<string, ElicitRequestParams.PrimitiveSchemaDefinition>()
+                        {
+                            ["rating"] = new ElicitRequestParams.NumberSchema
+                            {
+                                Title = "Rating",
+                                Description = "Rate your experience (1-5)",
+                                Minimum = 1,
+                                Maximum = 5,
+                            },
+                            ["comments"] = new ElicitRequestParams.StringSchema
+                            {
+                                Title = "Comments",
+                                Description = "Additional comments (optional)",
+                                MaxLength = 500,
+                            },
+                            ["recommend"] = new ElicitRequestParams.BooleanSchema
+                            {
+                                Title = "Would you recommend this?",
+                                Description = "Would you recommend this to others?",
+                            }
+                        },
+                        Required = new List<string> { "rating", "recommend" }
+                    }
+                };
+
+                break;
+
+            default:
+                throw new Exception($"Unknown info type: ${infoType}");
+
+        }
+
+
+        var result = await thisServer.ElicitAsync(elicitRequestParams);
+        var textResult = result.Action switch
+        {
+            "accept" => $"Thank you! Collected ${infoType} information: {JsonSerializer.Serialize(result.Content, McpJsonUtilities.DefaultOptions.GetTypeInfo(typeof(IDictionary<string, JsonElement>)))}",
+            "decline" => "No information was collected. User declined ${infoType} information request.",
+            "cancel" => "Information collection was cancelled by the user.",
+            _ => "Error collecting ${infoType} information: ${error}"
+        };
+
+        return new CallToolResult()
+        { 
+            Content = [ new TextContentBlock { Text = textResult } ],
+        };
+    }
+}

samples/AspNetCoreMcpServer/appsettings.json

@@ -5,5 +5,9 @@
       "Microsoft.AspNetCore": "Warning"
     }
   },
-  "AllowedHosts": "*"
+  "AllowedHosts": "*",
+  "EventStore": {
+    "CleanupJobRunFrequencyInMinutes": 30,
+    "EventsRetentionDurationInMinutes": 60
+  }
 }