feat: add support for custom server notifications

#556 introduced support for
custom client notifications, so this PR makes the complementary change, adding
support for custom server notifications.

MCP clients, particularly ones that offer "experimental" capabilities,
may wish to handle custom server notifications that are not part of the
standard MCP specification. This change introduces a new
`CustomServerNotification` type that allows a client to process
such custom notifications.

- introduces `CustomServerNotification` to carry arbitrary methods/params while
  still preserving meta/extensions; wires it into the `ServerNotification` union
  and `serde` so `params` can be decoded with `params_as`
- allows client handlers to receive custom notifications via a new
  `on_custom_notification` hook
- adds integration coverage that sends a custom server notification end-to-end
  and asserts the client sees the method and payload

Test:

```shell
cargo test -p rmcp --features client test_custom_server_notification_reaches_client
```

Author: bolinfest

crates/rmcp/src/handler/client.rs

@@ -56,6 +56,9 @@ impl<H: ClientHandler> Service<RoleClient> for H {
             ServerNotification::PromptListChangedNotification(_notification_no_param) => {
                 self.on_prompt_list_changed(context).await
             }
+            ServerNotification::CustomServerNotification(notification) => {
+                self.on_custom_notification(notification, context).await
+            }
         };
         Ok(())
     }
@@ -166,6 +169,14 @@ pub trait ClientHandler: Sized + Send + Sync + 'static {
     ) -> impl Future<Output = ()> + Send + '_ {
         std::future::ready(())
     }
+    fn on_custom_notification(
+        &self,
+        notification: CustomServerNotification,
+        context: NotificationContext<RoleClient>,
+    ) -> impl Future<Output = ()> + Send + '_ {
+        let _ = (notification, context);
+        std::future::ready(())
+    }
 
     fn get_info(&self) -> ClientInfo {
         ClientInfo::default()

crates/rmcp/src/model.rs

@@ -661,6 +661,40 @@ impl CustomClientNotification {
     }
 }
 
+/// A catch-all notification the server can use to send custom messages to a client.
+///
+/// This preserves the raw `method` name and `params` payload so handlers can
+/// deserialize them into domain-specific types.
+#[derive(Debug, Clone)]
+#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
+pub struct CustomServerNotification {
+    pub method: String,
+    pub params: Option<Value>,
+    /// extensions will carry anything possible in the context, including [`Meta`]
+    ///
+    /// this is similar with the Extensions in `http` crate
+    #[cfg_attr(feature = "schemars", schemars(skip))]
+    pub extensions: Extensions,
+}
+
+impl CustomServerNotification {
+    pub fn new(method: impl Into<String>, params: Option<Value>) -> Self {
+        Self {
+            method: method.into(),
+            params,
+            extensions: Extensions::default(),
+        }
+    }
+
+    /// Deserialize `params` into a strongly-typed structure.
+    pub fn params_as<T: DeserializeOwned>(&self) -> Result<Option<T>, serde_json::Error> {
+        self.params
+            .as_ref()
+            .map(|params| serde_json::from_value(params.clone()))
+            .transpose()
+    }
+}
+
 const_string!(InitializeResultMethod = "initialize");
 /// # Initialization
 /// This request is sent from the client to the server when it first connects, asking it to begin initialization.
@@ -1817,7 +1851,8 @@ ts_union!(
     | ResourceUpdatedNotification
     | ResourceListChangedNotification
     | ToolListChangedNotification
-    | PromptListChangedNotification;
+    | PromptListChangedNotification
+    | CustomServerNotification;
 );
 
 ts_union!(
@@ -1927,6 +1962,38 @@ mod tests {
         assert_eq!(json, raw);
     }
 
+    #[test]
+    fn test_custom_server_notification_roundtrip() {
+        let raw = json!( {
+            "jsonrpc": JsonRpcVersion2_0,
+            "method": "notifications/custom-server",
+            "params": {"hello": "world"},
+        });
+
+        let message: ServerJsonRpcMessage =
+            serde_json::from_value(raw.clone()).expect("invalid notification");
+        match &message {
+            ServerJsonRpcMessage::Notification(JsonRpcNotification {
+                notification: ServerNotification::CustomServerNotification(notification),
+                ..
+            }) => {
+                assert_eq!(notification.method, "notifications/custom-server");
+                assert_eq!(
+                    notification
+                        .params
+                        .as_ref()
+                        .and_then(|p| p.get("hello"))
+                        .expect("hello present"),
+                    "world"
+                );
+            }
+            _ => panic!("Expected custom server notification"),
+        }
+
+        let json = serde_json::to_value(message).expect("valid json");
+        assert_eq!(json, raw);
+    }
+
     #[test]
     fn test_request_conversion() {
         let raw = json!( {

crates/rmcp/src/model/meta.rs

@@ -4,8 +4,9 @@ use serde::{Deserialize, Serialize};
 use serde_json::Value;
 
 use super::{
-    ClientNotification, ClientRequest, CustomClientNotification, Extensions, JsonObject,
-    JsonRpcMessage, NumberOrString, ProgressToken, ServerNotification, ServerRequest,
+    ClientNotification, ClientRequest, CustomClientNotification, CustomServerNotification,
+    Extensions, JsonObject, JsonRpcMessage, NumberOrString, ProgressToken, ServerNotification,
+    ServerRequest,
 };
 
 pub trait GetMeta {
@@ -38,6 +39,26 @@ impl GetMeta for CustomClientNotification {
     }
 }
 
+impl GetExtensions for CustomServerNotification {
+    fn extensions(&self) -> &Extensions {
+        &self.extensions
+    }
+    fn extensions_mut(&mut self) -> &mut Extensions {
+        &mut self.extensions
+    }
+}
+
+impl GetMeta for CustomServerNotification {
+    fn get_meta_mut(&mut self) -> &mut Meta {
+        self.extensions_mut().get_or_insert_default()
+    }
+    fn get_meta(&self) -> &Meta {
+        self.extensions()
+            .get::<Meta>()
+            .unwrap_or(Meta::static_empty())
+    }
+}
+
 macro_rules! variant_extension {
     (
         $Enum: ident {
@@ -117,6 +138,7 @@ variant_extension! {
         ResourceListChangedNotification
         ToolListChangedNotification
         PromptListChangedNotification
+        CustomServerNotification
     }
 }
 #[derive(Debug, Serialize, Deserialize, Clone, Default, PartialEq)]

crates/rmcp/src/model/serde_impl.rs

@@ -3,8 +3,8 @@ use std::borrow::Cow;
 use serde::{Deserialize, Serialize};
 
 use super::{
-    CustomClientNotification, Extensions, Meta, Notification, NotificationNoParam, Request,
-    RequestNoParam, RequestOptionalParam,
+    CustomClientNotification, CustomServerNotification, Extensions, Meta, Notification,
+    NotificationNoParam, Request, RequestNoParam, RequestOptionalParam,
 };
 #[derive(Serialize, Deserialize)]
 struct WithMeta<'a, P> {
@@ -302,6 +302,59 @@ impl<'de> Deserialize<'de> for CustomClientNotification {
     }
 }
 
+impl Serialize for CustomServerNotification {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: serde::Serializer,
+    {
+        let extensions = &self.extensions;
+        let _meta = extensions.get::<Meta>().map(Cow::Borrowed);
+        let params = self.params.as_ref();
+
+        let params = if _meta.is_some() || params.is_some() {
+            Some(WithMeta {
+                _meta,
+                _rest: &self.params,
+            })
+        } else {
+            None
+        };
+
+        ProxyOptionalParam::serialize(
+            &ProxyOptionalParam {
+                method: &self.method,
+                params,
+            },
+            serializer,
+        )
+    }
+}
+
+impl<'de> Deserialize<'de> for CustomServerNotification {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: serde::Deserializer<'de>,
+    {
+        let body =
+            ProxyOptionalParam::<'_, _, Option<serde_json::Value>>::deserialize(deserializer)?;
+        let mut params = None;
+        let mut _meta = None;
+        if let Some(body_params) = body.params {
+            params = body_params._rest;
+            _meta = body_params._meta.map(|m| m.into_owned());
+        }
+        let mut extensions = Extensions::new();
+        if let Some(meta) = _meta {
+            extensions.insert(meta);
+        }
+        Ok(CustomServerNotification {
+            extensions,
+            method: body.method,
+            params,
+        })
+    }
+}
+
 #[cfg(test)]
 mod test {
     use serde_json::json;

crates/rmcp/tests/test_message_schema/server_json_rpc_message_schema.json

@@ -392,6 +392,19 @@
         "content"
       ]
     },
+    "CustomServerNotification": {
+      "description": "A catch-all notification the server can use to send custom messages to a client.\n\nThis preserves the raw `method` name and `params` payload so handlers can\ndeserialize them into domain-specific types.",
+      "type": "object",
+      "properties": {
+        "method": {
+          "type": "string"
+        },
+        "params": true
+      },
+      "required": [
+        "method"
+      ]
+    },
     "CancelledNotificationMethod": {
       "type": "string",
       "format": "const",
@@ -977,6 +990,9 @@
         },
         {
           "$ref": "#/definitions/NotificationNoParam3"
+        },
+        {
+          "$ref": "#/definitions/CustomServerNotification"
         }
       ],
       "required": [

crates/rmcp/tests/test_message_schema/server_json_rpc_message_schema_current.json

@@ -392,6 +392,19 @@
         "content"
       ]
     },
+    "CustomServerNotification": {
+      "description": "A catch-all notification the server can use to send custom messages to a client.\n\nThis preserves the raw `method` name and `params` payload so handlers can\ndeserialize them into domain-specific types.",
+      "type": "object",
+      "properties": {
+        "method": {
+          "type": "string"
+        },
+        "params": true
+      },
+      "required": [
+        "method"
+      ]
+    },
     "CancelledNotificationMethod": {
       "type": "string",
       "format": "const",
@@ -977,6 +990,9 @@
         },
         {
           "$ref": "#/definitions/NotificationNoParam3"
+        },
+        {
+          "$ref": "#/definitions/CustomServerNotification"
         }
       ],
       "required": [

crates/rmcp/tests/test_notification.rs

@@ -3,8 +3,9 @@ use std::sync::Arc;
 use rmcp::{
     ClientHandler, ServerHandler, ServiceExt,
     model::{
-        ClientNotification, CustomClientNotification, ResourceUpdatedNotificationParam,
-        ServerCapabilities, ServerInfo, SubscribeRequestParam,
+        ClientNotification, CustomClientNotification, CustomServerNotification,
+        ResourceUpdatedNotificationParam, ServerCapabilities, ServerInfo, ServerNotification,
+        SubscribeRequestParam,
     },
 };
 use serde_json::json;
@@ -165,3 +166,76 @@ async fn test_custom_client_notification_reaches_server() -> anyhow::Result<()>
     client.cancel().await?;
     Ok(())
 }
+
+struct CustomServerNotifier;
+
+impl ServerHandler for CustomServerNotifier {
+    async fn on_initialized(&self, context: rmcp::service::NotificationContext<rmcp::RoleServer>) {
+        let peer = context.peer.clone();
+        tokio::spawn(async move {
+            peer.send_notification(ServerNotification::CustomServerNotification(
+                CustomServerNotification::new(
+                    "notifications/custom-test",
+                    Some(json!({ "hello": "world" })),
+                ),
+            ))
+            .await
+            .expect("send custom notification");
+        });
+    }
+}
+
+struct CustomClient {
+    receive_signal: Arc<Notify>,
+    payload: Arc<Mutex<Option<CustomNotificationPayload>>>,
+}
+
+impl ClientHandler for CustomClient {
+    async fn on_custom_notification(
+        &self,
+        notification: CustomServerNotification,
+        _context: rmcp::service::NotificationContext<rmcp::RoleClient>,
+    ) {
+        let CustomServerNotification { method, params, .. } = notification;
+        let mut payload = self.payload.lock().await;
+        *payload = Some((method, params));
+        self.receive_signal.notify_one();
+    }
+}
+
+#[tokio::test]
+async fn test_custom_server_notification_reaches_client() -> anyhow::Result<()> {
+    let _ = tracing_subscriber::registry()
+        .with(
+            tracing_subscriber::EnvFilter::try_from_default_env()
+                .unwrap_or_else(|_| "debug".to_string().into()),
+        )
+        .with(tracing_subscriber::fmt::layer())
+        .try_init();
+
+    let (server_transport, client_transport) = tokio::io::duplex(4096);
+    tokio::spawn(async move {
+        let server = CustomServerNotifier {}.serve(server_transport).await?;
+        server.waiting().await?;
+        anyhow::Ok(())
+    });
+
+    let receive_signal = Arc::new(Notify::new());
+    let payload = Arc::new(Mutex::new(None));
+
+    let client = CustomClient {
+        receive_signal: receive_signal.clone(),
+        payload: payload.clone(),
+    }
+    .serve(client_transport)
+    .await?;
+
+    tokio::time::timeout(std::time::Duration::from_secs(5), receive_signal.notified()).await?;
+
+    let (method, params) = payload.lock().await.clone().expect("payload set");
+    assert_eq!("notifications/custom-test", method);
+    assert_eq!(Some(json!({ "hello": "world" })), params);
+
+    client.cancel().await?;
+    Ok(())
+}