feat: MCP client sdk

crates/mcp-client/Cargo.toml

@@ -14,10 +14,13 @@ 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"
 
 [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,81 @@
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::sync::Mutex;
+
+use mcp_client::{
+    client::{ClientCapabilities, ClientInfo, McpClient},
+    service::{ServiceError, TransportService},
+    transport::StdioTransport,
+};
+use tower::{ServiceBuilder, ServiceExt};
+use tower::timeout::TimeoutLayer;
+use tracing_subscriber::EnvFilter;
+use tower::util::BoxService;
+use mcp_core::protocol::JsonRpcMessage;
+
+// Define a type alias for the boxed service using BoxService
+type BoxedService = BoxService<JsonRpcMessage, JsonRpcMessage, ServiceError>;
+
+#[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())
+                .add_directive("reqwest_eventsource=debug".parse().unwrap()),
+        )
+        .init();
+
+    // Create two separate clients with stdio transport
+    let client1 = create_client("client1", "1.0.0")?;
+    let client2 = create_client("client2", "1.0.0")?;
+
+    // Initialize both clients
+    let mut clients: Vec<McpClient<BoxedService>> = vec![client1, client2];
+
+    // 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 each client
+    for (i, client) in clients.iter_mut().enumerate() {
+        let tools = client.list_tools().await?;
+        println!("\nClient {} tools: {:?}", i + 1, tools);
+    }
+
+    Ok(())
+}
+
+fn create_client(
+    _name: &str,
+    _version: &str,
+) -> Result<McpClient<BoxService<JsonRpcMessage, JsonRpcMessage, ServiceError>>, Box<dyn std::error::Error>> {
+    // Create the transport
+    let transport = Arc::new(Mutex::new(StdioTransport::new("uvx", ["mcp-server-git"])));
+
+    // Build service with middleware including timeout
+    let service = ServiceBuilder::new()
+        .layer(TimeoutLayer::new(Duration::from_secs(30)))
+        .service(TransportService::new(Arc::clone(&transport)))
+        .map_err(|e: Box<dyn std::error::Error + Send + Sync>| {
+            if e.is::<tower::timeout::error::Elapsed>() {
+                ServiceError::Timeout(tower::timeout::error::Elapsed::new())
+            } else {
+                ServiceError::Other(e.to_string())
+            }
+        })
+        .boxed(); // Box the service to create a BoxService
+
+    // Create the client
+    Ok(McpClient::new(service))
+}

crates/mcp-client/examples/sse.rs

@@ -0,0 +1,73 @@
+use anyhow::Result;
+use mcp_client::client::{ClientCapabilities, ClientInfo, McpClient};
+use mcp_client::{
+    service::{ServiceError, TransportService},
+    transport::SseTransport,
+};
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::sync::Mutex;
+use tower::timeout::TimeoutLayer;
+use tower::{ServiceBuilder, ServiceExt};
+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("reqwest_eventsource=debug".parse().unwrap()),
+        )
+        .init();
+
+    // Create the base transport as Arc<Mutex<SseTransport>>
+    let transport = Arc::new(Mutex::new(SseTransport::new("http://localhost:8000/sse")?));
+
+    // Build service with middleware including timeout
+    let service = ServiceBuilder::new()
+        .layer(TimeoutLayer::new(Duration::from_secs(30)))
+        .service(TransportService::new(Arc::clone(&transport)))
+        .map_err(|e: Box<dyn std::error::Error + Send + Sync>| {
+            if e.is::<tower::timeout::error::Elapsed>() {
+                ServiceError::Timeout(tower::timeout::error::Elapsed::new())
+            } else {
+                ServiceError::Other(e.to_string())
+            }
+        });
+
+    // Create client
+    let mut client = McpClient::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,66 @@
+use anyhow::Result;
+use mcp_client::client::{ClientCapabilities, ClientInfo, Error as ClientError, McpClient};
+use mcp_client::{
+    service::{ServiceError, TransportService},
+    transport::StdioTransport,
+};
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::sync::Mutex;
+use tower::timeout::TimeoutLayer;
+use tower::{ServiceBuilder, ServiceExt};
+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("reqwest_eventsource=debug".parse().unwrap()),
+        )
+        .init();
+
+    // Create the base transport as Arc<Mutex<StdioTransport>>
+    let transport = Arc::new(Mutex::new(StdioTransport::new("uvx", ["mcp-server-git"])));
+
+    // Build service with middleware including timeout
+    let service = ServiceBuilder::new()
+        .layer(TimeoutLayer::new(Duration::from_secs(30)))
+        .service(TransportService::new(Arc::clone(&transport)))
+        .map_err(|e: Box<dyn std::error::Error + Send + Sync>| {
+            if e.is::<tower::timeout::error::Elapsed>() {
+                ServiceError::Timeout(tower::timeout::error::Elapsed::new())
+            } else {
+                ServiceError::Other(e.to_string())
+            }
+        });
+
+    // Create client
+    let mut client = McpClient::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' wtih arguments = {"repo_path": "."}
+    let tool_result = client
+        .call_tool("git_status", serde_json::json!({ "repo_path": "." }))
+        .await?;
+    println!("Tool result: {tool_result:?}");
+
+    Ok(())
+}