fix(server): remove initialized notification gate to support Streamable HTTP (#788)
* fix(server): remove initialized notification gate to support Streamable HTTP The server's init handshake loop fatally rejected any request arriving before the `notifications/initialized` message. This breaks Streamable HTTP clients where each JSON-RPC message is a separate POST with no ordering guarantee — `tools/list` can easily arrive before `initialized`. Remove the ~40-line wait loop and enter `serve_inner` immediately after sending `InitializeResult`. The `initialized` notification is now handled as a regular notification by the main service loop, matching the TypeScript SDK behavior (validated in typescript-sdk#578). Also remove the now-unreachable `ExpectedInitializedNotification` error variant from `ServerInitializeError`. Closes #783 Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> * fix(server): keep ExpectedInitializedNotification as deprecated Retain the variant for semver compatibility — removing it would be a breaking change caught by cargo-semver-checks. Mark it deprecated with a note that it is never constructed and will be removed in a future major release. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Anar Azadaliyev <anar.azadaliye@gmail.com> Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
parent
a7b570062e
commit
65d2b29da5
3 changed files with 78 additions and 54 deletions
2
.gitignore
vendored
2
.gitignore
vendored
|
|
@ -27,3 +27,5 @@ __pycache__/
|
||||||
# and can be added to the global gitignore or merged into this file. For a more nuclear
|
# and can be added to the global gitignore or merged into this file. For a more nuclear
|
||||||
# option (not recommended) you can uncomment the following to ignore the entire idea folder.
|
# option (not recommended) you can uncomment the following to ignore the entire idea folder.
|
||||||
#.idea/
|
#.idea/
|
||||||
|
node_modules/
|
||||||
|
.DS_Store
|
||||||
|
|
|
||||||
|
|
@ -53,6 +53,10 @@ pub enum ServerInitializeError {
|
||||||
#[error("expect initialized request, but received: {0:?}")]
|
#[error("expect initialized request, but received: {0:?}")]
|
||||||
ExpectedInitializeRequest(Option<ClientJsonRpcMessage>),
|
ExpectedInitializeRequest(Option<ClientJsonRpcMessage>),
|
||||||
|
|
||||||
|
#[deprecated(
|
||||||
|
since = "1.4.0",
|
||||||
|
note = "The server no longer gates on the initialized notification. This variant is never constructed and will be removed in a future major release."
|
||||||
|
)]
|
||||||
#[error("expect initialized notification, but received: {0:?}")]
|
#[error("expect initialized notification, but received: {0:?}")]
|
||||||
ExpectedInitializedNotification(Option<ClientJsonRpcMessage>),
|
ExpectedInitializedNotification(Option<ClientJsonRpcMessage>),
|
||||||
|
|
||||||
|
|
@ -243,49 +247,12 @@ where
|
||||||
ServerInitializeError::transport::<T>(error, "sending initialize response")
|
ServerInitializeError::transport::<T>(error, "sending initialize response")
|
||||||
})?;
|
})?;
|
||||||
|
|
||||||
// Wait for initialized notification. The MCP spec permits logging/setLevel and ping
|
// Enter the main service loop immediately after sending InitializeResult.
|
||||||
// before initialized; VS Code sends setLevel immediately after the initialize response.
|
// The initialized notification will be handled as a regular notification by serve_inner.
|
||||||
let notification = loop {
|
// This matches the TypeScript SDK behavior: no init gate, no waiting for initialized.
|
||||||
let msg = expect_next_message(&mut transport, "initialize notification").await?;
|
// Streamable HTTP has no ordering guarantee between POSTs, and the MCP spec uses
|
||||||
match msg {
|
// SHOULD NOT (not MUST NOT) for pre-initialized messages, so any request arriving
|
||||||
ClientJsonRpcMessage::Notification(n)
|
// before initialized is processed normally.
|
||||||
if matches!(
|
|
||||||
n.notification,
|
|
||||||
ClientNotification::InitializedNotification(_)
|
|
||||||
) =>
|
|
||||||
{
|
|
||||||
break n.notification;
|
|
||||||
}
|
|
||||||
ClientJsonRpcMessage::Request(req)
|
|
||||||
if matches!(
|
|
||||||
req.request,
|
|
||||||
ClientRequest::SetLevelRequest(_) | ClientRequest::PingRequest(_)
|
|
||||||
) =>
|
|
||||||
{
|
|
||||||
transport
|
|
||||||
.send(ServerJsonRpcMessage::response(
|
|
||||||
ServerResult::EmptyResult(EmptyResult {}),
|
|
||||||
req.id,
|
|
||||||
))
|
|
||||||
.await
|
|
||||||
.map_err(|error| {
|
|
||||||
ServerInitializeError::transport::<T>(error, "sending pre-init response")
|
|
||||||
})?;
|
|
||||||
}
|
|
||||||
other => {
|
|
||||||
return Err(ServerInitializeError::ExpectedInitializedNotification(
|
|
||||||
Some(other),
|
|
||||||
));
|
|
||||||
}
|
|
||||||
}
|
|
||||||
};
|
|
||||||
let context = NotificationContext {
|
|
||||||
meta: notification.get_meta().clone(),
|
|
||||||
extensions: notification.extensions().clone(),
|
|
||||||
peer: peer.clone(),
|
|
||||||
};
|
|
||||||
let _ = service.handle_notification(notification, context).await;
|
|
||||||
// Continue processing service
|
|
||||||
Ok(serve_inner(service, transport, peer, peer_rx, ct))
|
Ok(serve_inner(service, transport, peer, peer_rx, ct))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -6,7 +6,6 @@ use common::handlers::TestServer;
|
||||||
use rmcp::{
|
use rmcp::{
|
||||||
ServiceExt,
|
ServiceExt,
|
||||||
model::{ClientJsonRpcMessage, ServerJsonRpcMessage, ServerResult},
|
model::{ClientJsonRpcMessage, ServerJsonRpcMessage, ServerResult},
|
||||||
service::ServerInitializeError,
|
|
||||||
transport::{IntoTransport, Transport},
|
transport::{IntoTransport, Transport},
|
||||||
};
|
};
|
||||||
|
|
||||||
|
|
@ -54,7 +53,7 @@ async fn do_initialize(client: &mut impl Transport<rmcp::RoleClient>) {
|
||||||
let _response = client.receive().await.unwrap();
|
let _response = client.receive().await.unwrap();
|
||||||
}
|
}
|
||||||
|
|
||||||
// Server responds with EmptyResult to setLevel received before initialized.
|
// Server handles setLevel sent before initialized notification (processed by serve_inner).
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
async fn server_init_set_level_response_is_empty_result() {
|
async fn server_init_set_level_response_is_empty_result() {
|
||||||
let (server_transport, client_transport) = tokio::io::duplex(4096);
|
let (server_transport, client_transport) = tokio::io::duplex(4096);
|
||||||
|
|
@ -64,7 +63,14 @@ async fn server_init_set_level_response_is_empty_result() {
|
||||||
do_initialize(&mut client).await;
|
do_initialize(&mut client).await;
|
||||||
client.send(set_level_request(2)).await.unwrap();
|
client.send(set_level_request(2)).await.unwrap();
|
||||||
|
|
||||||
let response = client.receive().await.unwrap();
|
// The handler may send logging notifications before the response;
|
||||||
|
// skip notifications to find the EmptyResult response.
|
||||||
|
let response = loop {
|
||||||
|
let msg = client.receive().await.unwrap();
|
||||||
|
if matches!(msg, ServerJsonRpcMessage::Response(_)) {
|
||||||
|
break msg;
|
||||||
|
}
|
||||||
|
};
|
||||||
assert!(
|
assert!(
|
||||||
matches!(
|
matches!(
|
||||||
response,
|
response,
|
||||||
|
|
@ -85,7 +91,13 @@ async fn server_init_succeeds_after_set_level_before_initialized() {
|
||||||
|
|
||||||
do_initialize(&mut client).await;
|
do_initialize(&mut client).await;
|
||||||
client.send(set_level_request(2)).await.unwrap();
|
client.send(set_level_request(2)).await.unwrap();
|
||||||
let _response = client.receive().await.unwrap();
|
// Skip notifications until we get the response
|
||||||
|
loop {
|
||||||
|
let msg = client.receive().await.unwrap();
|
||||||
|
if matches!(msg, ServerJsonRpcMessage::Response(_)) {
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
}
|
||||||
client.send(initialized_notification()).await.unwrap();
|
client.send(initialized_notification()).await.unwrap();
|
||||||
|
|
||||||
let result = server_handle.await.unwrap();
|
let result = server_handle.await.unwrap();
|
||||||
|
|
@ -179,23 +191,66 @@ async fn server_init_succeeds_after_ping_before_initialized() {
|
||||||
result.unwrap().cancel().await.unwrap();
|
result.unwrap().cancel().await.unwrap();
|
||||||
}
|
}
|
||||||
|
|
||||||
// Server returns ExpectedInitializedNotification for any other message before initialized.
|
// Server buffers tools/list sent before initialized and processes it after initialization.
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
async fn server_init_rejects_unexpected_message_before_initialized() {
|
async fn server_init_buffers_request_before_initialized() {
|
||||||
let (server_transport, client_transport) = tokio::io::duplex(4096);
|
let (server_transport, client_transport) = tokio::io::duplex(4096);
|
||||||
let server_handle =
|
let server_handle =
|
||||||
tokio::spawn(async move { TestServer::new().serve(server_transport).await });
|
tokio::spawn(async move { TestServer::new().serve(server_transport).await });
|
||||||
let mut client = IntoTransport::<rmcp::RoleClient, _, _>::into_transport(client_transport);
|
let mut client = IntoTransport::<rmcp::RoleClient, _, _>::into_transport(client_transport);
|
||||||
|
|
||||||
do_initialize(&mut client).await;
|
do_initialize(&mut client).await;
|
||||||
|
// Send tools/list before initialized notification
|
||||||
client.send(list_tools_request(2)).await.unwrap();
|
client.send(list_tools_request(2)).await.unwrap();
|
||||||
|
// Now send initialized notification
|
||||||
|
client.send(initialized_notification()).await.unwrap();
|
||||||
|
|
||||||
|
// The buffered tools/list should be processed — expect a response
|
||||||
|
let response = client.receive().await.unwrap();
|
||||||
|
assert!(
|
||||||
|
matches!(response, ServerJsonRpcMessage::Response(_)),
|
||||||
|
"expected response for buffered tools/list, got: {response:?}"
|
||||||
|
);
|
||||||
|
|
||||||
let result = server_handle.await.unwrap();
|
let result = server_handle.await.unwrap();
|
||||||
assert!(
|
assert!(
|
||||||
matches!(
|
result.is_ok(),
|
||||||
result,
|
"server should initialize successfully when buffering pre-init messages"
|
||||||
Err(ServerInitializeError::ExpectedInitializedNotification(_))
|
|
||||||
),
|
|
||||||
"expected ExpectedInitializedNotification error"
|
|
||||||
);
|
);
|
||||||
|
result.unwrap().cancel().await.unwrap();
|
||||||
|
}
|
||||||
|
|
||||||
|
// Server buffers multiple requests before initialized and processes them in order.
|
||||||
|
#[tokio::test]
|
||||||
|
async fn server_init_buffers_multiple_requests_before_initialized() {
|
||||||
|
let (server_transport, client_transport) = tokio::io::duplex(4096);
|
||||||
|
let server_handle =
|
||||||
|
tokio::spawn(async move { TestServer::new().serve(server_transport).await });
|
||||||
|
let mut client = IntoTransport::<rmcp::RoleClient, _, _>::into_transport(client_transport);
|
||||||
|
|
||||||
|
do_initialize(&mut client).await;
|
||||||
|
// Send two requests before initialized
|
||||||
|
client.send(list_tools_request(2)).await.unwrap();
|
||||||
|
client.send(ping_request(3)).await.unwrap();
|
||||||
|
// Now send initialized notification
|
||||||
|
client.send(initialized_notification()).await.unwrap();
|
||||||
|
|
||||||
|
// Both buffered messages should get responses
|
||||||
|
let response1 = client.receive().await.unwrap();
|
||||||
|
let response2 = client.receive().await.unwrap();
|
||||||
|
assert!(
|
||||||
|
matches!(response1, ServerJsonRpcMessage::Response(_)),
|
||||||
|
"expected response for first buffered message, got: {response1:?}"
|
||||||
|
);
|
||||||
|
assert!(
|
||||||
|
matches!(response2, ServerJsonRpcMessage::Response(_)),
|
||||||
|
"expected response for second buffered message, got: {response2:?}"
|
||||||
|
);
|
||||||
|
|
||||||
|
let result = server_handle.await.unwrap();
|
||||||
|
assert!(
|
||||||
|
result.is_ok(),
|
||||||
|
"server should initialize successfully with multiple buffered messages"
|
||||||
|
);
|
||||||
|
result.unwrap().cancel().await.unwrap();
|
||||||
}
|
}
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue