example: add example in progress

Author: jokemanfire

examples/clients/README.md

@@ -4,6 +4,12 @@ This directory contains Model Context Protocol (MCP) client examples implemented
 
 ## Example List
 
+### Test scripts
+- `progress_test_client.py`: A client that communicates with an MCP server using progress notifications.
+```bash
+python3 progress_test_client.py
+```
+
 ### SSE Client (`sse.rs`)
 
 A client that communicates with an MCP server using Server-Sent Events (SSE) transport.

examples/clients/scripts/progress_test_client.py

@@ -0,0 +1,223 @@
+#!/usr/bin/env python3
+"""
+Simple Python client to test progress notification functionality
+"""
+
+import json
+import sys
+import subprocess
+import time
+import threading
+from typing import Dict, Any
+
+class MCPClient:
+    def __init__(self):
+        self.request_id = 1
+        self.process = None
+
+    def start_server(self):
+        """Start the server process"""
+        try:
+            self.process = subprocess.Popen(
+                ["cargo", "run", "--example", "servers_progress_demo_stdio"],
+                cwd="../../servers",
+                stdin=subprocess.PIPE,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                text=True,
+                bufsize=0
+            )
+            return True
+        except Exception as e:
+            print(f"Failed to start server: {e}")
+            return False
+
+    def send_message(self, message: Dict[str, Any]) -> str:
+        """Send message to server"""
+        if not self.process:
+            return None
+        
+        json_msg = json.dumps(message)
+        print(f"→ Sending: {json_msg}")
+        
+        try:
+            self.process.stdin.write(json_msg + "\n")
+            self.process.stdin.flush()
+            return json_msg
+        except Exception as e:
+            print(f"Failed to send message: {e}")
+            return None
+
+    def read_response(self, timeout=5):
+        """Read server response"""
+        if not self.process:
+            return None
+            
+        try:
+            # Read one line of response
+            line = self.process.stdout.readline()
+            if line:
+                print(f"← Received: {line.strip()}")
+                return json.loads(line.strip())
+        except json.JSONDecodeError as e:
+            print(f"JSON parsing error: {e}, raw data: {line}")
+        except Exception as e:
+            print(f"Failed to read response: {e}")
+        return None
+
+    def initialize(self):
+        """Initialize connection"""
+        init_message = {
+            "jsonrpc": "2.0",
+            "id": self.request_id,
+            "method": "initialize",
+            "params": {
+                "protocolVersion": "2024-11-05",
+                "capabilities": {},
+                "clientInfo": {
+                    "name": "progress-test-client",
+                    "version": "1.0.0"
+                }
+            }
+        }
+        self.request_id += 1
+        
+        self.send_message(init_message)
+        response = self.read_response()
+        
+        if response and response.get("result"):
+            print("✅ Initialization successful")
+            
+            # Send initialization complete notification
+            initialized_notification = {
+                "jsonrpc": "2.0",
+                "method": "notifications/initialized"
+            }
+            self.send_message(initialized_notification)
+            return True
+        else:
+            print("❌ Initialization failed")
+            return False
+
+    def list_tools(self):
+        """List available tools"""
+        list_tools_message = {
+            "jsonrpc": "2.0",
+            "id": self.request_id,
+            "method": "tools/list"
+        }
+        self.request_id += 1
+        
+        self.send_message(list_tools_message)
+        response = self.read_response()
+        
+        if response and response.get("result"):
+            tools = response["result"].get("tools", [])
+            print(f"✅ Available tools: {[tool['name'] for tool in tools]}")
+            return tools
+        else:
+            print("❌ Failed to get tool list")
+            return []
+
+    def call_stream_processor(self, record_count=5):
+        """Call the stream processor tool"""
+        call_tool_message = {
+            "jsonrpc": "2.0",
+            "id": self.request_id,
+            "method": "tools/call",
+            "params": {
+                "name": "stream_processor",
+                "arguments": {
+                    "record_count": record_count
+                }
+            }
+        }
+        self.request_id += 1
+        
+        print(f"\n🚀 Starting to process {record_count} records...")
+        self.send_message(call_tool_message)
+        
+        # Listen for progress notifications and final result
+        progress_count = 0
+        start_time = time.time()
+        
+        while True:
+            response = self.read_response(timeout=10)
+            if not response:
+                print("Timeout or connection lost")
+                break
+            
+            # Handle progress notifications
+            if response.get("method") == "notifications/progress":
+                params = response.get("params", {})
+                progress = params.get("progress", 0)
+                total = params.get("total", 0)
+                message = params.get("message", "")
+                token = params.get("progressToken", "")
+                
+                elapsed = time.time() - start_time
+                print(f"📊 Progress update [{token}]: {progress}/{total} - {message} (elapsed: {elapsed:.1f}s)")
+                progress_count += 1
+            
+            # Handle tool call result
+            elif "result" in response:
+                result = response["result"]
+                if result.get("content"):
+                    content = result["content"][0]["text"]
+                    elapsed = time.time() - start_time
+                    print(f"✅ Processing completed: {content}")
+                    print(f"📈 Total progress notifications received: {progress_count}")
+                    print(f"⏱️  Total time elapsed: {elapsed:.2f}s")
+                break
+            
+            # Handle errors
+            elif "error" in response:
+                error = response["error"]
+                print(f"❌ Error: {error}")
+                break
+
+    def stop_server(self):
+        """Stop the server"""
+        if self.process:
+            self.process.terminate()
+            self.process.wait()
+
+def main():
+    print("🧪 RMCP Progress Notification Test Client")
+    print("=" * 50)
+    
+    client = MCPClient()
+    
+    try:
+        # Start server
+        print("🔧 Starting server...")
+        if not client.start_server():
+            return
+        
+        time.sleep(1)  # Wait for server to start
+        
+        # Initialize connection
+        print("\n🤝 Initializing connection...")
+        if not client.initialize():
+            return
+        
+        # List tools
+        print("\n🔧 Getting available tools...")
+        tools = client.list_tools()
+        
+        # Test stream processor tool
+        if any(tool["name"] == "stream_processor" for tool in tools):
+            client.call_stream_processor(record_count=50)
+        else:
+            print("❌ stream_processor tool not found")
+        
+    except KeyboardInterrupt:
+        print("\n\n⏹️  User interrupted")
+    except Exception as e:
+        print(f"\n❌ Error occurred: {e}")
+    finally:
+        print("\n🔚 Closing connection...")
+        client.stop_server()
+
+if __name__ == "__main__":
+    main() 

examples/servers/Cargo.toml

@@ -1,5 +1,3 @@
-
-
 [package]
 name = "mcp-server-examples"
 version = "0.1.5"
@@ -9,6 +7,7 @@ publish = false
 [dependencies]
 rmcp = { path = "../../crates/rmcp", features = [
     "server",
+    "client",
     "transport-sse-server",
     "transport-io",
     "transport-streamable-http-server",
@@ -78,3 +77,7 @@ path = "src/simple_auth_sse.rs"
 [[example]]
 name = "counter_hyper_streamable_http"
 path = "src/counter_hyper_streamable_http.rs"
+
+[[example]]
+name = "servers_progress_demo_stdio"
+path = "src/progress_demo_stdio.rs"

examples/servers/src/common/mod.rs

@@ -1,3 +1,4 @@
 pub mod calculator;
 pub mod counter;
 pub mod generic_service;
+pub mod progress_demo;

examples/servers/src/common/progress_demo.rs

@@ -0,0 +1,110 @@
+use std::time::Duration;
+
+use rmcp::{
+    Error as McpError, RoleServer, ServerHandler, model::*, schemars, service::RequestContext, tool,
+};
+use serde_json::json;
+use tracing::{debug, info};
+
+#[derive(Clone)]
+pub struct ProgressDemo {}
+
+#[tool(tool_box)]
+impl ProgressDemo {
+    #[allow(dead_code)]
+    pub fn new() -> Self {
+        Self {}
+    }
+    #[tool(description = "Process data stream with progress updates")]
+    async fn stream_processor(
+        &self,
+        #[tool(param)]
+        #[schemars(description = "Number of records to simulate")]
+        record_count: u32,
+        ctx: RequestContext<RoleServer>,
+    ) -> Result<CallToolResult, McpError> {
+        if record_count == 0 {
+            return Ok(CallToolResult::success(vec![Content::text(
+                "No records to process",
+            )]));
+        }
+
+        info!("Starting stream processing of {} records", record_count);
+        let mut processed_records = Vec::new();
+        let mut counter = 0u32;
+
+        // simulate data stream processing
+        for i in 0..record_count {
+            // simulate getting record from data stream
+            tokio::time::sleep(Duration::from_millis(200)).await;
+
+            let record = format!("Record_{:04}", i + 1);
+            processed_records.push(record.clone());
+            counter += 1;
+
+            // create progress notification param
+            let progress_param = ProgressNotificationParam {
+                progress_token: ProgressToken(NumberOrString::Number(counter)),
+                progress: counter,
+                total: Some(record_count),
+                message: Some(record.clone()),
+            };
+
+            match ctx.peer.notify_progress(progress_param).await {
+                Ok(_) => {
+                    debug!("Processed record: {}", record);
+                }
+                Err(e) => {
+                    return Err(McpError::internal_error(
+                        format!("Failed to notify progress: {}", e),
+                        Some(json!({
+                            "record": record,
+                            "progress": counter,
+                            "error": e.to_string()
+                        })),
+                    ));
+                }
+            }
+        }
+
+        Ok(CallToolResult::success(vec![Content::text(format!(
+            "Processed {} records successfully",
+            processed_records.len()
+        ))]))
+    }
+}
+
+#[tool(tool_box)]
+impl ServerHandler for ProgressDemo {
+    fn get_info(&self) -> ServerInfo {
+        ServerInfo {
+            protocol_version: ProtocolVersion::V_2024_11_05,
+            capabilities: ServerCapabilities::builder()
+                .enable_tools()
+                .build(),
+            server_info: Implementation::from_build_env(),
+            instructions: Some(
+                "This server demonstrates progress notifications during long-running operations. \
+                 Use the tools to see real-time progress updates for batch processing, file downloads, \
+                 and stream processing operations.".to_string()
+            ),
+        }
+    }
+
+    async fn initialize(
+        &self,
+        _request: ClientInfo,
+        _context: RequestContext<RoleServer>,
+    ) -> Result<InitializeResult, McpError> {
+        Ok(InitializeResult {
+            protocol_version: ProtocolVersion::V_2024_11_05,
+            capabilities: ServerCapabilities::builder().enable_tools().build(),
+            server_info: Implementation::from_build_env(),
+            instructions: Some(
+                "Progress notification demo server initialized. \
+                 Ready to demonstrate real-time progress updates."
+                    .to_string(),
+            ),
+        })
+    }
+}

examples/servers/src/progress_demo_stdio.rs

@@ -0,0 +1,22 @@
+use rmcp::{ServiceExt, transport::stdio};
+use tracing_subscriber::EnvFilter;
+
+mod common;
+use common::progress_demo::ProgressDemo;
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    // Initialize tracing for better debugging
+    tracing_subscriber::fmt()
+        .with_env_filter(EnvFilter::from_default_env())
+        .init();
+
+    let server = ProgressDemo::new();
+    let service = server.serve(stdio()).await.inspect_err(|e| {
+        tracing::error!("serving error: {:?}", e);
+    })?;
+
+    service.waiting().await?;
+
+    Ok(())
+}