using Microsoft.Extensions.Logging;
using ModelContextProtocol.Protocol;
using ModelContextProtocol.Server;
using System.Diagnostics;
using System.Diagnostics.CodeAnalysis;
using System.Text.Json;
using System.Text.Json.Nodes;
namespace ModelContextProtocol.Client;
///
/// Represents an instance of a Model Context Protocol (MCP) client session that connects to and communicates with an MCP server.
///
public abstract partial class McpClient : McpSession
{
/// Creates an , connecting it to the specified server.
/// The transport instance used to communicate with the server.
///
/// A client configuration object that specifies client capabilities and protocol version.
/// If , details based on the current process are used.
///
/// A logger factory for creating loggers for clients.
/// The to monitor for cancellation requests. The default is .
/// An that's connected to the specified server.
/// is .
/// An error occurred while connecting to the server over HTTP.
/// The server returned an error response during initialization.
///
///
/// When using an HTTP-based transport (such as ), this method may throw
/// if there is a problem establishing the connection to the MCP server.
///
///
/// If the server requires authentication and credentials are not provided or are invalid, an
/// with an HTTP 401 Unauthorized status code will be thrown.
/// To authenticate with a protected server, configure the
/// property of the transport with appropriate credentials before calling this method.
///
///
public static async Task CreateAsync(
IClientTransport clientTransport,
McpClientOptions? clientOptions = null,
ILoggerFactory? loggerFactory = null,
CancellationToken cancellationToken = default)
{
Throw.IfNull(clientTransport);
var transport = await clientTransport.ConnectAsync(cancellationToken).ConfigureAwait(false);
var endpointName = clientTransport.Name;
var clientSession = new McpClientImpl(transport, endpointName, clientOptions, loggerFactory);
try
{
await clientSession.ConnectAsync(cancellationToken).ConfigureAwait(false);
}
catch (Exception ex) when (ex is not OperationCanceledException and not ClientTransportClosedException)
{
// ConnectAsync already disposed the session (which includes awaiting Completion).
// Check if the transport provided structured completion details indicating
// why the transport closed that aren't already in the original exception chain.
Debug.Assert(clientSession.Completion.IsCompleted, "Completion should already be finished after ConnectAsync's DisposeAsync.");
var completionDetails = await clientSession.Completion.ConfigureAwait(false);
// If the transport closed with a non-graceful error (e.g., server process exited)
// and the completion details carry an exception that's NOT already in the original
// exception chain, throw a ClientTransportClosedException with the structured details so
// callers can programmatically inspect the closure reason (exit code, stderr, etc.).
// When the same exception is already in the chain (e.g., HttpRequestException from
// an HTTP transport), the original exception is more appropriate to re-throw.
if (completionDetails.Exception is { } detailsException &&
!ExceptionChainContains(ex, detailsException))
{
throw new ClientTransportClosedException(completionDetails);
}
throw;
}
return clientSession;
}
///
/// Returns if is the same object as
/// or any exception in its chain.
///
private static bool ExceptionChainContains(Exception exception, Exception target)
{
for (Exception? current = exception; current is not null; current = current.InnerException)
{
if (ReferenceEquals(current, target))
{
return true;
}
}
return false;
}
///
/// Recreates an using an existing transport session without sending a new initialize request.
///
/// The transport instance already configured to connect to the target server.
/// The metadata captured from the original session that should be applied when resuming.
/// Optional client settings that should mirror those used to create the original session.
/// An optional logger factory for diagnostics.
/// The to monitor for cancellation requests. The default is .
/// An bound to the resumed session.
/// , , , or is .
public static async Task ResumeSessionAsync(
IClientTransport clientTransport,
ResumeClientSessionOptions resumeOptions,
McpClientOptions? clientOptions = null,
ILoggerFactory? loggerFactory = null,
CancellationToken cancellationToken = default)
{
Throw.IfNull(clientTransport);
Throw.IfNull(resumeOptions);
Throw.IfNull(resumeOptions.ServerCapabilities);
Throw.IfNull(resumeOptions.ServerInfo);
var transport = await clientTransport.ConnectAsync(cancellationToken).ConfigureAwait(false);
McpClientImpl clientSession = new(transport, clientTransport.Name, clientOptions, loggerFactory);
clientSession.ResumeSession(resumeOptions);
return clientSession;
}
///
/// Sends a ping request to verify server connectivity.
///
/// Optional request options including metadata, serialization settings, and progress tracking.
/// The to monitor for cancellation requests. The default is .
/// A task containing the ping result.
/// The server cannot be reached or returned an error response.
public ValueTask PingAsync(RequestOptions? options = null, CancellationToken cancellationToken = default)
{
return PingAsync(
new PingRequestParams
{
Meta = options?.GetMetaForRequest()
},
cancellationToken);
}
///
/// Sends a ping request to verify server connectivity.
///
/// The request parameters to send in the request.
/// The to monitor for cancellation requests. The default is .
/// A task containing the ping result.
/// is .
/// The server cannot be reached or returned an error response.
public ValueTask PingAsync(
PingRequestParams requestParams,
CancellationToken cancellationToken = default)
{
Throw.IfNull(requestParams);
return SendRequestAsync(
RequestMethods.Ping,
requestParams,
McpJsonUtilities.JsonContext.Default.PingRequestParams,
McpJsonUtilities.JsonContext.Default.PingResult,
cancellationToken: cancellationToken);
}
///
/// Retrieves a list of available tools from the server.
///
/// Optional request options including metadata, serialization settings, and progress tracking.
/// The to monitor for cancellation requests. The default is .
/// A list of all available tools as instances.
/// The request failed or the server returned an error response.
///
/// This overload aggregates every page into a single list and does not surface the per-result caching hints
/// ( and ). To read those hints,
/// use the overload, which returns the
/// raw for each page.
///
public async ValueTask> ListToolsAsync(
RequestOptions? options = null,
CancellationToken cancellationToken = default)
{
ToolCacheClearing?.Invoke();
List? tools = null;
ListToolsRequestParams requestParams = new() { Meta = options?.GetMetaForRequest() };
do
{
var toolResults = await ListToolsAsync(requestParams, cancellationToken).ConfigureAwait(false);
tools ??= new(toolResults.Tools.Count);
foreach (var tool in toolResults.Tools)
{
// Validate x-mcp-header annotations per SEP-2243. The spec requires Streamable HTTP
// clients to exclude tools with invalid annotations and permits other transports
// (e.g., stdio) to ignore the annotations entirely. This client validates on all
// transports so a malformed definition is rejected consistently regardless of transport.
if (!McpHeaderExtractor.ValidateToolSchema(tool, out var rejectionReason))
{
ToolRejected?.Invoke(tool, rejectionReason!);
continue;
}
ToolDiscovered?.Invoke(tool);
tools.Add(new(this, tool, options?.JsonSerializerOptions));
}
requestParams.Cursor = toolResults.NextCursor;
}
while (requestParams.Cursor is not null);
return tools;
}
///
/// Invoked when a tool definition is discovered from a tools/list response.
///
internal Action? ToolDiscovered;
///
/// Invoked when a tool definition is rejected due to invalid x-mcp-header annotations.
///
internal Action? ToolRejected;
///
/// Invoked before enumerating tools to clear any previously cached tool definitions.
///
internal Action? ToolCacheClearing;
///
/// Retrieves a list of available tools from the server.
///
/// The request parameters to send in the request.
/// The to monitor for cancellation requests. The default is .
/// The result of the request as provided by the server.
/// is .
/// The request failed or the server returned an error response.
///
/// The overload retrieves all tools by automatically handling pagination.
/// This overload works with the lower-level and , returning the raw result from the server.
/// Any pagination needs to be managed by the caller.
///
public ValueTask ListToolsAsync(
ListToolsRequestParams requestParams,
CancellationToken cancellationToken = default)
{
Throw.IfNull(requestParams);
return SendRequestAsync(
RequestMethods.ToolsList,
requestParams,
McpJsonUtilities.JsonContext.Default.ListToolsRequestParams,
McpJsonUtilities.JsonContext.Default.ListToolsResult,
cancellationToken: cancellationToken);
}
///
/// Retrieves a list of available prompts from the server.
///
/// Optional request options including metadata, serialization settings, and progress tracking.
/// The to monitor for cancellation requests. The default is .
/// A list of all available prompts as instances.
/// The request failed or the server returned an error response.
///
/// This overload aggregates every page into a single list and does not surface the per-result caching hints
/// ( and ). To read those hints,
/// use the overload, which returns the
/// raw for each page.
///
public async ValueTask> ListPromptsAsync(
RequestOptions? options = null,
CancellationToken cancellationToken = default)
{
List? prompts = null;
ListPromptsRequestParams requestParams = new() { Meta = options?.GetMetaForRequest() };
do
{
var promptResults = await ListPromptsAsync(requestParams, cancellationToken).ConfigureAwait(false);
prompts ??= new(promptResults.Prompts.Count);
foreach (var prompt in promptResults.Prompts)
{
prompts.Add(new(this, prompt));
}
requestParams.Cursor = promptResults.NextCursor;
}
while (requestParams.Cursor is not null);
return prompts;
}
///
/// Retrieves a list of available prompts from the server.
///
/// The request parameters to send in the request.
/// The to monitor for cancellation requests. The default is .
/// The result of the request as provided by the server.
/// is .
/// The request failed or the server returned an error response.
///
/// The overload retrieves all prompts by automatically handling pagination.
/// This overload works with the lower-level and , returning the raw result from the server.
/// Any pagination needs to be managed by the caller.
///
public ValueTask ListPromptsAsync(
ListPromptsRequestParams requestParams,
CancellationToken cancellationToken = default)
{
Throw.IfNull(requestParams);
return SendRequestAsync(
RequestMethods.PromptsList,
requestParams,
McpJsonUtilities.JsonContext.Default.ListPromptsRequestParams,
McpJsonUtilities.JsonContext.Default.ListPromptsResult,
cancellationToken: cancellationToken);
}
///
/// Retrieves a specific prompt from the MCP server.
///
/// The name of the prompt to retrieve.
/// Optional arguments for the prompt. The dictionary keys are parameter names, and the values are the argument values.
/// Optional request options including metadata, serialization settings, and progress tracking.
/// The to monitor for cancellation requests. The default is .
/// A task containing the prompt's result with content and messages.
/// is .
/// is empty or composed entirely of whitespace.
/// The request failed or the server returned an error response.
public ValueTask GetPromptAsync(
string name,
IReadOnlyDictionary? arguments = null,
RequestOptions? options = null,
CancellationToken cancellationToken = default)
{
Throw.IfNullOrWhiteSpace(name);
var serializerOptions = options?.JsonSerializerOptions ?? McpJsonUtilities.DefaultOptions;
serializerOptions.MakeReadOnly();
return GetPromptAsync(
new GetPromptRequestParams
{
Name = name,
Arguments = ToArgumentsDictionary(arguments, serializerOptions),
Meta = options?.GetMetaForRequest(),
},
cancellationToken);
}
///
/// Retrieves a specific prompt from the MCP server.
///
/// The request parameters to send in the request.
/// The to monitor for cancellation requests. The default is .
/// The result of the request as provided by the server.
/// is .
/// The request failed or the server returned an error response.
public ValueTask GetPromptAsync(
GetPromptRequestParams requestParams,
CancellationToken cancellationToken = default)
{
Throw.IfNull(requestParams);
return SendRequestAsync(
RequestMethods.PromptsGet,
requestParams,
McpJsonUtilities.JsonContext.Default.GetPromptRequestParams,
McpJsonUtilities.JsonContext.Default.GetPromptResult,
cancellationToken: cancellationToken);
}
///
/// Retrieves a list of available resource templates from the server.
///
/// Optional request options including metadata, serialization settings, and progress tracking.
/// The to monitor for cancellation requests. The default is .
/// A list of all available resource templates as instances.
/// The request failed or the server returned an error response.
///
/// This overload aggregates every page into a single list and does not surface the per-result caching hints
/// ( and ). To read those hints,
/// use the overload, which returns the
/// raw for each page.
///
public async ValueTask> ListResourceTemplatesAsync(
RequestOptions? options = null,
CancellationToken cancellationToken = default)
{
List? resourceTemplates = null;
ListResourceTemplatesRequestParams requestParams = new() { Meta = options?.GetMetaForRequest() };
do
{
var templateResults = await ListResourceTemplatesAsync(requestParams, cancellationToken).ConfigureAwait(false);
resourceTemplates ??= new(templateResults.ResourceTemplates.Count);
foreach (var template in templateResults.ResourceTemplates)
{
resourceTemplates.Add(new(this, template));
}
requestParams.Cursor = templateResults.NextCursor;
}
while (requestParams.Cursor is not null);
return resourceTemplates;
}
///
/// Retrieves a list of available resource templates from the server.
///
/// The request parameters to send in the request.
/// The to monitor for cancellation requests. The default is .
/// The result of the request as provided by the server.
/// is .
/// The request failed or the server returned an error response.
///
/// The overload retrieves all resource templates by automatically handling pagination.
/// This overload works with the lower-level and , returning the raw result from the server.
/// Any pagination needs to be managed by the caller.
///
public ValueTask ListResourceTemplatesAsync(
ListResourceTemplatesRequestParams requestParams,
CancellationToken cancellationToken = default)
{
Throw.IfNull(requestParams);
return SendRequestAsync(
RequestMethods.ResourcesTemplatesList,
requestParams,
McpJsonUtilities.JsonContext.Default.ListResourceTemplatesRequestParams,
McpJsonUtilities.JsonContext.Default.ListResourceTemplatesResult,
cancellationToken: cancellationToken);
}
///
/// Retrieves a list of available resources from the server.
///
/// Optional request options including metadata, serialization settings, and progress tracking.
/// The to monitor for cancellation requests. The default is .
/// A list of all available resources as instances.
/// The request failed or the server returned an error response.
///
/// This overload aggregates every page into a single list and does not surface the per-result caching hints
/// ( and ). To read those hints,
/// use the overload, which returns the
/// raw for each page.
///
public async ValueTask> ListResourcesAsync(
RequestOptions? options = null,
CancellationToken cancellationToken = default)
{
List? resources = null;
ListResourcesRequestParams requestParams = new() { Meta = options?.GetMetaForRequest() };
do
{
var resourceResults = await ListResourcesAsync(requestParams, cancellationToken).ConfigureAwait(false);
resources ??= new(resourceResults.Resources.Count);
foreach (var resource in resourceResults.Resources)
{
resources.Add(new(this, resource));
}
requestParams.Cursor = resourceResults.NextCursor;
}
while (requestParams.Cursor is not null);
return resources;
}
///
/// Retrieves a list of available resources from the server.
///
/// The request parameters to send in the request.
/// The to monitor for cancellation requests. The default is .
/// The result of the request as provided by the server.
/// is .
/// The request failed or the server returned an error response.
///
/// The overload retrieves all resources by automatically handling pagination.
/// This overload works with the lower-level and , returning the raw result from the server.
/// Any pagination needs to be managed by the caller.
///
public ValueTask ListResourcesAsync(
ListResourcesRequestParams requestParams,
CancellationToken cancellationToken = default)
{
Throw.IfNull(requestParams);
return SendRequestAsync(
RequestMethods.ResourcesList,
requestParams,
McpJsonUtilities.JsonContext.Default.ListResourcesRequestParams,
McpJsonUtilities.JsonContext.Default.ListResourcesResult,
cancellationToken: cancellationToken);
}
///
/// Reads a resource from the server.
///
/// The URI of the resource.
/// Optional request options including metadata, serialization settings, and progress tracking.
/// The to monitor for cancellation requests. The default is .
/// is .
/// The request failed or the server returned an error response.
public ValueTask ReadResourceAsync(
Uri uri, RequestOptions? options = null, CancellationToken cancellationToken = default)
{
Throw.IfNull(uri);
return ReadResourceAsync(uri.AbsoluteUri, options, cancellationToken);
}
///
/// Reads a resource from the server.
///
/// The URI of the resource.
/// Optional request options including metadata, serialization settings, and progress tracking.
/// The to monitor for cancellation requests. The default is .
/// is .
/// is empty or composed entirely of whitespace.
/// The request failed or the server returned an error response.
public ValueTask ReadResourceAsync(
string uri, RequestOptions? options = null, CancellationToken cancellationToken = default)
{
Throw.IfNullOrWhiteSpace(uri);
return ReadResourceAsync(new ReadResourceRequestParams
{
Uri = uri,
Meta = options?.GetMetaForRequest(),
}, cancellationToken);
}
///
/// Reads a resource from the server.
///
/// The URI template of the resource.
/// Arguments to use to format .
/// Optional request options including metadata, serialization settings, and progress tracking.
/// The to monitor for cancellation requests. The default is .
/// or is .
/// is empty or composed entirely of whitespace.
/// The request failed or the server returned an error response.
public ValueTask ReadResourceAsync(
string uriTemplate, IReadOnlyDictionary arguments, RequestOptions? options = null, CancellationToken cancellationToken = default)
{
Throw.IfNullOrWhiteSpace(uriTemplate);
Throw.IfNull(arguments);
return ReadResourceAsync(
new ReadResourceRequestParams
{
Uri = UriTemplate.FormatUri(uriTemplate, arguments),
Meta = options?.GetMetaForRequest(),
},
cancellationToken);
}
///
/// Reads a resource from the server.
///
/// The request parameters to send in the request.
/// The to monitor for cancellation requests. The default is .
/// The result of the request.
/// is .
/// The request failed or the server returned an error response.
public ValueTask ReadResourceAsync(
ReadResourceRequestParams requestParams,
CancellationToken cancellationToken = default)
{
Throw.IfNull(requestParams);
return SendRequestAsync(
RequestMethods.ResourcesRead,
requestParams,
McpJsonUtilities.JsonContext.Default.ReadResourceRequestParams,
McpJsonUtilities.JsonContext.Default.ReadResourceResult,
cancellationToken: cancellationToken);
}
///
/// Requests completion suggestions for a prompt argument or resource reference.
///
/// The reference object specifying the type and optional URI or name.
/// The name of the argument for which completions are requested.
/// The current value of the argument, used to filter relevant completions.
/// Optional request options including metadata, serialization settings, and progress tracking.
/// The to monitor for cancellation requests. The default is .
/// A containing completion suggestions.
/// or is .
/// is empty or composed entirely of whitespace.
/// The request failed or the server returned an error response.
public ValueTask CompleteAsync(
Reference reference, string argumentName, string argumentValue,
RequestOptions? options = null, CancellationToken cancellationToken = default)
{
Throw.IfNull(reference);
Throw.IfNullOrWhiteSpace(argumentName);
return CompleteAsync(
new CompleteRequestParams
{
Ref = reference,
Argument = new() { Name = argumentName, Value = argumentValue },
Meta = options?.GetMetaForRequest(),
},
cancellationToken);
}
///
/// Requests completion suggestions for a prompt argument or resource reference.
///
/// The request parameters to send in the request.
/// The to monitor for cancellation requests. The default is .
/// The result of the request.
/// is .
/// The request failed or the server returned an error response.
public ValueTask CompleteAsync(
CompleteRequestParams requestParams,
CancellationToken cancellationToken = default)
{
Throw.IfNull(requestParams);
return SendRequestAsync(
RequestMethods.CompletionComplete,
requestParams,
McpJsonUtilities.JsonContext.Default.CompleteRequestParams,
McpJsonUtilities.JsonContext.Default.CompleteResult,
cancellationToken: cancellationToken);
}
///
/// Subscribes to a resource on the server to receive notifications when it changes.
///
/// The URI of the resource to subscribe to.
/// Optional request options including metadata, serialization settings, and progress tracking.
/// The to monitor for cancellation requests. The default is .
/// A task that represents the asynchronous operation.
/// is .
/// The request failed or the server returned an error response.
public Task SubscribeToResourceAsync(Uri uri, RequestOptions? options = null, CancellationToken cancellationToken = default)
{
Throw.IfNull(uri);
return SubscribeToResourceAsync(uri.AbsoluteUri, options, cancellationToken);
}
///
/// Subscribes to a resource on the server to receive notifications when it changes.
///
/// The URI of the resource to which to subscribe.
/// Optional request options including metadata, serialization settings, and progress tracking.
/// The to monitor for cancellation requests. The default is .
/// A task that represents the asynchronous operation.
/// is .
/// is empty or composed entirely of whitespace.
/// The request failed or the server returned an error response.
public Task SubscribeToResourceAsync(string uri, RequestOptions? options = null, CancellationToken cancellationToken = default)
{
Throw.IfNullOrWhiteSpace(uri);
return SubscribeToResourceAsync(
new SubscribeRequestParams
{
Uri = uri,
Meta = options?.GetMetaForRequest(),
},
cancellationToken);
}
///
/// Subscribes to a resource on the server to receive notifications when it changes.
///
/// The request parameters to send in the request.
/// The to monitor for cancellation requests. The default is .
/// The result of the request.
/// is .
/// The request failed or the server returned an error response.
///
///
/// This method subscribes to resource update notifications but does not register a handler.
/// To receive notifications, you must separately call
/// with and filter for the specific resource URI.
/// To unsubscribe, call and dispose the handler registration.
///
///
/// For a simpler API that handles both subscription and notification registration in a single call,
/// use .
///
///
public Task SubscribeToResourceAsync(
SubscribeRequestParams requestParams,
CancellationToken cancellationToken = default)
{
Throw.IfNull(requestParams);
return SendRequestAsync(
RequestMethods.ResourcesSubscribe,
requestParams,
McpJsonUtilities.JsonContext.Default.SubscribeRequestParams,
McpJsonUtilities.JsonContext.Default.EmptyResult,
cancellationToken: cancellationToken).AsTask();
}
///
/// Subscribes to a resource on the server and registers a handler for notifications when it changes.
///
/// The URI of the resource to which to subscribe.
/// The handler to invoke when the resource is updated. It receives for the subscribed resource.
/// Optional request options including metadata, serialization settings, and progress tracking.
/// The to monitor for cancellation requests. The default is .
///
/// A task that completes with an that, when disposed, unsubscribes from the resource
/// and removes the notification handler.
///
/// or is .
/// The request failed or the server returned an error response.
///
///
/// This method provides a convenient way to subscribe to resource updates and handle notifications in a single call.
/// The returned manages both the subscription and the notification handler registration.
/// When disposed, it automatically unsubscribes from the resource and removes the handler.
///
///
/// The handler will only be invoked for notifications related to the specified resource URI.
/// Notifications for other resources are filtered out automatically.
///
///
public Task SubscribeToResourceAsync(
Uri uri,
Func handler,
RequestOptions? options = null,
CancellationToken cancellationToken = default)
{
Throw.IfNull(uri);
return SubscribeToResourceAsync(uri.AbsoluteUri, handler, options, cancellationToken);
}
///
/// Subscribes to a resource on the server and registers a handler for notifications when it changes.
///
/// The URI of the resource to which to subscribe.
/// The handler to invoke when the resource is updated. It receives for the subscribed resource.
/// Optional request options including metadata, serialization settings, and progress tracking.
/// The to monitor for cancellation requests. The default is .
///
/// A task that completes with an that, when disposed, unsubscribes from the resource
/// and removes the notification handler.
///
/// or is .
/// is empty or composed entirely of whitespace.
/// The request failed or the server returned an error response.
///
///
/// This method provides a convenient way to subscribe to resource updates and handle notifications in a single call.
/// The returned manages both the subscription and the notification handler registration.
/// When disposed, it automatically unsubscribes from the resource and removes the handler.
///
///
/// The handler will only be invoked for notifications related to the specified resource URI.
/// Notifications for other resources are filtered out automatically.
///
///
public async Task SubscribeToResourceAsync(
string uri,
Func handler,
RequestOptions? options = null,
CancellationToken cancellationToken = default)
{
Throw.IfNullOrWhiteSpace(uri);
Throw.IfNull(handler);
// Register a notification handler that filters for this specific resource
IAsyncDisposable handlerRegistration = RegisterNotificationHandler(
NotificationMethods.ResourceUpdatedNotification,
async (notification, ct) =>
{
if (JsonSerializer.Deserialize(notification.Params, McpJsonUtilities.JsonContext.Default.ResourceUpdatedNotificationParams) is { } resourceUpdate &&
UriTemplate.UriTemplateComparer.Instance.Equals(resourceUpdate.Uri, uri))
{
await handler(resourceUpdate, ct).ConfigureAwait(false);
}
});
try
{
// Subscribe to the resource
await SubscribeToResourceAsync(uri, options, cancellationToken).ConfigureAwait(false);
}
catch
{
// If subscription fails, unregister the handler before propagating the exception
await handlerRegistration.DisposeAsync().ConfigureAwait(false);
throw;
}
// Return a disposable that unsubscribes and removes the handler
return new ResourceSubscription(this, uri, handlerRegistration, options);
}
///
/// Manages a resource subscription, handling both unsubscription and handler disposal.
///
private sealed class ResourceSubscription : IAsyncDisposable
{
private readonly McpClient _client;
private readonly string _uri;
private readonly IAsyncDisposable _handlerRegistration;
private readonly RequestOptions? _options;
private int _disposed;
public ResourceSubscription(McpClient client, string uri, IAsyncDisposable handlerRegistration, RequestOptions? options)
{
_client = client;
_uri = uri;
_handlerRegistration = handlerRegistration;
_options = options;
}
public async ValueTask DisposeAsync()
{
if (Interlocked.Exchange(ref _disposed, 1) != 0)
{
return;
}
try
{
// Unsubscribe from the resource
await _client.UnsubscribeFromResourceAsync(_uri, _options, CancellationToken.None).ConfigureAwait(false);
}
finally
{
// Dispose the notification handler registration
await _handlerRegistration.DisposeAsync().ConfigureAwait(false);
}
}
}
///
/// Unsubscribes from a resource on the server to stop receiving notifications about its changes.
///
/// The URI of the resource to unsubscribe from.
/// Optional request options including metadata, serialization settings, and progress tracking.
/// The to monitor for cancellation requests. The default is .
/// A task that represents the asynchronous operation.
/// is .
/// The request failed or the server returned an error response.
public Task UnsubscribeFromResourceAsync(Uri uri, RequestOptions? options = null, CancellationToken cancellationToken = default)
{
Throw.IfNull(uri);
return UnsubscribeFromResourceAsync(uri.AbsoluteUri, options, cancellationToken);
}
///
/// Unsubscribes from a resource on the server to stop receiving notifications about its changes.
///
/// The URI of the resource to unsubscribe from.
/// Optional request options including metadata, serialization settings, and progress tracking.
/// The to monitor for cancellation requests. The default is .
/// A task that represents the asynchronous operation.
/// is .
/// is empty or composed entirely of whitespace.
/// The request failed or the server returned an error response.
public Task UnsubscribeFromResourceAsync(string uri, RequestOptions? options = null, CancellationToken cancellationToken = default)
{
Throw.IfNullOrWhiteSpace(uri);
return UnsubscribeFromResourceAsync(
new UnsubscribeRequestParams
{
Uri = uri,
Meta = options?.GetMetaForRequest()
},
cancellationToken);
}
///
/// Unsubscribes from a resource on the server to stop receiving notifications about its changes.
///
/// The request parameters to send in the request.
/// The to monitor for cancellation requests. The default is .
/// The result of the request.
/// is .
/// The request failed or the server returned an error response.
public Task UnsubscribeFromResourceAsync(
UnsubscribeRequestParams requestParams,
CancellationToken cancellationToken = default)
{
Throw.IfNull(requestParams);
return SendRequestAsync(
RequestMethods.ResourcesUnsubscribe,
requestParams,
McpJsonUtilities.JsonContext.Default.UnsubscribeRequestParams,
McpJsonUtilities.JsonContext.Default.EmptyResult,
cancellationToken: cancellationToken).AsTask();
}
///
/// Invokes a tool on the server.
///
/// The name of the tool to call on the server.
/// An optional dictionary of arguments to pass to the tool.
/// An optional progress reporter for server notifications.
/// Optional request options including metadata, serialization settings, and progress tracking.
/// The to monitor for cancellation requests. The default is .
/// The from the tool execution.
/// is .
/// The request failed or the server returned an error response.
///
/// This overload supports the tasks extension transparently. If the server responds with a
/// task handle rather than an immediate result, this method polls tasks/get until the
/// task completes, dispatching any entries through
/// the client's registered sampling and elicitation handlers along the way. Use
/// to disable automatic polling.
///
public ValueTask CallToolAsync(
string toolName,
IReadOnlyDictionary? arguments = null,
IProgress? progress = null,
RequestOptions? options = null,
CancellationToken cancellationToken = default)
{
Throw.IfNull(toolName);
var serializerOptions = options?.JsonSerializerOptions ?? McpJsonUtilities.DefaultOptions;
serializerOptions.MakeReadOnly();
if (progress is null)
{
return CallToolAsync(
new CallToolRequestParams
{
Name = toolName,
Arguments = ToArgumentsDictionary(arguments, serializerOptions),
Meta = options?.GetMetaForRequest(),
},
cancellationToken);
}
return SendRequestWithProgressAsync(toolName, arguments, progress, options?.GetMetaForRequest(), serializerOptions, cancellationToken);
async ValueTask SendRequestWithProgressAsync(
string toolName,
IReadOnlyDictionary? arguments,
IProgress progress,
JsonObject? meta,
JsonSerializerOptions serializerOptions,
CancellationToken cancellationToken)
{
ProgressToken progressToken = new(Guid.NewGuid().ToString("N"));
await using var _ = RegisterNotificationHandler(NotificationMethods.ProgressNotification,
(notification, cancellationToken) =>
{
if (JsonSerializer.Deserialize(notification.Params, McpJsonUtilities.JsonContext.Default.ProgressNotificationParams) is { } pn &&
pn.ProgressToken == progressToken)
{
progress.Report(pn.Progress);
}
return default;
}).ConfigureAwait(false);
JsonObject metaWithProgress = meta is not null ? (JsonObject)meta.DeepClone() : [];
metaWithProgress["progressToken"] = progressToken.ToString();
return await CallToolAsync(
new()
{
Name = toolName,
Arguments = ToArgumentsDictionary(arguments, serializerOptions),
Meta = metaWithProgress,
},
cancellationToken).ConfigureAwait(false);
}
}
///
/// Invokes a tool on the server.
///
/// The request parameters to send in the request.
/// The to monitor for cancellation requests. The default is .
/// The result of the request.
/// is .
/// The request failed or the server returned an error response.
///
/// This method automatically includes the io.modelcontextprotocol/tasks extension capability
/// in the request metadata. If the server returns a task handle instead of an immediate result,
/// this method transparently polls tasks/get until the task completes, fails, or is cancelled.
/// Use
/// to receive the raw without automatic polling.
///
public async ValueTask CallToolAsync(
CallToolRequestParams requestParams,
CancellationToken cancellationToken = default)
{
Throw.IfNull(requestParams);
var augmented = await CallToolRawAsync(requestParams, cancellationToken).ConfigureAwait(false);
if (!augmented.IsTask)
{
return augmented.Result!;
}
return await PollTaskToCompletionAsync(augmented.TaskCreated!, cancellationToken).ConfigureAwait(false);
}
///
/// Polls a task until it reaches a terminal state and returns the final .
///
private async ValueTask PollTaskToCompletionAsync(
CreateTaskResult taskCreated,
CancellationToken cancellationToken)
{
// If the server claims InputRequired but never publishes new input requests after we have
// already responded to everything it asked for, treat that as a stuck task. The client
// can still cancel earlier via cancellationToken; this guard prevents an unbounded poll
// loop when the server is misbehaving. The threshold is configurable via
// McpClientOptions.MaxConsecutiveStuckPolls.
int maxConsecutiveStuckPolls = MaxConsecutiveStuckPolls;
string taskId = taskCreated.TaskId;
long pollIntervalMs = taskCreated.PollIntervalMs ?? 1000;
HashSet? resolvedRequestKeys = null;
bool isFirstPoll = true;
int consecutiveStuckPolls = 0;
while (true)
{
// Skip the delay before the first poll: many tasks complete almost immediately and we
// don't want to pay the poll interval as gratuitous latency.
if (!isFirstPoll)
{
await Task.Delay(TimeSpan.FromMilliseconds(pollIntervalMs), cancellationToken).ConfigureAwait(false);
}
isFirstPoll = false;
var taskResult = await GetTaskAsync(taskId, cancellationToken).ConfigureAwait(false);
// Update poll interval if the server changed it.
if (taskResult.PollIntervalMs is { } newInterval)
{
pollIntervalMs = newInterval;
}
switch (taskResult)
{
case CompletedTaskResult completed:
return JsonSerializer.Deserialize(completed.Result, McpJsonUtilities.JsonContext.Default.CallToolResult)
?? throw new JsonException("Failed to deserialize CallToolResult from completed task.");
case FailedTaskResult failed:
throw new McpException($"Task '{taskId}' failed: {failed.Error}");
case CancelledTaskResult:
throw new OperationCanceledException($"Task '{taskId}' was cancelled by the server.");
case InputRequiredTaskResult inputRequired:
// Dedup: only resolve input requests we haven't already responded to.
var newRequests = new Dictionary();
if (inputRequired.InputRequests is { } incomingRequests)
{
foreach (var kvp in incomingRequests)
{
if (resolvedRequestKeys is null || !resolvedRequestKeys.Contains(kvp.Key))
{
newRequests[kvp.Key] = kvp.Value;
}
}
}
if (newRequests.Count > 0)
{
consecutiveStuckPolls = 0;
IDictionary inputResponses;
try
{
inputResponses = await ResolveInputRequestsAsync(newRequests, cancellationToken).ConfigureAwait(false);
}
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
{
throw;
}
catch
{
// The input handler failed (e.g., ElicitationHandler threw or no handler was registered).
// Best-effort cancel of the server-side task so it doesn't stay stuck in InputRequired
// until TTL expires.
try
{
await CancelTaskAsync(taskId, CancellationToken.None).ConfigureAwait(false);
}
catch
{
// Swallow secondary failures; we're already propagating the original exception.
}
throw;
}
await UpdateTaskAsync(new UpdateTaskRequestParams
{
TaskId = taskId,
InputResponses = inputResponses,
}, cancellationToken).ConfigureAwait(false);
resolvedRequestKeys ??= new HashSet(StringComparer.Ordinal);
foreach (var key in inputResponses.Keys)
{
resolvedRequestKeys.Add(key);
}
}
else if (++consecutiveStuckPolls >= maxConsecutiveStuckPolls)
{
// Best-effort cancel of the server-side task so it doesn't leak until TTL expires.
try
{
await CancelTaskAsync(taskId, CancellationToken.None).ConfigureAwait(false);
}
catch
{
// Swallow secondary failures; we're already propagating an exception.
}
throw new McpException(
$"Task '{taskId}' has remained in '{McpTaskStatus.InputRequired}' for {maxConsecutiveStuckPolls} consecutive polls " +
"without publishing new input requests after all previously requested inputs were resolved.");
}
break;
case WorkingTaskResult:
// Continue polling.
consecutiveStuckPolls = 0;
break;
default:
throw new McpException(
$"Unexpected task result type '{taskResult.GetType().Name}' for task '{taskId}'.");
}
}
}
///
/// Invokes a tool on the server with task extension support, returning the raw response
/// without automatic polling. The caller is responsible for handling task lifecycle.
///
/// The request parameters to send. The tasks extension capability will be injected into the request metadata.
/// The to monitor for cancellation requests. The default is .
/// A that is either an immediate result or a task handle.
/// is .
/// The request failed or the server returned an error response.
///
///
/// Unlike , this method does not
/// automatically poll for task completion. If the server returns a ,
/// the caller must manage polling via .
///
///
public async ValueTask> CallToolRawAsync(
CallToolRequestParams requestParams,
CancellationToken cancellationToken = default)
{
Throw.IfNull(requestParams);
var paramsWithMeta = new CallToolRequestParams
{
Name = requestParams.Name,
Arguments = requestParams.Arguments,
Meta = GetMetaWithTaskCapability(requestParams.Meta),
};
JsonRpcRequest jsonRpcRequest = new()
{
Method = RequestMethods.ToolsCall,
Params = JsonSerializer.SerializeToNode(paramsWithMeta, McpJsonUtilities.JsonContext.Default.CallToolRequestParams),
};
JsonRpcResponse response = await SendRequestAsync(jsonRpcRequest, cancellationToken).ConfigureAwait(false);
// Discriminate based on resultType field.
if (response.Result is JsonObject resultObj &&
resultObj.TryGetPropertyValue("resultType", out var resultTypeNode) &&
resultTypeNode?.GetValue() == "task")
{
var taskCreated = resultObj.Deserialize(McpJsonUtilities.JsonContext.Default.CreateTaskResult)
?? throw new JsonException("Failed to deserialize CreateTaskResult from response.");
return new ResultOrCreatedTask(taskCreated);
}
var callToolResult = JsonSerializer.Deserialize(response.Result, McpJsonUtilities.JsonContext.Default.CallToolResult)
?? throw new JsonException("Failed to deserialize CallToolResult from response.");
return new ResultOrCreatedTask(callToolResult);
}
///
/// Sets the logging level for the server to control which log messages are sent to the client.
///
/// The minimum severity level of log messages to receive from the server.
/// Optional request options including metadata, serialization settings, and progress tracking.
/// The to monitor for cancellation requests. The default is .
/// A task representing the asynchronous operation.
/// The request failed or the server returned an error response.
[Obsolete(Obsoletions.DeprecatedLogging_Message, DiagnosticId = Obsoletions.Deprecated_DiagnosticId, UrlFormat = Obsoletions.Deprecated_Url)]
public Task SetLoggingLevelAsync(LogLevel level, RequestOptions? options = null, CancellationToken cancellationToken = default) =>
SetLoggingLevelAsync(McpServerImpl.ToLoggingLevel(level), options, cancellationToken);
///
/// Sets the logging level for the server to control which log messages are sent to the client.
///
/// The minimum severity level of log messages to receive from the server.
/// Optional request options including metadata, serialization settings, and progress tracking.
/// The to monitor for cancellation requests. The default is .
/// A task representing the asynchronous operation.
/// The request failed or the server returned an error response.
[Obsolete(Obsoletions.DeprecatedLogging_Message, DiagnosticId = Obsoletions.Deprecated_DiagnosticId, UrlFormat = Obsoletions.Deprecated_Url)]
public Task SetLoggingLevelAsync(LoggingLevel level, RequestOptions? options = null, CancellationToken cancellationToken = default)
{
return SetLoggingLevelAsync(
new SetLevelRequestParams
{
Level = level,
Meta = options?.GetMetaForRequest()
},
cancellationToken);
}
///
/// Sets the logging level for the server to control which log messages are sent to the client.
///
/// The request parameters to send in the request.
/// The to monitor for cancellation requests. The default is .
/// The result of the request.
/// is .
/// The request failed or the server returned an error response.
[Obsolete(Obsoletions.DeprecatedLogging_Message, DiagnosticId = Obsoletions.Deprecated_DiagnosticId, UrlFormat = Obsoletions.Deprecated_Url)]
public Task SetLoggingLevelAsync(
SetLevelRequestParams requestParams,
CancellationToken cancellationToken = default)
{
Throw.IfNull(requestParams);
return SendRequestAsync(
RequestMethods.LoggingSetLevel,
requestParams,
McpJsonUtilities.JsonContext.Default.SetLevelRequestParams,
McpJsonUtilities.JsonContext.Default.EmptyResult,
cancellationToken: cancellationToken).AsTask();
}
///
/// Retrieves the current state of a task from the server.
///
/// The stable identifier of the task to retrieve.
/// The to monitor for cancellation requests. The default is .
/// A subtype representing the current task state.
/// is .
/// The request failed or the server returned an error response.
public ValueTask GetTaskAsync(
string taskId,
CancellationToken cancellationToken = default)
{
Throw.IfNull(taskId);
return GetTaskAsync(new GetTaskRequestParams { TaskId = taskId }, cancellationToken);
}
///
/// Retrieves the current state of a task from the server.
///
/// The request parameters to send in the request.
/// The to monitor for cancellation requests. The default is .
/// A subtype representing the current task state.
/// is .
/// The request failed or the server returned an error response.
public ValueTask GetTaskAsync(
GetTaskRequestParams requestParams,
CancellationToken cancellationToken = default)
{
Throw.IfNull(requestParams);
return SendRequestAsync(
RequestMethods.TasksGet,
requestParams,
McpJsonUtilities.JsonContext.Default.GetTaskRequestParams,
McpJsonUtilities.JsonContext.Default.GetTaskResult,
cancellationToken: cancellationToken);
}
///
/// Provides input responses to a task that is in the state.
///
/// The request parameters containing the task ID and input responses.
/// The to monitor for cancellation requests. The default is .
/// The result acknowledging the update.
/// is .
/// The request failed or the server returned an error response.
public ValueTask UpdateTaskAsync(
UpdateTaskRequestParams requestParams,
CancellationToken cancellationToken = default)
{
Throw.IfNull(requestParams);
return SendRequestAsync(
RequestMethods.TasksUpdate,
requestParams,
McpJsonUtilities.JsonContext.Default.UpdateTaskRequestParams,
McpJsonUtilities.JsonContext.Default.UpdateTaskResult,
cancellationToken: cancellationToken);
}
///
/// Requests cancellation of an in-progress task on the server.
///
/// The stable identifier of the task to cancel.
/// The to monitor for cancellation requests. The default is .
/// The result acknowledging the cancellation request.
/// is .
/// The request failed or the server returned an error response.
public ValueTask CancelTaskAsync(
string taskId,
CancellationToken cancellationToken = default)
{
Throw.IfNull(taskId);
return CancelTaskAsync(new CancelTaskRequestParams { TaskId = taskId }, cancellationToken);
}
///
/// Requests cancellation of an in-progress task on the server.
///
/// The request parameters to send in the request.
/// The to monitor for cancellation requests. The default is .
/// The result acknowledging the cancellation request.
/// is .
/// The request failed or the server returned an error response.
public ValueTask CancelTaskAsync(
CancelTaskRequestParams requestParams,
CancellationToken cancellationToken = default)
{
Throw.IfNull(requestParams);
return SendRequestAsync(
RequestMethods.TasksCancel,
requestParams,
McpJsonUtilities.JsonContext.Default.CancelTaskRequestParams,
McpJsonUtilities.JsonContext.Default.CancelTaskResult,
cancellationToken: cancellationToken);
}
/// Converts a dictionary with values to a dictionary with values.
private static Dictionary? ToArgumentsDictionary(
IReadOnlyDictionary? arguments, JsonSerializerOptions options)
{
var typeInfo = options.GetTypeInfo();
Dictionary? result = null;
if (arguments is not null)
{
result = new(arguments.Count);
foreach (var kvp in arguments)
{
result.Add(kvp.Key, kvp.Value is JsonElement je ? je : JsonSerializer.SerializeToElement(kvp.Value, typeInfo));
}
}
return result;
}
// Per SEP-2663 ยง51, the per-request opt-in uses the SEP-2575 capabilities envelope:
// _meta/io.modelcontextprotocol/clientCapabilities/extensions/io.modelcontextprotocol/tasks = {}
// TODO: replace the literal with a shared NotificationMethods.ClientCapabilitiesMetaKey once
// the SEP-2575 plumbing lands and drop the local consts.
private const string ClientCapabilitiesMetaKey = "io.modelcontextprotocol/clientCapabilities";
private const string ExtensionsKey = "extensions";
private static JsonObject GetMetaWithTaskCapability(JsonObject? existingMeta)
{
JsonObject meta = existingMeta is not null
? (JsonObject)existingMeta.DeepClone()
: [];
if (meta[ClientCapabilitiesMetaKey] is not JsonObject capsRoot)
{
capsRoot = [];
meta[ClientCapabilitiesMetaKey] = capsRoot;
}
if (capsRoot[ExtensionsKey] is not JsonObject extensionsRoot)
{
extensionsRoot = [];
capsRoot[ExtensionsKey] = extensionsRoot;
}
extensionsRoot.TryAdd(McpExtensions.Tasks, new JsonObject());
return meta;
}
}