admin/Playa
1
0
Fork 0
forked from admin/VolcanicAgenticCore
Code Pull Requests Activity

feat(mcp): process-wide tool catalogue discovery

Browse Source 
McpProcess starts a server through the SDK stdio client under the startup
timeout and keeps a stderr tail for its errors. McpCatalogCache runs one shared
discovery per server config in a temp dir, with the environment from the process
spawn policy. It keeps successes for the process lifetime and failures for 60 s.

Deviation from the plan: the discovery process is disposed and its temp dir
deleted in the background, after the catalogue is returned. SDK 2.2 waits the
whole ShutdownTimeout (5 s) with stdin still open before it kills the process,
so an awaited dispose held every first activation for 5 s. A new test
(Discovery_returns_without_waiting_for_the_server_shutdown) covers this. Spec
section 3 gets the new GetAsync signature, the detached discovery and the
background cleanup. Section 5 now describes the SDK's shutdown order correctly.
This commit is contained in:
EugeneTes
2026-09-26 20:31:43 +02:00
parent 52f9d88575
commit ed026ac6c8
7 changed files with 456 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
extensions/Volcanic.AgenticCore.Extensions.Mcp/McpCatalog.cs Normal file
View File

@@ -0,0 +1,24 @@
using System.Text.Json;
using ModelContextProtocol.Protocol;
namespace Volcanic.AgenticCore.Extensions.Mcp;
/// <summary>One server tool: the name on the wire, the name the agent sees, and its schema verbatim.</summary>
internal sealed record McpCatalogTool(string McpName, string ExposedName, string? Description, string? Title, JsonElement InputSchema);
/// <summary>A server's <c>tools/list</c> result, discovered once per host process.</summary>
internal sealed record McpCatalog(string ServerName, string? ServerInfoName, string? ServerVersion, IReadOnlyList<McpCatalogTool> Tools)
{
public static McpCatalog From(string serverName, McpServerConfig config, Implementation? info, IReadOnlyList<Tool> tools)
{
var exposed = ToolNaming.Expose([.. tools.Select(t => t.Name)], config.ToolPrefix);
return new McpCatalog(serverName, info?.Name, info?.Version,
[.. tools.Select((t, i) => new McpCatalogTool(t.Name, exposed[i], t.Description, t.Annotations?.Title ?? t.Title, t.InputSchema.Clone()))]);
}
}
/// <summary>A server could not be started or didn't answer the handshake. The message includes its stderr tail.</summary>
internal sealed class McpServerStartException(string message, Exception? inner = null) : Exception(message, inner);
/// <summary>A running server exited; the next call starts a new process.</summary>
internal sealed class McpServerExitedException(string message) : Exception(message);

126
extensions/Volcanic.AgenticCore.Extensions.Mcp/McpCatalogCache.cs Normal file
View File

@@ -0,0 +1,126 @@
using Microsoft.Extensions.Logging;
using ModelContextProtocol.Client;
using Volcanic.AgenticCore.Abstractions.Processes;
namespace Volcanic.AgenticCore.Extensions.Mcp;
/// <summary>
/// Process-wide cache of <c>tools/list</c> results, keyed by server name and a digest of its unexpanded configuration.
/// Concurrent requests share one discovery. Success is kept for the process lifetime, failure for <see cref="FailureCooldown"/>.
/// </summary>
public sealed class McpCatalogCache
{
/// <summary>How long a failed discovery is remembered before the next request tries again.</summary>
public static readonly TimeSpan FailureCooldown = TimeSpan.FromSeconds(60);
/// <summary>Explicit process-wide state, like <c>ClaudeLoginCoordinator.Shared</c>.</summary>
public static McpCatalogCache Shared { get; } = new();
private readonly TimeProvider _time;
private readonly Lock _lock = new();
private readonly Dictionary<string, Task<McpCatalog>> _tasks = new(StringComparer.Ordinal);
private readonly Dictionary<string, DateTimeOffset> _failedAt = new(StringComparer.Ordinal);
/// <summary>A cache on the system clock. Hosts normally use <see cref="Shared"/>.</summary>
public McpCatalogCache() : this(TimeProvider.System) { }
internal McpCatalogCache(TimeProvider time) => _time = time;
/// <summary>
/// The server's catalogue, discovering it if nothing is cached or a failure's cooldown has passed. The shared discovery
/// runs detached from any caller's token; each caller waits with its own. Throws <see cref="McpServerStartException"/>.
/// </summary>
internal Task<McpCatalog> GetAsync(string name, McpServerConfig config, IProcessSpawnPolicy processes, Func<string, string?> getEnv, ILogger logger, CancellationToken cancellationToken)
{
var key = McpLaunch.CacheKey(name, config);
Task<McpCatalog> task;
lock (_lock)
{
_tasks.TryGetValue(key, out var existing);
var expired = existing is { IsFaulted: true }
&& _failedAt.TryGetValue(key, out var failedAt)
&& _time.GetUtcNow() - failedAt >= FailureCooldown;
if (existing is null || expired)
{
_failedAt.Remove(key);
existing = Task.Run(() => DiscoverAsync(key, name, config, processes, getEnv, logger), CancellationToken.None);
_tasks[key] = existing;
}
task = existing;
}
return task.WaitAsync(cancellationToken);
}
private async Task<McpCatalog> DiscoverAsync(string key, string name, McpServerConfig config, IProcessSpawnPolicy processes, Func<string, string?> getEnv, ILogger logger)
{
var dir = Path.Combine(Path.GetTempPath(), "volcanic-mcp-discovery", Guid.NewGuid().ToString("N"));
Directory.CreateDirectory(dir);
var stderr = new StderrTail();
McpProcess? process = null;
try
{
ProcessSpawnContext spawn;
try { spawn = processes.Prepare(McpLaunch.BuildRequest(name, config, dir, getEnv)); }
catch (Exception ex) { throw new McpServerStartException($"{name}: process environment hook failed: {ex.Message}", ex); }
process = await McpProcess.StartAsync(name, config, spawn, stderr, logger, CancellationToken.None).ConfigureAwait(false);
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(config.StartupTimeoutSeconds));
IList<McpClientTool> tools;
try { tools = await process.Client.ListToolsAsync(cancellationToken: cts.Token).ConfigureAwait(false); }
catch (OperationCanceledException)
{
throw new McpServerStartException($"{name} MCP server did not list its tools within {config.StartupTimeoutSeconds}s{McpProcess.TailSuffix(stderr)}");
}
catch (Exception ex)
{
throw new McpServerStartException($"{name} MCP server failed to list its tools: {ex.Message}{McpProcess.TailSuffix(stderr)}", ex);
}
var catalog = McpCatalog.From(name, config, process.Client.ServerInfo, [.. tools.Select(t => t.ProtocolTool)]);
logger.LogInformation("MCP server {Name}: {Count} tools ({Server} {Version})", name, catalog.Tools.Count, catalog.ServerInfoName, catalog.ServerVersion);
return catalog;
}
catch (Exception ex)
{
lock (_lock) _failedAt[key] = _time.GetUtcNow(); // recorded before the task faults
logger.LogWarning("MCP server {Name} is unavailable: {Message}", name, ex.Message);
if (ex is McpServerStartException) throw;
throw new McpServerStartException($"{name} MCP server discovery failed: {ex.Message}", ex);
}
finally
{
// Detached: the SDK spends ShutdownTimeout on every dispose, and no caller needs to wait for it.
_ = Task.Run(() => CleanUpAsync(name, process, dir, logger), CancellationToken.None);
}
}
private static async Task CleanUpAsync(string name, McpProcess? process, string dir, ILogger logger)
{
try
{
if (process is not null) await process.DisposeAsync().ConfigureAwait(false);
}
catch (Exception ex)
{
logger.LogDebug(ex, "MCP server {Name}: stopping the discovery process failed", name);
}
// Windows can hold the cwd for a moment after the process exits.
for (var attempt = 1; ; attempt++)
{
try
{
if (Directory.Exists(dir)) Directory.Delete(dir, recursive: true);
return;
}
catch (Exception ex) when (ex is IOException or UnauthorizedAccessException)
{
if (attempt == 5)
{
logger.LogDebug(ex, "MCP server {Name}: could not delete the discovery directory {Dir}", name, dir);
return;
}
await Task.Delay(TimeSpan.FromMilliseconds(200 * attempt)).ConfigureAwait(false);
}
}
}
}

87
extensions/Volcanic.AgenticCore.Extensions.Mcp/McpProcess.cs Normal file
View File

@@ -0,0 +1,87 @@
using System.ComponentModel;
using System.Diagnostics;
using Microsoft.Extensions.Logging;
using ModelContextProtocol.Client;
using ModelContextProtocol.Protocol;
using Volcanic.AgenticCore.Abstractions.Processes;
namespace Volcanic.AgenticCore.Extensions.Mcp;
/// <summary>A started server: the SDK client over a stdio transport. Used for discovery and for session processes.</summary>
internal sealed class McpProcess : IAsyncDisposable
{
private static readonly TimeSpan StderrGrace = TimeSpan.FromMilliseconds(300);
private int _disposed;
private McpProcess(McpClient client) => Client = client;
public McpClient Client { get; }
public Task<ClientCompletionDetails> Completion => Client.Completion;
public bool HasExited => Client.Completion.IsCompleted;
/// <summary>Spawns the prepared process and runs the handshake under <c>startupTimeoutSeconds</c>. Stderr goes to <paramref name="stderr"/> and the log.</summary>
public static async Task<McpProcess> StartAsync(string name, McpServerConfig config, ProcessSpawnContext spawn, StderrTail stderr, ILogger logger, CancellationToken cancellationToken)
{
var timeout = TimeSpan.FromSeconds(config.StartupTimeoutSeconds);
var transport = new StdioClientTransport(McpLaunch.TransportOptions(name, spawn, line =>
{
stderr.Add(line);
logger.Log(config.StderrLevel, "{Server} mcp: {Line}", name, line);
}));
using var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
cts.CancelAfter(timeout);
try
{
var client = await McpClient.CreateAsync(
transport,
new McpClientOptions { ClientInfo = new Implementation { Name = "volcanic-agentic-core", Version = "1.0.0" }, InitializationTimeout = timeout },
loggerFactory: null,
cts.Token).ConfigureAwait(false);
return new McpProcess(client);
}
catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested)
{
throw new McpServerStartException($"{name} MCP server did not start within {config.StartupTimeoutSeconds}s{TailSuffix(stderr)}");
}
catch (Exception ex) when (ex is not OperationCanceledException)
{
await Task.Delay(StderrGrace, CancellationToken.None).ConfigureAwait(false); // let the dying process's last lines arrive
throw new McpServerStartException($"{name} MCP server failed to start: {ex.Message}{TailSuffix(stderr)}", ex);
}
}
public static string TailSuffix(StderrTail stderr) =>
stderr.Lines is { Count: > 0 } lines ? "\n" + string.Join('\n', lines) : "";
/// <summary>The completion details if the process ends within <paramref name="grace"/>, else null.</summary>
public async Task<ClientCompletionDetails?> WaitForExitAsync(TimeSpan grace)
{
var finished = await Task.WhenAny(Completion, Task.Delay(grace)).ConfigureAwait(false);
return finished == Completion ? await Completion.ConfigureAwait(false) : null;
}
/// <summary>
/// Idempotent. The SDK (2.2) waits up to <c>ShutdownTimeout</c> for the process to exit, with stdin still open, then
/// kills its tree; so a live server always costs that long. Any leftover tree is killed here.
/// </summary>
public async ValueTask DisposeAsync()
{
if (Interlocked.Exchange(ref _disposed, 1) == 1) return;
try { await Client.DisposeAsync().ConfigureAwait(false); }
catch (Exception) { /* a dead transport may throw on dispose; the process is handled below */ }
if (Client.Completion.IsCompletedSuccessfully && Client.Completion.Result is StdioClientCompletionDetails { ProcessId: { } pid })
KillTree(pid);
}
private static void KillTree(int pid)
{
try
{
using var process = Process.GetProcessById(pid);
if (!process.HasExited) process.Kill(entireProcessTree: true);
}
catch (ArgumentException) { }
catch (InvalidOperationException) { }
catch (Win32Exception) { }
}
}