Harden tool discovery and loading against concurrency (#4057)

* Harden tool discovery and loading against concurrency

* Fix double free
This commit is contained in:
Alan Zimmer authored and GitHub committed 2026-10-30 13:12:27 -04:00
1 parent 286a2b8b6e
commit 37cd285a28
14 files changed
+766 -95

No files matched your search

@@ -231,6 +231,228 @@ public class BaseDiscoveryStrategyTests
await provider2.Received(1).CreateClientAsync(Arg.Any<McpClientOptions>(), Arg.Any<CancellationToken>());
}
[Fact]
public async Task GetOrCreateClientAsync_ConcurrentCalls_CreateOneClient()
{
await using var client = LoopbackMcpClient.Create(_ => null, out var tracker);
var creation = new TaskCompletionSource<McpClient>(TaskCreationOptions.RunContinuationsAsynchronously);
var provider = CreateMockServerProvider("external");
provider.CreateClientAsync(Arg.Any<McpClientOptions>(), Arg.Any<CancellationToken>())
.Returns(creation.Task);
await using var strategy = CreateMockStrategy(provider);
var tasks = Enumerable.Range(0, 16)
.Select(index => strategy.GetOrCreateClientAsync(
index % 2 == 0 ? "external" : "EXTERNAL", cancellationToken: TestContext.Current.CancellationToken))
.ToArray();
creation.SetResult(client);
var clients = await Task.WhenAll(tasks);
Assert.All(clients, result => Assert.Same(client, result));
await provider.Received(1).CreateClientAsync(Arg.Any<McpClientOptions>(), Arg.Any<CancellationToken>());
await strategy.DisposeAsync();
Assert.Equal(1, tracker.DisposeCount);
}
[Fact]
public async Task GetOrCreateClientAsync_ConcurrentServers_InitializeIndependently()
{
const int serverCount = 16;
var cancellationToken = TestContext.Current.CancellationToken;
using var starts = new Barrier(serverCount);
var clients = Enumerable.Range(0, serverCount).Select(_ => LoopbackMcpClient.Create(_ => null)).ToArray();
var providers = Enumerable.Range(0, serverCount).Select(index =>
{
var provider = CreateMockServerProvider($"external{index}");
provider.CreateClientAsync(Arg.Any<McpClientOptions>(), Arg.Any<CancellationToken>()).Returns(_ =>
{
Assert.True(starts.SignalAndWait(TimeSpan.FromSeconds(30), cancellationToken));
return clients[index];
});
return provider;
}).ToArray();
await using var strategy = CreateMockStrategy(providers);
var tasks = Enumerable.Range(0, serverCount).Select(index => Task.Factory.StartNew(
() => strategy.GetOrCreateClientAsync($"external{index}", cancellationToken: cancellationToken),
cancellationToken, TaskCreationOptions.LongRunning, TaskScheduler.Default).Unwrap()).ToArray();
var results = await Task.WhenAll(tasks);
for (var index = 0; index < serverCount; index++)
{
Assert.Same(clients[index], results[index]);
Assert.Same(clients[index], await strategy.GetOrCreateClientAsync(
$"EXTERNAL{index}", cancellationToken: cancellationToken));
await providers[index].Received(1).CreateClientAsync(Arg.Any<McpClientOptions>(), Arg.Any<CancellationToken>());
}
}
[Theory]
[InlineData(true)]
[InlineData(false)]
public async Task GetOrCreateClientAsync_CallerCancellation_DoesNotCancelSharedCreation(bool cancelFirstCaller)
{
await using var client = LoopbackMcpClient.Create(_ => null);
var creation = new TaskCompletionSource<McpClient>(TaskCreationOptions.RunContinuationsAsynchronously);
var provider = CreateMockServerProvider("external");
var initializationToken = CancellationToken.None;
var creations = 0;
provider.CreateClientAsync(Arg.Any<McpClientOptions>(), Arg.Any<CancellationToken>()).Returns(call =>
{
var token = call.Arg<CancellationToken>();
if (Interlocked.Increment(ref creations) == 1)
{
initializationToken = token;
}
return creation.Task.WaitAsync(token);
});
await using var strategy = CreateMockStrategy(provider);
using var callerCancellation = CancellationTokenSource.CreateLinkedTokenSource(TestContext.Current.CancellationToken);
var firstCall = strategy.GetOrCreateClientAsync("external",
cancellationToken: cancelFirstCaller ? callerCancellation.Token : TestContext.Current.CancellationToken);
var secondCall = strategy.GetOrCreateClientAsync("EXTERNAL",
cancellationToken: cancelFirstCaller ? TestContext.Current.CancellationToken : callerCancellation.Token);
try
{
await callerCancellation.CancelAsync();
await Assert.ThrowsAnyAsync<OperationCanceledException>(() => cancelFirstCaller ? firstCall : secondCall);
Assert.False(initializationToken.IsCancellationRequested);
}
finally
{
creation.TrySetResult(client);
}
Assert.Same(client, await (cancelFirstCaller ? secondCall : firstCall));
Assert.Same(client, await strategy.GetOrCreateClientAsync("external",
cancellationToken: TestContext.Current.CancellationToken));
Assert.Equal(1, creations);
}
[Theory]
[InlineData(false)]
[InlineData(true)]
public async Task GetOrCreateClientAsync_FailedCreation_AllowsRetry(bool canceled)
{
await using var client = LoopbackMcpClient.Create(_ => null);
var failure = new InvalidOperationException("Client creation failed.");
var provider = CreateMockServerProvider("external");
var creations = 0;
provider.CreateClientAsync(Arg.Any<McpClientOptions>(), Arg.Any<CancellationToken>()).Returns(_ =>
Interlocked.Increment(ref creations) == 1
? canceled
? Task.FromCanceled<McpClient>(new CancellationToken(true))
: Task.FromException<McpClient>(failure)
: Task.FromResult(client));
await using var strategy = CreateMockStrategy(provider);
if (canceled)
{
await Assert.ThrowsAnyAsync<OperationCanceledException>(() =>
strategy.GetOrCreateClientAsync("external", cancellationToken: TestContext.Current.CancellationToken));
}
else
{
Assert.Same(failure, await Assert.ThrowsAsync<InvalidOperationException>(() =>
strategy.GetOrCreateClientAsync("external", cancellationToken: TestContext.Current.CancellationToken)));
}
Assert.Same(client, await strategy.GetOrCreateClientAsync("external",
cancellationToken: TestContext.Current.CancellationToken));
Assert.Equal(2, creations);
}
[Fact]
public async Task GetOrCreateClientAsync_ConcurrentFailure_AllowsFreshAttempt()
{
await using var client = LoopbackMcpClient.Create(_ => null);
var creation = new TaskCompletionSource<McpClient>(TaskCreationOptions.RunContinuationsAsynchronously);
var failure = new InvalidOperationException("Shared initialization failed.");
var provider = CreateMockServerProvider("external");
var attempts = 0;
provider.CreateClientAsync(Arg.Any<McpClientOptions>(), Arg.Any<CancellationToken>()).Returns(_ =>
Interlocked.Increment(ref attempts) == 1 ? creation.Task : Task.FromResult(client));
await using var strategy = CreateMockStrategy(provider);
var callers = Enumerable.Range(0, 8).Select(_ => strategy.GetOrCreateClientAsync(
"external", cancellationToken: TestContext.Current.CancellationToken)).ToArray();
creation.SetException(failure);
Assert.Same(failure, await Assert.ThrowsAsync<InvalidOperationException>(() => Task.WhenAll(callers)));
Assert.All(callers, caller => Assert.True(caller.IsFaulted));
Assert.Same(client, await strategy.GetOrCreateClientAsync(
"external", cancellationToken: TestContext.Current.CancellationToken));
Assert.Equal(2, attempts);
}
[Fact]
public async Task GetOrCreateClientAsync_CanceledBeforeCreation_DoesNotStartProvider()
{
var provider = CreateMockServerProvider("external");
await using var strategy = CreateMockStrategy(provider);
using var cancellation = new CancellationTokenSource();
await cancellation.CancelAsync();
await Assert.ThrowsAnyAsync<OperationCanceledException>(() =>
strategy.GetOrCreateClientAsync("external", cancellationToken: cancellation.Token));
await provider.DidNotReceive().CreateClientAsync(Arg.Any<McpClientOptions>(), Arg.Any<CancellationToken>());
}
[Theory]
[InlineData(false)]
[InlineData(true)]
public async Task DisposeAsync_ConcurrentCalls_WaitForInFlightCreation(bool honorsCancellation)
{
await using var client = LoopbackMcpClient.Create(_ => null, out var tracker);
var creation = new TaskCompletionSource<McpClient>(TaskCreationOptions.RunContinuationsAsynchronously);
var initializationToken = CancellationToken.None;
var provider = CreateMockServerProvider("external");
provider.CreateClientAsync(Arg.Any<McpClientOptions>(), Arg.Any<CancellationToken>()).Returns(call =>
{
initializationToken = call.Arg<CancellationToken>();
return honorsCancellation ? creation.Task.WaitAsync(initializationToken) : creation.Task;
});
await using var strategy = CreateMockStrategy(provider);
var pendingClient = strategy.GetOrCreateClientAsync("external", cancellationToken: TestContext.Current.CancellationToken);
var firstDisposal = strategy.DisposeAsync().AsTask();
var secondDisposal = strategy.DisposeAsync().AsTask();
try
{
Assert.Same(firstDisposal, secondDisposal);
if (!honorsCancellation)
{
Assert.False(firstDisposal.IsCompleted);
}
else
{
await Assert.ThrowsAnyAsync<OperationCanceledException>(() =>
pendingClient.WaitAsync(TimeSpan.FromSeconds(30), TestContext.Current.CancellationToken));
}
await Assert.ThrowsAsync<ObjectDisposedException>(() =>
strategy.GetOrCreateClientAsync("external", cancellationToken: TestContext.Current.CancellationToken));
}
finally
{
creation.TrySetResult(client);
}
if (honorsCancellation)
{
await Assert.ThrowsAnyAsync<OperationCanceledException>(() => pendingClient);
}
else
{
Assert.Same(client, await pendingClient);
}
await Task.WhenAll(firstDisposal, secondDisposal);
Assert.True(initializationToken.IsCancellationRequested);
Assert.Equal(honorsCancellation ? 0 : 1, tracker.DisposeCount);
}
[Fact]
public async Task FindServerProvider_WithNullName_ThrowsArgumentNullException()
{
@@ -409,12 +409,13 @@ public class CompositeToolLoaderTests
public async Task CallToolHandler_ConcurrentCallsWithoutListingFirst_InitializesOnlyOnce()
{
var logger = Substitute.For<ILogger<CompositeToolLoader>>();
var initialization = new TaskCompletionSource<ListToolsResult>(TaskCreationOptions.RunContinuationsAsynchronously);
// Setup a loader that has the tool we want to call
var mockLoader = Substitute.For<IToolLoader>();
var testTool = CreateTestTool("test-tool");
mockLoader.ListToolsHandler(Arg.Any<RequestContext<ListToolsRequestParams>>(), Arg.Any<CancellationToken>())
.Returns(new ListToolsResult { Tools = [testTool] });
.Returns(_ => new ValueTask<ListToolsResult>(initialization.Task));
var expectedResult = new CallToolResult
{
@@ -437,6 +438,8 @@ public class CompositeToolLoaderTests
tasks.Add(toolLoader.CallToolHandler(callRequest, TestContext.Current.CancellationToken).AsTask());
}
Assert.All(tasks, task => Assert.False(task.IsCompleted));
initialization.SetResult(new ListToolsResult { Tools = [testTool] });
var results = await Task.WhenAll(tasks);
// Verify all calls succeeded
@@ -603,6 +606,43 @@ public class CompositeToolLoaderTests
await mockLoader.Received(1).DisposeAsync();
}
[Theory]
[InlineData(false)]
[InlineData(true)]
public async Task DisposeAsync_ConcurrentCalls_AwaitOneCleanup(bool cleanupFails)
{
var cleanup = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
var mockLoader = Substitute.For<IToolLoader>();
mockLoader.DisposeAsync().Returns(_ => new ValueTask(cleanup.Task));
var logger = Substitute.For<ILogger<CompositeToolLoader>>();
var compositeLoader = new CompositeToolLoader([mockLoader], logger);
var disposals = Enumerable.Range(0, 16).Select(_ => compositeLoader.DisposeAsync().AsTask()).ToArray();
try
{
Assert.All(disposals, disposal =>
{
Assert.Same(disposals[0], disposal);
Assert.False(disposal.IsCompleted);
});
}
finally
{
if (cleanupFails)
{
cleanup.TrySetException(new InvalidOperationException("Cleanup failed."));
}
else
{
cleanup.TrySetResult();
}
await Task.WhenAll(disposals);
}
await compositeLoader.DisposeAsync();
_ = mockLoader.Received(1).DisposeAsync();
}
[Fact]
public async Task DisposeAsync_WithSingleChildLoader_ShouldDisposeChild()
{
@@ -679,18 +679,22 @@ public sealed class NamespaceToolLoaderTests : IAsyncDisposable
// Arrange
var loader = new NamespaceToolLoader(new CommandContextAccessor(), _commandFactory, _configuration, _logger);
var toolName = GetFirstAvailableNamespace();
var cancellationToken = TestContext.Current.CancellationToken;
using var startBarrier = new Barrier(10);
// Act - Simulate concurrent access
var tasks = Enumerable.Range(0, 10).Select(async _ =>
{
var request = McpTestUtilities.CreateToolCallRequest(toolName, new Dictionary<string, object?>
var tasks = Enumerable.Range(0, 10).Select(_ => Task.Factory.StartNew(async () =>
{
["learn"] = true,
["intent"] = "concurrent test"
});
var request = McpTestUtilities.CreateToolCallRequest(toolName, new Dictionary<string, object?>
{
["learn"] = true,
["intent"] = "concurrent test"
});
Assert.True(startBarrier.SignalAndWait(TimeSpan.FromSeconds(30), cancellationToken));
return await loader.CallToolHandler(request, TestContext.Current.CancellationToken);
});
return await loader.CallToolHandler(request, cancellationToken);
}, cancellationToken, TaskCreationOptions.LongRunning, TaskScheduler.Default).Unwrap())
.ToArray();
var results = await Task.WhenAll(tasks);
@@ -710,8 +714,150 @@ public sealed class NamespaceToolLoaderTests : IAsyncDisposable
});
}
[Theory]
[InlineData(true)]
[InlineData(false)]
public async Task GetChildToolList_ConcurrentColdCalls_PublishStableLists(bool sameNamespace)
{
const int workerCount = 16;
var cancellationToken = TestContext.Current.CancellationToken;
var namespaceNames = Enumerable.Range(0, sameNamespace ? 1 : workerCount)
.Select(index => $"namespace{index}")
.ToArray();
using var misses = new Barrier(workerCount);
var synchronizeMisses = true;
var factory = CreateNamespaceCommandFactory(namespaceNames, commands =>
{
// Hold every cold caller after its cache miss, before any result can be published.
if (Volatile.Read(ref synchronizeMisses))
{
Assert.True(misses.SignalAndWait(TimeSpan.FromSeconds(30), cancellationToken));
}
return commands;
});
var loader = new NamespaceToolLoader(new CommandContextAccessor(), factory, _configuration, _logger);
var tasks = Enumerable.Range(0, workerCount).Select(index => Task.Factory.StartNew(() =>
{
var namespaceName = namespaceNames[index % namespaceNames.Length];
var request = McpTestUtilities.CreateToolCallRequest(namespaceName);
IReadOnlyList<Tool> tools = loader.GetChildToolList(request, namespaceName);
return (NamespaceName: namespaceName, Tools: tools);
}, cancellationToken, TaskCreationOptions.LongRunning, TaskScheduler.Default))
.ToArray();
var results = await Task.WhenAll(tasks);
Volatile.Write(ref synchronizeMisses, false);
foreach (var result in results)
{
Assert.Equal($"{result.NamespaceName}_get", Assert.Single(result.Tools).Name);
for (var pass = 0; pass < 3; pass++)
{
var namespaceName = result.NamespaceName.ToUpperInvariant();
var request = McpTestUtilities.CreateToolCallRequest(namespaceName);
Assert.Same(result.Tools, loader.GetChildToolList(request, namespaceName));
}
}
factory.Received(workerCount).GroupCommands(Arg.Any<string[]>());
}
[Fact]
public async Task DisposeAsync_ClearsCaches()
public async Task GetChildToolList_LoadsOnlyRequestedNamespaceAndReturnsReadOnlyList()
{
var factory = CreateNamespaceCommandFactory(["storage", "keyvault"]);
var loader = new NamespaceToolLoader(new CommandContextAccessor(), factory, _configuration, _logger);
var namespaces = await loader.ListToolsHandler(
McpTestUtilities.CreateToolListRequest(), TestContext.Current.CancellationToken);
Assert.Equal(["storage", "keyvault"], namespaces.Tools.Select(tool => tool.Name));
factory.DidNotReceive().GroupCommands(Arg.Any<string[]>());
var request = McpTestUtilities.CreateToolCallRequest("storage");
var tools = loader.GetChildToolList(request, "storage");
Assert.Equal("storage_get", Assert.Single(tools).Name);
Assert.True(Assert.IsAssignableFrom<ICollection<Tool>>(tools).IsReadOnly);
Assert.Same(tools, loader.GetChildToolList(request, "STORAGE"));
factory.Received(1).GroupCommands(Arg.Any<string[]>());
factory.DidNotReceive().GroupCommands(Arg.Is<string[]>(names => names[0] == "keyvault"));
}
[Theory]
[InlineData(false)]
[InlineData(true)]
public void GetChildToolList_RetriesAfterCommandLookupFailure(bool returnNull)
{
var attempts = 0;
var failure = new InvalidOperationException("Command lookup failed.");
var factory = CreateNamespaceCommandFactory(["storage"], commands =>
{
if (Interlocked.Increment(ref attempts) == 1)
{
if (returnNull)
{
return null!;
}
throw failure;
}
return commands;
});
var loader = new NamespaceToolLoader(new CommandContextAccessor(), factory, _configuration, _logger);
var request = McpTestUtilities.CreateToolCallRequest("storage");
if (returnNull)
{
Assert.Empty(loader.GetChildToolList(request, "storage"));
}
else
{
Assert.Same(failure, Assert.Throws<InvalidOperationException>(
() => loader.GetChildToolList(request, "storage")));
}
var tools = loader.GetChildToolList(request, "storage");
Assert.Equal("storage_get", Assert.Single(tools).Name);
Assert.Same(tools, loader.GetChildToolList(request, "storage"));
Assert.Equal(2, attempts);
}
[Fact]
public void GetChildToolList_RetriesAfterSchemaGenerationFailure()
{
var factory = CreateNamespaceCommandFactory(["storage"]);
var command = factory.RootGroup.SubGroup[0].Commands["storage_get"].Command;
var underlyingCommand = command.GetCommand();
var attempts = 0;
var failure = new InvalidOperationException("Schema generation failed.");
command.GetCommand().Returns(_ =>
{
if (Interlocked.Increment(ref attempts) == 1)
{
throw failure;
}
return underlyingCommand;
});
var loader = new NamespaceToolLoader(new CommandContextAccessor(), factory, _configuration, _logger);
var request = McpTestUtilities.CreateToolCallRequest("storage");
Assert.Same(failure, Assert.Throws<InvalidOperationException>(
() => loader.GetChildToolList(request, "storage")));
var tools = loader.GetChildToolList(request, "storage");
Assert.Equal("storage_get", Assert.Single(tools).Name);
Assert.Same(tools, loader.GetChildToolList(request, "storage"));
factory.Received(2).GroupCommands(Arg.Any<string[]>());
}
[Fact]
public async Task DisposeAsync_CanBeCalledRepeatedly()
{
// Arrange
var loader = new NamespaceToolLoader(new CommandContextAccessor(), _commandFactory, _configuration, _logger);
@@ -728,9 +874,7 @@ public sealed class NamespaceToolLoaderTests : IAsyncDisposable
// Act
await loader.DisposeAsync();
// Assert - No exception should be thrown
// Cache clearing is internal, but disposal should complete successfully
await loader.DisposeAsync();
}
[Fact]
@@ -1397,6 +1541,31 @@ public sealed class NamespaceToolLoaderTests : IAsyncDisposable
// Helper methods
private static ICommandFactory CreateNamespaceCommandFactory(
string[] namespaceNames,
Func<IReadOnlyDictionary<string, CommandRegistration>, IReadOnlyDictionary<string, CommandRegistration>>? getCommands = null)
{
var root = new CommandGroup("root", "Root command group");
foreach (var namespaceName in namespaceNames)
{
var group = new CommandGroup(namespaceName, "Namespace commands");
var commandName = $"{namespaceName}_get";
group.AddCommand(commandName, new CommandRegistration(CreateRoutingCommand(commandName), namespaceName));
root.AddSubGroup(group);
}
var factory = Substitute.For<ICommandFactory>();
factory.RootGroup.Returns(root);
factory.GroupCommands(Arg.Any<string[]>()).Returns(call =>
{
var namespaceName = call.Arg<string[]>()[0];
var commands = root.SubGroup.Single(group =>
string.Equals(group.Name, namespaceName, StringComparison.OrdinalIgnoreCase)).Commands;
return getCommands == null ? commands : getCommands(commands);
});
return factory;
}
private NamespaceToolLoader CreateRoutingLoader(
Dictionary<string, IBaseCommand> commands,
ServerRuntimeConfiguration? configuration = null,
@@ -797,6 +797,49 @@ public class RegistryToolLoaderTests
_ = mockDiscoveryStrategy.Received(1).GetOrCreateClientAsync("server-3", Arg.Any<McpClientOptions>(), Arg.Any<CancellationToken>());
}
[Fact]
public async Task ListAndCallTools_ConcurrentInitialization_PublishesCompleteMap()
{
await using var client = new MockMcpClientBuilder()
.AddTool("test-tool", "Test tool", "Tool response")
.Build();
var creation = new TaskCompletionSource<McpClient>(TaskCreationOptions.RunContinuationsAsynchronously);
var provider = Substitute.For<IMcpServerProvider>();
provider.CreateMetadata().Returns(new McpServerMetadata("external", "external", "External server"));
var discovery = Substitute.For<IMcpDiscoveryStrategy>();
discovery.DiscoverServersAsync(Arg.Any<CancellationToken>()).Returns([provider]);
discovery.GetOrCreateClientAsync("external", Arg.Any<McpClientOptions>(), Arg.Any<CancellationToken>())
.Returns(creation.Task);
await using var loader = CreateToolLoader(discovery);
var lists = Enumerable.Range(0, 8).Select(_ => loader.ListToolsHandler(
McpTestUtilities.CreateToolListRequest(), TestContext.Current.CancellationToken).AsTask()).ToArray();
var calls = Enumerable.Range(0, 8).Select(_ => loader.CallToolHandler(
McpTestUtilities.CreateToolCallRequest("test-tool"), TestContext.Current.CancellationToken).AsTask()).ToArray();
try
{
Assert.All(lists, task => Assert.False(task.IsCompleted));
Assert.All(calls, task => Assert.False(task.IsCompleted));
}
finally
{
creation.TrySetResult(client);
}
var listResults = await Task.WhenAll(lists);
var callResults = await Task.WhenAll(calls);
Assert.All(listResults, result => Assert.Equal("test-tool", Assert.Single(result.Tools).Name));
Assert.All(callResults, result =>
{
Assert.NotEqual(true, result.IsError);
Assert.Equal("Tool response", Assert.IsType<TextContentBlock>(Assert.Single(result.Content)).Text);
});
await discovery.Received(1).DiscoverServersAsync(Arg.Any<CancellationToken>());
await discovery.Received(1).GetOrCreateClientAsync(
"external", Arg.Any<McpClientOptions>(), Arg.Any<CancellationToken>());
}
[Fact]
public async Task ListToolsHandler_WhenCancellationOccursDuringInitialization_AllowsRetry()
{
@@ -864,6 +864,66 @@ public class SingleProxyToolLoaderTests
Assert.Equal(content1, content2);
}
[Theory]
[InlineData(null)]
[InlineData(StructuredOutputMode.Duplicated)]
[InlineData(StructuredOutputMode.Compact)]
public async Task CallToolHandler_ConcurrentRootLearning_PublishesOneSnapshot(StructuredOutputMode? mode)
{
var (toolLoader, factory, discovery) = CreateToolLoader(
useRealDiscovery: false, new ServerRuntimeConfiguration { StructuredOutputMode = mode });
factory.RootGroup.Returns(new CommandGroup("root", "Root command group"));
var firstDiscovery = new TaskCompletionSource<IEnumerable<IMcpServerProvider>>(TaskCreationOptions.RunContinuationsAsynchronously);
var secondDiscovery = new TaskCompletionSource<IEnumerable<IMcpServerProvider>>(TaskCreationOptions.RunContinuationsAsynchronously);
var discoveries = 0;
discovery.DiscoverServersAsync(Arg.Any<CancellationToken>()).Returns(_ =>
Interlocked.Increment(ref discoveries) == 1 ? firstDiscovery.Task : secondDiscovery.Task);
var firstProvider = Substitute.For<IMcpServerProvider>();
firstProvider.CreateMetadata().Returns(new McpServerMetadata("first", "first", "First snapshot"));
var secondProvider = Substitute.For<IMcpServerProvider>();
secondProvider.CreateMetadata().Returns(new McpServerMetadata("second", "second", "Second snapshot"));
var firstRequest = McpTestUtilities.CreateToolCallRequest("azure", new Dictionary<string, object?> { ["learn"] = true });
var secondRequest = McpTestUtilities.CreateToolCallRequest("azure", new Dictionary<string, object?> { ["learn"] = true });
var firstCall = toolLoader.CallToolHandler(firstRequest, TestContext.Current.CancellationToken).AsTask();
var secondCall = toolLoader.CallToolHandler(secondRequest, TestContext.Current.CancellationToken).AsTask();
firstDiscovery.SetResult([firstProvider]);
var firstResult = await firstCall;
secondDiscovery.SetResult([secondProvider]);
var secondResult = await secondCall;
var cachedResult = await toolLoader.CallToolHandler(firstRequest, TestContext.Current.CancellationToken);
var firstText = Assert.IsType<TextContentBlock>(Assert.Single(firstResult.Content)).Text;
var firstPayload = firstResult.StructuredContent?.GetRawText() ?? firstText;
Assert.Contains("first", firstPayload);
Assert.DoesNotContain("second", firstPayload);
Assert.Equal(firstText, Assert.IsType<TextContentBlock>(Assert.Single(secondResult.Content)).Text);
Assert.Equal(firstText, Assert.IsType<TextContentBlock>(Assert.Single(cachedResult.Content)).Text);
Assert.Equal(firstResult.StructuredContent?.GetRawText(), secondResult.StructuredContent?.GetRawText());
Assert.Equal(2, discoveries);
}
[Fact]
public async Task CallToolHandler_RootDiscoveryFailure_AllowsRetry()
{
var (toolLoader, factory, discovery) = CreateToolLoader(useRealDiscovery: false);
factory.RootGroup.Returns(new CommandGroup("root", "Root command group"));
var attempts = 0;
var failure = new InvalidOperationException("Discovery failed.");
discovery.DiscoverServersAsync(Arg.Any<CancellationToken>()).Returns(_ =>
Interlocked.Increment(ref attempts) == 1
? Task.FromException<IEnumerable<IMcpServerProvider>>(failure)
: Task.FromResult<IEnumerable<IMcpServerProvider>>([]));
var request = McpTestUtilities.CreateToolCallRequest("azure", new Dictionary<string, object?> { ["learn"] = true });
Assert.Same(failure, await Assert.ThrowsAsync<InvalidOperationException>(() =>
toolLoader.CallToolHandler(request, TestContext.Current.CancellationToken).AsTask()));
var result = await toolLoader.CallToolHandler(request, TestContext.Current.CancellationToken);
Assert.NotEmpty(result.Content);
Assert.Equal(2, attempts);
}
[Fact]
public void SingleProxyToolLoader_Constructor_ThrowsOnNullArguments()
{
@@ -21,7 +21,10 @@ public abstract class BaseDiscoveryStrategy(ILogger logger) : IMcpDiscoveryStrat
/// </summary>
protected readonly Dictionary<string, McpClient> _clientCache = new(StringComparer.OrdinalIgnoreCase);
private bool _disposed = false;
private readonly object _clientCacheLock = new();
private readonly Dictionary<string, Task<McpClient>> _clientCreations = new(StringComparer.OrdinalIgnoreCase);
private readonly CancellationTokenSource _clientCreationCancellation = new();
private TaskCompletionSource? _disposeCompletion;
/// <inheritdoc/>
public abstract Task<IEnumerable<IMcpServerProvider>> DiscoverServersAsync(CancellationToken cancellationToken);
@@ -57,31 +60,117 @@ public abstract class BaseDiscoveryStrategy(ILogger logger) : IMcpDiscoveryStrat
throw new ArgumentNullException(nameof(name), "Server name cannot be null or empty.");
}
if (_clientCache.TryGetValue(name, out var client))
Task<McpClient> creationTask;
TaskCompletionSource<McpClient>? creation = null;
lock (_clientCacheLock)
{
return client;
ObjectDisposedException.ThrowIf(_disposeCompletion != null, this);
if (_clientCache.TryGetValue(name, out var client))
{
return client;
}
cancellationToken.ThrowIfCancellationRequested();
if (!_clientCreations.TryGetValue(name, out creationTask!))
{
creation = new TaskCompletionSource<McpClient>(TaskCreationOptions.RunContinuationsAsynchronously);
creationTask = creation.Task;
_clientCreations.Add(name, creationTask);
}
}
var serverProvider = await FindServerProviderAsync(name, cancellationToken);
client = await serverProvider.CreateClientAsync(clientOptions ?? new McpClientOptions(), cancellationToken);
_clientCache[name] = client;
if (creation != null)
{
_ = CreateClientAsync(name, clientOptions, creation);
}
return client;
return await creationTask.WaitAsync(cancellationToken);
}
private async Task CreateClientAsync(string name, McpClientOptions? clientOptions, TaskCompletionSource<McpClient> creation)
{
try
{
var cancellationToken = _clientCreationCancellation.Token;
cancellationToken.ThrowIfCancellationRequested();
var serverProvider = await FindServerProviderAsync(name, cancellationToken);
var client = await serverProvider.CreateClientAsync(clientOptions ?? new McpClientOptions(), cancellationToken);
lock (_clientCacheLock)
{
_clientCache[name] = client;
_clientCreations.Remove(name);
}
creation.SetResult(client);
}
catch (OperationCanceledException ex)
{
_logger.LogDebug(ex, "Client initialization canceled for MCP server {ServerName}.", name);
lock (_clientCacheLock)
{
_clientCreations.Remove(name);
}
creation.SetCanceled(ex.CancellationToken);
}
catch (Exception ex)
{
_logger.LogError(ex, "Failed to initialize client for MCP server {ServerName}.", name);
lock (_clientCacheLock)
{
_clientCreations.Remove(name);
}
creation.SetException(ex);
}
}
/// <summary>
/// Disposes all cached MCP clients with double disposal protection.
/// Cancels pending initialization and disposes cached clients once, sharing completion with concurrent callers.
/// </summary>
public async ValueTask DisposeAsync()
public ValueTask DisposeAsync()
{
if (_disposed)
TaskCompletionSource completion;
Task<McpClient>[] creations;
lock (_clientCacheLock)
{
return;
if (_disposeCompletion != null)
{
return new ValueTask(_disposeCompletion.Task);
}
completion = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
_disposeCompletion = completion;
creations = [.. _clientCreations.Values];
}
_ = DisposeAsyncOnce(creations, completion);
return new ValueTask(completion.Task);
}
private async Task DisposeAsyncOnce(Task<McpClient>[] creations, TaskCompletionSource completion)
{
try
{
// First, let derived classes dispose their resources (isolated from base cleanup)
try
{
await _clientCreationCancellation.CancelAsync();
}
catch (Exception ex)
{
_logger.LogError(ex, "Failed to cancel pending MCP client initialization.");
}
try
{
await Task.WhenAll(creations);
}
catch (OperationCanceledException) when (_clientCreationCancellation.IsCancellationRequested)
{
_logger.LogDebug("Pending MCP client initialization was canceled during disposal.");
}
catch (Exception ex)
{
_logger.LogWarning(ex, "MCP client initialization failed during disposal.");
}
try
{
await DisposeAsyncCore();
@@ -91,8 +180,14 @@ public abstract class BaseDiscoveryStrategy(ILogger logger) : IMcpDiscoveryStrat
_logger.LogError(ex, "Error occurred while disposing derived resources in discovery strategy {StrategyType}", GetType().Name);
}
// Then dispose our own critical resources using best-effort approach
var clientDisposalTasks = _clientCache.Values.Select(async client =>
McpClient[] clients;
lock (_clientCacheLock)
{
clients = [.. _clientCache.Values];
_clientCache.Clear();
}
var clientDisposalTasks = clients.Select(async client =>
{
try
{
@@ -103,9 +198,7 @@ public abstract class BaseDiscoveryStrategy(ILogger logger) : IMcpDiscoveryStrat
_logger.LogError(ex, "Failed to dispose MCP client in discovery strategy {StrategyType}", GetType().Name);
}
});
await Task.WhenAll(clientDisposalTasks);
_clientCache.Clear();
}
catch (Exception ex)
{
@@ -115,7 +208,8 @@ public abstract class BaseDiscoveryStrategy(ILogger logger) : IMcpDiscoveryStrat
}
finally
{
_disposed = true;
_clientCreationCancellation.Dispose();
completion.SetResult();
}
}
@@ -26,6 +26,10 @@ public interface IMcpDiscoveryStrategy : IAsyncDisposable
/// <summary>
/// Gets an MCP client for the specified server.
/// </summary>
/// <remarks>
/// Concurrent calls for the same server share initialization. Cancellation stops only the caller's wait;
/// shared initialization is canceled when the strategy is disposed. Failed initialization can be retried.
/// </remarks>
/// <param name="name">The name of the server to get a client for.</param>
/// <param name="clientOptions">Optional client configuration options. If null, default options are used.</param>
/// <param name="cancellationToken">A cancellation token.</param>
@@ -42,7 +42,7 @@ public abstract class BaseToolLoader(ILogger logger) : IToolLoader
}
}
private bool _disposed = false;
private TaskCompletionSource? _disposeCompletion;
/// <summary>
/// Handles requests to list all tools available in the MCP server.
@@ -153,15 +153,29 @@ public abstract class BaseToolLoader(ILogger logger) : IToolLoader
}
/// <summary>
/// Disposes resources owned by this tool loader with double disposal protection.
/// Disposes owned resources once, sharing cleanup completion with concurrent callers.
/// </summary>
public async ValueTask DisposeAsync()
public ValueTask DisposeAsync()
{
if (_disposed)
var existing = Volatile.Read(ref _disposeCompletion);
if (existing != null)
{
return;
return new ValueTask(existing.Task);
}
var completion = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
existing = Interlocked.CompareExchange(ref _disposeCompletion, completion, null);
if (existing != null)
{
return new ValueTask(existing.Task);
}
_ = DisposeAsyncOnce(completion);
return new ValueTask(completion.Task);
}
private async Task DisposeAsyncOnce(TaskCompletionSource completion)
{
try
{
await DisposeAsyncCore();
@@ -174,7 +188,7 @@ public abstract class BaseToolLoader(ILogger logger) : IToolLoader
}
finally
{
_disposed = true;
completion.SetResult();
}
}
@@ -21,7 +21,7 @@ public sealed class CompositeToolLoader(IEnumerable<IToolLoader> toolLoaders, IL
internal readonly IEnumerable<IToolLoader> _toolLoaders = InitializeToolLoaders(toolLoaders);
private readonly Dictionary<string, IToolLoader> _toolLoaderMap = [];
private readonly SemaphoreSlim _initializationSemaphore = new(1, 1);
private bool _isInitialized = false;
private volatile bool _isInitialized = false;
// Server-side performance cache only. This should not be treated as protocol freshness metadata.
private List<Tool>? _cachedTools;
@@ -1,6 +1,7 @@
// Copyright (c) Microsoft Corporation.
// Licensed under the MIT License.
using System.Collections.Frozen;
using System.CommandLine;
using System.Diagnostics;
using System.Net;
@@ -45,7 +46,9 @@ public sealed class NamespaceToolLoader(
private readonly ICommandFactory _commandFactory = commandFactory ?? throw new ArgumentNullException(nameof(commandFactory));
private readonly IOptions<ServerRuntimeConfiguration> _configuration = configuration ?? throw new ArgumentNullException(nameof(configuration));
private readonly Lazy<IReadOnlyList<string>> _availableNamespaces = new(() =>
private readonly Lazy<(
IReadOnlyList<string> Names,
FrozenDictionary<string, Func<IReadOnlyList<Tool>>> ToolLists)> _namespaceCatalog = new(() =>
{
IEnumerable<CommandGroup> allSubGroups = commandFactory.RootGroup.SubGroup;
@@ -58,10 +61,13 @@ public sealed class NamespaceToolLoader(
configuration.Value.Namespace.Contains(group.Name, StringComparer.OrdinalIgnoreCase));
}
return [.. allSubGroups.Select(group => group.Name)];
IReadOnlyList<string> names = [.. allSubGroups.Select(group => group.Name)];
return (names, names.ToFrozenDictionary(
name => name,
name => CreateChildToolListAccessor(commandFactory, configuration, logger, name),
StringComparer.OrdinalIgnoreCase));
});
private readonly Dictionary<string, List<Tool>> _cachedToolLists = new(StringComparer.OrdinalIgnoreCase);
private ListToolsResult? _cachedListToolsResult;
private bool StructuredOutputEnabled => _configuration.Value.StructuredOutputMode != null;
@@ -118,7 +124,7 @@ public sealed class NamespaceToolLoader(
return ValueTask.FromResult(_cachedListToolsResult);
}
var namespaces = _availableNamespaces.Value;
var namespaces = _namespaceCatalog.Value.Names;
var allToolsResponse = new ListToolsResult
{
Tools = []
@@ -176,7 +182,7 @@ public sealed class NamespaceToolLoader(
throw new ArgumentNullException(nameof(request.Params.Name), "Tool name cannot be null or empty.");
}
string tool = _availableNamespaces.Value.FirstOrDefault(namespaceName =>
string tool = _namespaceCatalog.Value.Names.FirstOrDefault(namespaceName =>
string.Equals(namespaceName, request.Params.Name, StringComparison.OrdinalIgnoreCase)) ?? request.Params.Name;
var args = request.Params.Arguments;
string? intent = null;
@@ -614,48 +620,61 @@ public sealed class NamespaceToolLoader(
}
/// <summary>
/// Gets the available tools from the namespace commands and caches the result for subsequent requests.
/// Gets the lazily cached, read-only tools for a namespace.
/// </summary>
internal List<Tool> GetChildToolList(RequestContext<CallToolRequestParams> request, string namespaceName)
internal IReadOnlyList<Tool> GetChildToolList(RequestContext<CallToolRequestParams> request, string namespaceName)
{
// Check cache first
if (_cachedToolLists.TryGetValue(namespaceName, out var cachedList))
{
return cachedList;
}
if (string.IsNullOrWhiteSpace(request.Params?.Name))
{
throw new ArgumentNullException(nameof(request.Params.Name), "Tool name cannot be null or empty.");
}
var namespaces = _availableNamespaces.Value;
if (!namespaces.Any(ns => string.Equals(ns, namespaceName, StringComparison.OrdinalIgnoreCase)))
var catalog = _namespaceCatalog.Value;
if (!catalog.ToolLists.TryGetValue(namespaceName, out var getTools))
{
Activity.Current?.SetTag(TagName.ToolArea, TagConstants.Unknown)
.SetTag(TagName.ToolName, TagConstants.Unknown);
var availableList = string.Join(", ", namespaces);
var availableList = string.Join(", ", catalog.Names);
throw new KeyNotFoundException($"The namespace '{namespaceName}' was not found. Available namespaces: {availableList}");
}
var namespaceCommands = _commandFactory.GroupCommands([namespaceName]);
if (namespaceCommands == null)
{
_logger.LogWarning("No commands found for namespace: {Namespace}", namespaceName);
return [];
}
var list = namespaceCommands
.Where(kvp => !_configuration.Value.ReadOnly || kvp.Value.Command.Metadata.ReadOnly)
.Where(kvp => !_configuration.Value.IsHttpMode || !kvp.Value.Command.Metadata.LocalRequired)
.Select(kvp => CreateToolFromCommand(kvp.Key, kvp.Value.Command))
.ToList();
// Cache for subsequent requests
_cachedToolLists[namespaceName] = list;
return list;
return getTools();
}
private static Func<IReadOnlyList<Tool>> CreateChildToolListAccessor(
ICommandFactory commandFactory,
IOptions<ServerRuntimeConfiguration> configuration,
ILogger<NamespaceToolLoader> logger,
string namespaceName)
{
IReadOnlyList<Tool>? cachedList = null;
return () =>
{
var existingList = Volatile.Read(ref cachedList);
if (existingList != null)
{
return existingList;
}
var namespaceCommands = commandFactory.GroupCommands([namespaceName]);
if (namespaceCommands == null)
{
logger.LogWarning("No commands found for namespace: {Namespace}", namespaceName);
return [];
}
IReadOnlyList<Tool> list = namespaceCommands
.Where(kvp => !configuration.Value.ReadOnly || kvp.Value.Command.Metadata.ReadOnly)
.Where(kvp => !configuration.Value.IsHttpMode || !kvp.Value.Command.Metadata.LocalRequired)
.Select(kvp => CreateToolFromCommand(kvp.Key, kvp.Value.Command))
.ToList()
.AsReadOnly();
// Publish only successful metadata so failed lookups and schema generation remain retryable.
return Interlocked.CompareExchange(ref cachedList, list, null) ?? list;
};
}
internal static Dictionary<string, JsonElement> GetParametersFromArgs(IDictionary<string, JsonElement>? args)
{
if (args == null)
@@ -699,7 +718,7 @@ public sealed class NamespaceToolLoader(
RequestContext<CallToolRequestParams> request,
string intent,
string namespaceName,
List<Tool> availableTools,
IReadOnlyList<Tool> availableTools,
CancellationToken cancellationToken)
{
#pragma warning disable MCP9005 // Sampling APIs remain for backward compatibility during migration.
@@ -780,13 +799,4 @@ public sealed class NamespaceToolLoader(
#pragma warning restore MCP9005
}
/// <summary>
/// Disposes resources owned by this tool loader.
/// Clears the cached tool lists dictionary.
/// </summary>
protected override ValueTask DisposeAsyncCore()
{
_cachedToolLists.Clear();
return ValueTask.CompletedTask;
}
}
@@ -30,7 +30,7 @@ public sealed class RegistryToolLoader(
private readonly List<McpClient> _discoveredClients = [];
private readonly Dictionary<McpClient, string?> _clientPrefixMap = [];
private readonly SemaphoreSlim _initializationSemaphore = new(1, 1);
private bool _isInitialized = false;
private volatile bool _isInitialized = false;
/// <summary>
/// Gets or sets the client options used when creating MCP clients.
@@ -0,0 +1,8 @@
// Copyright (c) Microsoft Corporation.
// Licensed under the MIT License.
using ModelContextProtocol.Protocol;
namespace Microsoft.Mcp.Core.Areas.Server.Commands.ToolLoading;
internal sealed record RootToolsSnapshot(IReadOnlyList<Tool> Tools, string Json);
@@ -57,7 +57,7 @@ public sealed class SingleProxyToolLoader(
private readonly JsonElement _toolSchema = BuildToolSchema(serverConfiguration!.Value.ShortName);
private bool StructuredOutputEnabled => _configuration.Value.StructuredOutputMode != null;
private (List<Tool> Tools, string Json)? _cachedTools;
private RootToolsSnapshot? _cachedTools;
private readonly ConcurrentDictionary<string, (List<ToolCommandInfo> Commands, string Json)> _cachedToolCommands = new(StringComparer.OrdinalIgnoreCase);
private readonly ConcurrentDictionary<string, IList<Tool>> _cachedCommandFactoryTools = new(StringComparer.OrdinalIgnoreCase);
private readonly ConcurrentDictionary<string, IList<Tool>> _cachedDiscoveryTools = new(StringComparer.OrdinalIgnoreCase);
@@ -207,11 +207,12 @@ public sealed class SingleProxyToolLoader(
/// Gets and caches all of the <see cref="IAreaSetup"/>'s available in the server.
/// </summary>
/// <param name="cancellationToken">A cancellation token.</param>
private async Task InitializeRootToolsCacheAsync(CancellationToken cancellationToken)
private async Task<RootToolsSnapshot> GetRootToolsSnapshotAsync(CancellationToken cancellationToken)
{
if (_cachedTools != null)
var existing = Volatile.Read(ref _cachedTools);
if (existing != null)
{
return;
return existing;
}
var tools = new List<Tool>();
@@ -262,8 +263,8 @@ public sealed class SingleProxyToolLoader(
}
var json = JsonSerializer.Serialize(tools.Select(t => new ToolCommandInfo(t, false)), ServerJsonContext.Default.IEnumerableToolCommandInfo);
_cachedTools = (tools, json);
return;
var snapshot = new RootToolsSnapshot(tools.AsReadOnly(), json);
return Interlocked.CompareExchange(ref _cachedTools, snapshot, null) ?? snapshot;
}
/// <summary>
@@ -386,22 +387,22 @@ public sealed class SingleProxyToolLoader(
{
activity?.SetTag(TagName.IsServerCommandInvoked, false)
.SetTag(TagName.IsLearn, true);
await InitializeRootToolsCacheAsync(cancellationToken);
var rootTools = await GetRootToolsSnapshotAsync(cancellationToken);
var contentText = $"""
Here are the available tools.
Next, identify the tool you want to learn about and run again with the "learn" argument and the "tool" name to get a list of available commands and their parameters.
{_cachedTools!.Value.Json}
{rootTools.Json}
""";
var learnResponse = StructuredOutputHelper.CreateCallToolResult(
_configuration.Value.StructuredOutputMode,
() => contentText,
() => AggregateStructuredOutput.CreateToolList(_cachedTools!.Value.Json));
() => AggregateStructuredOutput.CreateToolList(rootTools.Json));
var response = learnResponse;
if (SupportsSampling(request.Server) && !string.IsNullOrWhiteSpace(intent))
{
var toolName = await GetToolNameFromIntentAsync(request, intent, cancellationToken);
var availableTool = _cachedTools!.Value.Tools.FirstOrDefault(candidate =>
var toolName = await GetToolNameFromIntentAsync(request, intent, rootTools.Json, cancellationToken);
var availableTool = rootTools.Tools.FirstOrDefault(candidate =>
string.Equals(candidate.Name, toolName, StringComparison.OrdinalIgnoreCase));
if (availableTool != null)
{
@@ -849,7 +850,8 @@ public sealed class SingleProxyToolLoader(
}, cancellationToken: cancellationToken);
}
private async Task<string?> GetToolNameFromIntentAsync(RequestContext<CallToolRequestParams> request, string intent, CancellationToken cancellationToken)
private async Task<string?> GetToolNameFromIntentAsync(
RequestContext<CallToolRequestParams> request, string intent, string availableToolsJson, CancellationToken cancellationToken)
{
#pragma warning disable MCP9005 // Sampling APIs remain for backward compatibility during migration.
await NotifyProgressAsync(request, $"Learning about {_displayName} capabilities...", cancellationToken);
@@ -872,7 +874,7 @@ public sealed class SingleProxyToolLoader(
{intent}
Available Tools:
{_cachedTools!.Value.Json}
{availableToolsJson}
"""
}]
}
@@ -988,7 +990,7 @@ public sealed class SingleProxyToolLoader(
_cachedCommandFactoryTools.Clear();
_cachedDiscoveryTools.Clear();
_cachedToolCommands.Clear();
_cachedTools = null;
Interlocked.Exchange(ref _cachedTools, null);
await ValueTask.CompletedTask;
}
@@ -0,0 +1,5 @@
changes:
- section: "Bugs Fixed"
description: "Fixed concurrent first calls to namespace tools corrupting the tool-list cache and causing subsequent calls to fail until the server restarts."
- section: "Bugs Fixed"
description: "Fixed concurrent external MCP client initialization, single-tool discovery cache publication, and overlapping tool-loader disposal. Canceling a caller no longer cancels shared client initialization."