From dbbb5b0f7f298f69840e8bba18c5a7bb5f1e5f9f Mon Sep 17 00:00:00 2001 From: daniel Date: Mon, 19 Oct 2026 12:13:02 +0100 Subject: [PATCH 1/3] fix(mcp): honor advertised catalog subscription capabilities --- crates/agentkit-mcp/src/lib.rs | 67 +++++++++++-- crates/agentkit-mcp/tests/modern_http.rs | 115 ++++++++++++++++++++++- 2 files changed, 167 insertions(+), 15 deletions(-) diff --git a/crates/agentkit-mcp/src/lib.rs b/crates/agentkit-mcp/src/lib.rs index 56de871..b487e75 100644 --- a/crates/agentkit-mcp/src/lib.rs +++ b/crates/agentkit-mcp/src/lib.rs @@ -1648,14 +1648,18 @@ impl McpConnection { if !resources.is_empty() { filter.resource_subscriptions = Some(resources.iter().cloned().collect()); } - Some( - subscriptions::SubscriptionController::start( - new_peer.clone(), - filter, - candidate_sink.clone(), + if subscription_filter_is_empty(&filter) { + None + } else { + Some( + subscriptions::SubscriptionController::start( + new_peer.clone(), + filter, + candidate_sink.clone(), + ) + .await?, ) - .await?, - ) + } } else { // Legacy subscriptions are session-scoped and must be replayed on // the candidate before retiring the authoritative old transport. @@ -1771,6 +1775,9 @@ impl McpConnection { if !resources.is_empty() { filter.resource_subscriptions = Some(resources.iter().cloned().collect()); } + if subscription_filter_is_empty(&filter) { + return Ok(()); + } let controller = subscriptions::SubscriptionController::start( self.peer(), filter, @@ -1813,6 +1820,20 @@ impl McpConnection { if let Some(controller) = self.subscriptions.lock().await.as_ref() { return controller.subscribe_resource(uri).await; } + if self.modern() { + // Resource-only servers have no catalog stream to extend. Start a + // modern stream with the explicit URI, never a legacy subscribe RPC. + let mut filter = self.subscription_filter(); + filter.resource_subscriptions = Some(vec![uri]); + let controller = subscriptions::SubscriptionController::start( + self.peer(), + filter, + self.sink.clone(), + ) + .await?; + *self.subscriptions.lock().await = Some(controller); + return Ok(()); + } self.peer() .subscribe(rmcp_model::SubscribeRequestParams::new(uri.clone())) .await @@ -1836,6 +1857,10 @@ impl McpConnection { if let Some(controller) = self.subscriptions.lock().await.as_ref() { return controller.unsubscribe_resource(uri).await; } + if self.modern() { + // No modern controller means no resource subscription exists. + return Ok(()); + } self.peer() .unsubscribe(rmcp_model::UnsubscribeRequestParams::new(uri.clone())) .await @@ -2185,12 +2210,34 @@ impl McpConnection { fn filter_for_capabilities(capabilities: &McpServerCapabilities) -> McpSubscriptionFilter { let mut filter = McpSubscriptionFilter::default(); - filter.tools_list_changed = capabilities.tools.as_ref().map(|_| true); - filter.resources_list_changed = capabilities.resources.as_ref().map(|_| true); - filter.prompts_list_changed = capabilities.prompts.as_ref().map(|_| true); + filter.tools_list_changed = capabilities + .tools + .as_ref() + .and_then(|c| c.list_changed) + .filter(|enabled| *enabled); + filter.resources_list_changed = capabilities + .resources + .as_ref() + .and_then(|c| c.list_changed) + .filter(|enabled| *enabled); + filter.prompts_list_changed = capabilities + .prompts + .as_ref() + .and_then(|c| c.list_changed) + .filter(|enabled| *enabled); filter } +fn subscription_filter_is_empty(filter: &McpSubscriptionFilter) -> bool { + filter.tools_list_changed != Some(true) + && filter.resources_list_changed != Some(true) + && filter.prompts_list_changed != Some(true) + && filter + .resource_subscriptions + .as_ref() + .is_none_or(Vec::is_empty) +} + fn client_lifecycle() -> ClientLifecycleMode { ClientLifecycleMode::Auto { // Auto must retry discovery at a mutually supported version before an diff --git a/crates/agentkit-mcp/tests/modern_http.rs b/crates/agentkit-mcp/tests/modern_http.rs index c56e2c8..f8091b4 100644 --- a/crates/agentkit-mcp/tests/modern_http.rs +++ b/crates/agentkit-mcp/tests/modern_http.rs @@ -148,6 +148,22 @@ async fn rpc( } else { capabilities.build() }; + let capabilities = match token { + "Bearer caps-missing" => json!({}), + "Bearer caps-empty" => json!({"tools":{},"resources":{},"prompts":{}}), + "Bearer caps-false" => { + json!({"tools":{"listChanged":false},"resources":{"listChanged":false},"prompts":{"listChanged":false}}) + } + "Bearer caps-resource" => json!({"resources":{"subscribe":true}}), + "Bearer caps-mixed" => { + json!({"tools":{"listChanged":true},"resources":{"listChanged":false},"prompts":{}}) + } + "Bearer caps-true" => { + json!({"tools":{"listChanged":true},"resources":{"listChanged":true},"prompts":{"listChanged":true}}) + } + _ => serde_json::to_value(capabilities).unwrap(), + }; + let capabilities = serde_json::from_value(capabilities).unwrap(); serde_json::to_value(DiscoverResult::new( vec![ProtocolVersion::V_2026_07_28], capabilities, @@ -155,6 +171,15 @@ async fn rpc( .unwrap() } "subscriptions/listen" => { + if matches!( + token, + "Bearer caps-missing" | "Bearer caps-empty" | "Bearer caps-false" + ) { + return Json(json!({"jsonrpc":"2.0", "id":id, "error":{ + "code":-32601, "message":"subscriptions not supported" + }})) + .into_response(); + } let (frames, receiver) = mpsc::unbounded_channel(); let filter = message["params"]["notifications"].clone(); if token != "Bearer hold" && !state.hold_ack.load(Ordering::SeqCst) { @@ -323,7 +348,7 @@ async fn auto_http_discovery_stream_metadata_and_auth_replacement_restore_resour let mut events = connection.subscribe_events(); replacement.notify("notifications/resources/updated", json!({"uri":"memo:keep"})); assert!(matches!(events.recv().await.unwrap(), McpServerEvent::ResourceUpdated(params) if params.uri == "memo:keep")); - connection.close().await.unwrap(); + connection.close().await; bounded(replacement.frames.closed()).await; }).await; } @@ -359,7 +384,7 @@ async fn failed_and_cancelled_auth_reconnect_leave_old_http_stream_usable() { McpServerEvent::ToolListChanged )); connection.discover().await.unwrap(); - connection.close().await.unwrap(); + connection.close().await; bounded(active.frames.closed()).await; }) .await; @@ -436,7 +461,7 @@ async fn modern_to_legacy_and_legacy_to_legacy_replay_committed_resource_uris() connection.resolve_auth(auth("replacement")).await.unwrap(); let modern = fixture.listen().await; assert_eq!(modern.filter["resourceSubscriptions"], json!(["memo:keep"])); - connection.close().await.unwrap(); + connection.close().await; bounded(modern.frames.closed()).await; }) .await; @@ -470,7 +495,7 @@ async fn rejected_legacy_resource_replay_keeps_authoritative_modern_connection() active.notify("notifications/resources/updated", json!({"uri":"memo:keep"})); assert!(matches!(events.recv().await.unwrap(), McpServerEvent::ResourceUpdated(params) if params.uri == "memo:keep")); connection.discover().await.unwrap(); - connection.close().await.unwrap(); + connection.close().await; bounded(active.frames.closed()).await; }).await; } @@ -545,7 +570,7 @@ async fn auth_reconnect_partial_ack_preserves_committed_http_subscription() { assert_eq!(request["method"], "resources/list"); assert_eq!(headers.get("authorization").unwrap(), "Bearer initial"); drop(requests); - connection.close().await.unwrap(); + connection.close().await; bounded(active.frames.closed()).await; }).await; } @@ -736,3 +761,83 @@ async fn auto_http_no_common_version_does_not_initialize() { }) .await; } + +#[tokio::test] +async fn advertised_catalog_flags_gate_listen_on_connect_discover_and_reconnect() { + bounded(async { + for profile in ["missing", "empty", "false", "mixed", "true"] { + let mut fixture = Fixture::start().await; + let token = format!("caps-{profile}"); + if let McpTransportBinding::StreamableHttp(binding) = &mut fixture.config.transport { + *binding = binding.clone().with_bearer_token(&token); + } + let connection = McpConnection::connect(&fixture.config).await.unwrap(); + let expected = match profile { + "mixed" => Some(json!({"toolsListChanged":true})), + "true" => Some(json!({"toolsListChanged":true,"resourcesListChanged":true,"promptsListChanged":true})), + _ => None, + }; + if let Some(filter) = &expected { + assert_eq!(&fixture.listen().await.filter, filter); + } else { + assert!(connection.subscription_status().await.is_none()); + } + connection.discover().await.unwrap(); + connection.discover().await.unwrap(); + connection.resolve_auth(auth(&token)).await.unwrap(); + connection.discover().await.unwrap(); + if let Some(filter) = &expected { + assert_eq!(&fixture.listen().await.filter, filter); + } + let requests = fixture.requests.lock().await; + assert_eq!(requests.iter().filter(|(_, r)| r["method"] == "subscriptions/listen").count(), if expected.is_some() { 2 } else { 0 }, "{profile}"); + drop(requests); + connection.close().await; + } + }).await; +} + +#[tokio::test] +async fn resource_only_server_starts_modern_stream_lazily_and_replays_on_reconnect() { + bounded(async { + let mut fixture = Fixture::start().await; + if let McpTransportBinding::StreamableHttp(binding) = &mut fixture.config.transport { + *binding = binding.clone().with_bearer_token("caps-resource"); + } + let connection = McpConnection::connect(&fixture.config).await.unwrap(); + connection.discover().await.unwrap(); + connection + .unsubscribe_resource("memo:absent") + .await + .unwrap(); + assert!(connection.subscription_status().await.is_none()); + assert!( + !fixture + .requests + .lock() + .await + .iter() + .any(|(_, r)| r["method"] == "subscriptions/listen") + ); + connection.subscribe_resource("memo:one").await.unwrap(); + assert_eq!( + fixture.listen().await.filter, + json!({"resourceSubscriptions":["memo:one"]}) + ); + connection + .resolve_auth(auth("caps-resource")) + .await + .unwrap(); + assert_eq!( + fixture.listen().await.filter, + json!({"resourceSubscriptions":["memo:one"]}) + ); + connection.unsubscribe_resource("memo:one").await.unwrap(); + assert!(!fixture.requests.lock().await.iter().any(|(_, r)| matches!( + r["method"].as_str(), + Some("resources/subscribe" | "resources/unsubscribe") + ))); + connection.close().await; + }) + .await; +} From 3cbd01144cebfc05cc1ee39d7c4b0fe2c8c74490 Mon Sep 17 00:00:00 2001 From: daniel Date: Mon, 19 Oct 2026 12:14:43 +0100 Subject: [PATCH 2/3] test(mcp): retain HTTP stream fixtures through capability assertions --- crates/agentkit-mcp/tests/modern_http.rs | 41 +++++++++++++++--------- 1 file changed, 26 insertions(+), 15 deletions(-) diff --git a/crates/agentkit-mcp/tests/modern_http.rs b/crates/agentkit-mcp/tests/modern_http.rs index f8091b4..30080b3 100644 --- a/crates/agentkit-mcp/tests/modern_http.rs +++ b/crates/agentkit-mcp/tests/modern_http.rs @@ -348,7 +348,7 @@ async fn auto_http_discovery_stream_metadata_and_auth_replacement_restore_resour let mut events = connection.subscribe_events(); replacement.notify("notifications/resources/updated", json!({"uri":"memo:keep"})); assert!(matches!(events.recv().await.unwrap(), McpServerEvent::ResourceUpdated(params) if params.uri == "memo:keep")); - connection.close().await; + connection.close().await.unwrap(); bounded(replacement.frames.closed()).await; }).await; } @@ -384,7 +384,7 @@ async fn failed_and_cancelled_auth_reconnect_leave_old_http_stream_usable() { McpServerEvent::ToolListChanged )); connection.discover().await.unwrap(); - connection.close().await; + connection.close().await.unwrap(); bounded(active.frames.closed()).await; }) .await; @@ -461,7 +461,7 @@ async fn modern_to_legacy_and_legacy_to_legacy_replay_committed_resource_uris() connection.resolve_auth(auth("replacement")).await.unwrap(); let modern = fixture.listen().await; assert_eq!(modern.filter["resourceSubscriptions"], json!(["memo:keep"])); - connection.close().await; + connection.close().await.unwrap(); bounded(modern.frames.closed()).await; }) .await; @@ -495,7 +495,7 @@ async fn rejected_legacy_resource_replay_keeps_authoritative_modern_connection() active.notify("notifications/resources/updated", json!({"uri":"memo:keep"})); assert!(matches!(events.recv().await.unwrap(), McpServerEvent::ResourceUpdated(params) if params.uri == "memo:keep")); connection.discover().await.unwrap(); - connection.close().await; + connection.close().await.unwrap(); bounded(active.frames.closed()).await; }).await; } @@ -570,7 +570,7 @@ async fn auth_reconnect_partial_ack_preserves_committed_http_subscription() { assert_eq!(request["method"], "resources/list"); assert_eq!(headers.get("authorization").unwrap(), "Bearer initial"); drop(requests); - connection.close().await; + connection.close().await.unwrap(); bounded(active.frames.closed()).await; }).await; } @@ -777,22 +777,30 @@ async fn advertised_catalog_flags_gate_listen_on_connect_discover_and_reconnect( "true" => Some(json!({"toolsListChanged":true,"resourcesListChanged":true,"promptsListChanged":true})), _ => None, }; - if let Some(filter) = &expected { - assert_eq!(&fixture.listen().await.filter, filter); + let initial = if let Some(filter) = &expected { + let stream = fixture.listen().await; + assert_eq!(&stream.filter, filter); + Some(stream) } else { assert!(connection.subscription_status().await.is_none()); - } + None + }; connection.discover().await.unwrap(); connection.discover().await.unwrap(); connection.resolve_auth(auth(&token)).await.unwrap(); connection.discover().await.unwrap(); - if let Some(filter) = &expected { - assert_eq!(&fixture.listen().await.filter, filter); - } + let replacement = if let Some(filter) = &expected { + let stream = fixture.listen().await; + assert_eq!(&stream.filter, filter); + Some(stream) + } else { + None + }; let requests = fixture.requests.lock().await; assert_eq!(requests.iter().filter(|(_, r)| r["method"] == "subscriptions/listen").count(), if expected.is_some() { 2 } else { 0 }, "{profile}"); drop(requests); - connection.close().await; + connection.close().await.unwrap(); + drop((initial, replacement)); } }).await; } @@ -820,16 +828,18 @@ async fn resource_only_server_starts_modern_stream_lazily_and_replays_on_reconne .any(|(_, r)| r["method"] == "subscriptions/listen") ); connection.subscribe_resource("memo:one").await.unwrap(); + let initial = fixture.listen().await; assert_eq!( - fixture.listen().await.filter, + initial.filter, json!({"resourceSubscriptions":["memo:one"]}) ); connection .resolve_auth(auth("caps-resource")) .await .unwrap(); + let replacement = fixture.listen().await; assert_eq!( - fixture.listen().await.filter, + replacement.filter, json!({"resourceSubscriptions":["memo:one"]}) ); connection.unsubscribe_resource("memo:one").await.unwrap(); @@ -837,7 +847,8 @@ async fn resource_only_server_starts_modern_stream_lazily_and_replays_on_reconne r["method"].as_str(), Some("resources/subscribe" | "resources/unsubscribe") ))); - connection.close().await; + connection.close().await.unwrap(); + drop((initial, replacement)); }) .await; } From 30c5b758f9f3102d3fe5c2e6285e453a0c89e99a Mon Sep 17 00:00:00 2001 From: daniel Date: Mon, 19 Oct 2026 12:20:24 +0100 Subject: [PATCH 3/3] fix(mcp): use current transport sink for lazy subscriptions --- crates/agentkit-mcp/src/lib.rs | 20 ++++++++------------ crates/agentkit-mcp/tests/modern_http.rs | 6 ++++++ 2 files changed, 14 insertions(+), 12 deletions(-) diff --git a/crates/agentkit-mcp/src/lib.rs b/crates/agentkit-mcp/src/lib.rs index b487e75..9303df6 100644 --- a/crates/agentkit-mcp/src/lib.rs +++ b/crates/agentkit-mcp/src/lib.rs @@ -1778,12 +1778,10 @@ impl McpConnection { if subscription_filter_is_empty(&filter) { return Ok(()); } - let controller = subscriptions::SubscriptionController::start( - self.peer(), - filter, - self.sink.clone(), - ) - .await?; + let mut sink = self.sink.clone(); + sink.active = self.inner.lock().await.service().active.clone(); + let controller = + subscriptions::SubscriptionController::start(self.peer(), filter, sink).await?; *subscriptions = Some(controller); } Ok(()) @@ -1825,12 +1823,10 @@ impl McpConnection { // modern stream with the explicit URI, never a legacy subscribe RPC. let mut filter = self.subscription_filter(); filter.resource_subscriptions = Some(vec![uri]); - let controller = subscriptions::SubscriptionController::start( - self.peer(), - filter, - self.sink.clone(), - ) - .await?; + let mut sink = self.sink.clone(); + sink.active = self.inner.lock().await.service().active.clone(); + let controller = + subscriptions::SubscriptionController::start(self.peer(), filter, sink).await?; *self.subscriptions.lock().await = Some(controller); return Ok(()); } diff --git a/crates/agentkit-mcp/tests/modern_http.rs b/crates/agentkit-mcp/tests/modern_http.rs index 30080b3..530ddee 100644 --- a/crates/agentkit-mcp/tests/modern_http.rs +++ b/crates/agentkit-mcp/tests/modern_http.rs @@ -827,12 +827,18 @@ async fn resource_only_server_starts_modern_stream_lazily_and_replays_on_reconne .iter() .any(|(_, r)| r["method"] == "subscriptions/listen") ); + // Reconnect before the first stream exists: lazy startup must use the + // replacement transport's active notification sink. + connection.resolve_auth(auth("caps-resource")).await.unwrap(); + let mut events = connection.subscribe_events(); connection.subscribe_resource("memo:one").await.unwrap(); let initial = fixture.listen().await; assert_eq!( initial.filter, json!({"resourceSubscriptions":["memo:one"]}) ); + initial.notify("notifications/resources/updated", json!({"uri":"memo:one"})); + assert!(matches!(events.recv().await.unwrap(), McpServerEvent::ResourceUpdated(params) if params.uri == "memo:one")); connection .resolve_auth(auth("caps-resource")) .await