Co-authored-by: Stephen Halter <halter73@gmail.com> Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Co-authored-by: tarekgh <tarek.mahmoud.sayed@gmail.com>
This commit is contained in:
parent
6787c0cc19
commit
26f60ab7f4
10 changed files with 894 additions and 101 deletions
|
|
@ -733,6 +733,12 @@ public abstract partial class McpClient : McpSession
|
|||
/// For a simpler API that handles both subscription and notification registration in a single call,
|
||||
/// use <see cref="SubscribeToResourceAsync(Uri, Func{ResourceUpdatedNotificationParams, CancellationToken, ValueTask}, RequestOptions?, CancellationToken)"/>.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// The 2026-07-28 protocol revision (SEP-2575) removed <c>resources/subscribe</c> in favor of
|
||||
/// <see cref="RequestMethods.SubscriptionsListen"/> with <c>resourceSubscriptions</c>. On a session that
|
||||
/// negotiated that revision or later, this method throws an <see cref="McpException"/> with
|
||||
/// <see cref="McpErrorCode.MethodNotFound"/>.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public Task SubscribeToResourceAsync(
|
||||
SubscribeRequestParams requestParams,
|
||||
|
|
@ -931,6 +937,12 @@ public abstract partial class McpClient : McpSession
|
|||
/// <returns>The result of the request.</returns>
|
||||
/// <exception cref="ArgumentNullException"><paramref name="requestParams"/> is <see langword="null"/>.</exception>
|
||||
/// <exception cref="McpException">The request failed or the server returned an error response.</exception>
|
||||
/// <remarks>
|
||||
/// The 2026-07-28 protocol revision (SEP-2575) removed <c>resources/unsubscribe</c> in favor of
|
||||
/// <see cref="RequestMethods.SubscriptionsListen"/> with <c>resourceSubscriptions</c>. On a session that
|
||||
/// negotiated that revision or later, this method throws an <see cref="McpException"/> with
|
||||
/// <see cref="McpErrorCode.MethodNotFound"/>.
|
||||
/// </remarks>
|
||||
public Task UnsubscribeFromResourceAsync(
|
||||
UnsubscribeRequestParams requestParams,
|
||||
CancellationToken cancellationToken = default)
|
||||
|
|
|
|||
|
|
@ -29,14 +29,16 @@ public abstract partial class McpClient : McpSession
|
|||
/// <remarks>
|
||||
/// <para>
|
||||
/// This property provides identification details about the connected server, including its name and version.
|
||||
/// It is populated during the initialization handshake and is available after a successful connection.
|
||||
/// It is populated during the initialization handshake or from <c>server/discover</c> result metadata.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// This information can be useful for logging, debugging, compatibility checks, and displaying server
|
||||
/// information to users.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
/// <exception cref="InvalidOperationException">The client is not connected.</exception>
|
||||
/// <exception cref="InvalidOperationException">
|
||||
/// The client is not connected, or the server omitted the optional server identity metadata.
|
||||
/// </exception>
|
||||
public abstract Implementation ServerInfo { get; }
|
||||
|
||||
/// <summary>
|
||||
|
|
|
|||
|
|
@ -170,7 +170,8 @@ internal sealed partial class McpClientImpl : McpClient
|
|||
public override ServerCapabilities ServerCapabilities => _serverCapabilities ?? throw new InvalidOperationException("The client is not connected.");
|
||||
|
||||
/// <inheritdoc/>
|
||||
public override Implementation ServerInfo => _serverInfo ?? throw new InvalidOperationException("The client is not connected.");
|
||||
public override Implementation ServerInfo => _serverInfo ?? throw new InvalidOperationException(
|
||||
"The client is not connected, or the connected server did not provide optional server identity metadata.");
|
||||
|
||||
/// <inheritdoc/>
|
||||
public override string? ServerInstructions => _serverInstructions;
|
||||
|
|
@ -432,15 +433,19 @@ internal sealed partial class McpClientImpl : McpClient
|
|||
}
|
||||
else
|
||||
{
|
||||
var discoveredServerInfo = GetServerInfoFromDiscover(discoverResult!);
|
||||
|
||||
if (_logger.IsEnabled(LogLevel.Information))
|
||||
{
|
||||
LogServerCapabilitiesReceived(_endpointName,
|
||||
capabilities: JsonSerializer.Serialize(discoverResult!.Capabilities, McpJsonUtilities.JsonContext.Default.ServerCapabilities),
|
||||
serverInfo: JsonSerializer.Serialize(discoverResult.ServerInfo, McpJsonUtilities.JsonContext.Default.Implementation));
|
||||
serverInfo: discoveredServerInfo is null
|
||||
? "(none)"
|
||||
: JsonSerializer.Serialize(discoveredServerInfo, McpJsonUtilities.JsonContext.Default.Implementation));
|
||||
}
|
||||
|
||||
_serverCapabilities = discoverResult!.Capabilities;
|
||||
_serverInfo = discoverResult.ServerInfo;
|
||||
_serverInfo = discoveredServerInfo;
|
||||
_serverInstructions = discoverResult.Instructions;
|
||||
}
|
||||
|
||||
|
|
@ -484,6 +489,30 @@ internal sealed partial class McpClientImpl : McpClient
|
|||
LogClientConnected(_endpointName);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Resolves the server identity from a <c>server/discover</c> result. The 2026-07-28 revision carries
|
||||
/// <c>serverInfo</c> in the result's <c>_meta/io.modelcontextprotocol/serverInfo</c> field rather than
|
||||
/// the result body. Identity is optional, so a server that omits it yields <see langword="null"/>.
|
||||
/// </summary>
|
||||
private static Implementation? GetServerInfoFromDiscover(DiscoverResult discoverResult)
|
||||
{
|
||||
if (discoverResult.Meta is { } meta &&
|
||||
meta.TryGetPropertyValue(MetaKeys.ServerInfo, out JsonNode? serverInfoNode))
|
||||
{
|
||||
if (serverInfoNode is null)
|
||||
{
|
||||
throw new JsonException(
|
||||
$"Discover result metadata '{MetaKeys.ServerInfo}' must contain a server implementation.");
|
||||
}
|
||||
|
||||
return JsonSerializer.Deserialize(serverInfoNode, McpJsonUtilities.JsonContext.Default.Implementation)
|
||||
?? throw new JsonException(
|
||||
$"Discover result metadata '{MetaKeys.ServerInfo}' must contain a server implementation.");
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Performs the initialize handshake (initialize request + initialized notification),
|
||||
/// records the negotiated protocol version, and stores the server capabilities/info/instructions.
|
||||
|
|
|
|||
|
|
@ -135,7 +135,7 @@ public sealed class McpClientTool : AIFunction
|
|||
// would lose that information. So, we only do the translation if there is no additional information to preserve.
|
||||
if (result.IsError is not true &&
|
||||
result.StructuredContent is null &&
|
||||
result.Meta is not { Count: > 0 })
|
||||
!HasApplicationResultMetadata(result.Meta))
|
||||
{
|
||||
switch (result.Content.Count)
|
||||
{
|
||||
|
|
@ -150,6 +150,26 @@ public sealed class McpClientTool : AIFunction
|
|||
return JsonSerializer.SerializeToElement(result, McpJsonUtilities.JsonContext.Default.CallToolResult);
|
||||
}
|
||||
|
||||
private static bool HasApplicationResultMetadata(JsonObject? meta)
|
||||
{
|
||||
if (meta is null)
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
foreach (var property in meta)
|
||||
{
|
||||
// Server identity is protocol metadata already exposed through McpClient.ServerInfo.
|
||||
// It should not force an otherwise simple tool result into its JSON envelope.
|
||||
if (!string.Equals(property.Key, MetaKeys.ServerInfo, StringComparison.Ordinal))
|
||||
{
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Invokes the tool on the server.
|
||||
/// </summary>
|
||||
|
|
|
|||
|
|
@ -31,12 +31,6 @@ public sealed class DiscoverResult : Result, ICacheableResult
|
|||
[JsonPropertyName("capabilities")]
|
||||
public required ServerCapabilities Capabilities { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets information about the server implementation.
|
||||
/// </summary>
|
||||
[JsonPropertyName("serverInfo")]
|
||||
public required Implementation ServerInfo { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets optional instructions describing how to use the server and its features.
|
||||
/// </summary>
|
||||
|
|
|
|||
|
|
@ -24,6 +24,16 @@ public static class MetaKeys
|
|||
/// </remarks>
|
||||
public const string ClientInfo = "io.modelcontextprotocol/clientInfo";
|
||||
|
||||
/// <summary>
|
||||
/// The metadata key used to identify the server software in a response's <c>_meta</c> field.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Introduced by the 2026-07-28 protocol revision. Servers SHOULD identify themselves by
|
||||
/// carrying an <see cref="Protocol.Implementation"/> under this key in every result's
|
||||
/// <c>_meta</c>.
|
||||
/// </remarks>
|
||||
public const string ServerInfo = "io.modelcontextprotocol/serverInfo";
|
||||
|
||||
/// <summary>
|
||||
/// The metadata key used to declare client capabilities in a request's <c>_meta</c> field.
|
||||
/// </summary>
|
||||
|
|
|
|||
|
|
@ -146,10 +146,12 @@ internal sealed partial class McpServerImpl : McpServer
|
|||
ServerCapabilities.Resources.ListChanged = null;
|
||||
}
|
||||
|
||||
// And initialize the session. The built-in meta-reading filter runs ahead of any
|
||||
// user-supplied incoming filters; see PrependMetaReadingFilter for what it records and why.
|
||||
// And initialize the session. The built-in protocol metadata filters run ahead of any
|
||||
// user-supplied message filters.
|
||||
var incomingMessageFilter = PrependMetaReadingFilter(BuildMessageFilterPipeline(options.Filters.Message.IncomingFilters));
|
||||
var outgoingMessageFilter = BuildMessageFilterPipeline(options.Filters.Message.OutgoingFilters);
|
||||
var outgoingMessageFilter = PrependServerInfoFilter(
|
||||
BuildMessageFilterPipeline(options.Filters.Message.OutgoingFilters),
|
||||
options.ServerInfo ?? DefaultImplementation);
|
||||
|
||||
_sessionHandler = new McpSessionHandler(
|
||||
isServer: true,
|
||||
|
|
@ -169,12 +171,13 @@ internal sealed partial class McpServerImpl : McpServer
|
|||
/// validates the per-request protocol version, before delegating to the user-supplied incoming filters.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Under the 2026-07-28 protocol revision (SEP-2575) there is no <c>initialize</c> handshake, so these values
|
||||
/// MUST be populated per-request. Per-request client capabilities and client info are consumed request-scoped
|
||||
/// by <see cref="DestinationBoundMcpServer"/> and are not read from server-wide state by request handlers. The
|
||||
/// shared <see cref="_clientInfo"/> write below is best-effort and used only to derive the session endpoint
|
||||
/// name for logging/telemetry. For initialize-handshake clients the per-request values are absent and the built-in
|
||||
/// filter is a no-op (the values were captured during the initialize handler).
|
||||
/// Under the 2026-07-28 protocol revision (SEP-2575) there is no <c>initialize</c> handshake, so the protocol
|
||||
/// version and client capabilities MUST be populated per-request. Client info is optional. Per-request client
|
||||
/// capabilities and client info are consumed request-scoped by <see cref="DestinationBoundMcpServer"/> and are
|
||||
/// not read from server-wide state by request handlers. The shared <see cref="_clientInfo"/> write below is
|
||||
/// best-effort and used only to derive the session endpoint name for logging/telemetry. For initialize-handshake
|
||||
/// clients the per-request values are absent and the built-in filter is a no-op (the values were captured during
|
||||
/// the initialize handler).
|
||||
/// </remarks>
|
||||
private JsonRpcMessageFilter PrependMetaReadingFilter(JsonRpcMessageFilter inner)
|
||||
{
|
||||
|
|
@ -204,9 +207,14 @@ internal sealed partial class McpServerImpl : McpServer
|
|||
// with an UnsupportedProtocolVersionError (-32022) carrying the supported list.
|
||||
if (!_supportedProtocolVersions.Contains(protocolVersion))
|
||||
{
|
||||
var supportedVersions =
|
||||
hasProtocolVersionMeta && _perRequestMetadataProtocolVersions.Length > 0 ?
|
||||
_perRequestMetadataProtocolVersions :
|
||||
_supportedProtocolVersions;
|
||||
|
||||
throw new UnsupportedProtocolVersionException(
|
||||
requested: protocolVersion,
|
||||
supported: _supportedProtocolVersions);
|
||||
supported: supportedVersions);
|
||||
}
|
||||
|
||||
if (McpProtocolVersions.RequiresPerRequestMetadata(protocolVersion))
|
||||
|
|
@ -214,7 +222,6 @@ internal sealed partial class McpServerImpl : McpServer
|
|||
ValidateRequiredPerRequestMetadata(
|
||||
protocolVersion,
|
||||
hasProtocolVersionMeta,
|
||||
context.ClientInfo is not null,
|
||||
context.ClientCapabilities is not null);
|
||||
}
|
||||
else if (McpProtocolVersions.SupportsInitializeHandshake(protocolVersion))
|
||||
|
|
@ -293,7 +300,6 @@ internal sealed partial class McpServerImpl : McpServer
|
|||
private static void ValidateRequiredPerRequestMetadata(
|
||||
string protocolVersion,
|
||||
bool hasProtocolVersionMeta,
|
||||
bool hasClientInfoMeta,
|
||||
bool hasClientCapabilitiesMeta)
|
||||
{
|
||||
if (!hasProtocolVersionMeta)
|
||||
|
|
@ -301,10 +307,7 @@ internal sealed partial class McpServerImpl : McpServer
|
|||
ThrowMissingPerRequestMetadata(protocolVersion, MetaKeys.ProtocolVersion);
|
||||
}
|
||||
|
||||
if (!hasClientInfoMeta)
|
||||
{
|
||||
ThrowMissingPerRequestMetadata(protocolVersion, MetaKeys.ClientInfo);
|
||||
}
|
||||
// clientInfo is optional: requests whose _meta omits it are served, not rejected.
|
||||
|
||||
if (!hasClientCapabilitiesMeta)
|
||||
{
|
||||
|
|
@ -344,8 +347,46 @@ internal sealed partial class McpServerImpl : McpServer
|
|||
paramsObj["_meta"] is JsonObject metaObj &&
|
||||
metaObj.ContainsKey(key);
|
||||
|
||||
/// <summary>
|
||||
/// Adds the server identity to every successful result on per-request-metadata protocol revisions.
|
||||
/// The filter runs before application filters so they can inspect or intentionally remove the metadata.
|
||||
/// </summary>
|
||||
private JsonRpcMessageFilter PrependServerInfoFilter(JsonRpcMessageFilter inner, Implementation serverInfo)
|
||||
{
|
||||
JsonRpcMessageFilter serverInfoFilter = next => async (message, cancellationToken) =>
|
||||
{
|
||||
if (message is JsonRpcResponse { Result: JsonObject result } &&
|
||||
McpProtocolVersions.RequiresPerRequestMetadata(
|
||||
message.Context?.ProtocolVersion ?? _negotiatedProtocolVersion))
|
||||
{
|
||||
if (result["_meta"] is not JsonObject meta)
|
||||
{
|
||||
meta = new JsonObject();
|
||||
result["_meta"] = meta;
|
||||
}
|
||||
|
||||
meta[MetaKeys.ServerInfo] = JsonSerializer.SerializeToNode(
|
||||
serverInfo,
|
||||
McpJsonUtilities.JsonContext.Default.Implementation);
|
||||
}
|
||||
|
||||
await next(message, cancellationToken).ConfigureAwait(false);
|
||||
};
|
||||
|
||||
return next => serverInfoFilter(inner(next));
|
||||
}
|
||||
|
||||
private void ValidateInitializeRequestBoundary(JsonRpcRequest request)
|
||||
{
|
||||
// Per-request-metadata revisions (SEP-2575) removed the initialize handshake entirely:
|
||||
// the request is for a method the server does not implement on that revision.
|
||||
if (McpProtocolVersions.RequiresPerRequestMetadata(request.Context?.ProtocolVersion))
|
||||
{
|
||||
throw new McpProtocolException(
|
||||
$"Method '{RequestMethods.Initialize}' is not available on protocol version '{request.Context?.ProtocolVersion}'. Use '{RequestMethods.ServerDiscover}' and per-request metadata instead.",
|
||||
McpErrorCode.MethodNotFound);
|
||||
}
|
||||
|
||||
if (request.Context?.ProtocolVersion is { } protocolVersion &&
|
||||
!McpProtocolVersions.SupportsInitializeHandshake(protocolVersion))
|
||||
{
|
||||
|
|
@ -415,14 +456,32 @@ internal sealed partial class McpServerImpl : McpServer
|
|||
McpErrorCode.MethodNotFound);
|
||||
}
|
||||
|
||||
if (usesPerRequestMetadata && request.Method == RequestMethods.LoggingSetLevel)
|
||||
if (usesPerRequestMetadata &&
|
||||
request.Method is RequestMethods.Ping or RequestMethods.LoggingSetLevel
|
||||
or RequestMethods.ResourcesSubscribe or RequestMethods.ResourcesUnsubscribe)
|
||||
{
|
||||
var replacement = GetRemovedMethodReplacementHint(request.Method);
|
||||
throw new McpProtocolException(
|
||||
$"The method '{RequestMethods.LoggingSetLevel}' is not available on protocol version '{request.Context?.ProtocolVersion ?? NegotiatedProtocolVersion}'. Use per-request _meta/{MetaKeys.LogLevel} instead.",
|
||||
$"The method '{request.Method}' is not available on protocol version '{request.Context?.ProtocolVersion ?? NegotiatedProtocolVersion}'." +
|
||||
(replacement is null ? "" : $" {replacement}"),
|
||||
McpErrorCode.MethodNotFound);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Returns guidance on the per-request-metadata replacement for a method that SEP-2575 removed,
|
||||
/// or <see langword="null"/> when the method has no direct replacement. Surfaced in the
|
||||
/// <see cref="McpErrorCode.MethodNotFound"/> error so a client that still calls the legacy RPC
|
||||
/// (for example <c>resources/subscribe</c>) learns how to migrate.
|
||||
/// </summary>
|
||||
private static string? GetRemovedMethodReplacementHint(string method) => method switch
|
||||
{
|
||||
RequestMethods.LoggingSetLevel => $"Use the per-request '_meta/{MetaKeys.LogLevel}' field instead.",
|
||||
RequestMethods.ResourcesSubscribe or RequestMethods.ResourcesUnsubscribe =>
|
||||
$"Use '{RequestMethods.SubscriptionsListen}' with 'resourceSubscriptions' instead.",
|
||||
_ => null,
|
||||
};
|
||||
|
||||
/// <inheritdoc/>
|
||||
public override string? SessionId => _sessionTransport.SessionId;
|
||||
|
||||
|
|
@ -636,7 +695,6 @@ internal sealed partial class McpServerImpl : McpServer
|
|||
{
|
||||
SupportedVersions = [.. _perRequestMetadataProtocolVersions],
|
||||
Capabilities = ServerCapabilities ?? new(),
|
||||
ServerInfo = options.ServerInfo ?? DefaultImplementation,
|
||||
Instructions = options.ServerInstructions,
|
||||
// Spec PR #2855 makes ttlMs and cacheScope required on DiscoverResult. Default to
|
||||
// the safest values (immediately stale, not shareable) so existing servers keep
|
||||
|
|
|
|||
|
|
@ -14,7 +14,8 @@ public sealed class McpServerOptions
|
|||
/// Gets or sets information about this server implementation, including its name and version.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// This information is sent to the client during initialization or discovery to identify the server.
|
||||
/// This information is sent in the initialization result on handshake-based protocol revisions and in
|
||||
/// every successful result's metadata on per-request-metadata revisions.
|
||||
/// It's displayed in client logs and can be used for debugging and compatibility checks.
|
||||
/// </remarks>
|
||||
public Implementation? ServerInfo { get; set; }
|
||||
|
|
|
|||
|
|
@ -15,7 +15,8 @@ internal sealed partial class StreamableHttpPostTransport(
|
|||
StreamableHttpServerTransport parentTransport,
|
||||
Stream responseStream,
|
||||
CancellationToken sessionCancellationToken,
|
||||
ILogger logger) : ITransport
|
||||
ILogger logger,
|
||||
Func<JsonRpcMessage?, ValueTask>? onResponseStarting = null) : ITransport
|
||||
{
|
||||
private readonly SemaphoreSlim _messageLock = new(1, 1);
|
||||
private readonly TaskCompletionSource<bool> _httpResponseTcs = new(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
|
|
@ -29,6 +30,7 @@ internal sealed partial class StreamableHttpPostTransport(
|
|||
private RequestId _pendingRequest;
|
||||
private bool _finalResponseMessageSent;
|
||||
private bool _httpResponseCompleted;
|
||||
private bool _httpResponseStarted;
|
||||
|
||||
public ChannelReader<JsonRpcMessage> MessageReader => throw new NotSupportedException("JsonRpcMessage.Context.RelatedTransport should only be used for sending messages.");
|
||||
|
||||
|
|
@ -50,8 +52,13 @@ internal sealed partial class StreamableHttpPostTransport(
|
|||
_pendingRequest = request.Id;
|
||||
message.Context.RelatedTransport = this;
|
||||
|
||||
// Invoke the initialize request handler if applicable.
|
||||
if (request.Method == RequestMethods.Initialize)
|
||||
// Invoke the initialize request handler if applicable. On a per-request-metadata
|
||||
// protocol revision (2026-07-28+) initialize is a removed method: skip the eager
|
||||
// params deserialization (whose required properties would throw on an arbitrary
|
||||
// payload) and let the session's protocol boundary reject the request with
|
||||
// Method not found.
|
||||
if (request.Method == RequestMethods.Initialize &&
|
||||
!McpProtocolVersions.RequiresPerRequestMetadata(message.Context.ProtocolVersion))
|
||||
{
|
||||
var initializeRequest = JsonSerializer.Deserialize(request.Params, McpJsonUtilities.JsonContext.Default.InitializeRequestParams);
|
||||
await parentTransport.HandleInitializeRequestAsync(initializeRequest).ConfigureAwait(false);
|
||||
|
|
@ -69,32 +76,117 @@ internal sealed partial class StreamableHttpPostTransport(
|
|||
return false;
|
||||
}
|
||||
|
||||
CancellationTokenSource? deferredFlushCts = null;
|
||||
Task? deferredFlushTask = null;
|
||||
bool deferHeaderFlush = false;
|
||||
using (await _messageLock.LockAsync(cancellationToken).ConfigureAwait(false))
|
||||
{
|
||||
var primingItem = await TryStartSseEventStreamAsync(_pendingRequest).ConfigureAwait(false);
|
||||
if (primingItem.HasValue)
|
||||
{
|
||||
await NotifyResponseStartingAsync(firstMessage: null).ConfigureAwait(false);
|
||||
await _httpSseWriter.WriteAsync(primingItem.Value, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
else
|
||||
else if (onResponseStarting is null)
|
||||
{
|
||||
// If there's no priming write, flush the stream to ensure HTTP response headers are
|
||||
// sent to the client now that the server is ready to process the request.
|
||||
// This prevents HttpClient timeout for long-running requests.
|
||||
await responseStream.FlushAsync(cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
else
|
||||
{
|
||||
deferHeaderFlush = true;
|
||||
}
|
||||
|
||||
// Ensure that we've sent the priming event before processing the incoming request.
|
||||
await parentTransport.MessageWriter.WriteAsync(message, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
// Wait for the response to be written before returning from the handler.
|
||||
// This keeps the HTTP response open until the final response message is sent.
|
||||
await _httpResponseTcs.Task.WaitAsync(cancellationToken).ConfigureAwait(false);
|
||||
if (deferHeaderFlush)
|
||||
{
|
||||
// Defer the flush (and the header commit it implies) so the callback can still choose
|
||||
// the HTTP status line for an immediate JSON-RPC error. Start the bounded grace period
|
||||
// only after the request has been queued for dispatch.
|
||||
deferredFlushCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
|
||||
deferredFlushTask = DeferredHeaderFlushAsync(deferredFlushCts.Token);
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
// Wait for the response to be written before returning from the handler.
|
||||
// This keeps the HTTP response open until the final response message is sent.
|
||||
await _httpResponseTcs.Task.WaitAsync(cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
finally
|
||||
{
|
||||
if (deferredFlushCts is not null)
|
||||
{
|
||||
deferredFlushCts.Cancel();
|
||||
await deferredFlushTask!.ConfigureAwait(false);
|
||||
deferredFlushCts.Dispose();
|
||||
}
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Bounds the deferred header flush: after a short grace window, flushes the response headers
|
||||
/// if no response message has been written yet. Immediate rejections land well inside the
|
||||
/// window, so the response-starting callback can still map their JSON-RPC error codes onto the
|
||||
/// HTTP status line; a handler that runs longer commits the headers here so clients see them
|
||||
/// promptly (long-running tool calls must not trip HttpClient's response timeout).
|
||||
/// </summary>
|
||||
private async Task DeferredHeaderFlushAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
try
|
||||
{
|
||||
await Task.Delay(DeferredHeaderFlushGrace, cancellationToken).ConfigureAwait(false);
|
||||
using var _ = await _messageLock.LockAsync(cancellationToken).ConfigureAwait(false);
|
||||
if (!_httpResponseStarted && !_httpResponseCompleted)
|
||||
{
|
||||
await NotifyResponseStartingAsync(firstMessage: null).ConfigureAwait(false);
|
||||
await responseStream.FlushAsync(cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
|
||||
{
|
||||
// The response was written or the request ended before the grace window elapsed.
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
// Surface the failure to the awaiting HandlePostAsync when possible. If the response
|
||||
// future has already been resolved (the response started or completed on another path),
|
||||
// TrySetException is a no-op, so log here to keep the deferred-flush failure diagnosable.
|
||||
if (!_httpResponseTcs.TrySetException(ex))
|
||||
{
|
||||
LogDeferredHeaderFlushFailed(ex);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>How long the response-header flush may be deferred waiting for the first response message.</summary>
|
||||
internal static readonly TimeSpan DeferredHeaderFlushGrace = TimeSpan.FromMilliseconds(250);
|
||||
|
||||
/// <summary>
|
||||
/// Invokes the response-starting callback exactly once, immediately before the first write to
|
||||
/// the HTTP response stream, so the HTTP application can still set the response status line.
|
||||
/// </summary>
|
||||
private async ValueTask NotifyResponseStartingAsync(JsonRpcMessage? firstMessage)
|
||||
{
|
||||
if (_httpResponseStarted)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
_httpResponseStarted = true;
|
||||
if (onResponseStarting is not null)
|
||||
{
|
||||
await onResponseStarting(firstMessage).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
|
||||
public async Task SendMessageAsync(JsonRpcMessage message, CancellationToken cancellationToken = default)
|
||||
{
|
||||
Throw.IfNull(message);
|
||||
|
|
@ -130,6 +222,7 @@ internal sealed partial class StreamableHttpPostTransport(
|
|||
|
||||
try
|
||||
{
|
||||
await NotifyResponseStartingAsync(message).ConfigureAwait(false);
|
||||
await _httpSseWriter.WriteAsync(item, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
catch (Exception ex) when (!cancellationToken.IsCancellationRequested)
|
||||
|
|
@ -173,6 +266,7 @@ internal sealed partial class StreamableHttpPostTransport(
|
|||
// Write to the response stream if it still exists.
|
||||
if (!_httpResponseCompleted)
|
||||
{
|
||||
await NotifyResponseStartingAsync(firstMessage: null).ConfigureAwait(false);
|
||||
await _httpSseWriter.WriteAsync(primingItem, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
|
|
@ -248,4 +342,7 @@ internal sealed partial class StreamableHttpPostTransport(
|
|||
|
||||
[LoggerMessage(Level = LogLevel.Warning, Message = "Failed to dispose SSE event stream writer.")]
|
||||
private partial void LogStoreStreamDisposalFailed(Exception exception);
|
||||
|
||||
[LoggerMessage(Level = LogLevel.Warning, Message = "Failed to flush deferred Streamable HTTP response headers.")]
|
||||
private partial void LogDeferredHeaderFlushFailed(Exception exception);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -125,7 +125,7 @@ public sealed partial class StreamableHttpServerTransport : ITransport
|
|||
/// to the SSE response stream until cancellation is requested or the transport is disposed.
|
||||
/// </summary>
|
||||
/// <param name="sseResponseStream">The response stream to write MCP JSON-RPC messages as SSE events to.</param>
|
||||
/// <param name="cancellationToken">The <see cref="CancellationToken"/> to monitor for cancellation requests. The default is <see cref="CancellationToken.None"/>.</param>
|
||||
/// <param name="cancellationToken">The <see cref="CancellationToken"/> to monitor for cancellation requests.</param>
|
||||
/// <returns>A task representing the send loop that writes JSON-RPC messages to the SSE response stream.</returns>
|
||||
/// <exception cref="ArgumentNullException"><paramref name="sseResponseStream"/> is <see langword="null"/>.</exception>
|
||||
/// <exception cref="InvalidOperationException">
|
||||
|
|
@ -207,12 +207,39 @@ public sealed partial class StreamableHttpServerTransport : ITransport
|
|||
/// If an authenticated <see cref="ClaimsPrincipal"/> sent the message, that can be included in the <see cref="JsonRpcMessage.Context"/>.
|
||||
/// No other part of the context should be set.
|
||||
/// </remarks>
|
||||
public async Task<bool> HandlePostRequestAsync(JsonRpcMessage message, Stream responseStream, CancellationToken cancellationToken = default)
|
||||
public Task<bool> HandlePostRequestAsync(JsonRpcMessage message, Stream responseStream, CancellationToken cancellationToken = default)
|
||||
=> HandlePostRequestAsync(message, responseStream, onResponseStarting: null, cancellationToken);
|
||||
|
||||
/// <summary>
|
||||
/// Handles a Streamable HTTP POST request, processing the JSON-RPC message and writing any
|
||||
/// JSON-RPC responses to the response stream.
|
||||
/// This overload additionally reports the first JSON-RPC message written to the response via
|
||||
/// <paramref name="onResponseStarting"/>, before any response bytes are written, so the HTTP
|
||||
/// application can still choose the response status line (SEP-2575 maps some JSON-RPC error
|
||||
/// codes to HTTP statuses). When <paramref name="onResponseStarting"/> is provided, the eager
|
||||
/// response-header flush that normally precedes request processing is deferred until that first
|
||||
/// message; the callback receives <see langword="null"/> when the first write is not a JSON-RPC
|
||||
/// message (e.g. a resumability priming event).
|
||||
/// The status line can only be influenced by the FIRST write: when a handler streams a
|
||||
/// notification (e.g. progress) before failing, or runs past the transport's bounded
|
||||
/// header-flush grace window, the status is already committed and a later JSON-RPC error
|
||||
/// rides the committed status.
|
||||
/// </summary>
|
||||
/// <param name="message">The JSON-RPC message to process.</param>
|
||||
/// <param name="responseStream">The response stream to write any JSON-RPC responses to.</param>
|
||||
/// <param name="onResponseStarting">Callback invoked once, immediately before the first write to <paramref name="responseStream"/>.</param>
|
||||
/// <param name="cancellationToken">The <see cref="CancellationToken"/> to monitor for cancellation requests. The default is <see cref="CancellationToken.None"/>.</param>
|
||||
/// <returns>
|
||||
/// <see langword="true"/> if data was written to the response body.
|
||||
/// <see langword="false"/> if nothing was written because the request body did not contain any <see cref="JsonRpcRequest"/> messages to respond to.
|
||||
/// </returns>
|
||||
/// <exception cref="ArgumentNullException"><paramref name="message"/> or <paramref name="responseStream"/> is <see langword="null"/>.</exception>
|
||||
public async Task<bool> HandlePostRequestAsync(JsonRpcMessage message, Stream responseStream, Func<JsonRpcMessage?, ValueTask>? onResponseStarting, CancellationToken cancellationToken)
|
||||
{
|
||||
Throw.IfNull(message);
|
||||
Throw.IfNull(responseStream);
|
||||
|
||||
var postTransport = new StreamableHttpPostTransport(this, responseStream, _transportDisposedCts.Token, _logger);
|
||||
var postTransport = new StreamableHttpPostTransport(this, responseStream, _transportDisposedCts.Token, _logger, onResponseStarting);
|
||||
using var postCts = CancellationTokenSource.CreateLinkedTokenSource(_transportDisposedCts.Token, cancellationToken);
|
||||
await using (postTransport.ConfigureAwait(false))
|
||||
{
|
||||
|
|
|
|||
Loading…
Reference in a new issue