Hare/yoi
1
1
Fork 0
Code Actions Packages Releases Activity

mcp: handle list changed notifications

Browse Source
This commit is contained in:
Hare
2026-06-20 19:24:59 +09:00
parent d31b89072d
commit e33dee192c
4 changed files with 459 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/mcp/src/stdio.rs
+118 -2
View File
@@ -1,4 +1,4 @@
use std::collections::{BTreeMap, VecDeque};
use std::collections::{BTreeMap, BTreeSet, VecDeque};
use std::env;
use std::fmt;
use std::path::PathBuf;
@@ -350,6 +350,67 @@ pub struct McpContentBlock {
pub fields: BTreeMap<String, Value>,
}
/// MCP list surface whose `notifications/*/list_changed` signal was observed.
///
/// The notification is only a freshness signal. The stdio client records this
/// bounded enum state and deliberately ignores notification params so a server
/// cannot inject resource/prompt content or alter model-visible tool schemas
/// through an out-of-band notification.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
pub enum McpListChangedKind {
Tools,
Resources,
Prompts,
}
impl McpListChangedKind {
fn from_notification_method(method: &str) -> Option<Self> {
match method {
"notifications/tools/list_changed" => Some(Self::Tools),
"notifications/resources/list_changed" => Some(Self::Resources),
"notifications/prompts/list_changed" => Some(Self::Prompts),
_ => None,
}
}
pub fn notification_method(self) -> &'static str {
match self {
Self::Tools => "notifications/tools/list_changed",
Self::Resources => "notifications/resources/list_changed",
Self::Prompts => "notifications/prompts/list_changed",
}
}
pub fn list_method(self) -> &'static str {
match self {
Self::Tools => "tools/list",
Self::Resources => "resources/list",
Self::Prompts => "prompts/list",
}
}
}
/// Bounded snapshot of list-change signals observed from one stdio server.
#[derive(Debug, Clone)]
pub struct McpListChangedSnapshot {
pub server_name: String,
kinds: BTreeSet<McpListChangedKind>,
}
impl McpListChangedSnapshot {
pub fn is_empty(&self) -> bool {
self.kinds.is_empty()
}
pub fn contains(&self, kind: McpListChangedKind) -> bool {
self.kinds.contains(&kind)
}
pub fn kinds(&self) -> impl Iterator<Item = McpListChangedKind> + '_ {
self.kinds.iter().copied()
}
}
/// A resolved, explicit local stdio MCP server process specification.
#[derive(Clone)]
pub struct McpStdioServerSpec {
@@ -515,6 +576,7 @@ pub struct McpStdioClient {
limits: McpStdioLimits,
redactor: Redactor,
diagnostics: Arc<Mutex<BoundedDiagnostics>>,
list_changes: Arc<Mutex<BoundedListChanged>>,
stdin: Arc<Mutex<Option<ChildStdin>>>,
child: Option<Child>,
responses: mpsc::Receiver<ReaderEvent>,
@@ -607,6 +669,7 @@ impl McpStdioClient {
limits.max_diagnostic_lines,
redactor.clone(),
)));
let list_changes = Arc::new(Mutex::new(BoundedListChanged::new(spec.name.clone())));
let (tx, rx) = mpsc::channel(16);
let reader_task = spawn_stdout_reader(
spec.name.clone(),
@@ -615,6 +678,7 @@ impl McpStdioClient {
tx,
limits.clone(),
redactor.clone(),
list_changes.clone(),
);
let stderr_task = spawn_stderr_reader(stderr, diagnostics.clone(), limits.clone());
@@ -623,6 +687,7 @@ impl McpStdioClient {
limits,
redactor,
diagnostics,
list_changes,
stdin,
child: Some(child),
responses: rx,
@@ -808,6 +873,21 @@ impl McpStdioClient {
self.diagnostics.lock().await.snapshot()
}
/// Return bounded list-change signals observed so far for this connection.
///
/// This is diagnostic/freshness state only. It never contains notification
/// params and must not be used to mutate an active run's model-visible tool
/// schema outside an explicit safe boundary.
pub async fn snapshot_list_changes(&self) -> McpListChangedSnapshot {
self.list_changes.lock().await.snapshot()
}
/// Clear observed list-change signals before an explicit safe-boundary
/// refresh. New notifications received after this call will be recorded.
pub async fn clear_list_changes(&self) {
self.list_changes.lock().await.clear();
}
pub async fn request<T: for<'de> Deserialize<'de>>(
&mut self,
phase: McpPhase,
@@ -1235,6 +1315,7 @@ fn spawn_stdout_reader(
tx: mpsc::Sender<ReaderEvent>,
limits: McpStdioLimits,
redactor: Redactor,
list_changes: Arc<Mutex<BoundedListChanged>>,
) -> JoinHandle<()> {
tokio::spawn(async move {
let mut stdout = BufReader::new(stdout);
@@ -1248,6 +1329,7 @@ fn spawn_stdout_reader(
&tx,
&limits,
&redactor,
&list_changes,
message,
)
.await
@@ -1290,6 +1372,7 @@ async fn handle_incoming_message(
tx: &mpsc::Sender<ReaderEvent>,
limits: &McpStdioLimits,
redactor: &Redactor,
list_changes: &Arc<Mutex<BoundedListChanged>>,
message: IncomingMessage,
) {
if message.method.is_some() && message.id.is_some() {
@@ -1315,7 +1398,10 @@ async fn handle_incoming_message(
return;
}
if message.method.is_some() {
if let Some(method) = message.method.as_deref() {
if let Some(kind) = McpListChangedKind::from_notification_method(method) {
list_changes.lock().await.mark(kind);
}
let _ = tx.send(ReaderEvent::Notification).await;
return;
}
@@ -1342,6 +1428,36 @@ async fn handle_incoming_message(
.await;
}
#[derive(Debug)]
struct BoundedListChanged {
server_name: String,
kinds: BTreeSet<McpListChangedKind>,
}
impl BoundedListChanged {
fn new(server_name: String) -> Self {
Self {
server_name,
kinds: BTreeSet::new(),
}
}
fn mark(&mut self, kind: McpListChangedKind) {
self.kinds.insert(kind);
}
fn clear(&mut self) {
self.kinds.clear();
}
fn snapshot(&self) -> McpListChangedSnapshot {
McpListChangedSnapshot {
server_name: self.server_name.clone(),
kinds: self.kinds.clone(),
}
}
}
fn spawn_stderr_reader(
stderr: ChildStderr,
diagnostics: Arc<Mutex<BoundedDiagnostics>>,
crates/mcp/tests/fixtures/mock_server.rs
Vendored
+31
View File
@@ -16,6 +16,7 @@ fn main() {
"tools-call-forbidden" => tools_call_forbidden(),
"fail-init" => fail_init(),
"sampling" => sampling_request(),
"list-changed-all" => list_changed_all(),
"shutdown-hang" => shutdown_hang(),
other => panic!("unknown mock mode: {other}"),
}
@@ -223,6 +224,36 @@ fn sampling_request() {
assert_eq!(response["error"]["code"], -32601);
}
fn list_changed_all() {
let init = read_json();
write_json(json!({
"jsonrpc": "2.0",
"id": init["id"],
"result": initialize_result(),
}));
let initialized = read_json();
assert_eq!(initialized["method"], "notifications/initialized");
for method in [
"notifications/tools/list_changed",
"notifications/resources/list_changed",
"notifications/prompts/list_changed",
] {
write_json(json!({
"jsonrpc": "2.0",
"method": method,
"params": {
"malicious_instruction": "INJECT_ME_FROM_LIST_CHANGED_PARAMS"
}
}));
}
let shutdown = read_json();
assert_eq!(shutdown["method"], "shutdown");
write_json(json!({"jsonrpc":"2.0", "id": shutdown["id"], "result": {}}));
let notification = read_json();
assert_eq!(notification["method"], "exit");
}
fn shutdown_hang() {
let init = read_json();
write_json(json!({
crates/mcp/tests/stdio_lifecycle.rs
+32 -2
View File
@@ -1,8 +1,8 @@
use std::time::Duration;
use mcp::stdio::{
CallToolRequest, McpErrorKind, McpPhase, McpStdioClient, McpStdioLimits, McpStdioServerSpec,
McpToolListLimits,
CallToolRequest, McpErrorKind, McpListChangedKind, McpPhase, McpStdioClient, McpStdioLimits,
McpStdioServerSpec, McpToolListLimits,
};
fn mock_server(mode: &str) -> McpStdioServerSpec {
@@ -239,6 +239,36 @@ async fn shutdown_terminates_or_kills_uncooperative_server() {
assert!(shutdown.terminated || shutdown.killed);
}
#[tokio::test]
async fn list_changed_notifications_record_bounded_kind_only_state() {
let mut client = McpStdioClient::connect(mock_server("list-changed-all"), tight_limits())
.await
.expect("initialize succeeds");
tokio::time::sleep(Duration::from_millis(50)).await;
let snapshot = client.snapshot_list_changes().await;
assert_eq!(snapshot.server_name, "mock");
assert!(snapshot.contains(McpListChangedKind::Tools));
assert!(snapshot.contains(McpListChangedKind::Resources));
assert!(snapshot.contains(McpListChangedKind::Prompts));
let methods: Vec<&'static str> = snapshot
.kinds()
.map(McpListChangedKind::notification_method)
.collect();
assert_eq!(
methods,
vec![
"notifications/tools/list_changed",
"notifications/resources/list_changed",
"notifications/prompts/list_changed"
]
);
client.clear_list_changes().await;
assert!(client.snapshot_list_changes().await.is_empty());
client.shutdown().await.expect("shutdown succeeds");
}
#[tokio::test]
async fn sampling_requests_fail_closed_and_are_not_advertised() {
let mut client = McpStdioClient::connect(mock_server("sampling"), tight_limits())