using ModelContextProtocol.Protocol;
using System.Collections.Concurrent;
using System.Collections.Immutable;
using System.Diagnostics.CodeAnalysis;
using System.Text.Json;
namespace ModelContextProtocol.Server;
///
/// Provides an in-memory implementation of for development and testing scenarios.
///
///
///
/// This implementation stores all task state in memory using immutable snapshots and
/// compare-and-swap updates for thread safety without locks.
/// Tasks are not persisted across process restarts.
///
///
/// For production scenarios requiring durability, session isolation, or TTL-based cleanup,
/// implement a custom .
///
///
[Experimental(Experimentals.Extensions_DiagnosticId, UrlFormat = Experimentals.Extensions_Url)]
public class InMemoryMcpTaskStore : IMcpTaskStore
{
private readonly ConcurrentDictionary _tasks = new();
///
/// Gets or sets the default poll interval in milliseconds for new tasks.
///
/// The default is 1000 milliseconds.
public long DefaultPollIntervalMs { get; set; } = 1000;
///
/// Gets or sets the default time-to-live for new tasks, or for unlimited.
///
public TimeSpan? DefaultTimeToLive { get; set; }
///
public Task CreateTaskAsync(CancellationToken cancellationToken = default)
{
var taskId = Guid.NewGuid().ToString("N");
var now = DateTimeOffset.UtcNow;
var info = new McpTaskInfo(taskId, McpTaskStatus.Working, now, now, DefaultTimeToLive, DefaultPollIntervalMs);
_tasks[taskId] = info;
return Task.FromResult(info);
}
///
public Task GetTaskAsync(string taskId, CancellationToken cancellationToken = default)
{
_tasks.TryGetValue(taskId, out var info);
return Task.FromResult(info);
}
///
public Task SetCompletedAsync(string taskId, JsonElement result, CancellationToken cancellationToken = default)
{
Update(taskId, entry => IsTerminal(entry.Status)
? entry
: entry with
{
Status = McpTaskStatus.Completed,
Result = result,
LastUpdatedAt = DateTimeOffset.UtcNow,
});
return Task.CompletedTask;
}
///
public Task SetFailedAsync(string taskId, JsonElement error, CancellationToken cancellationToken = default)
{
Update(taskId, entry => IsTerminal(entry.Status)
? entry
: entry with
{
Status = McpTaskStatus.Failed,
Error = error,
LastUpdatedAt = DateTimeOffset.UtcNow,
});
return Task.CompletedTask;
}
///
public Task SetCancelledAsync(string taskId, CancellationToken cancellationToken = default)
{
if (!_tasks.TryGetValue(taskId, out var entry) || IsTerminal(entry.Status))
{
return Task.FromResult(false);
}
bool transitioned = false;
Update(taskId, e =>
{
if (IsTerminal(e.Status))
{
return e;
}
transitioned = true;
return e with { Status = McpTaskStatus.Cancelled, LastUpdatedAt = DateTimeOffset.UtcNow };
});
return Task.FromResult(transitioned);
}
///
public event Action? InputResponseReceived;
///
public Task ResolveInputRequestsAsync(
string taskId,
IDictionary inputResponses,
CancellationToken cancellationToken = default)
{
bool wasTerminal = false;
Update(taskId, entry =>
{
if (IsTerminal(entry.Status))
{
wasTerminal = true;
return entry;
}
var requests = entry.InputRequests as ImmutableDictionary
?? entry.InputRequests?.ToImmutableDictionary()
?? ImmutableDictionary.Empty;
foreach (var key in inputResponses.Keys)
{
requests = requests.Remove(key);
}
var status = requests.IsEmpty ? McpTaskStatus.Working : entry.Status;
return entry with
{
InputRequests = requests,
Status = status,
LastUpdatedAt = DateTimeOffset.UtcNow,
};
});
if (wasTerminal)
{
// Drop responses targeting a terminal task — there are no listeners that can act on them.
return Task.CompletedTask;
}
foreach (var kvp in inputResponses)
{
InputResponseReceived?.Invoke(new InputResponseReceivedEventArgs
{
TaskId = taskId,
RequestId = kvp.Key,
Response = kvp.Value,
});
}
return Task.CompletedTask;
}
///
public Task SetInputRequestsAsync(
string taskId,
IDictionary inputRequests,
CancellationToken cancellationToken = default)
{
Update(taskId, entry =>
{
if (IsTerminal(entry.Status))
{
return entry;
}
var requests = entry.InputRequests as ImmutableDictionary
?? entry.InputRequests?.ToImmutableDictionary()
?? ImmutableDictionary.Empty;
foreach (var kvp in inputRequests)
{
requests = requests.SetItem(kvp.Key, kvp.Value);
}
return entry with
{
InputRequests = requests,
Status = McpTaskStatus.InputRequired,
LastUpdatedAt = DateTimeOffset.UtcNow,
};
});
return Task.CompletedTask;
}
private static bool IsTerminal(McpTaskStatus status) =>
status is McpTaskStatus.Completed or McpTaskStatus.Failed or McpTaskStatus.Cancelled;
private void Update(string taskId, Func transform)
{
SpinWait spin = default;
while (true)
{
if (!_tasks.TryGetValue(taskId, out var current))
{
throw new InvalidOperationException($"Task '{taskId}' not found.");
}
var updated = transform(current);
if (ReferenceEquals(updated, current) || _tasks.TryUpdate(taskId, updated, current))
{
return;
}
spin.SpinOnce();
}
}
}