headless_lms_mcp/protocol/
progress.rs

1//! Progress notifications on the call's response stream. A no-op unless the client sent a
2//! progress token and the transport streams the response.
3
4use rmcp::model::{
5    JsonRpcNotification, JsonRpcVersion2_0, ProgressNotification, ProgressNotificationParam,
6    ProgressToken,
7};
8use serde_json::Value;
9use tokio::sync::mpsc::UnboundedSender;
10
11/// Sends progress notifications for one call.
12#[derive(Debug, Clone, Default)]
13pub struct ProgressReporter {
14    token: Option<ProgressToken>,
15    sink: Option<UnboundedSender<Value>>,
16}
17
18impl ProgressReporter {
19    /// `sink` receives complete JSON-RPC notification messages for the transport to stream.
20    pub fn new(token: Option<ProgressToken>, sink: Option<UnboundedSender<Value>>) -> Self {
21        Self { token, sink }
22    }
23
24    /// Reports `progress` out of `total` (when known). Never fails the call.
25    pub fn report(&self, progress: f64, total: Option<f64>, message: Option<&str>) {
26        let (Some(token), Some(sink)) = (&self.token, &self.sink) else {
27            return;
28        };
29        let mut params = ProgressNotificationParam::new(token.clone(), progress);
30        params.total = total;
31        params.message = message.map(str::to_string);
32        let notification = JsonRpcNotification {
33            jsonrpc: JsonRpcVersion2_0,
34            notification: ProgressNotification::new(params),
35        };
36        if let Ok(notification) = serde_json::to_value(notification) {
37            let _ = sink.send(notification);
38        }
39    }
40}
41
42#[cfg(test)]
43mod tests {
44    use rmcp::model::NumberOrString;
45    use serde_json::json;
46
47    use super::*;
48
49    #[test]
50    fn sends_only_with_token_and_sink() {
51        let (sender, mut receiver) = tokio::sync::mpsc::unbounded_channel();
52        ProgressReporter::new(None, Some(sender.clone())).report(1.0, None, None);
53        assert!(receiver.try_recv().is_err());
54        let token = ProgressToken(NumberOrString::String("t".into()));
55        ProgressReporter::new(Some(token), Some(sender)).report(1.0, Some(4.0), Some("validating"));
56        let sent = receiver.try_recv().unwrap();
57        assert_eq!(sent["params"]["progressToken"], json!("t"));
58        assert_eq!(sent["params"]["total"], json!(4.0));
59    }
60}
headless_lms_mcp/protocol/progress.rs

headless_lms_mcp/protocol/
progress.rs
