Add diagnostics for messages dropped on the GET SSE stream (#1609)
This commit is contained in:
parent
acb4cc9fd1
commit
ed07fdb6f4
3 changed files with 255 additions and 4 deletions
|
|
@ -54,9 +54,18 @@ public abstract partial class McpServer : McpSession
|
||||||
/// <exception cref="InvalidOperationException">The client does not support sampling.</exception>
|
/// <exception cref="InvalidOperationException">The client does not support sampling.</exception>
|
||||||
/// <exception cref="McpException">The request failed or the client returned an error response.</exception>
|
/// <exception cref="McpException">The request failed or the client returned an error response.</exception>
|
||||||
/// <remarks>
|
/// <remarks>
|
||||||
|
/// <para>
|
||||||
|
/// When the server is using the Streamable HTTP transport, prefer calling this method on the
|
||||||
|
/// <see cref="McpServer"/> instance available via <c>RequestContext</c> from inside a tool, prompt,
|
||||||
|
/// or resource handler. That routes the request through the originating POST response stream via
|
||||||
|
/// <see cref="JsonRpcMessageContext.RelatedTransport"/>, which is always open for the duration of
|
||||||
|
/// the request, rather than relying on the optional standalone GET SSE stream.
|
||||||
|
/// </para>
|
||||||
|
/// <para>
|
||||||
/// When called during task-augmented tool execution, this method automatically updates the task
|
/// When called during task-augmented tool execution, this method automatically updates the task
|
||||||
/// status to <see cref="McpTaskStatus.InputRequired"/> while waiting for the client response,
|
/// status to <see cref="McpTaskStatus.InputRequired"/> while waiting for the client response,
|
||||||
/// then returns to <see cref="McpTaskStatus.Working"/> when the response is received.
|
/// then returns to <see cref="McpTaskStatus.Working"/> when the response is received.
|
||||||
|
/// </para>
|
||||||
/// </remarks>
|
/// </remarks>
|
||||||
public async ValueTask<CreateMessageResult> SampleAsync(
|
public async ValueTask<CreateMessageResult> SampleAsync(
|
||||||
CreateMessageRequestParams requestParams,
|
CreateMessageRequestParams requestParams,
|
||||||
|
|
@ -252,6 +261,13 @@ public abstract partial class McpServer : McpSession
|
||||||
/// <param name="serializerOptions">The <see cref="JsonSerializerOptions"/> to use for serialization. If <see langword="null"/>, <see cref="McpJsonUtilities.DefaultOptions"/> is used.</param>
|
/// <param name="serializerOptions">The <see cref="JsonSerializerOptions"/> to use for serialization. If <see langword="null"/>, <see cref="McpJsonUtilities.DefaultOptions"/> is used.</param>
|
||||||
/// <returns>The <see cref="IChatClient"/> that can be used to issue sampling requests to the client.</returns>
|
/// <returns>The <see cref="IChatClient"/> that can be used to issue sampling requests to the client.</returns>
|
||||||
/// <exception cref="InvalidOperationException">The client does not support sampling.</exception>
|
/// <exception cref="InvalidOperationException">The client does not support sampling.</exception>
|
||||||
|
/// <remarks>
|
||||||
|
/// When the server is using the Streamable HTTP transport, prefer obtaining this chat client from the
|
||||||
|
/// <see cref="McpServer"/> instance available via <c>RequestContext</c> from inside a tool, prompt,
|
||||||
|
/// or resource handler. That routes sampling requests through the originating POST response stream via
|
||||||
|
/// <see cref="JsonRpcMessageContext.RelatedTransport"/>, which is always open for the duration of
|
||||||
|
/// the request, rather than relying on the optional standalone GET SSE stream.
|
||||||
|
/// </remarks>
|
||||||
public IChatClient AsSamplingChatClient(JsonSerializerOptions? serializerOptions = null)
|
public IChatClient AsSamplingChatClient(JsonSerializerOptions? serializerOptions = null)
|
||||||
{
|
{
|
||||||
ThrowIfSamplingUnsupported();
|
ThrowIfSamplingUnsupported();
|
||||||
|
|
@ -273,6 +289,13 @@ public abstract partial class McpServer : McpSession
|
||||||
/// <exception cref="ArgumentNullException"><paramref name="requestParams"/> is <see langword="null"/>.</exception>
|
/// <exception cref="ArgumentNullException"><paramref name="requestParams"/> is <see langword="null"/>.</exception>
|
||||||
/// <exception cref="InvalidOperationException">The client does not support roots.</exception>
|
/// <exception cref="InvalidOperationException">The client does not support roots.</exception>
|
||||||
/// <exception cref="McpException">The request failed or the client returned an error response.</exception>
|
/// <exception cref="McpException">The request failed or the client returned an error response.</exception>
|
||||||
|
/// <remarks>
|
||||||
|
/// When the server is using the Streamable HTTP transport, prefer calling this method on the
|
||||||
|
/// <see cref="McpServer"/> instance available via <c>RequestContext</c> from inside a tool, prompt,
|
||||||
|
/// or resource handler. That routes the request through the originating POST response stream via
|
||||||
|
/// <see cref="JsonRpcMessageContext.RelatedTransport"/>, which is always open for the duration of
|
||||||
|
/// the request, rather than relying on the optional standalone GET SSE stream.
|
||||||
|
/// </remarks>
|
||||||
public ValueTask<ListRootsResult> RequestRootsAsync(
|
public ValueTask<ListRootsResult> RequestRootsAsync(
|
||||||
ListRootsRequestParams requestParams,
|
ListRootsRequestParams requestParams,
|
||||||
CancellationToken cancellationToken = default)
|
CancellationToken cancellationToken = default)
|
||||||
|
|
@ -298,9 +321,18 @@ public abstract partial class McpServer : McpSession
|
||||||
/// <exception cref="InvalidOperationException">The client does not support elicitation.</exception>
|
/// <exception cref="InvalidOperationException">The client does not support elicitation.</exception>
|
||||||
/// <exception cref="McpException">The request failed or the client returned an error response.</exception>
|
/// <exception cref="McpException">The request failed or the client returned an error response.</exception>
|
||||||
/// <remarks>
|
/// <remarks>
|
||||||
|
/// <para>
|
||||||
|
/// When the server is using the Streamable HTTP transport, prefer calling this method on the
|
||||||
|
/// <see cref="McpServer"/> instance available via <c>RequestContext</c> from inside a tool, prompt,
|
||||||
|
/// or resource handler. That routes the request through the originating POST response stream via
|
||||||
|
/// <see cref="JsonRpcMessageContext.RelatedTransport"/>, which is always open for the duration of
|
||||||
|
/// the request, rather than relying on the optional standalone GET SSE stream.
|
||||||
|
/// </para>
|
||||||
|
/// <para>
|
||||||
/// When called during task-augmented tool execution, this method automatically updates the task
|
/// When called during task-augmented tool execution, this method automatically updates the task
|
||||||
/// status to <see cref="McpTaskStatus.InputRequired"/> while waiting for user input,
|
/// status to <see cref="McpTaskStatus.InputRequired"/> while waiting for user input,
|
||||||
/// then returns to <see cref="McpTaskStatus.Working"/> when the response is received.
|
/// then returns to <see cref="McpTaskStatus.Working"/> when the response is received.
|
||||||
|
/// </para>
|
||||||
/// </remarks>
|
/// </remarks>
|
||||||
public async ValueTask<ElicitResult> ElicitAsync(
|
public async ValueTask<ElicitResult> ElicitAsync(
|
||||||
ElicitRequestParams requestParams,
|
ElicitRequestParams requestParams,
|
||||||
|
|
|
||||||
|
|
@ -221,6 +221,34 @@ public sealed partial class StreamableHttpServerTransport : ITransport
|
||||||
}
|
}
|
||||||
|
|
||||||
/// <inheritdoc/>
|
/// <inheritdoc/>
|
||||||
|
/// <remarks>
|
||||||
|
/// <para>
|
||||||
|
/// This method sends server-to-client messages via the standalone SSE stream opened by an
|
||||||
|
/// optional HTTP GET request (see <see cref="HandleGetRequestAsync(Stream, CancellationToken)"/>).
|
||||||
|
/// </para>
|
||||||
|
/// <para>
|
||||||
|
/// <strong>This is generally the wrong channel for server-to-client requests.</strong> Requests
|
||||||
|
/// sent via the GET stream depend on the client keeping a long-lived GET open, have no per-request
|
||||||
|
/// correlation to a caller, and race with GET startup and teardown. When called from inside a
|
||||||
|
/// tool, prompt, or resource handler, use the <see cref="McpServer"/> instance available via
|
||||||
|
/// <c>RequestContext</c> instead — it routes through the originating POST response stream via
|
||||||
|
/// <see cref="JsonRpcMessageContext.RelatedTransport"/>, which is always open for the duration of
|
||||||
|
/// the request. A <see cref="LogLevel.Warning"/> diagnostic is emitted whenever a
|
||||||
|
/// <see cref="JsonRpcRequest"/> is sent through this method.
|
||||||
|
/// </para>
|
||||||
|
/// <para>
|
||||||
|
/// If no GET SSE stream has yet been opened on this session, behavior depends on the message kind:
|
||||||
|
/// <see cref="JsonRpcRequest"/> messages throw <see cref="InvalidOperationException"/> because the
|
||||||
|
/// awaiting caller has no way to receive a response; <see cref="JsonRpcNotification"/> messages are
|
||||||
|
/// dropped (notifications are best-effort and the spec does not require clients to issue a GET)
|
||||||
|
/// and a <see cref="LogLevel.Debug"/> diagnostic is logged; other messages are dropped and a
|
||||||
|
/// <see cref="LogLevel.Warning"/> diagnostic is logged.
|
||||||
|
/// </para>
|
||||||
|
/// </remarks>
|
||||||
|
/// <exception cref="InvalidOperationException">
|
||||||
|
/// <see cref="Stateless"/> is <see langword="true"/>, or <paramref name="message"/> is a
|
||||||
|
/// <see cref="JsonRpcRequest"/> and no GET SSE stream has been opened on this session.
|
||||||
|
/// </exception>
|
||||||
public async Task SendMessageAsync(JsonRpcMessage message, CancellationToken cancellationToken = default)
|
public async Task SendMessageAsync(JsonRpcMessage message, CancellationToken cancellationToken = default)
|
||||||
{
|
{
|
||||||
Throw.IfNull(message);
|
Throw.IfNull(message);
|
||||||
|
|
@ -234,9 +262,32 @@ public sealed partial class StreamableHttpServerTransport : ITransport
|
||||||
|
|
||||||
if (!_getHttpRequestStarted)
|
if (!_getHttpRequestStarted)
|
||||||
{
|
{
|
||||||
// Clients are not required to make a GET request for unsolicited messages.
|
switch (message)
|
||||||
// If no GET request has been made, drop the message.
|
{
|
||||||
return;
|
case JsonRpcRequest request:
|
||||||
|
throw new InvalidOperationException(
|
||||||
|
$"Cannot send server-to-client JSON-RPC request '{request.Method}' because no GET SSE stream has been opened on this session " +
|
||||||
|
$"(SessionId: '{SessionId}'). " +
|
||||||
|
"Inside a tool, prompt, or resource handler, use the IMcpServer instance from RequestContext (or any IMcpServer obtained via DI from a request-scoped service provider) so the request is routed through the originating POST response stream via JsonRpcMessageContext.RelatedTransport. " +
|
||||||
|
"The standalone GET SSE stream is optional for clients and is not a reliable channel for server-to-client requests.");
|
||||||
|
|
||||||
|
case JsonRpcNotification notification:
|
||||||
|
// Clients are not required to make a GET request for unsolicited messages.
|
||||||
|
// If no GET request has been made, drop the notification (best-effort).
|
||||||
|
LogNotificationDroppedNoGetStream(notification.Method, SessionId ?? string.Empty);
|
||||||
|
return;
|
||||||
|
|
||||||
|
default:
|
||||||
|
// JsonRpcResponse / JsonRpcError generally flow through the originating POST response
|
||||||
|
// stream, so receiving one here without a GET is unexpected. Log loudly and drop.
|
||||||
|
LogMessageDroppedNoGetStream(message.GetType().Name, GetMessageId(message), SessionId ?? string.Empty);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if (message is JsonRpcRequest openRequest)
|
||||||
|
{
|
||||||
|
LogServerRequestOverGetStream(openRequest.Method, SessionId ?? string.Empty);
|
||||||
}
|
}
|
||||||
|
|
||||||
Debug.Assert(_httpResponseTcs is not null);
|
Debug.Assert(_httpResponseTcs is not null);
|
||||||
|
|
@ -263,6 +314,9 @@ public sealed partial class StreamableHttpServerTransport : ITransport
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private static string GetMessageId(JsonRpcMessage message) =>
|
||||||
|
message is JsonRpcMessageWithId withId ? withId.Id.ToString() : string.Empty;
|
||||||
|
|
||||||
/// <inheritdoc/>
|
/// <inheritdoc/>
|
||||||
public async ValueTask DisposeAsync()
|
public async ValueTask DisposeAsync()
|
||||||
{
|
{
|
||||||
|
|
@ -323,4 +377,18 @@ public sealed partial class StreamableHttpServerTransport : ITransport
|
||||||
|
|
||||||
return sseEventStreamWriter;
|
return sseEventStreamWriter;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
[LoggerMessage(Level = LogLevel.Warning,
|
||||||
|
Message = "Sending server-to-client JSON-RPC request '{Method}' over the standalone GET SSE stream (SessionId: '{SessionId}'). " +
|
||||||
|
"Consider using the IMcpServer instance from RequestContext inside a tool, prompt, or resource handler so the request is routed through the originating POST response stream via JsonRpcMessageContext.RelatedTransport, which is more reliable than the optional GET SSE stream.")]
|
||||||
|
private partial void LogServerRequestOverGetStream(string method, string sessionId);
|
||||||
|
|
||||||
|
[LoggerMessage(Level = LogLevel.Debug,
|
||||||
|
Message = "Dropping server-to-client JSON-RPC notification '{Method}' because no GET SSE stream has been opened on this session (SessionId: '{SessionId}').")]
|
||||||
|
private partial void LogNotificationDroppedNoGetStream(string method, string sessionId);
|
||||||
|
|
||||||
|
[LoggerMessage(Level = LogLevel.Warning,
|
||||||
|
Message = "Dropping unexpected server-to-client {MessageType} (Id: '{MessageId}') because no GET SSE stream has been opened on this session (SessionId: '{SessionId}'). " +
|
||||||
|
"Responses normally flow through the originating POST response stream via JsonRpcMessageContext.RelatedTransport.")]
|
||||||
|
private partial void LogMessageDroppedNoGetStream(string messageType, string messageId, string sessionId);
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -384,7 +384,8 @@ public class StreamableHttpServerConformanceTests(ITestOutputHelper outputHelper
|
||||||
public async Task SendNotificationAsync_DoesNotThrow_WhenNoGetRequestHasBeenMade()
|
public async Task SendNotificationAsync_DoesNotThrow_WhenNoGetRequestHasBeenMade()
|
||||||
{
|
{
|
||||||
// Clients are not required to make a GET request for unsolicited messages.
|
// Clients are not required to make a GET request for unsolicited messages.
|
||||||
// If no GET request has been made, the messages should be dropped rather than throwing.
|
// If no GET request has been made, the messages should be dropped rather than throwing,
|
||||||
|
// and the drop should be visible as a Debug-level log so it can be diagnosed.
|
||||||
McpServer? server = null;
|
McpServer? server = null;
|
||||||
|
|
||||||
Builder.Services.AddMcpServer()
|
Builder.Services.AddMcpServer()
|
||||||
|
|
@ -409,6 +410,156 @@ public class StreamableHttpServerConformanceTests(ITestOutputHelper outputHelper
|
||||||
var exception = await Record.ExceptionAsync(() =>
|
var exception = await Record.ExceptionAsync(() =>
|
||||||
server.SendNotificationAsync("test-method", TestContext.Current.CancellationToken));
|
server.SendNotificationAsync("test-method", TestContext.Current.CancellationToken));
|
||||||
Assert.Null(exception);
|
Assert.Null(exception);
|
||||||
|
|
||||||
|
Assert.Contains(MockLoggerProvider.LogMessages, log =>
|
||||||
|
log.Category == typeof(StreamableHttpServerTransport).FullName &&
|
||||||
|
log.LogLevel == LogLevel.Debug &&
|
||||||
|
log.Message.Contains("test-method") &&
|
||||||
|
log.Message.Contains("no GET SSE stream"));
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task SendRequestAsync_Throws_WhenNoGetRequestHasBeenMade()
|
||||||
|
{
|
||||||
|
// A server-to-client request sent before any GET SSE stream is opened can never
|
||||||
|
// receive a response, so the transport should fail fast with InvalidOperationException
|
||||||
|
// instead of silently dropping the message and leaving the caller hanging on the TCS
|
||||||
|
// registered by SendRequestAsync.
|
||||||
|
McpServer? server = null;
|
||||||
|
|
||||||
|
Builder.Services.AddMcpServer()
|
||||||
|
.WithHttpTransport(options =>
|
||||||
|
{
|
||||||
|
#pragma warning disable MCPEXP002 // RunSessionHandler is experimental
|
||||||
|
options.RunSessionHandler = (httpContext, mcpServer, cancellationToken) =>
|
||||||
|
{
|
||||||
|
server = mcpServer;
|
||||||
|
return mcpServer.RunAsync(cancellationToken);
|
||||||
|
};
|
||||||
|
#pragma warning restore MCPEXP002
|
||||||
|
});
|
||||||
|
|
||||||
|
await StartAsync();
|
||||||
|
|
||||||
|
await CallInitializeAndValidateAsync();
|
||||||
|
Assert.NotNull(server);
|
||||||
|
|
||||||
|
var request = new JsonRpcRequest
|
||||||
|
{
|
||||||
|
Method = "roots/list",
|
||||||
|
Id = new RequestId(42),
|
||||||
|
};
|
||||||
|
|
||||||
|
var ex = await Assert.ThrowsAsync<InvalidOperationException>(() =>
|
||||||
|
server.SendRequestAsync(request, TestContext.Current.CancellationToken));
|
||||||
|
|
||||||
|
Assert.Contains("roots/list", ex.Message);
|
||||||
|
Assert.Contains("no GET SSE stream", ex.Message);
|
||||||
|
Assert.Contains("RequestContext", ex.Message);
|
||||||
|
Assert.Contains("RelatedTransport", ex.Message);
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task SendMessageAsync_LogsWarning_OnUnexpectedResponse_WhenNoGetRequestHasBeenMade()
|
||||||
|
{
|
||||||
|
// Responses normally ride the originating POST response stream via RelatedTransport, so
|
||||||
|
// receiving one through the GET path without an open GET is unexpected. The message is
|
||||||
|
// dropped (preserving best-effort semantics) but a warning is logged so the situation is
|
||||||
|
// visible.
|
||||||
|
McpServer? server = null;
|
||||||
|
|
||||||
|
Builder.Services.AddMcpServer()
|
||||||
|
.WithHttpTransport(options =>
|
||||||
|
{
|
||||||
|
#pragma warning disable MCPEXP002 // RunSessionHandler is experimental
|
||||||
|
options.RunSessionHandler = (httpContext, mcpServer, cancellationToken) =>
|
||||||
|
{
|
||||||
|
server = mcpServer;
|
||||||
|
return mcpServer.RunAsync(cancellationToken);
|
||||||
|
};
|
||||||
|
#pragma warning restore MCPEXP002
|
||||||
|
});
|
||||||
|
|
||||||
|
await StartAsync();
|
||||||
|
|
||||||
|
await CallInitializeAndValidateAsync();
|
||||||
|
Assert.NotNull(server);
|
||||||
|
|
||||||
|
var response = new JsonRpcResponse
|
||||||
|
{
|
||||||
|
Id = new RequestId(7),
|
||||||
|
Result = new JsonObject(),
|
||||||
|
};
|
||||||
|
|
||||||
|
var exception = await Record.ExceptionAsync(() =>
|
||||||
|
server.SendMessageAsync(response, TestContext.Current.CancellationToken));
|
||||||
|
Assert.Null(exception);
|
||||||
|
|
||||||
|
Assert.Contains(MockLoggerProvider.LogMessages, log =>
|
||||||
|
log.Category == typeof(StreamableHttpServerTransport).FullName &&
|
||||||
|
log.LogLevel == LogLevel.Warning &&
|
||||||
|
log.Message.Contains(nameof(JsonRpcResponse)) &&
|
||||||
|
log.Message.Contains("no GET SSE stream"));
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task SendRequestAsync_LogsWarning_WhenGetRequestIsOpen()
|
||||||
|
{
|
||||||
|
// Even when the GET SSE stream is open and the request is delivered, server-to-client
|
||||||
|
// requests sent via the GET path are fragile (no per-request correlation, depend on a
|
||||||
|
// long-lived GET, race with startup/teardown). A warning is logged to direct callers at
|
||||||
|
// the RequestContext.RelatedTransport channel instead, without changing behavior.
|
||||||
|
McpServer? server = null;
|
||||||
|
|
||||||
|
Builder.Services.AddMcpServer()
|
||||||
|
.WithHttpTransport(options =>
|
||||||
|
{
|
||||||
|
#pragma warning disable MCPEXP002 // RunSessionHandler is experimental
|
||||||
|
options.RunSessionHandler = (httpContext, mcpServer, cancellationToken) =>
|
||||||
|
{
|
||||||
|
server = mcpServer;
|
||||||
|
return mcpServer.RunAsync(cancellationToken);
|
||||||
|
};
|
||||||
|
#pragma warning restore MCPEXP002
|
||||||
|
});
|
||||||
|
|
||||||
|
await StartAsync();
|
||||||
|
|
||||||
|
await CallInitializeAndValidateAsync();
|
||||||
|
Assert.NotNull(server);
|
||||||
|
|
||||||
|
using var getResponse = await HttpClient.GetAsync("", HttpCompletionOption.ResponseHeadersRead, TestContext.Current.CancellationToken);
|
||||||
|
Assert.Equal(HttpStatusCode.OK, getResponse.StatusCode);
|
||||||
|
|
||||||
|
// Send a request via the GET stream and assert it lands on the wire (proving behavior is unchanged).
|
||||||
|
// SendRequestAsync awaits a response that the test never produces, so use a CTS to cancel after
|
||||||
|
// confirming wire delivery.
|
||||||
|
var request = new JsonRpcRequest
|
||||||
|
{
|
||||||
|
Method = "roots/list",
|
||||||
|
Id = new RequestId(99),
|
||||||
|
};
|
||||||
|
|
||||||
|
using var requestCts = CancellationTokenSource.CreateLinkedTokenSource(TestContext.Current.CancellationToken);
|
||||||
|
var sendTask = server.SendRequestAsync(request, requestCts.Token);
|
||||||
|
|
||||||
|
await foreach (var sseEvent in ReadSseAsync(getResponse.Content))
|
||||||
|
{
|
||||||
|
var received = JsonSerializer.Deserialize(sseEvent, GetJsonTypeInfo<JsonRpcRequest>());
|
||||||
|
Assert.NotNull(received);
|
||||||
|
Assert.Equal("roots/list", received.Method);
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
|
||||||
|
// Cancel the awaited response so SendRequestAsync completes — the wire delivery has already happened.
|
||||||
|
requestCts.Cancel();
|
||||||
|
await Assert.ThrowsAnyAsync<OperationCanceledException>(() => sendTask);
|
||||||
|
|
||||||
|
Assert.Contains(MockLoggerProvider.LogMessages, log =>
|
||||||
|
log.Category == typeof(StreamableHttpServerTransport).FullName &&
|
||||||
|
log.LogLevel == LogLevel.Warning &&
|
||||||
|
log.Message.Contains("roots/list") &&
|
||||||
|
log.Message.Contains("RequestContext"));
|
||||||
}
|
}
|
||||||
|
|
||||||
[Fact]
|
[Fact]
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue