modelcontextprotocol/rust-sdk
Watch
1
0
Fork
You've already forked rust-sdk
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

feat: include reqwest in transport-streamble-http-client feature (#376)

Browse source 
* fix: add reqwest dependency to transport-streamable-http-client feature

- Fix compilation error when using transport-streamable-http-client feature due to missing dependency
- Move From<reqwest::Error> implementation to reqwest module

* feat(rmcp): enhance transport features by decoupling reqwest

- Added reqwest features for reqwest-based implementations.
- Updated documentation
- Modified error handling in SSE transport to use `String` for content type.
- Updated examples to include new features

* feat(rmcp): enhance transport features by decoupling reqwest

- Added reqwest features for reqwest-based implementations
- Updated documentation
- Modified error handling in SSE transport to use `String`
- Updated examples to include new features
This commit is contained in:
Julián Montes de Oca 2025-08-20 00:43:18 -03:00 • committed by GitHub
commit 7f20a87825
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
1 changed files with 326 additions and 29 deletions
Download patch file Download diff file Expand all files Collapse all files

168
crates/rmcp/src/transport/streamable_http_client.rs
View file

@ -50,17 +50,12 @@ pub enum StreamableHttpError<E: std::error::Error + Send + Sync + 'static> {
Auth(#[from] crate::transport::auth::AuthError),
}
impl From<reqwest::Error> for StreamableHttpError<reqwest::Error> {
fn from(e: reqwest::Error) -> Self {
StreamableHttpError::Client(e)
}
}
#[derive(Debug, Clone, Error)]
pub enum StreamableHttpProtocolError {
#[error("Missing session id in response")]
MissingSessionIdInResponse,
}
pub enum StreamableHttpPostResponse {
Accepted,
Json(ServerJsonRpcMessage, Option<String>),
@ -490,9 +485,170 @@ impl<C: StreamableHttpClient> Worker for StreamableHttpClientWorker<C> {
}
}
/// A client-agnostic HTTP transport for RMCP that supports streaming responses.
///
/// This transport allows you to choose your preferred HTTP client implementation
/// by implementing the [`StreamableHttpClient`] trait. The transport handles
/// session management, SSE streaming, and automatic reconnection.
///
/// # Usage
///
/// ## Using reqwest
///
/// ```rust,no_run
/// use rmcp::transport::StreamableHttpClientTransport;
///
/// // Enable the reqwest feature in Cargo.toml:
/// // rmcp = { version = "0.5", features = ["transport-streamable-http-client-reqwest"] }
///
/// let transport = StreamableHttpClientTransport::from_uri("http://localhost:8000/mcp");
/// ```
///
/// ## Using a custom HTTP client
///
/// ```rust,no_run
/// use rmcp::transport::streamable_http_client::{
/// StreamableHttpClient,
/// StreamableHttpClientTransport,
/// StreamableHttpClientTransportConfig
/// };
/// use std::sync::Arc;
/// use futures::stream::BoxStream;
/// use rmcp::model::ClientJsonRpcMessage;
/// use sse_stream::{Sse, Error as SseError};
///
/// #[derive(Clone)]
/// struct MyHttpClient;
///
/// #[derive(Debug, thiserror::Error)]
/// struct MyError;
///
/// impl std::fmt::Display for MyError {
/// fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
/// write!(f, "MyError")
/// }
/// }
///
/// impl StreamableHttpClient for MyHttpClient {
/// type Error = MyError;
///
/// async fn post_message(
/// &self,
/// _uri: Arc<str>,
/// _message: ClientJsonRpcMessage,
/// _session_id: Option<Arc<str>>,
/// _auth_header: Option<String>,
/// ) -> Result<rmcp::transport::streamable_http_client::StreamableHttpPostResponse, rmcp::transport::streamable_http_client::StreamableHttpError<Self::Error>> {
/// todo!()
/// }
///
/// async fn delete_session(
/// &self,
/// _uri: Arc<str>,
/// _session_id: Arc<str>,
/// _auth_header: Option<String>,
/// ) -> Result<(), rmcp::transport::streamable_http_client::StreamableHttpError<Self::Error>> {
/// todo!()
/// }
///
/// async fn get_stream(
/// &self,
/// _uri: Arc<str>,
/// _session_id: Arc<str>,
/// _last_event_id: Option<String>,
/// _auth_header: Option<String>,
/// ) -> Result<BoxStream<'static, Result<Sse, SseError>>, rmcp::transport::streamable_http_client::StreamableHttpError<Self::Error>> {
/// todo!()
/// }
/// }
///
/// let transport = StreamableHttpClientTransport::with_client(
/// MyHttpClient,
/// StreamableHttpClientTransportConfig::with_uri("http://localhost:8000/mcp")
/// );
/// ```
///
/// # Feature Flags
///
/// - `transport-streamable-http-client`: Base feature providing the generic transport infrastructure
/// - `transport-streamable-http-client-reqwest`: Includes reqwest HTTP client support with convenience methods
pub type StreamableHttpClientTransport<C> = WorkerTransport<StreamableHttpClientWorker<C>>;
impl<C: StreamableHttpClient> StreamableHttpClientTransport<C> {
/// Creates a new transport with a custom HTTP client implementation.
///
/// This method allows you to use any HTTP client that implements the [`StreamableHttpClient`] trait.
/// Use this when you want to use a custom HTTP client or when the reqwest feature is not enabled.
///
/// # Arguments
///
/// * `client` - Your HTTP client implementation
/// * `config` - Transport configuration including the server URI
///
/// # Example
///
/// ```rust,no_run
/// use rmcp::transport::streamable_http_client::{
/// StreamableHttpClient,
/// StreamableHttpClientTransport,
/// StreamableHttpClientTransportConfig
/// };
/// use std::sync::Arc;
/// use futures::stream::BoxStream;
/// use rmcp::model::ClientJsonRpcMessage;
/// use sse_stream::{Sse, Error as SseError};
///
/// // Define your custom client
/// #[derive(Clone)]
/// struct MyHttpClient;
///
/// #[derive(Debug, thiserror::Error)]
/// struct MyError;
///
/// impl std::fmt::Display for MyError {
/// fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
/// write!(f, "MyError")
/// }
/// }
///
/// impl StreamableHttpClient for MyHttpClient {
/// type Error = MyError;
///
/// async fn post_message(
/// &self,
/// _uri: Arc<str>,
/// _message: ClientJsonRpcMessage,
/// _session_id: Option<Arc<str>>,
/// _auth_header: Option<String>,
/// ) -> Result<rmcp::transport::streamable_http_client::StreamableHttpPostResponse, rmcp::transport::streamable_http_client::StreamableHttpError<Self::Error>> {
/// todo!()
/// }
///
/// async fn delete_session(
/// &self,
/// _uri: Arc<str>,
/// _session_id: Arc<str>,
/// _auth_header: Option<String>,
/// ) -> Result<(), rmcp::transport::streamable_http_client::StreamableHttpError<Self::Error>> {
/// todo!()
/// }
///
/// async fn get_stream(
/// &self,
/// _uri: Arc<str>,
/// _session_id: Arc<str>,
/// _last_event_id: Option<String>,
/// _auth_header: Option<String>,
/// ) -> Result<BoxStream<'static, Result<Sse, SseError>>, rmcp::transport::streamable_http_client::StreamableHttpError<Self::Error>> {
/// todo!()
/// }
/// }
///
/// let transport = StreamableHttpClientTransport::with_client(
/// MyHttpClient,
/// StreamableHttpClientTransportConfig::with_uri("http://localhost:8000/mcp")
/// );
/// ```
pub fn with_client(client: C, config: StreamableHttpClientTransportConfig) -> Self {
let worker = StreamableHttpClientWorker::new(client, config);
WorkerTransport::spawn(worker)