feat: MCP client sdk matching pending requests

crates/mcp-client/Cargo.toml

@@ -6,18 +6,24 @@ edition = "2021"
 [dependencies]
 mcp-core = { path = "../mcp-core" }
 tokio = { version = "1", features = ["full"] }
-reqwest = { version = "0.12.9", default-features = false, features = ["json", "stream", "rustls-tls"] }
-reqwest-eventsource = "0.6.0"
+reqwest = { version = "0.11", default-features = false, features = ["json", "stream", "rustls-tls"] }
+eventsource-client = "0.12.0"
+futures = "0.3"
 futures-util = "0.3"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 clap = { version = "4.5", features = ["derive"] }
 async-trait = "0.1.83"
 url = "2.5.4"
+thiserror = "1.0"
 anyhow = "1.0"
 tracing = "0.1"
 tracing-subscriber = { version = "0.3", features = ["env-filter"] }
 tokio-retry = "0.3"
+tower = { version = "0.4", features = ["timeout", "util"] }
+tower-service = "0.3"
+tokio-util = { version = "0.7", features = ["io-util", "io"] }
+rand = "0.8"
 
 [dev-dependencies]
 warp = "0.3"

crates/mcp-client/README.md

@@ -1,13 +1,11 @@
-## Testing stdio
+## Testing stdio transport
 
 ```bash
-cargo run -p mcp_client -- --mode git
-cargo run -p mcp_client -- --mode echo
-
-cargo run -p mcp_client --bin stdio
+cargo run -p mcp-client --example stdio
 ```
 
-## Testing SSE
+## Testing SSE transport
+
+1. Start the MCP server in one terminal: `fastmcp run -t sse echo.py`
+2. Run the client example in new terminal: `cargo run -p mcp-client --example sse`
 
-1. Start the MCP server: `fastmcp run -t sse echo.py`
-2. Run the client: `cargo run -p mcp_client --bin sse`

crates/mcp-client/examples/clients.rs

@@ -0,0 +1,144 @@
+use mcp_client::{
+    client::{ClientCapabilities, ClientInfo, McpClient, McpClientImpl},
+    service::TransportService,
+    transport::{SseTransport, StdioTransport},
+};
+use rand::Rng;
+use rand::SeedableRng;
+use std::sync::Arc;
+use std::time::Duration;
+use tower::ServiceBuilder;
+use tracing_subscriber::EnvFilter;
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    // Initialize logging
+    tracing_subscriber::fmt()
+        .with_env_filter(
+            EnvFilter::from_default_env().add_directive("mcp_client=debug".parse().unwrap()),
+        )
+        .init();
+
+    // Create two separate clients with stdio transport
+    let client1 = create_stdio_client("client1", "1.0.0")?;
+    let client2 = create_stdio_client("client2", "1.0.0")?;
+    let client3 = create_sse_client("client3", "1.0.0")?;
+
+    // Initialize both clients
+    let mut clients: Vec<Box<dyn McpClient + Send + Sync>> = Vec::new();
+    clients.push(client1);
+    clients.push(client2);
+    clients.push(client3);
+
+    // Initialize all clients
+    for (i, client) in clients.iter_mut().enumerate() {
+        let info = ClientInfo {
+            name: format!("example-client-{}", i + 1),
+            version: "1.0.0".to_string(),
+        };
+        let capabilities = ClientCapabilities::default();
+
+        println!("\nInitializing client {}", i + 1);
+        let init_result = client.initialize(info, capabilities).await?;
+        println!("Client {} initialized: {:?}", i + 1, init_result);
+    }
+
+    // List tools for all clients
+    for (i, client) in clients.iter_mut().enumerate() {
+        let tools = client.list_tools().await?;
+        println!("\nClient {} tools: {:?}", i + 1, tools);
+    }
+
+    println!("\n\n----------------------------------\n\n");
+
+    // Wrap clients in Arc before spawning tasks
+    let clients = Arc::new(clients);
+    let mut handles = vec![];
+
+    for i in 0..20 {
+        let clients = Arc::clone(&clients);
+        let handle = tokio::spawn(async move {
+            // let mut rng = rand::thread_rng();
+            let mut rng = rand::rngs::StdRng::from_entropy();
+            tokio::time::sleep(Duration::from_millis(rng.gen_range(5..50))).await;
+
+            // Randomly select an operation
+            match rng.gen_range(0..4) {
+                0 => {
+                    println!("\n{i}: Listing tools for client 1 (stdio)");
+                    match clients[0].list_tools().await {
+                        Ok(tools) => {
+                            println!("  {i}: -> Got tools, first one: {:?}", tools.tools.first())
+                        }
+                        Err(e) => println!("  {i}: -> Error: {}", e),
+                    }
+                }
+                1 => {
+                    println!("\n{i}: Listing tools for client 3 (sse)");
+                    match clients[2].list_tools().await {
+                        Ok(tools) => {
+                            println!("  {i}: -> Got tools, first one: {:?}", tools.tools.first())
+                        }
+                        Err(e) => println!("  {i}: -> Error: {}", e),
+                    }
+                }
+                2 => {
+                    println!("\n{i}: Calling tool for client 2 (stdio)");
+                    match clients[1]
+                        .call_tool("git_status", serde_json::json!({ "repo_path": "." }))
+                        .await
+                    {
+                        Ok(result) => println!(
+                            "  {i}: -> Tool execution result, is_error: {:?}",
+                            result.is_error
+                        ),
+                        Err(e) => println!("  {i}: -> Error: {}", e),
+                    }
+                }
+                3 => {
+                    println!("\n{i}: Calling tool for client 3 (sse)");
+                    match clients[2]
+                            .call_tool(
+                                "echo_tool",
+                                serde_json::json!({ "message": "Client with SSE transport - calling a tool" }),
+                            )
+                            .await
+                        {
+                        Ok(result) => println!("  {i}: -> Tool execution result, is_error: {:?}", result.is_error),
+                        Err(e) => println!("  {i}: -> Error: {}", e),
+                    }
+                }
+                _ => unreachable!(),
+            }
+            Ok::<(), Box<dyn std::error::Error + Send + Sync>>(())
+        });
+        handles.push(handle);
+    }
+
+    // Wait for all tasks to complete
+    for handle in handles {
+        handle.await.unwrap().unwrap();
+    }
+
+    Ok(())
+}
+
+fn create_stdio_client(
+    _name: &str,
+    _version: &str,
+) -> Result<Box<dyn McpClient + Send + Sync>, Box<dyn std::error::Error>> {
+    let transport = StdioTransport::new("uvx", vec!["mcp-server-git".to_string()]);
+    // TODO: Add timeout middleware
+    let service = ServiceBuilder::new().service(TransportService::new(transport));
+    Ok(Box::new(McpClientImpl::new(service)))
+}
+
+fn create_sse_client(
+    _name: &str,
+    _version: &str,
+) -> Result<Box<dyn McpClient + Send + Sync>, Box<dyn std::error::Error>> {
+    let transport = SseTransport::new("http://localhost:8000/sse");
+    // TODO: Add timeout middleware
+    let service = ServiceBuilder::new().service(TransportService::new(transport));
+    Ok(Box::new(McpClientImpl::new(service)))
+}

crates/mcp-client/examples/sse.rs

@@ -0,0 +1,59 @@
+use anyhow::Result;
+use mcp_client::client::{ClientCapabilities, ClientInfo, McpClient, McpClientImpl};
+use mcp_client::{service::TransportService, transport::SseTransport};
+use std::time::Duration;
+use tower::ServiceBuilder;
+use tracing_subscriber::EnvFilter;
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    // Initialize logging
+    tracing_subscriber::fmt()
+        .with_env_filter(
+            EnvFilter::from_default_env()
+                .add_directive("mcp_client=debug".parse().unwrap())
+                .add_directive("eventsource_client=debug".parse().unwrap()),
+        )
+        .init();
+
+    // Create the base transport
+    let transport = SseTransport::new("http://localhost:8000/sse");
+
+    // Build service
+    // TODO: Add timeout middleware
+    let service = ServiceBuilder::new().service(TransportService::new(transport));
+
+    // Create client
+    let client = McpClientImpl::new(service);
+    println!("Client created\n");
+
+    // Initialize
+    let server_info = client
+        .initialize(
+            ClientInfo {
+                name: "test-client".into(),
+                version: "1.0.0".into(),
+            },
+            ClientCapabilities::default(),
+        )
+        .await?;
+    println!("Connected to server: {server_info:?}\n");
+
+    // Sleep for 100ms to allow the server to start - surprisingly this is required!
+    tokio::time::sleep(Duration::from_millis(100)).await;
+
+    // List tools
+    let tools = client.list_tools().await?;
+    println!("Available tools: {tools:?}\n");
+
+    // Call tool
+    let tool_result = client
+        .call_tool(
+            "echo_tool",
+            serde_json::json!({ "message": "Client with SSE transport - calling a tool" }),
+        )
+        .await?;
+    println!("Tool result: {tool_result:?}");
+
+    Ok(())
+}

crates/mcp-client/examples/stdio.rs

@@ -0,0 +1,53 @@
+use anyhow::Result;
+use mcp_client::client::{
+    ClientCapabilities, ClientInfo, Error as ClientError, McpClient, McpClientImpl,
+};
+use mcp_client::{service::TransportService, transport::StdioTransport};
+use tower::ServiceBuilder;
+use tracing_subscriber::EnvFilter;
+
+#[tokio::main]
+async fn main() -> Result<(), ClientError> {
+    // Initialize logging
+    tracing_subscriber::fmt()
+        .with_env_filter(
+            EnvFilter::from_default_env()
+                .add_directive("mcp_client=debug".parse().unwrap())
+                .add_directive("eventsource_client=debug".parse().unwrap()),
+        )
+        .init();
+
+    // Create the transport
+    let transport = StdioTransport::new("uvx", vec!["mcp-server-git".to_string()]);
+
+    // Build service
+    // TODO: Add timeout middleware
+    let service = ServiceBuilder::new().service(TransportService::new(transport));
+
+    // Create client
+    let client = McpClientImpl::new(service);
+
+    // Initialize
+    let server_info = client
+        .initialize(
+            ClientInfo {
+                name: "test-client".into(),
+                version: "1.0.0".into(),
+            },
+            ClientCapabilities::default(),
+        )
+        .await?;
+    println!("Connected to server: {server_info:?}\n");
+
+    // List tools
+    let tools = client.list_tools().await?;
+    println!("Available tools: {tools:?}\n");
+
+    // Call tool 'git_status' with arguments = {"repo_path": "."}
+    let tool_result = client
+        .call_tool("git_status", serde_json::json!({ "repo_path": "." }))
+        .await?;
+    println!("Tool result: {tool_result:?}\n");
+
+    Ok(())
+}

crates/mcp-client/examples/stdio_integration.rs

@@ -0,0 +1,73 @@
+// This example shows how to use the mcp-client crate to interact with a server that has a simple counter tool.
+// The server is started by running `cargo run -p mcp-server` in the root of the mcp-server crate.
+use anyhow::Result;
+use mcp_client::client::{
+    ClientCapabilities, ClientInfo, Error as ClientError, McpClient, McpClientImpl,
+};
+use mcp_client::{service::TransportService, transport::StdioTransport};
+use tower::ServiceBuilder;
+use tracing_subscriber::EnvFilter;
+
+#[tokio::main]
+async fn main() -> Result<(), ClientError> {
+    // Initialize logging
+    tracing_subscriber::fmt()
+        .with_env_filter(
+            EnvFilter::from_default_env()
+                .add_directive("mcp_client=debug".parse().unwrap())
+                .add_directive("eventsource_client=debug".parse().unwrap()),
+        )
+        .init();
+
+    // Create the transport
+    let transport = StdioTransport::new(
+        "cargo",
+        vec!["run", "-p", "mcp-server"]
+            .into_iter()
+            .map(|s| s.to_string())
+            .collect(),
+    );
+
+    // Build service
+    // TODO: Add timeout middleware
+    let service = ServiceBuilder::new().service(TransportService::new(transport));
+
+    // Create client
+    let client = McpClientImpl::new(service);
+
+    // Initialize
+    let server_info = client
+        .initialize(
+            ClientInfo {
+                name: "test-client".into(),
+                version: "1.0.0".into(),
+            },
+            ClientCapabilities::default(),
+        )
+        .await?;
+    println!("Connected to server: {server_info:?}\n");
+
+    // List tools
+    let tools = client.list_tools().await?;
+    println!("Available tools: {tools:?}\n");
+
+    // Call tool 'increment' tool 3 times
+    for _ in 0..3 {
+        let increment_result = client.call_tool("increment", serde_json::json!({})).await?;
+        println!("Tool result for 'increment': {increment_result:?}\n");
+    }
+
+    // Call tool 'get_value'
+    let get_value_result = client.call_tool("get_value", serde_json::json!({})).await?;
+    println!("Tool result for 'get_value': {get_value_result:?}\n");
+
+    // Call tool 'decrement' once
+    let decrement_result = client.call_tool("decrement", serde_json::json!({})).await?;
+    println!("Tool result for 'decrement': {decrement_result:?}\n");
+
+    // Call tool 'get_value'
+    let get_value_result = client.call_tool("get_value", serde_json::json!({})).await?;
+    println!("Tool result for 'get_value': {get_value_result:?}\n");
+
+    Ok(())
+}