refactor: Transport trait and worker transport, and streamable http client with those new features. (#167)

* refactor: a transport trait and streamable http client

* test: test with js streamable http server

* refactor: separate sse stream connection

* fix(test): streamable http client test with js

* fix(test): wait longer time for js server startup

Author: 4t145

.github/workflows/ci.yml

@@ -157,15 +157,16 @@ jobs:
       - uses: actions/checkout@v3
 
       - name: Install Rust
-        uses: dtolnay/rust-toolchain@stable
+        uses: dtolnay/rust-toolchain@nightly
 
       - uses: Swatinem/rust-cache@v2
 
       - name: Generate documentation
         run: |
-          cargo doc --no-deps -p rmcp -p rmcp-macros
+          cargo +nightly doc --no-deps -p rmcp -p rmcp-macros --all-features
         env:
-          RUSTDOCFLAGS: -Dwarnings
+          RUSTDOCFLAGS: --cfg docsrs -Dwarnings
+          RUSTFLAGS: --cfg docsrs 
 
   release:
     name: Release crates

README.md

@@ -45,18 +45,24 @@ use tokio::io::{stdin, stdout};
 let transport = (stdin(), stdout());
 ```
 
-The transport type must implemented [`IntoTransport`](crate::transport::IntoTransport) trait, which allow split into a sink and a stream.
+#### Transport
+The transport type must implemented [`Transport`] trait, which allow it send message concurrently and receive message sequentially.
+There are 3 pairs of standard transport types:
 
-For client, the sink item is [`ClientJsonRpcMessage`](crate::model::ClientJsonRpcMessage) and stream item is [`ServerJsonRpcMessage`](crate::model::ServerJsonRpcMessage)
+| transport         | client                                                    | server                                                |
+|:-:                |:-:                                                        |:-:                                                    |
+| std IO            | [`child_process::TokioChildProcess`]                      | [`io::stdio`]                                         |
+| streamable http   | [`streamable_http_client::StreamableHttpClientTransport`] | [`streamable_http_server::session::create_session`]   |
+| sse               | [`sse_client::SseClientTransport`]                        | [`sse_server::SseServer`]                             |
 
-For server, the sink item is [`ServerJsonRpcMessage`](crate::model::ServerJsonRpcMessage) and stream item is [`ClientJsonRpcMessage`](crate::model::ClientJsonRpcMessage)
+#### [IntoTransport](`IntoTransport`) trait
+[`IntoTransport`] is a helper trait that implicitly convert a type into a transport type.
 
-##### These types is automatically implemented [`IntoTransport`](crate::transport::IntoTransport) trait
-
-1. The types that already implement both [`Sink`](futures::Sink) and [`Stream`](futures::Stream) trait.
-2. A tuple of sink `Tx` and stream `Rx`: `(Tx, Rx)`.
-3. The type that implement both [`tokio::io::AsyncRead`] and [`tokio::io::AsyncWrite`] trait.
-4. A tuple of [`tokio::io::AsyncRead`] `R `and [`tokio::io::AsyncWrite`] `W`: `(R, W)`.
+These types is automatically implemented [`IntoTransport`] trait
+1. A type that already implement both [`futures::Sink`] and [`futures::Stream`] trait, or a tuple `(Tx, Rx)`  where `Tx` is [`futures::Sink`] and `Rx` is [`futures::Stream`].
+2. A type that implement both [`tokio::io::AsyncRead`] and [`tokio::io::AsyncWrite`] trait. or a tuple `(R, W)` where `R` is [`tokio::io::AsyncRead`] and `W` is [`tokio::io::AsyncWrite`].
+3. A type that implement [Worker](`worker::Worker`) trait.
+4. A type that implement [`Transport`] trait.
 
 For example, you can see how we build a transport through TCP stream or http upgrade so easily. [examples](examples/README.md)
 </details>
@@ -194,13 +200,15 @@ See [examples](examples/README.md)
 - `macros`: macros default
 - `schemars`: implement `JsonSchema` for all model structs
 
-### Transports
+#### Transports
 
 - `transport-io`: Server stdio transport
 - `transport-sse-server`: Server SSE transport
 - `transport-child-process`: Client stdio transport
-- `transport-sse`: Client sse transport
+- `transport-sse-client`: Client sse transport
 - `transport-streamable-http-server` streamable http server transport
+- `transport-streamable-client-server` streamable http server transport
+
 
 ## Related Resources
 

crates/rmcp/Cargo.toml

@@ -11,6 +11,7 @@ documentation = "https://docs.rs/rmcp"
 
 [package.metadata.docs.rs]
 all-features = true
+rustdoc-args = ["--cfg", "docsrs"]
 
 [dependencies]
 serde = { version = "1.0", features = ["derive", "rc"] }
@@ -38,7 +39,7 @@ reqwest = { version = "0.12", default-features = false, features = [
     "json",
     "stream",
 ], optional = true }
-sse-stream = { version = "0.1.3", optional = true }
+sse-stream = { version = "0.1.4", optional = true }
 url = { version = "2.4", optional = true }
 
 # For tower compatibility
@@ -62,9 +63,26 @@ default = ["base64", "macros", "server"]
 client = []
 server = ["transport-async-rw", "dep:schemars"]
 macros = ["dep:rmcp-macros", "dep:paste"]
-__transport-sse = ["dep:reqwest", "dep:sse-stream", "dep:url"]
+
+# reqwest http client
+reqwest = ["dep:reqwest"]
+
+# SSE client
+client-side-sse = ["dep:sse-stream"]
+
+__transport-sse = ["reqwest", "client-side-sse", "dep:url", "transport-worker"]
+
+transport-worker = ["dep:tokio-stream"]
+
 transport-sse = ["__transport-sse", "reqwest?/rustls-tls"]
 transport-sse-tls-no-provider = ["__transport-sse", "reqwest?/rustls-tls-no-provider"]
+
+# Streamable HTTP client
+__transport-streamable-http-client = ["reqwest", "client-side-sse", "transport-worker", "dep:url"]
+transport-streamable-http-client = ["__transport-streamable-http-client", "reqwest?/rustls-tls"]
+transport-streamable-http-client-tls-no-provider = ["__transport-streamable-http-client", "reqwest?/rustls-tls-no-provider"]
+
+
 transport-async-rw = ["tokio/io-util", "tokio-util/codec"]
 transport-io = ["transport-async-rw", "tokio/io-std"]
 transport-child-process = ["transport-async-rw", "tokio/process"]
@@ -114,7 +132,7 @@ path = "tests/test_with_python.rs"
 
 [[test]]
 name = "test_with_js"
-required-features = ["server", "client", "transport-sse-server", "transport-child-process", "transport-streamable-http-server"]
+required-features = ["server", "client", "transport-sse-server", "transport-child-process", "transport-streamable-http-server", "transport-streamable-http-client"]
 path = "tests/test_with_js.rs"
 
 [[test]]

crates/rmcp/src/handler.rs

@@ -1,4 +1,6 @@
 #[cfg(feature = "client")]
+#[cfg_attr(docsrs, doc(cfg(feature = "client")))]
 pub mod client;
 #[cfg(feature = "server")]
+#[cfg_attr(docsrs, doc(cfg(feature = "server")))]
 pub mod server;

crates/rmcp/src/handler/server/tool.rs

@@ -431,6 +431,7 @@ impl<S> ToolBox<S> {
 }
 
 #[cfg(feature = "macros")]
+#[cfg_attr(docsrs, doc(cfg(feature = "macros")))]
 #[macro_export]
 macro_rules! tool_box {
     (@pin_add $callee: ident, $attr: expr, $f: expr) => {

crates/rmcp/src/lib.rs

@@ -1,3 +1,5 @@
+#![cfg_attr(docsrs, feature(doc_cfg))]
+#![cfg_attr(docsrs, allow(unused_attributes))]
 //! The official Rust SDK for the Model Context Protocol (MCP).
 //!
 //! The MCP is a protocol that allows AI assistants to communicate with other
@@ -93,29 +95,40 @@ pub use error::Error;
 /// Basic data types in MCP specification
 pub mod model;
 #[cfg(any(feature = "client", feature = "server"))]
+#[cfg_attr(docsrs, doc(cfg(any(feature = "client", feature = "server"))))]
 pub mod service;
 #[cfg(feature = "client")]
+#[cfg_attr(docsrs, doc(cfg(feature = "client")))]
 pub use handler::client::ClientHandler;
 #[cfg(feature = "server")]
+#[cfg_attr(docsrs, doc(cfg(feature = "server")))]
 pub use handler::server::ServerHandler;
 #[cfg(any(feature = "client", feature = "server"))]
+#[cfg_attr(docsrs, doc(cfg(any(feature = "client", feature = "server"))))]
 pub use service::{Peer, Service, ServiceError, ServiceExt};
 #[cfg(feature = "client")]
+#[cfg_attr(docsrs, doc(cfg(feature = "client")))]
 pub use service::{RoleClient, serve_client};
 #[cfg(feature = "server")]
+#[cfg_attr(docsrs, doc(cfg(feature = "server")))]
 pub use service::{RoleServer, serve_server};
 
 pub mod handler;
 pub mod transport;
 
 // re-export
 #[cfg(all(feature = "macros", feature = "server"))]
+#[cfg_attr(docsrs, doc(cfg(all(feature = "macros", feature = "server"))))]
 pub use paste::paste;
 #[cfg(all(feature = "macros", feature = "server"))]
+#[cfg_attr(docsrs, doc(cfg(all(feature = "macros", feature = "server"))))]
 pub use rmcp_macros::tool;
 #[cfg(all(feature = "macros", feature = "server"))]
+#[cfg_attr(docsrs, doc(cfg(all(feature = "macros", feature = "server"))))]
 pub use schemars;
 #[cfg(feature = "macros")]
+#[cfg_attr(docsrs, doc(cfg(feature = "macros")))]
 pub use serde;
 #[cfg(feature = "macros")]
+#[cfg_attr(docsrs, doc(cfg(feature = "macros")))]
 pub use serde_json;

crates/rmcp/src/model.rs

@@ -36,6 +36,7 @@ pub fn object(value: serde_json::Value) -> JsonObject {
 
 /// Use this macro just like [`serde_json::json!`]
 #[cfg(feature = "macros")]
+#[cfg_attr(docsrs, doc(cfg(feature = "macros")))]
 #[macro_export]
 macro_rules! object {
     ({$($tt:tt)*}) => {