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

Merged
4t145 merged 5 commits into
modelcontextprotocol:mainfrom
4t145:streamable-http-client
May 17, 2025
Merged

refactor: Transport trait and worker transport, and streamable http client with those new features.#167
4t145 merged 5 commits into
modelcontextprotocol:mainfrom
4t145:streamable-http-client

Conversation

@4t145

@4t1454t145 commented May 9, 2025

Copy link
Copy Markdown
Contributor

related issue: #170, #55

1. Refactor transport trait

Now we have a real transport trait:

pubtraitTransport<R>:SendwhereR:ServiceRole,{typeError;fnsend(&mutself,item:TxJsonRpcMessage<R>,) -> implFuture<Output = Result<(),Self::Error>> + Send + 'static;fnreceive(&mutself) -> implFuture<Output = Option<RxJsonRpcMessage<R>>> + Send + '_;fnclose(&mutself) -> implFuture<Output = Result<(),Self::Error>> + Send;}

and this new transport trait is compatible with old Sink + Stream constrainment, with a thread shared sink.

pubstructSinkStreamTransport<Si,St>{stream:St,sink:Arc<Mutex<Si>>,}impl<Si,St>SinkStreamTransport<Si,St>{pubfnnew(sink:Si,stream:St) -> Self{Self{
stream,sink:Arc::new(Mutex::new(sink)),}}}impl<Role:ServiceRole,Si,St>Transport<Role>forSinkStreamTransport<Si,St>whereSt:Send + Stream<Item = RxJsonRpcMessage<Role>> + Unpin,Si:Send + Sink<TxJsonRpcMessage<Role>> + Unpin + 'static,{typeError = Si::Error;fnsend(&mutself,item:TxJsonRpcMessage<Role>,) -> implFuture<Output = Result<(),Self::Error>> + Send + 'static{use futures::SinkExt;let lock = self.sink.clone();asyncmove{letmut write = lock.lock().await;
write.send(item).await}}fnreceive(&mutself) -> implFuture<Output = Option<RxJsonRpcMessage<Role>>> + '_{use futures::StreamExt;self.stream.next()}asyncfnclose(&mutself) -> Result<(),Self::Error>{Ok(())}}

Based on those changes, we can make serve_inner loop not blocked by sending request.

  1. New worker transport type:
    For the case those we need to create a new worker task to run the transport, there's a new trait Worker, you can implement Worker and hand it over to WorkerTransport
pubtraitWorker:Sized + Send + 'static{typeError: std::error::Error + Send + Sync + 'static;typeRole:ServiceRole;fnerr_closed() -> Self::Error;fnerr_join(e: tokio::task::JoinError) -> Self::Error;fnrun(self,context:WorkerContext<Self>,) -> implFuture<Output = Result<(),WorkerQuitReason>> + Send;fnconfig(&self) -> WorkerConfig{WorkerConfig::default()}}

And with this new trait, I refactor many existed implementation based on it.

  1. Streamable http client
    Implemented streamable http client transport based on new worker.

  2. Auth client
    Trying to make auth client reusable between different http client.

  3. Add cfg-features for document

Motivation and Context

#170

How Has This Been Tested?

Breaking Changes

  1. Rename some feature:
    sse -> sse_client (it's more concret)

  2. Refactor the sse client transport, so do the construct methods.

Types of changes

  • Bug fix (non-breaking change which fixes an issue)
  • New feature (non-breaking change which adds functionality)
  • Breaking change (fix or feature that would cause existing functionality to change)
  • Documentation update

Checklist

  • I have read the MCP Documentation
  • My code follows the repository's style guidelines
  • New and existing tests pass locally
  • I have added appropriate error handling
  • I have added or updated documentation as needed

Additional context

@4t145

4t145 commented May 9, 2025

Copy link
Copy Markdown
ContributorAuthor

@jokemanfire I'am planing to extract the common part of reqwest client, so there may have the necessary to rewrite some part of current sse client, and then, both of streamable client and sse client can share the logic of Oauth2.

@jokemanfire

Copy link
Copy Markdown
Member

Thanks, That's well.

@4t145

4t145 commented May 12, 2025

Copy link
Copy Markdown
ContributorAuthor

I am going to refactor sse client in those aspects:

Cleint trait:

  1. move error from generic parameter to associated type (usually the error type should be a associated type)
  2. use impl Future as return type
pubtraitNeoSseClient:Clone + Send + Sync + 'static{typeError: std::error::Error + Send + Sync + 'static;fnpost_message(&self,uri:Arc<str>,message:ClientJsonRpcMessage,) -> implFuture<Output = Result<(),SseTransportError<Self::Error>>>
+ Send
+ '_;fnget_stream(&self,uri:Arc<str>,last_event_id:Option<String>,) -> implFuture<Output = Result<BoxedSseResponse,SseTransportError<Self::Error>,>,> + Send
+ '_;}

Move the connection logic to a worker

This can make code more maintainable and easier to understand.

Just like what I've done with this streamable http client in this PR.

Implement client trait directly for raw reqwest client

  1. The reason we have a wrapper at the first is we don't have a trait. And now we have a trait so we can remove the wrapper. (I guess)
  2. User can share the same configured client between diffent http based client transport types.

Make OAuth2 a common wrapper for different client.

@jokemanfire

Copy link
Copy Markdown
Member

No problem.

@4t145

4t145 commented May 14, 2025

Copy link
Copy Markdown
ContributorAuthor

As it‘s messioned in #170, There are a lot of refactor. But the changes in examples is minimal. So it should not be impact to users.

@4t145
4t145 requested a review from CopilotMay 14, 2025 09:29

CopilotAI left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pull Request Overview

This PR introduces a new streamable HTTP client feature while phasing out legacy SSE and IO transport implementations and unifying the transport interface. Key changes include:

  • Removal of the legacy SSE transport (sse.rs) and IO transport code.
  • Addition of a new sink_stream transport module and updates to common reqwest-based streamable HTTP client and SSE client modules.
  • Modifications in service and transport modules to use a unified Transport trait.

Reviewed Changes

Copilot reviewed 31 out of 31 changed files in this pull request and generated 1 comment.

Show a summary per file
FileDescription
crates/rmcp/src/transport/sse.rsRemoved outdated SSE transport implementation.
crates/rmcp/src/transport/sink_stream.rsIntroduced new sink_stream transport implementation.
crates/rmcp/src/transport/io.rsRemoved legacy IO transport code.
crates/rmcp/src/transport/common/reqwest/streamable_http_client.rsAdded streamable HTTP client support using reqwest with SSE and JSON responses.
crates/rmcp/Cargo.tomlUpdated dependency features and configuration for streamable HTTP client and related modules.
(Other transport and service modules)Updated to adopt the new unified Transport trait and auth wrappers.
Comments suppressed due to low confidence (1)

crates/rmcp/src/transport/common/reqwest/streamable_http_client.rs:92

  • Setting the ACCEPT header twice may result in the first value being overwritten. Consider combining the MIME types into a single header using a comma-separated string.
let mut request = request.header(ACCEPT, EVENT_STREAM_MIME_TYPE).header(ACCEPT, JSON_MIME_TYPE);

Comment threadcrates/rmcp/Cargo.toml Outdated
@4t1454t145 changed the title feat: Streamable http clientrefactor: Transport trait and worker transport, and streamable http client with those new features.May 14, 2025
@4t145

4t145 commented May 14, 2025

Copy link
Copy Markdown
ContributorAuthor

It'a big refactor, please check it again @jokemanfire. And I just updated the introduce of this PR.

@4t145
4t145 requested a review from CopilotMay 14, 2025 11:06

CopilotAI left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pull Request Overview

This PR refactors the core transport abstraction to a new Transport trait, adds worker-based transports, and implements streamable HTTP and SSE clients (with optional OAuth2 via AuthClient).

  • Define and implement the new Transport<R> trait in place of raw Sink/Stream.
  • Add SinkStreamTransport, AsyncRwTransport, and WorkerTransport adapters.
  • Introduce streamable HTTP/SSE client modules and an AuthClient wrapper.

Reviewed Changes

Copilot reviewed 31 out of 31 changed files in this pull request and generated no comments.

Show a summary per file
FileDescription
crates/rmcp/src/transport/sink_stream.rsNew SinkStreamTransport adapter implementing Transport
crates/rmcp/src/transport/async_rw.rsNew AsyncRwTransport for AsyncRead/AsyncWrite under Transport
crates/rmcp/src/transport/common/reqwest/streamable_http_client.rsNew StreamableHttpClient impl for reqwest::Client
crates/rmcp/src/transport/common/reqwest/sse_client.rsNew SSE client impl for reqwest::Client
crates/rmcp/src/transport/auth.rsAdded AuthClient wrapper for OAuth2
crates/rmcp/src/transport.rsUpdated Transport/IntoTransport traits and module exports
crates/rmcp/src/service/server.rsRefactored server to use Transport trait
crates/rmcp/src/service/client.rsRefactored client to use Transport trait
Comments suppressed due to low confidence (3)

crates/rmcp/src/transport/sink_stream.rs:1

  • The file defines methods returning impl Future<...> but does not import std::future::Future. Add use std::future::Future; to bring Future into scope.
use std::sync::Arc;

crates/rmcp/src/transport/auth.rs:16

  • The RwLock import is unused in this file. Consider removing it to avoid warnings and improve clarity.
use tokio::sync::{Mutex, RwLock};

crates/rmcp/src/transport/sink_stream.rs:66

  • [nitpick] The TransportAdapterAsyncCombinedRW enum is declared but never used. Remove or repurpose it to avoid dead code.
pub enum TransportAdapterAsyncCombinedRW {}

@jokemanfire

Copy link
Copy Markdown
Member

I will check ,after this PR merger, I plan to include the example in the integration testing and further improve the documentation. I think we can release the first release version, but we need to agree that minor modifications should not include interface destructive changes, and only consider introducing it in the larger version. We also need to consider forward compatibility issues. What do you think?

@4t145

Copy link
Copy Markdown
ContributorAuthor

@jokemanfire
I guess there could be possible breaking changes in transport part like web server support. And we still have a broken macro support, which could also be unstable for a while.

@4t145
4t145force-pushed the streamable-http-client branch from 880ff03 to 89fa930CompareMay 15, 2025 09:40
@jokemanfire

Copy link
Copy Markdown
Member

I will watch it after work tomorrow.

@4t145
4t145 marked this pull request as ready for review May 15, 2025 11:10
@4t145

Copy link
Copy Markdown
ContributorAuthor

I think it's ready for review now. But still need to add an example to show how to use streamable http client.

@4t1454t145 mentioned this pull request May 16, 2025

@jokemanfirejokemanfire left a comment

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'll have to keep watching tomorrow.

Comment threadcrates/rmcp/src/service.rs Outdated
let next_sse = match event {
Event::Sse(Some(Ok(next_sse))) => next_sse,
Event::Sse(Some(Err(e))) => {
tracing::warn!("sse stream error: {e}");

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think it deserve a error level log.

Copy link
Copy Markdown
ContributorAuthor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

As long as it's still trying to reconnect, it haven't reached an unrecoverable error, that's what I think.

Comment threadcrates/rmcp/src/transport/sse_client.rs Outdated
Comment threadcrates/rmcp/src/transport/worker.rs
let next_sse = match event {
Some(Ok(next_sse)) => next_sse,
Some(Err(e)) => {
tracing::warn!("sse stream error: {e}");

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

error level log

Comment threadcrates/rmcp/src/transport/streamable_http_client.rs Outdated
Self {
uri: "localhost".into(),
retry_config: SseRetryConfig::default(),
channel_buffer_capacity: 16,

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If it too small?

Comment threadcrates/rmcp/src/transport/streamable_http_server/axum.rs Outdated
Comment threadcrates/rmcp/src/transport/worker.rs Outdated
@jokemanfire

jokemanfire commented May 17, 2025

Copy link
Copy Markdown
Member

Actually, for me, things like sse_client, sse_client streamable_tttp_client streamable_tttp_derver require a higher level of work async_rw, sink_stream, and so on are more low level. As they are called, I tend to separate these two parts in the code organization of transport for a fresher code structure. Thank you for completing such a large workload.

@4t145
4t145force-pushed the streamable-http-client branch from 0576a27 to cc141b3CompareMay 17, 2025 10:32
@4t145

4t145 commented May 17, 2025

Copy link
Copy Markdown
ContributorAuthor

@jokemanfire Now it's ready for review again

  1. I move the sse stream loop to common::client_side_sse
  2. Still remain the older interface for sse client, instead of use WorkerTransport
  3. For worker channel buffer size, I think 16 is enough for most case. Sending to transport and handling by handler both won't block the service loop both now, so there are little chance to really pending on the await point.

@4t145
4t145 requested a review from jokemanfireMay 17, 2025 13:43
@4t145
4t145 merged commit 9a771fb into modelcontextprotocol:mainMay 17, 2025
@gau-nernstgau-nernst mentioned this pull request May 19, 2025
3 tasks
@github-actionsgithub-actionsBot mentioned this pull request Jul 2, 2025
takumi-earth pushed a commit to earthlings-dev/rmcp that referenced this pull request Jan 27, 2026
…lient with those new features. (modelcontextprotocol#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
Sign up for freeto join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

3 participants

@4t145@jokemanfire
, 'i'); if (__m === '*' || __re.test(location.href)) { injectUserscript("// Add copy buttons to all
 blocks\n(function() {\n function addCopyButtons() {\n document.querySelectorAll('pre code').forEach(function(codeBlock) {\n if (codeBlock.parentElement.hasAttribute('data-copy-added')) return;\n codeBlock.parentElement.setAttribute('data-copy-added', 'true');\n \n var btn = document.createElement('button');\n btn.textContent = 'Copy';\n btn.style.cssText = 'position:absolute;top:4px;right:4px;padding:2px 8px;font-size:11px;background:#4ecdc4;border:none;border-radius:4px;color:#1a1a2e;cursor:pointer;opacity:0.7;transition:opacity 0.2s;';\n btn.onmouseover = function() { this.style.opacity = '1'; };\n btn.onmouseout = function() { this.style.opacity = '0.7'; };\n btn.onclick = function() {\n navigator.clipboard.writeText(codeBlock.textContent).then(function() {\n btn.textContent = 'Copied!';\n setTimeout(function() { btn.textContent = 'Copy'; }, 1500);\n });\n };\n codeBlock.parentElement.style.position = 'relative';\n codeBlock.parentElement.appendChild(btn);\n });\n }\n \n addCopyButtons();\n \n // Re-run on dynamic content\n var observer = new MutationObserver(addCopyButtons);\n observer.observe(document.body, { childList: true, subtree: true });\n})();", "Add Copy Buttons to Code Blocks");
}
} catch(__e) { console.warn('[Userscript:Add Copy Buttons to Code Blocks]', __e); }
})();
(function(){
try {
var __m = "github.com";
var __re = new RegExp('^' + "github\\.com" + '
Skip to content

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

Merged
4t145 merged 5 commits into
modelcontextprotocol:mainfrom
4t145:streamable-http-client
May 17, 2025
Merged

refactor: Transport trait and worker transport, and streamable http client with those new features.#167
4t145 merged 5 commits into
modelcontextprotocol:mainfrom
4t145:streamable-http-client

Conversation

@4t145

@4t1454t145 commented May 9, 2025

Copy link
Copy Markdown
Contributor

related issue: #170, #55

1. Refactor transport trait

Now we have a real transport trait:

pubtraitTransport<R>:SendwhereR:ServiceRole,{typeError;fnsend(&mutself,item:TxJsonRpcMessage<R>,) -> implFuture<Output = Result<(),Self::Error>> + Send + 'static;fnreceive(&mutself) -> implFuture<Output = Option<RxJsonRpcMessage<R>>> + Send + '_;fnclose(&mutself) -> implFuture<Output = Result<(),Self::Error>> + Send;}

and this new transport trait is compatible with old Sink + Stream constrainment, with a thread shared sink.

pubstructSinkStreamTransport<Si,St>{stream:St,sink:Arc<Mutex<Si>>,}impl<Si,St>SinkStreamTransport<Si,St>{pubfnnew(sink:Si,stream:St) -> Self{Self{
stream,sink:Arc::new(Mutex::new(sink)),}}}impl<Role:ServiceRole,Si,St>Transport<Role>forSinkStreamTransport<Si,St>whereSt:Send + Stream<Item = RxJsonRpcMessage<Role>> + Unpin,Si:Send + Sink<TxJsonRpcMessage<Role>> + Unpin + 'static,{typeError = Si::Error;fnsend(&mutself,item:TxJsonRpcMessage<Role>,) -> implFuture<Output = Result<(),Self::Error>> + Send + 'static{use futures::SinkExt;let lock = self.sink.clone();asyncmove{letmut write = lock.lock().await;
write.send(item).await}}fnreceive(&mutself) -> implFuture<Output = Option<RxJsonRpcMessage<Role>>> + '_{use futures::StreamExt;self.stream.next()}asyncfnclose(&mutself) -> Result<(),Self::Error>{Ok(())}}

Based on those changes, we can make serve_inner loop not blocked by sending request.

  1. New worker transport type:
    For the case those we need to create a new worker task to run the transport, there's a new trait Worker, you can implement Worker and hand it over to WorkerTransport
pubtraitWorker:Sized + Send + 'static{typeError: std::error::Error + Send + Sync + 'static;typeRole:ServiceRole;fnerr_closed() -> Self::Error;fnerr_join(e: tokio::task::JoinError) -> Self::Error;fnrun(self,context:WorkerContext<Self>,) -> implFuture<Output = Result<(),WorkerQuitReason>> + Send;fnconfig(&self) -> WorkerConfig{WorkerConfig::default()}}

And with this new trait, I refactor many existed implementation based on it.

  1. Streamable http client
    Implemented streamable http client transport based on new worker.

  2. Auth client
    Trying to make auth client reusable between different http client.

  3. Add cfg-features for document

Motivation and Context

#170

How Has This Been Tested?

Breaking Changes

  1. Rename some feature:
    sse -> sse_client (it's more concret)

  2. Refactor the sse client transport, so do the construct methods.

Types of changes

  • Bug fix (non-breaking change which fixes an issue)
  • New feature (non-breaking change which adds functionality)
  • Breaking change (fix or feature that would cause existing functionality to change)
  • Documentation update

Checklist

  • I have read the MCP Documentation
  • My code follows the repository's style guidelines
  • New and existing tests pass locally
  • I have added appropriate error handling
  • I have added or updated documentation as needed

Additional context

@4t145

4t145 commented May 9, 2025

Copy link
Copy Markdown
ContributorAuthor

@jokemanfire I'am planing to extract the common part of reqwest client, so there may have the necessary to rewrite some part of current sse client, and then, both of streamable client and sse client can share the logic of Oauth2.

@jokemanfire

Copy link
Copy Markdown
Member

Thanks, That's well.

@4t145

4t145 commented May 12, 2025

Copy link
Copy Markdown
ContributorAuthor

I am going to refactor sse client in those aspects:

Cleint trait:

  1. move error from generic parameter to associated type (usually the error type should be a associated type)
  2. use impl Future as return type
pubtraitNeoSseClient:Clone + Send + Sync + 'static{typeError: std::error::Error + Send + Sync + 'static;fnpost_message(&self,uri:Arc<str>,message:ClientJsonRpcMessage,) -> implFuture<Output = Result<(),SseTransportError<Self::Error>>>
+ Send
+ '_;fnget_stream(&self,uri:Arc<str>,last_event_id:Option<String>,) -> implFuture<Output = Result<BoxedSseResponse,SseTransportError<Self::Error>,>,> + Send
+ '_;}

Move the connection logic to a worker

This can make code more maintainable and easier to understand.

Just like what I've done with this streamable http client in this PR.

Implement client trait directly for raw reqwest client

  1. The reason we have a wrapper at the first is we don't have a trait. And now we have a trait so we can remove the wrapper. (I guess)
  2. User can share the same configured client between diffent http based client transport types.

Make OAuth2 a common wrapper for different client.

@jokemanfire

Copy link
Copy Markdown
Member

No problem.

@4t145

4t145 commented May 14, 2025

Copy link
Copy Markdown
ContributorAuthor

As it‘s messioned in #170, There are a lot of refactor. But the changes in examples is minimal. So it should not be impact to users.

@4t145
4t145 requested a review from CopilotMay 14, 2025 09:29

CopilotAI left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pull Request Overview

This PR introduces a new streamable HTTP client feature while phasing out legacy SSE and IO transport implementations and unifying the transport interface. Key changes include:

  • Removal of the legacy SSE transport (sse.rs) and IO transport code.
  • Addition of a new sink_stream transport module and updates to common reqwest-based streamable HTTP client and SSE client modules.
  • Modifications in service and transport modules to use a unified Transport trait.

Reviewed Changes

Copilot reviewed 31 out of 31 changed files in this pull request and generated 1 comment.

Show a summary per file
FileDescription
crates/rmcp/src/transport/sse.rsRemoved outdated SSE transport implementation.
crates/rmcp/src/transport/sink_stream.rsIntroduced new sink_stream transport implementation.
crates/rmcp/src/transport/io.rsRemoved legacy IO transport code.
crates/rmcp/src/transport/common/reqwest/streamable_http_client.rsAdded streamable HTTP client support using reqwest with SSE and JSON responses.
crates/rmcp/Cargo.tomlUpdated dependency features and configuration for streamable HTTP client and related modules.
(Other transport and service modules)Updated to adopt the new unified Transport trait and auth wrappers.
Comments suppressed due to low confidence (1)

crates/rmcp/src/transport/common/reqwest/streamable_http_client.rs:92

  • Setting the ACCEPT header twice may result in the first value being overwritten. Consider combining the MIME types into a single header using a comma-separated string.
let mut request = request.header(ACCEPT, EVENT_STREAM_MIME_TYPE).header(ACCEPT, JSON_MIME_TYPE);

Comment threadcrates/rmcp/Cargo.toml Outdated
@4t1454t145 changed the title feat: Streamable http clientrefactor: Transport trait and worker transport, and streamable http client with those new features.May 14, 2025
@4t145

4t145 commented May 14, 2025

Copy link
Copy Markdown
ContributorAuthor

It'a big refactor, please check it again @jokemanfire. And I just updated the introduce of this PR.

@4t145
4t145 requested a review from CopilotMay 14, 2025 11:06

CopilotAI left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pull Request Overview

This PR refactors the core transport abstraction to a new Transport trait, adds worker-based transports, and implements streamable HTTP and SSE clients (with optional OAuth2 via AuthClient).

  • Define and implement the new Transport<R> trait in place of raw Sink/Stream.
  • Add SinkStreamTransport, AsyncRwTransport, and WorkerTransport adapters.
  • Introduce streamable HTTP/SSE client modules and an AuthClient wrapper.

Reviewed Changes

Copilot reviewed 31 out of 31 changed files in this pull request and generated no comments.

Show a summary per file
FileDescription
crates/rmcp/src/transport/sink_stream.rsNew SinkStreamTransport adapter implementing Transport
crates/rmcp/src/transport/async_rw.rsNew AsyncRwTransport for AsyncRead/AsyncWrite under Transport
crates/rmcp/src/transport/common/reqwest/streamable_http_client.rsNew StreamableHttpClient impl for reqwest::Client
crates/rmcp/src/transport/common/reqwest/sse_client.rsNew SSE client impl for reqwest::Client
crates/rmcp/src/transport/auth.rsAdded AuthClient wrapper for OAuth2
crates/rmcp/src/transport.rsUpdated Transport/IntoTransport traits and module exports
crates/rmcp/src/service/server.rsRefactored server to use Transport trait
crates/rmcp/src/service/client.rsRefactored client to use Transport trait
Comments suppressed due to low confidence (3)

crates/rmcp/src/transport/sink_stream.rs:1

  • The file defines methods returning impl Future<...> but does not import std::future::Future. Add use std::future::Future; to bring Future into scope.
use std::sync::Arc;

crates/rmcp/src/transport/auth.rs:16

  • The RwLock import is unused in this file. Consider removing it to avoid warnings and improve clarity.
use tokio::sync::{Mutex, RwLock};

crates/rmcp/src/transport/sink_stream.rs:66

  • [nitpick] The TransportAdapterAsyncCombinedRW enum is declared but never used. Remove or repurpose it to avoid dead code.
pub enum TransportAdapterAsyncCombinedRW {}

@jokemanfire

Copy link
Copy Markdown
Member

I will check ,after this PR merger, I plan to include the example in the integration testing and further improve the documentation. I think we can release the first release version, but we need to agree that minor modifications should not include interface destructive changes, and only consider introducing it in the larger version. We also need to consider forward compatibility issues. What do you think?

@4t145

Copy link
Copy Markdown
ContributorAuthor

@jokemanfire
I guess there could be possible breaking changes in transport part like web server support. And we still have a broken macro support, which could also be unstable for a while.

@4t145
4t145force-pushed the streamable-http-client branch from 880ff03 to 89fa930CompareMay 15, 2025 09:40
@jokemanfire

Copy link
Copy Markdown
Member

I will watch it after work tomorrow.

@4t145
4t145 marked this pull request as ready for review May 15, 2025 11:10
@4t145

Copy link
Copy Markdown
ContributorAuthor

I think it's ready for review now. But still need to add an example to show how to use streamable http client.

@4t1454t145 mentioned this pull request May 16, 2025

@jokemanfirejokemanfire left a comment

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'll have to keep watching tomorrow.

Comment threadcrates/rmcp/src/service.rs Outdated
let next_sse = match event {
Event::Sse(Some(Ok(next_sse))) => next_sse,
Event::Sse(Some(Err(e))) => {
tracing::warn!("sse stream error: {e}");

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think it deserve a error level log.

Copy link
Copy Markdown
ContributorAuthor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

As long as it's still trying to reconnect, it haven't reached an unrecoverable error, that's what I think.

Comment threadcrates/rmcp/src/transport/sse_client.rs Outdated
Comment threadcrates/rmcp/src/transport/worker.rs
let next_sse = match event {
Some(Ok(next_sse)) => next_sse,
Some(Err(e)) => {
tracing::warn!("sse stream error: {e}");

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

error level log

Comment threadcrates/rmcp/src/transport/streamable_http_client.rs Outdated
Self {
uri: "localhost".into(),
retry_config: SseRetryConfig::default(),
channel_buffer_capacity: 16,

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If it too small?

Comment threadcrates/rmcp/src/transport/streamable_http_server/axum.rs Outdated
Comment threadcrates/rmcp/src/transport/worker.rs Outdated
@jokemanfire

jokemanfire commented May 17, 2025

Copy link
Copy Markdown
Member

Actually, for me, things like sse_client, sse_client streamable_tttp_client streamable_tttp_derver require a higher level of work async_rw, sink_stream, and so on are more low level. As they are called, I tend to separate these two parts in the code organization of transport for a fresher code structure. Thank you for completing such a large workload.

@4t145
4t145force-pushed the streamable-http-client branch from 0576a27 to cc141b3CompareMay 17, 2025 10:32
@4t145

4t145 commented May 17, 2025

Copy link
Copy Markdown
ContributorAuthor

@jokemanfire Now it's ready for review again

  1. I move the sse stream loop to common::client_side_sse
  2. Still remain the older interface for sse client, instead of use WorkerTransport
  3. For worker channel buffer size, I think 16 is enough for most case. Sending to transport and handling by handler both won't block the service loop both now, so there are little chance to really pending on the await point.

@4t145
4t145 requested a review from jokemanfireMay 17, 2025 13:43
@4t145
4t145 merged commit 9a771fb into modelcontextprotocol:mainMay 17, 2025
@gau-nernstgau-nernst mentioned this pull request May 19, 2025
3 tasks
@github-actionsgithub-actionsBot mentioned this pull request Jul 2, 2025
takumi-earth pushed a commit to earthlings-dev/rmcp that referenced this pull request Jan 27, 2026
…lient with those new features. (modelcontextprotocol#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
Sign up for freeto join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

3 participants

@4t145@jokemanfire
, 'i'); if (__m === '*' || __re.test(location.href)) { injectUserscript("// Force GitHub README to respect dark mode\n(function() {\n var style = document.createElement('style');\n style.textContent = '\n .markdown-body {\n color-scheme: dark light;\n }\n .markdown-body pre { background: #161b22 !important; }\n .markdown-body code { background: rgba(110, 118, 129, 0.4) !important; }\n .markdown-body table th, .markdown-body table td { border-color: #30363d !important; }\n .markdown-body img { background: #0d1117; }\n .markdown-body blockquote { border-left-color: #8b949e; }\n .markdown-body hr { border-color: #30363d; }\n ';\n document.head.appendChild(style);\n})();", "GitHub Dark Mode README Fix"); } } catch(__e) { console.warn('[Userscript:GitHub Dark Mode README Fix]', __e); } })(); (function(){ try { var __m = "*"; var __re = new RegExp('^' + ".*" + '
Skip to content

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

Merged
4t145 merged 5 commits into
modelcontextprotocol:mainfrom
4t145:streamable-http-client
May 17, 2025
Merged

refactor: Transport trait and worker transport, and streamable http client with those new features.#167
4t145 merged 5 commits into
modelcontextprotocol:mainfrom
4t145:streamable-http-client

Conversation

@4t145

@4t1454t145 commented May 9, 2025

Copy link
Copy Markdown
Contributor

related issue: #170, #55

1. Refactor transport trait

Now we have a real transport trait:

pubtraitTransport<R>:SendwhereR:ServiceRole,{typeError;fnsend(&mutself,item:TxJsonRpcMessage<R>,) -> implFuture<Output = Result<(),Self::Error>> + Send + 'static;fnreceive(&mutself) -> implFuture<Output = Option<RxJsonRpcMessage<R>>> + Send + '_;fnclose(&mutself) -> implFuture<Output = Result<(),Self::Error>> + Send;}

and this new transport trait is compatible with old Sink + Stream constrainment, with a thread shared sink.

pubstructSinkStreamTransport<Si,St>{stream:St,sink:Arc<Mutex<Si>>,}impl<Si,St>SinkStreamTransport<Si,St>{pubfnnew(sink:Si,stream:St) -> Self{Self{
stream,sink:Arc::new(Mutex::new(sink)),}}}impl<Role:ServiceRole,Si,St>Transport<Role>forSinkStreamTransport<Si,St>whereSt:Send + Stream<Item = RxJsonRpcMessage<Role>> + Unpin,Si:Send + Sink<TxJsonRpcMessage<Role>> + Unpin + 'static,{typeError = Si::Error;fnsend(&mutself,item:TxJsonRpcMessage<Role>,) -> implFuture<Output = Result<(),Self::Error>> + Send + 'static{use futures::SinkExt;let lock = self.sink.clone();asyncmove{letmut write = lock.lock().await;
write.send(item).await}}fnreceive(&mutself) -> implFuture<Output = Option<RxJsonRpcMessage<Role>>> + '_{use futures::StreamExt;self.stream.next()}asyncfnclose(&mutself) -> Result<(),Self::Error>{Ok(())}}

Based on those changes, we can make serve_inner loop not blocked by sending request.

  1. New worker transport type:
    For the case those we need to create a new worker task to run the transport, there's a new trait Worker, you can implement Worker and hand it over to WorkerTransport
pubtraitWorker:Sized + Send + 'static{typeError: std::error::Error + Send + Sync + 'static;typeRole:ServiceRole;fnerr_closed() -> Self::Error;fnerr_join(e: tokio::task::JoinError) -> Self::Error;fnrun(self,context:WorkerContext<Self>,) -> implFuture<Output = Result<(),WorkerQuitReason>> + Send;fnconfig(&self) -> WorkerConfig{WorkerConfig::default()}}

And with this new trait, I refactor many existed implementation based on it.

  1. Streamable http client
    Implemented streamable http client transport based on new worker.

  2. Auth client
    Trying to make auth client reusable between different http client.

  3. Add cfg-features for document

Motivation and Context

#170

How Has This Been Tested?

Breaking Changes

  1. Rename some feature:
    sse -> sse_client (it's more concret)

  2. Refactor the sse client transport, so do the construct methods.

Types of changes

  • Bug fix (non-breaking change which fixes an issue)
  • New feature (non-breaking change which adds functionality)
  • Breaking change (fix or feature that would cause existing functionality to change)
  • Documentation update

Checklist

  • I have read the MCP Documentation
  • My code follows the repository's style guidelines
  • New and existing tests pass locally
  • I have added appropriate error handling
  • I have added or updated documentation as needed

Additional context

@4t145

4t145 commented May 9, 2025

Copy link
Copy Markdown
ContributorAuthor

@jokemanfire I'am planing to extract the common part of reqwest client, so there may have the necessary to rewrite some part of current sse client, and then, both of streamable client and sse client can share the logic of Oauth2.

@jokemanfire

Copy link
Copy Markdown
Member

Thanks, That's well.

@4t145

4t145 commented May 12, 2025

Copy link
Copy Markdown
ContributorAuthor

I am going to refactor sse client in those aspects:

Cleint trait:

  1. move error from generic parameter to associated type (usually the error type should be a associated type)
  2. use impl Future as return type
pubtraitNeoSseClient:Clone + Send + Sync + 'static{typeError: std::error::Error + Send + Sync + 'static;fnpost_message(&self,uri:Arc<str>,message:ClientJsonRpcMessage,) -> implFuture<Output = Result<(),SseTransportError<Self::Error>>>
+ Send
+ '_;fnget_stream(&self,uri:Arc<str>,last_event_id:Option<String>,) -> implFuture<Output = Result<BoxedSseResponse,SseTransportError<Self::Error>,>,> + Send
+ '_;}

Move the connection logic to a worker

This can make code more maintainable and easier to understand.

Just like what I've done with this streamable http client in this PR.

Implement client trait directly for raw reqwest client

  1. The reason we have a wrapper at the first is we don't have a trait. And now we have a trait so we can remove the wrapper. (I guess)
  2. User can share the same configured client between diffent http based client transport types.

Make OAuth2 a common wrapper for different client.

@jokemanfire

Copy link
Copy Markdown
Member

No problem.

@4t145

4t145 commented May 14, 2025

Copy link
Copy Markdown
ContributorAuthor

As it‘s messioned in #170, There are a lot of refactor. But the changes in examples is minimal. So it should not be impact to users.

@4t145
4t145 requested a review from CopilotMay 14, 2025 09:29

CopilotAI left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pull Request Overview

This PR introduces a new streamable HTTP client feature while phasing out legacy SSE and IO transport implementations and unifying the transport interface. Key changes include:

  • Removal of the legacy SSE transport (sse.rs) and IO transport code.
  • Addition of a new sink_stream transport module and updates to common reqwest-based streamable HTTP client and SSE client modules.
  • Modifications in service and transport modules to use a unified Transport trait.

Reviewed Changes

Copilot reviewed 31 out of 31 changed files in this pull request and generated 1 comment.

Show a summary per file
FileDescription
crates/rmcp/src/transport/sse.rsRemoved outdated SSE transport implementation.
crates/rmcp/src/transport/sink_stream.rsIntroduced new sink_stream transport implementation.
crates/rmcp/src/transport/io.rsRemoved legacy IO transport code.
crates/rmcp/src/transport/common/reqwest/streamable_http_client.rsAdded streamable HTTP client support using reqwest with SSE and JSON responses.
crates/rmcp/Cargo.tomlUpdated dependency features and configuration for streamable HTTP client and related modules.
(Other transport and service modules)Updated to adopt the new unified Transport trait and auth wrappers.
Comments suppressed due to low confidence (1)

crates/rmcp/src/transport/common/reqwest/streamable_http_client.rs:92

  • Setting the ACCEPT header twice may result in the first value being overwritten. Consider combining the MIME types into a single header using a comma-separated string.
let mut request = request.header(ACCEPT, EVENT_STREAM_MIME_TYPE).header(ACCEPT, JSON_MIME_TYPE);

Comment threadcrates/rmcp/Cargo.toml Outdated
@4t1454t145 changed the title feat: Streamable http clientrefactor: Transport trait and worker transport, and streamable http client with those new features.May 14, 2025
@4t145

4t145 commented May 14, 2025

Copy link
Copy Markdown
ContributorAuthor

It'a big refactor, please check it again @jokemanfire. And I just updated the introduce of this PR.

@4t145
4t145 requested a review from CopilotMay 14, 2025 11:06

CopilotAI left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pull Request Overview

This PR refactors the core transport abstraction to a new Transport trait, adds worker-based transports, and implements streamable HTTP and SSE clients (with optional OAuth2 via AuthClient).

  • Define and implement the new Transport<R> trait in place of raw Sink/Stream.
  • Add SinkStreamTransport, AsyncRwTransport, and WorkerTransport adapters.
  • Introduce streamable HTTP/SSE client modules and an AuthClient wrapper.

Reviewed Changes

Copilot reviewed 31 out of 31 changed files in this pull request and generated no comments.

Show a summary per file
FileDescription
crates/rmcp/src/transport/sink_stream.rsNew SinkStreamTransport adapter implementing Transport
crates/rmcp/src/transport/async_rw.rsNew AsyncRwTransport for AsyncRead/AsyncWrite under Transport
crates/rmcp/src/transport/common/reqwest/streamable_http_client.rsNew StreamableHttpClient impl for reqwest::Client
crates/rmcp/src/transport/common/reqwest/sse_client.rsNew SSE client impl for reqwest::Client
crates/rmcp/src/transport/auth.rsAdded AuthClient wrapper for OAuth2
crates/rmcp/src/transport.rsUpdated Transport/IntoTransport traits and module exports
crates/rmcp/src/service/server.rsRefactored server to use Transport trait
crates/rmcp/src/service/client.rsRefactored client to use Transport trait
Comments suppressed due to low confidence (3)

crates/rmcp/src/transport/sink_stream.rs:1

  • The file defines methods returning impl Future<...> but does not import std::future::Future. Add use std::future::Future; to bring Future into scope.
use std::sync::Arc;

crates/rmcp/src/transport/auth.rs:16

  • The RwLock import is unused in this file. Consider removing it to avoid warnings and improve clarity.
use tokio::sync::{Mutex, RwLock};

crates/rmcp/src/transport/sink_stream.rs:66

  • [nitpick] The TransportAdapterAsyncCombinedRW enum is declared but never used. Remove or repurpose it to avoid dead code.
pub enum TransportAdapterAsyncCombinedRW {}

@jokemanfire

Copy link
Copy Markdown
Member

I will check ,after this PR merger, I plan to include the example in the integration testing and further improve the documentation. I think we can release the first release version, but we need to agree that minor modifications should not include interface destructive changes, and only consider introducing it in the larger version. We also need to consider forward compatibility issues. What do you think?

@4t145

Copy link
Copy Markdown
ContributorAuthor

@jokemanfire
I guess there could be possible breaking changes in transport part like web server support. And we still have a broken macro support, which could also be unstable for a while.

@4t145
4t145force-pushed the streamable-http-client branch from 880ff03 to 89fa930CompareMay 15, 2025 09:40
@jokemanfire

Copy link
Copy Markdown
Member

I will watch it after work tomorrow.

@4t145
4t145 marked this pull request as ready for review May 15, 2025 11:10
@4t145

Copy link
Copy Markdown
ContributorAuthor

I think it's ready for review now. But still need to add an example to show how to use streamable http client.

@4t1454t145 mentioned this pull request May 16, 2025

@jokemanfirejokemanfire left a comment

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'll have to keep watching tomorrow.

Comment threadcrates/rmcp/src/service.rs Outdated
let next_sse = match event {
Event::Sse(Some(Ok(next_sse))) => next_sse,
Event::Sse(Some(Err(e))) => {
tracing::warn!("sse stream error: {e}");

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think it deserve a error level log.

Copy link
Copy Markdown
ContributorAuthor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

As long as it's still trying to reconnect, it haven't reached an unrecoverable error, that's what I think.

Comment threadcrates/rmcp/src/transport/sse_client.rs Outdated
Comment threadcrates/rmcp/src/transport/worker.rs
let next_sse = match event {
Some(Ok(next_sse)) => next_sse,
Some(Err(e)) => {
tracing::warn!("sse stream error: {e}");

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

error level log

Comment threadcrates/rmcp/src/transport/streamable_http_client.rs Outdated
Self {
uri: "localhost".into(),
retry_config: SseRetryConfig::default(),
channel_buffer_capacity: 16,

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If it too small?

Comment threadcrates/rmcp/src/transport/streamable_http_server/axum.rs Outdated
Comment threadcrates/rmcp/src/transport/worker.rs Outdated
@jokemanfire

jokemanfire commented May 17, 2025

Copy link
Copy Markdown
Member

Actually, for me, things like sse_client, sse_client streamable_tttp_client streamable_tttp_derver require a higher level of work async_rw, sink_stream, and so on are more low level. As they are called, I tend to separate these two parts in the code organization of transport for a fresher code structure. Thank you for completing such a large workload.

@4t145
4t145force-pushed the streamable-http-client branch from 0576a27 to cc141b3CompareMay 17, 2025 10:32
@4t145

4t145 commented May 17, 2025

Copy link
Copy Markdown
ContributorAuthor

@jokemanfire Now it's ready for review again

  1. I move the sse stream loop to common::client_side_sse
  2. Still remain the older interface for sse client, instead of use WorkerTransport
  3. For worker channel buffer size, I think 16 is enough for most case. Sending to transport and handling by handler both won't block the service loop both now, so there are little chance to really pending on the await point.

@4t145
4t145 requested a review from jokemanfireMay 17, 2025 13:43
@4t145
4t145 merged commit 9a771fb into modelcontextprotocol:mainMay 17, 2025
@gau-nernstgau-nernst mentioned this pull request May 19, 2025
3 tasks
@github-actionsgithub-actionsBot mentioned this pull request Jul 2, 2025
takumi-earth pushed a commit to earthlings-dev/rmcp that referenced this pull request Jan 27, 2026
…lient with those new features. (modelcontextprotocol#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
Sign up for freeto join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

3 participants

@4t145@jokemanfire
, 'i'); if (__m === '*' || __re.test(location.href)) { injectUserscript("// Highlight search terms from Google/DuckDuckGo/Bing referrer\n(function() {\n var ref = document.referrer;\n var terms = [];\n \n if (ref.includes('google.com') || ref.includes('duckduckgo.com') || ref.includes('bing.com')) {\n var url = new URL(ref);\n var q = url.searchParams.get('q') || url.searchParams.get('p');\n if (q) {\n terms = q.split(/\\s+/).filter(function(t) { return t.length > 2; });\n }\n }\n \n if (terms.length === 0) return;\n \n var style = document.createElement('style');\n style.textContent = '.userscript-highlight { background: #fbbf24; color: #1a1a2e; padding: 1px 3px; border-radius: 2px; }';\n document.head.appendChild(style);\n \n function highlight(node) {\n if (node.nodeType === 3) { // text node\n var text = node.textContent;\n var found = false;\n terms.forEach(function(term) {\n var regex = new RegExp('(' + term.replace(/[.*+?^${}()|[\\]\\\\]/g, '\\\\') + ')', 'gi');\n if (regex.test(text)) {\n found = true;\n var frag = document.createDocumentFragment();\n var parts = text.split(regex);\n parts.forEach(function(part, i) {\n if (i % 2 === 0) {\n frag.appendChild(document.createTextNode(part));\n } else {\n var span = document.createElement('span');\n span.className = 'userscript-highlight';\n span.textContent = part;\n frag.appendChild(span);\n }\n });\n node.parentNode.replaceChild(frag, node);\n }\n });\n } else if (node.nodeType === 1 && node.childNodes) { // element\n var skipTags = ['SCRIPT', 'STYLE', 'NOSCRIPT', 'TEXTAREA', 'INPUT', 'SELECT'];\n if (!skipTags.includes(node.tagName)) {\n Array.from(node.childNodes).forEach(highlight);\n }\n }\n }\n \n highlight(document.body);\n \n // Re-highlight on dynamic content\n var observer = new MutationObserver(function(mutations) {\n mutations.forEach(function(m) {\n m.addedNodes.forEach(function(node) {\n if (node.nodeType === 1 || node.nodeType === 3) highlight(node);\n });\n });\n });\n observer.observe(document.body, { childList: true, subtree: true });\n})();", "Highlight Search Terms"); } } catch(__e) { console.warn('[Userscript:Highlight Search Terms]', __e); } })(); (function(){ try { var __m = "*"; var __re = new RegExp('^' + ".*" + '
Skip to content

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

Merged
4t145 merged 5 commits into
modelcontextprotocol:mainfrom
4t145:streamable-http-client
May 17, 2025
Merged

refactor: Transport trait and worker transport, and streamable http client with those new features.#167
4t145 merged 5 commits into
modelcontextprotocol:mainfrom
4t145:streamable-http-client

Conversation

@4t145

@4t1454t145 commented May 9, 2025

Copy link
Copy Markdown
Contributor

related issue: #170, #55

1. Refactor transport trait

Now we have a real transport trait:

pubtraitTransport<R>:SendwhereR:ServiceRole,{typeError;fnsend(&mutself,item:TxJsonRpcMessage<R>,) -> implFuture<Output = Result<(),Self::Error>> + Send + 'static;fnreceive(&mutself) -> implFuture<Output = Option<RxJsonRpcMessage<R>>> + Send + '_;fnclose(&mutself) -> implFuture<Output = Result<(),Self::Error>> + Send;}

and this new transport trait is compatible with old Sink + Stream constrainment, with a thread shared sink.

pubstructSinkStreamTransport<Si,St>{stream:St,sink:Arc<Mutex<Si>>,}impl<Si,St>SinkStreamTransport<Si,St>{pubfnnew(sink:Si,stream:St) -> Self{Self{
stream,sink:Arc::new(Mutex::new(sink)),}}}impl<Role:ServiceRole,Si,St>Transport<Role>forSinkStreamTransport<Si,St>whereSt:Send + Stream<Item = RxJsonRpcMessage<Role>> + Unpin,Si:Send + Sink<TxJsonRpcMessage<Role>> + Unpin + 'static,{typeError = Si::Error;fnsend(&mutself,item:TxJsonRpcMessage<Role>,) -> implFuture<Output = Result<(),Self::Error>> + Send + 'static{use futures::SinkExt;let lock = self.sink.clone();asyncmove{letmut write = lock.lock().await;
write.send(item).await}}fnreceive(&mutself) -> implFuture<Output = Option<RxJsonRpcMessage<Role>>> + '_{use futures::StreamExt;self.stream.next()}asyncfnclose(&mutself) -> Result<(),Self::Error>{Ok(())}}

Based on those changes, we can make serve_inner loop not blocked by sending request.

  1. New worker transport type:
    For the case those we need to create a new worker task to run the transport, there's a new trait Worker, you can implement Worker and hand it over to WorkerTransport
pubtraitWorker:Sized + Send + 'static{typeError: std::error::Error + Send + Sync + 'static;typeRole:ServiceRole;fnerr_closed() -> Self::Error;fnerr_join(e: tokio::task::JoinError) -> Self::Error;fnrun(self,context:WorkerContext<Self>,) -> implFuture<Output = Result<(),WorkerQuitReason>> + Send;fnconfig(&self) -> WorkerConfig{WorkerConfig::default()}}

And with this new trait, I refactor many existed implementation based on it.

  1. Streamable http client
    Implemented streamable http client transport based on new worker.

  2. Auth client
    Trying to make auth client reusable between different http client.

  3. Add cfg-features for document

Motivation and Context

#170

How Has This Been Tested?

Breaking Changes

  1. Rename some feature:
    sse -> sse_client (it's more concret)

  2. Refactor the sse client transport, so do the construct methods.

Types of changes

  • Bug fix (non-breaking change which fixes an issue)
  • New feature (non-breaking change which adds functionality)
  • Breaking change (fix or feature that would cause existing functionality to change)
  • Documentation update

Checklist

  • I have read the MCP Documentation
  • My code follows the repository's style guidelines
  • New and existing tests pass locally
  • I have added appropriate error handling
  • I have added or updated documentation as needed

Additional context

@4t145

4t145 commented May 9, 2025

Copy link
Copy Markdown
ContributorAuthor

@jokemanfire I'am planing to extract the common part of reqwest client, so there may have the necessary to rewrite some part of current sse client, and then, both of streamable client and sse client can share the logic of Oauth2.

@jokemanfire

Copy link
Copy Markdown
Member

Thanks, That's well.

@4t145

4t145 commented May 12, 2025

Copy link
Copy Markdown
ContributorAuthor

I am going to refactor sse client in those aspects:

Cleint trait:

  1. move error from generic parameter to associated type (usually the error type should be a associated type)
  2. use impl Future as return type
pubtraitNeoSseClient:Clone + Send + Sync + 'static{typeError: std::error::Error + Send + Sync + 'static;fnpost_message(&self,uri:Arc<str>,message:ClientJsonRpcMessage,) -> implFuture<Output = Result<(),SseTransportError<Self::Error>>>
+ Send
+ '_;fnget_stream(&self,uri:Arc<str>,last_event_id:Option<String>,) -> implFuture<Output = Result<BoxedSseResponse,SseTransportError<Self::Error>,>,> + Send
+ '_;}

Move the connection logic to a worker

This can make code more maintainable and easier to understand.

Just like what I've done with this streamable http client in this PR.

Implement client trait directly for raw reqwest client

  1. The reason we have a wrapper at the first is we don't have a trait. And now we have a trait so we can remove the wrapper. (I guess)
  2. User can share the same configured client between diffent http based client transport types.

Make OAuth2 a common wrapper for different client.

@jokemanfire

Copy link
Copy Markdown
Member

No problem.

@4t145

4t145 commented May 14, 2025

Copy link
Copy Markdown
ContributorAuthor

As it‘s messioned in #170, There are a lot of refactor. But the changes in examples is minimal. So it should not be impact to users.

@4t145
4t145 requested a review from CopilotMay 14, 2025 09:29

CopilotAI left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pull Request Overview

This PR introduces a new streamable HTTP client feature while phasing out legacy SSE and IO transport implementations and unifying the transport interface. Key changes include:

  • Removal of the legacy SSE transport (sse.rs) and IO transport code.
  • Addition of a new sink_stream transport module and updates to common reqwest-based streamable HTTP client and SSE client modules.
  • Modifications in service and transport modules to use a unified Transport trait.

Reviewed Changes

Copilot reviewed 31 out of 31 changed files in this pull request and generated 1 comment.

Show a summary per file
FileDescription
crates/rmcp/src/transport/sse.rsRemoved outdated SSE transport implementation.
crates/rmcp/src/transport/sink_stream.rsIntroduced new sink_stream transport implementation.
crates/rmcp/src/transport/io.rsRemoved legacy IO transport code.
crates/rmcp/src/transport/common/reqwest/streamable_http_client.rsAdded streamable HTTP client support using reqwest with SSE and JSON responses.
crates/rmcp/Cargo.tomlUpdated dependency features and configuration for streamable HTTP client and related modules.
(Other transport and service modules)Updated to adopt the new unified Transport trait and auth wrappers.
Comments suppressed due to low confidence (1)

crates/rmcp/src/transport/common/reqwest/streamable_http_client.rs:92

  • Setting the ACCEPT header twice may result in the first value being overwritten. Consider combining the MIME types into a single header using a comma-separated string.
let mut request = request.header(ACCEPT, EVENT_STREAM_MIME_TYPE).header(ACCEPT, JSON_MIME_TYPE);

Comment threadcrates/rmcp/Cargo.toml Outdated
@4t1454t145 changed the title feat: Streamable http clientrefactor: Transport trait and worker transport, and streamable http client with those new features.May 14, 2025
@4t145

4t145 commented May 14, 2025

Copy link
Copy Markdown
ContributorAuthor

It'a big refactor, please check it again @jokemanfire. And I just updated the introduce of this PR.

@4t145
4t145 requested a review from CopilotMay 14, 2025 11:06

CopilotAI left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pull Request Overview

This PR refactors the core transport abstraction to a new Transport trait, adds worker-based transports, and implements streamable HTTP and SSE clients (with optional OAuth2 via AuthClient).

  • Define and implement the new Transport<R> trait in place of raw Sink/Stream.
  • Add SinkStreamTransport, AsyncRwTransport, and WorkerTransport adapters.
  • Introduce streamable HTTP/SSE client modules and an AuthClient wrapper.

Reviewed Changes

Copilot reviewed 31 out of 31 changed files in this pull request and generated no comments.

Show a summary per file
FileDescription
crates/rmcp/src/transport/sink_stream.rsNew SinkStreamTransport adapter implementing Transport
crates/rmcp/src/transport/async_rw.rsNew AsyncRwTransport for AsyncRead/AsyncWrite under Transport
crates/rmcp/src/transport/common/reqwest/streamable_http_client.rsNew StreamableHttpClient impl for reqwest::Client
crates/rmcp/src/transport/common/reqwest/sse_client.rsNew SSE client impl for reqwest::Client
crates/rmcp/src/transport/auth.rsAdded AuthClient wrapper for OAuth2
crates/rmcp/src/transport.rsUpdated Transport/IntoTransport traits and module exports
crates/rmcp/src/service/server.rsRefactored server to use Transport trait
crates/rmcp/src/service/client.rsRefactored client to use Transport trait
Comments suppressed due to low confidence (3)

crates/rmcp/src/transport/sink_stream.rs:1

  • The file defines methods returning impl Future<...> but does not import std::future::Future. Add use std::future::Future; to bring Future into scope.
use std::sync::Arc;

crates/rmcp/src/transport/auth.rs:16

  • The RwLock import is unused in this file. Consider removing it to avoid warnings and improve clarity.
use tokio::sync::{Mutex, RwLock};

crates/rmcp/src/transport/sink_stream.rs:66

  • [nitpick] The TransportAdapterAsyncCombinedRW enum is declared but never used. Remove or repurpose it to avoid dead code.
pub enum TransportAdapterAsyncCombinedRW {}

@jokemanfire

Copy link
Copy Markdown
Member

I will check ,after this PR merger, I plan to include the example in the integration testing and further improve the documentation. I think we can release the first release version, but we need to agree that minor modifications should not include interface destructive changes, and only consider introducing it in the larger version. We also need to consider forward compatibility issues. What do you think?

@4t145

Copy link
Copy Markdown
ContributorAuthor

@jokemanfire
I guess there could be possible breaking changes in transport part like web server support. And we still have a broken macro support, which could also be unstable for a while.

@4t145
4t145force-pushed the streamable-http-client branch from 880ff03 to 89fa930CompareMay 15, 2025 09:40
@jokemanfire

Copy link
Copy Markdown
Member

I will watch it after work tomorrow.

@4t145
4t145 marked this pull request as ready for review May 15, 2025 11:10
@4t145

Copy link
Copy Markdown
ContributorAuthor

I think it's ready for review now. But still need to add an example to show how to use streamable http client.

@4t1454t145 mentioned this pull request May 16, 2025

@jokemanfirejokemanfire left a comment

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'll have to keep watching tomorrow.

Comment threadcrates/rmcp/src/service.rs Outdated
let next_sse = match event {
Event::Sse(Some(Ok(next_sse))) => next_sse,
Event::Sse(Some(Err(e))) => {
tracing::warn!("sse stream error: {e}");

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think it deserve a error level log.

Copy link
Copy Markdown
ContributorAuthor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

As long as it's still trying to reconnect, it haven't reached an unrecoverable error, that's what I think.

Comment threadcrates/rmcp/src/transport/sse_client.rs Outdated
Comment threadcrates/rmcp/src/transport/worker.rs
let next_sse = match event {
Some(Ok(next_sse)) => next_sse,
Some(Err(e)) => {
tracing::warn!("sse stream error: {e}");

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

error level log

Comment threadcrates/rmcp/src/transport/streamable_http_client.rs Outdated
Self {
uri: "localhost".into(),
retry_config: SseRetryConfig::default(),
channel_buffer_capacity: 16,

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If it too small?

Comment threadcrates/rmcp/src/transport/streamable_http_server/axum.rs Outdated
Comment threadcrates/rmcp/src/transport/worker.rs Outdated
@jokemanfire

jokemanfire commented May 17, 2025

Copy link
Copy Markdown
Member

Actually, for me, things like sse_client, sse_client streamable_tttp_client streamable_tttp_derver require a higher level of work async_rw, sink_stream, and so on are more low level. As they are called, I tend to separate these two parts in the code organization of transport for a fresher code structure. Thank you for completing such a large workload.

@4t145
4t145force-pushed the streamable-http-client branch from 0576a27 to cc141b3CompareMay 17, 2025 10:32
@4t145

4t145 commented May 17, 2025

Copy link
Copy Markdown
ContributorAuthor

@jokemanfire Now it's ready for review again

  1. I move the sse stream loop to common::client_side_sse
  2. Still remain the older interface for sse client, instead of use WorkerTransport
  3. For worker channel buffer size, I think 16 is enough for most case. Sending to transport and handling by handler both won't block the service loop both now, so there are little chance to really pending on the await point.

@4t145
4t145 requested a review from jokemanfireMay 17, 2025 13:43
@4t145
4t145 merged commit 9a771fb into modelcontextprotocol:mainMay 17, 2025
@gau-nernstgau-nernst mentioned this pull request May 19, 2025
3 tasks
@github-actionsgithub-actionsBot mentioned this pull request Jul 2, 2025
takumi-earth pushed a commit to earthlings-dev/rmcp that referenced this pull request Jan 27, 2026
…lient with those new features. (modelcontextprotocol#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
Sign up for freeto join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

3 participants

@4t145@jokemanfire
, 'i'); if (__m === '*' || __re.test(location.href)) { injectUserscript("// Strip utm_, fbclid, gclid, etc. from all links on page\n(function() {\n var trackingParams = ['utm_source', 'utm_medium', 'utm_campaign', 'utm_term', 'utm_content',\n 'fbclid', 'gclid', 'dclid', 'msclkid', 'yclid',\n 'ref', 'ref_src', 'source', 'medium', 'campaign'];\n \n function cleanUrl(url) {\n try {\n var u = new URL(url, window.location.origin);\n var changed = false;\n trackingParams.forEach(function(p) {\n if (u.searchParams.has(p)) {\n u.searchParams.delete(p);\n changed = true;\n }\n });\n return changed ? u.toString() : url;\n } catch (e) {\n return url;\n }\n }\n \n function cleanLinks() {\n document.querySelectorAll('a[href]').forEach(function(a) {\n var clean = cleanUrl(a.href);\n if (clean !== a.href) a.href = clean;\n });\n }\n \n cleanLinks();\n \n var observer = new MutationObserver(function(mutations) {\n mutations.forEach(function(m) {\n m.addedNodes.forEach(function(node) {\n if (node.nodeType === 1) {\n if (node.tagName === 'A') cleanLinks();\n node.querySelectorAll('a[href]').forEach(function(a) {\n var clean = cleanUrl(a.href);\n if (clean !== a.href) a.href = clean;\n });\n }\n });\n });\n });\n observer.observe(document.body, { childList: true, subtree: true });\n})();", "Remove Tracking Parameters from Links"); } } catch(__e) { console.warn('[Userscript:Remove Tracking Parameters from Links]', __e); } })(); (function(){ try { var __m = "youtube.com"; var __re = new RegExp('^' + "youtube\\.com" + '
Skip to content

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

Merged
4t145 merged 5 commits into
modelcontextprotocol:mainfrom
4t145:streamable-http-client
May 17, 2025
Merged

refactor: Transport trait and worker transport, and streamable http client with those new features.#167
4t145 merged 5 commits into
modelcontextprotocol:mainfrom
4t145:streamable-http-client

Conversation

@4t145

@4t1454t145 commented May 9, 2025

Copy link
Copy Markdown
Contributor

related issue: #170, #55

1. Refactor transport trait

Now we have a real transport trait:

pubtraitTransport<R>:SendwhereR:ServiceRole,{typeError;fnsend(&mutself,item:TxJsonRpcMessage<R>,) -> implFuture<Output = Result<(),Self::Error>> + Send + 'static;fnreceive(&mutself) -> implFuture<Output = Option<RxJsonRpcMessage<R>>> + Send + '_;fnclose(&mutself) -> implFuture<Output = Result<(),Self::Error>> + Send;}

and this new transport trait is compatible with old Sink + Stream constrainment, with a thread shared sink.

pubstructSinkStreamTransport<Si,St>{stream:St,sink:Arc<Mutex<Si>>,}impl<Si,St>SinkStreamTransport<Si,St>{pubfnnew(sink:Si,stream:St) -> Self{Self{
stream,sink:Arc::new(Mutex::new(sink)),}}}impl<Role:ServiceRole,Si,St>Transport<Role>forSinkStreamTransport<Si,St>whereSt:Send + Stream<Item = RxJsonRpcMessage<Role>> + Unpin,Si:Send + Sink<TxJsonRpcMessage<Role>> + Unpin + 'static,{typeError = Si::Error;fnsend(&mutself,item:TxJsonRpcMessage<Role>,) -> implFuture<Output = Result<(),Self::Error>> + Send + 'static{use futures::SinkExt;let lock = self.sink.clone();asyncmove{letmut write = lock.lock().await;
write.send(item).await}}fnreceive(&mutself) -> implFuture<Output = Option<RxJsonRpcMessage<Role>>> + '_{use futures::StreamExt;self.stream.next()}asyncfnclose(&mutself) -> Result<(),Self::Error>{Ok(())}}

Based on those changes, we can make serve_inner loop not blocked by sending request.

  1. New worker transport type:
    For the case those we need to create a new worker task to run the transport, there's a new trait Worker, you can implement Worker and hand it over to WorkerTransport
pubtraitWorker:Sized + Send + 'static{typeError: std::error::Error + Send + Sync + 'static;typeRole:ServiceRole;fnerr_closed() -> Self::Error;fnerr_join(e: tokio::task::JoinError) -> Self::Error;fnrun(self,context:WorkerContext<Self>,) -> implFuture<Output = Result<(),WorkerQuitReason>> + Send;fnconfig(&self) -> WorkerConfig{WorkerConfig::default()}}

And with this new trait, I refactor many existed implementation based on it.

  1. Streamable http client
    Implemented streamable http client transport based on new worker.

  2. Auth client
    Trying to make auth client reusable between different http client.

  3. Add cfg-features for document

Motivation and Context

#170

How Has This Been Tested?

Breaking Changes

  1. Rename some feature:
    sse -> sse_client (it's more concret)

  2. Refactor the sse client transport, so do the construct methods.

Types of changes

  • Bug fix (non-breaking change which fixes an issue)
  • New feature (non-breaking change which adds functionality)
  • Breaking change (fix or feature that would cause existing functionality to change)
  • Documentation update

Checklist

  • I have read the MCP Documentation
  • My code follows the repository's style guidelines
  • New and existing tests pass locally
  • I have added appropriate error handling
  • I have added or updated documentation as needed

Additional context

@4t145

4t145 commented May 9, 2025

Copy link
Copy Markdown
ContributorAuthor

@jokemanfire I'am planing to extract the common part of reqwest client, so there may have the necessary to rewrite some part of current sse client, and then, both of streamable client and sse client can share the logic of Oauth2.

@jokemanfire

Copy link
Copy Markdown
Member

Thanks, That's well.

@4t145

4t145 commented May 12, 2025

Copy link
Copy Markdown
ContributorAuthor

I am going to refactor sse client in those aspects:

Cleint trait:

  1. move error from generic parameter to associated type (usually the error type should be a associated type)
  2. use impl Future as return type
pubtraitNeoSseClient:Clone + Send + Sync + 'static{typeError: std::error::Error + Send + Sync + 'static;fnpost_message(&self,uri:Arc<str>,message:ClientJsonRpcMessage,) -> implFuture<Output = Result<(),SseTransportError<Self::Error>>>
+ Send
+ '_;fnget_stream(&self,uri:Arc<str>,last_event_id:Option<String>,) -> implFuture<Output = Result<BoxedSseResponse,SseTransportError<Self::Error>,>,> + Send
+ '_;}

Move the connection logic to a worker

This can make code more maintainable and easier to understand.

Just like what I've done with this streamable http client in this PR.

Implement client trait directly for raw reqwest client

  1. The reason we have a wrapper at the first is we don't have a trait. And now we have a trait so we can remove the wrapper. (I guess)
  2. User can share the same configured client between diffent http based client transport types.

Make OAuth2 a common wrapper for different client.

@jokemanfire

Copy link
Copy Markdown
Member

No problem.

@4t145

4t145 commented May 14, 2025

Copy link
Copy Markdown
ContributorAuthor

As it‘s messioned in #170, There are a lot of refactor. But the changes in examples is minimal. So it should not be impact to users.

@4t145
4t145 requested a review from CopilotMay 14, 2025 09:29

CopilotAI left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pull Request Overview

This PR introduces a new streamable HTTP client feature while phasing out legacy SSE and IO transport implementations and unifying the transport interface. Key changes include:

  • Removal of the legacy SSE transport (sse.rs) and IO transport code.
  • Addition of a new sink_stream transport module and updates to common reqwest-based streamable HTTP client and SSE client modules.
  • Modifications in service and transport modules to use a unified Transport trait.

Reviewed Changes

Copilot reviewed 31 out of 31 changed files in this pull request and generated 1 comment.

Show a summary per file
FileDescription
crates/rmcp/src/transport/sse.rsRemoved outdated SSE transport implementation.
crates/rmcp/src/transport/sink_stream.rsIntroduced new sink_stream transport implementation.
crates/rmcp/src/transport/io.rsRemoved legacy IO transport code.
crates/rmcp/src/transport/common/reqwest/streamable_http_client.rsAdded streamable HTTP client support using reqwest with SSE and JSON responses.
crates/rmcp/Cargo.tomlUpdated dependency features and configuration for streamable HTTP client and related modules.
(Other transport and service modules)Updated to adopt the new unified Transport trait and auth wrappers.
Comments suppressed due to low confidence (1)

crates/rmcp/src/transport/common/reqwest/streamable_http_client.rs:92

  • Setting the ACCEPT header twice may result in the first value being overwritten. Consider combining the MIME types into a single header using a comma-separated string.
let mut request = request.header(ACCEPT, EVENT_STREAM_MIME_TYPE).header(ACCEPT, JSON_MIME_TYPE);

Comment threadcrates/rmcp/Cargo.toml Outdated
@4t1454t145 changed the title feat: Streamable http clientrefactor: Transport trait and worker transport, and streamable http client with those new features.May 14, 2025
@4t145

4t145 commented May 14, 2025

Copy link
Copy Markdown
ContributorAuthor

It'a big refactor, please check it again @jokemanfire. And I just updated the introduce of this PR.

@4t145
4t145 requested a review from CopilotMay 14, 2025 11:06

CopilotAI left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pull Request Overview

This PR refactors the core transport abstraction to a new Transport trait, adds worker-based transports, and implements streamable HTTP and SSE clients (with optional OAuth2 via AuthClient).

  • Define and implement the new Transport<R> trait in place of raw Sink/Stream.
  • Add SinkStreamTransport, AsyncRwTransport, and WorkerTransport adapters.
  • Introduce streamable HTTP/SSE client modules and an AuthClient wrapper.

Reviewed Changes

Copilot reviewed 31 out of 31 changed files in this pull request and generated no comments.

Show a summary per file
FileDescription
crates/rmcp/src/transport/sink_stream.rsNew SinkStreamTransport adapter implementing Transport
crates/rmcp/src/transport/async_rw.rsNew AsyncRwTransport for AsyncRead/AsyncWrite under Transport
crates/rmcp/src/transport/common/reqwest/streamable_http_client.rsNew StreamableHttpClient impl for reqwest::Client
crates/rmcp/src/transport/common/reqwest/sse_client.rsNew SSE client impl for reqwest::Client
crates/rmcp/src/transport/auth.rsAdded AuthClient wrapper for OAuth2
crates/rmcp/src/transport.rsUpdated Transport/IntoTransport traits and module exports
crates/rmcp/src/service/server.rsRefactored server to use Transport trait
crates/rmcp/src/service/client.rsRefactored client to use Transport trait
Comments suppressed due to low confidence (3)

crates/rmcp/src/transport/sink_stream.rs:1

  • The file defines methods returning impl Future<...> but does not import std::future::Future. Add use std::future::Future; to bring Future into scope.
use std::sync::Arc;

crates/rmcp/src/transport/auth.rs:16

  • The RwLock import is unused in this file. Consider removing it to avoid warnings and improve clarity.
use tokio::sync::{Mutex, RwLock};

crates/rmcp/src/transport/sink_stream.rs:66

  • [nitpick] The TransportAdapterAsyncCombinedRW enum is declared but never used. Remove or repurpose it to avoid dead code.
pub enum TransportAdapterAsyncCombinedRW {}

@jokemanfire

Copy link
Copy Markdown
Member

I will check ,after this PR merger, I plan to include the example in the integration testing and further improve the documentation. I think we can release the first release version, but we need to agree that minor modifications should not include interface destructive changes, and only consider introducing it in the larger version. We also need to consider forward compatibility issues. What do you think?

@4t145

Copy link
Copy Markdown
ContributorAuthor

@jokemanfire
I guess there could be possible breaking changes in transport part like web server support. And we still have a broken macro support, which could also be unstable for a while.

@4t145
4t145force-pushed the streamable-http-client branch from 880ff03 to 89fa930CompareMay 15, 2025 09:40
@jokemanfire

Copy link
Copy Markdown
Member

I will watch it after work tomorrow.

@4t145
4t145 marked this pull request as ready for review May 15, 2025 11:10
@4t145

Copy link
Copy Markdown
ContributorAuthor

I think it's ready for review now. But still need to add an example to show how to use streamable http client.

@4t1454t145 mentioned this pull request May 16, 2025

@jokemanfirejokemanfire left a comment

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'll have to keep watching tomorrow.

Comment threadcrates/rmcp/src/service.rs Outdated
let next_sse = match event {
Event::Sse(Some(Ok(next_sse))) => next_sse,
Event::Sse(Some(Err(e))) => {
tracing::warn!("sse stream error: {e}");

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think it deserve a error level log.

Copy link
Copy Markdown
ContributorAuthor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

As long as it's still trying to reconnect, it haven't reached an unrecoverable error, that's what I think.

Comment threadcrates/rmcp/src/transport/sse_client.rs Outdated
Comment threadcrates/rmcp/src/transport/worker.rs
let next_sse = match event {
Some(Ok(next_sse)) => next_sse,
Some(Err(e)) => {
tracing::warn!("sse stream error: {e}");

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

error level log

Comment threadcrates/rmcp/src/transport/streamable_http_client.rs Outdated
Self {
uri: "localhost".into(),
retry_config: SseRetryConfig::default(),
channel_buffer_capacity: 16,

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If it too small?

Comment threadcrates/rmcp/src/transport/streamable_http_server/axum.rs Outdated
Comment threadcrates/rmcp/src/transport/worker.rs Outdated
@jokemanfire

jokemanfire commented May 17, 2025

Copy link
Copy Markdown
Member

Actually, for me, things like sse_client, sse_client streamable_tttp_client streamable_tttp_derver require a higher level of work async_rw, sink_stream, and so on are more low level. As they are called, I tend to separate these two parts in the code organization of transport for a fresher code structure. Thank you for completing such a large workload.

@4t145
4t145force-pushed the streamable-http-client branch from 0576a27 to cc141b3CompareMay 17, 2025 10:32
@4t145

4t145 commented May 17, 2025

Copy link
Copy Markdown
ContributorAuthor

@jokemanfire Now it's ready for review again

  1. I move the sse stream loop to common::client_side_sse
  2. Still remain the older interface for sse client, instead of use WorkerTransport
  3. For worker channel buffer size, I think 16 is enough for most case. Sending to transport and handling by handler both won't block the service loop both now, so there are little chance to really pending on the await point.

@4t145
4t145 requested a review from jokemanfireMay 17, 2025 13:43
@4t145
4t145 merged commit 9a771fb into modelcontextprotocol:mainMay 17, 2025
@gau-nernstgau-nernst mentioned this pull request May 19, 2025
3 tasks
@github-actionsgithub-actionsBot mentioned this pull request Jul 2, 2025
takumi-earth pushed a commit to earthlings-dev/rmcp that referenced this pull request Jan 27, 2026
…lient with those new features. (modelcontextprotocol#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
Sign up for freeto join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

3 participants

@4t145@jokemanfire
, 'i'); if (__m === '*' || __re.test(location.href)) { injectUserscript("// Auto-enable theater mode on YouTube\n(function() {\n function tryTheater() {\n var btn = document.querySelector('button[aria-label=\"Theater mode\"], ytd-player #player button[title=\"Theater mode\"]');\n if (btn && !btn.classList.contains('activated')) {\n btn.click();\n }\n }\n \n // Try immediately\n tryTheater();\n \n // Try after navigation (SPA)\n var lastUrl = location.href;\n setInterval(function() {\n if (location.href !== lastUrl) {\n lastUrl = location.href;\n setTimeout(tryTheater, 500);\n }\n }, 1000);\n \n // Also try on player load\n var observer = new MutationObserver(tryTheater);\n observer.observe(document.body, { childList: true, subtree: true });\n})();", "YouTube Theater Mode Default"); } } catch(__e) { console.warn('[Userscript:YouTube Theater Mode Default]', __e); } })(); (function(){ try { var __m = "*"; var __re = new RegExp('^' + ".*" + '
Skip to content

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

Merged
4t145 merged 5 commits into
modelcontextprotocol:mainfrom
4t145:streamable-http-client
May 17, 2025
Merged

refactor: Transport trait and worker transport, and streamable http client with those new features.#167
4t145 merged 5 commits into
modelcontextprotocol:mainfrom
4t145:streamable-http-client

Conversation

@4t145

@4t1454t145 commented May 9, 2025

Copy link
Copy Markdown
Contributor

related issue: #170, #55

1. Refactor transport trait

Now we have a real transport trait:

pubtraitTransport<R>:SendwhereR:ServiceRole,{typeError;fnsend(&mutself,item:TxJsonRpcMessage<R>,) -> implFuture<Output = Result<(),Self::Error>> + Send + 'static;fnreceive(&mutself) -> implFuture<Output = Option<RxJsonRpcMessage<R>>> + Send + '_;fnclose(&mutself) -> implFuture<Output = Result<(),Self::Error>> + Send;}

and this new transport trait is compatible with old Sink + Stream constrainment, with a thread shared sink.

pubstructSinkStreamTransport<Si,St>{stream:St,sink:Arc<Mutex<Si>>,}impl<Si,St>SinkStreamTransport<Si,St>{pubfnnew(sink:Si,stream:St) -> Self{Self{
stream,sink:Arc::new(Mutex::new(sink)),}}}impl<Role:ServiceRole,Si,St>Transport<Role>forSinkStreamTransport<Si,St>whereSt:Send + Stream<Item = RxJsonRpcMessage<Role>> + Unpin,Si:Send + Sink<TxJsonRpcMessage<Role>> + Unpin + 'static,{typeError = Si::Error;fnsend(&mutself,item:TxJsonRpcMessage<Role>,) -> implFuture<Output = Result<(),Self::Error>> + Send + 'static{use futures::SinkExt;let lock = self.sink.clone();asyncmove{letmut write = lock.lock().await;
write.send(item).await}}fnreceive(&mutself) -> implFuture<Output = Option<RxJsonRpcMessage<Role>>> + '_{use futures::StreamExt;self.stream.next()}asyncfnclose(&mutself) -> Result<(),Self::Error>{Ok(())}}

Based on those changes, we can make serve_inner loop not blocked by sending request.

  1. New worker transport type:
    For the case those we need to create a new worker task to run the transport, there's a new trait Worker, you can implement Worker and hand it over to WorkerTransport
pubtraitWorker:Sized + Send + 'static{typeError: std::error::Error + Send + Sync + 'static;typeRole:ServiceRole;fnerr_closed() -> Self::Error;fnerr_join(e: tokio::task::JoinError) -> Self::Error;fnrun(self,context:WorkerContext<Self>,) -> implFuture<Output = Result<(),WorkerQuitReason>> + Send;fnconfig(&self) -> WorkerConfig{WorkerConfig::default()}}

And with this new trait, I refactor many existed implementation based on it.

  1. Streamable http client
    Implemented streamable http client transport based on new worker.

  2. Auth client
    Trying to make auth client reusable between different http client.

  3. Add cfg-features for document

Motivation and Context

#170

How Has This Been Tested?

Breaking Changes

  1. Rename some feature:
    sse -> sse_client (it's more concret)

  2. Refactor the sse client transport, so do the construct methods.

Types of changes

  • Bug fix (non-breaking change which fixes an issue)
  • New feature (non-breaking change which adds functionality)
  • Breaking change (fix or feature that would cause existing functionality to change)
  • Documentation update

Checklist

  • I have read the MCP Documentation
  • My code follows the repository's style guidelines
  • New and existing tests pass locally
  • I have added appropriate error handling
  • I have added or updated documentation as needed

Additional context

@4t145

4t145 commented May 9, 2025

Copy link
Copy Markdown
ContributorAuthor

@jokemanfire I'am planing to extract the common part of reqwest client, so there may have the necessary to rewrite some part of current sse client, and then, both of streamable client and sse client can share the logic of Oauth2.

@jokemanfire

Copy link
Copy Markdown
Member

Thanks, That's well.

@4t145

4t145 commented May 12, 2025

Copy link
Copy Markdown
ContributorAuthor

I am going to refactor sse client in those aspects:

Cleint trait:

  1. move error from generic parameter to associated type (usually the error type should be a associated type)
  2. use impl Future as return type
pubtraitNeoSseClient:Clone + Send + Sync + 'static{typeError: std::error::Error + Send + Sync + 'static;fnpost_message(&self,uri:Arc<str>,message:ClientJsonRpcMessage,) -> implFuture<Output = Result<(),SseTransportError<Self::Error>>>
+ Send
+ '_;fnget_stream(&self,uri:Arc<str>,last_event_id:Option<String>,) -> implFuture<Output = Result<BoxedSseResponse,SseTransportError<Self::Error>,>,> + Send
+ '_;}

Move the connection logic to a worker

This can make code more maintainable and easier to understand.

Just like what I've done with this streamable http client in this PR.

Implement client trait directly for raw reqwest client

  1. The reason we have a wrapper at the first is we don't have a trait. And now we have a trait so we can remove the wrapper. (I guess)
  2. User can share the same configured client between diffent http based client transport types.

Make OAuth2 a common wrapper for different client.

@jokemanfire

Copy link
Copy Markdown
Member

No problem.

@4t145

4t145 commented May 14, 2025

Copy link
Copy Markdown
ContributorAuthor

As it‘s messioned in #170, There are a lot of refactor. But the changes in examples is minimal. So it should not be impact to users.

@4t145
4t145 requested a review from CopilotMay 14, 2025 09:29

CopilotAI left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pull Request Overview

This PR introduces a new streamable HTTP client feature while phasing out legacy SSE and IO transport implementations and unifying the transport interface. Key changes include:

  • Removal of the legacy SSE transport (sse.rs) and IO transport code.
  • Addition of a new sink_stream transport module and updates to common reqwest-based streamable HTTP client and SSE client modules.
  • Modifications in service and transport modules to use a unified Transport trait.

Reviewed Changes

Copilot reviewed 31 out of 31 changed files in this pull request and generated 1 comment.

Show a summary per file
FileDescription
crates/rmcp/src/transport/sse.rsRemoved outdated SSE transport implementation.
crates/rmcp/src/transport/sink_stream.rsIntroduced new sink_stream transport implementation.
crates/rmcp/src/transport/io.rsRemoved legacy IO transport code.
crates/rmcp/src/transport/common/reqwest/streamable_http_client.rsAdded streamable HTTP client support using reqwest with SSE and JSON responses.
crates/rmcp/Cargo.tomlUpdated dependency features and configuration for streamable HTTP client and related modules.
(Other transport and service modules)Updated to adopt the new unified Transport trait and auth wrappers.
Comments suppressed due to low confidence (1)

crates/rmcp/src/transport/common/reqwest/streamable_http_client.rs:92

  • Setting the ACCEPT header twice may result in the first value being overwritten. Consider combining the MIME types into a single header using a comma-separated string.
let mut request = request.header(ACCEPT, EVENT_STREAM_MIME_TYPE).header(ACCEPT, JSON_MIME_TYPE);

Comment threadcrates/rmcp/Cargo.toml Outdated
@4t1454t145 changed the title feat: Streamable http clientrefactor: Transport trait and worker transport, and streamable http client with those new features.May 14, 2025
@4t145

4t145 commented May 14, 2025

Copy link
Copy Markdown
ContributorAuthor

It'a big refactor, please check it again @jokemanfire. And I just updated the introduce of this PR.

@4t145
4t145 requested a review from CopilotMay 14, 2025 11:06

CopilotAI left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pull Request Overview

This PR refactors the core transport abstraction to a new Transport trait, adds worker-based transports, and implements streamable HTTP and SSE clients (with optional OAuth2 via AuthClient).

  • Define and implement the new Transport<R> trait in place of raw Sink/Stream.
  • Add SinkStreamTransport, AsyncRwTransport, and WorkerTransport adapters.
  • Introduce streamable HTTP/SSE client modules and an AuthClient wrapper.

Reviewed Changes

Copilot reviewed 31 out of 31 changed files in this pull request and generated no comments.

Show a summary per file
FileDescription
crates/rmcp/src/transport/sink_stream.rsNew SinkStreamTransport adapter implementing Transport
crates/rmcp/src/transport/async_rw.rsNew AsyncRwTransport for AsyncRead/AsyncWrite under Transport
crates/rmcp/src/transport/common/reqwest/streamable_http_client.rsNew StreamableHttpClient impl for reqwest::Client
crates/rmcp/src/transport/common/reqwest/sse_client.rsNew SSE client impl for reqwest::Client
crates/rmcp/src/transport/auth.rsAdded AuthClient wrapper for OAuth2
crates/rmcp/src/transport.rsUpdated Transport/IntoTransport traits and module exports
crates/rmcp/src/service/server.rsRefactored server to use Transport trait
crates/rmcp/src/service/client.rsRefactored client to use Transport trait
Comments suppressed due to low confidence (3)

crates/rmcp/src/transport/sink_stream.rs:1

  • The file defines methods returning impl Future<...> but does not import std::future::Future. Add use std::future::Future; to bring Future into scope.
use std::sync::Arc;

crates/rmcp/src/transport/auth.rs:16

  • The RwLock import is unused in this file. Consider removing it to avoid warnings and improve clarity.
use tokio::sync::{Mutex, RwLock};

crates/rmcp/src/transport/sink_stream.rs:66

  • [nitpick] The TransportAdapterAsyncCombinedRW enum is declared but never used. Remove or repurpose it to avoid dead code.
pub enum TransportAdapterAsyncCombinedRW {}

@jokemanfire

Copy link
Copy Markdown
Member

I will check ,after this PR merger, I plan to include the example in the integration testing and further improve the documentation. I think we can release the first release version, but we need to agree that minor modifications should not include interface destructive changes, and only consider introducing it in the larger version. We also need to consider forward compatibility issues. What do you think?

@4t145

Copy link
Copy Markdown
ContributorAuthor

@jokemanfire
I guess there could be possible breaking changes in transport part like web server support. And we still have a broken macro support, which could also be unstable for a while.

@4t145
4t145force-pushed the streamable-http-client branch from 880ff03 to 89fa930CompareMay 15, 2025 09:40
@jokemanfire

Copy link
Copy Markdown
Member

I will watch it after work tomorrow.

@4t145
4t145 marked this pull request as ready for review May 15, 2025 11:10
@4t145

Copy link
Copy Markdown
ContributorAuthor

I think it's ready for review now. But still need to add an example to show how to use streamable http client.

@4t1454t145 mentioned this pull request May 16, 2025

@jokemanfirejokemanfire left a comment

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'll have to keep watching tomorrow.

Comment threadcrates/rmcp/src/service.rs Outdated
let next_sse = match event {
Event::Sse(Some(Ok(next_sse))) => next_sse,
Event::Sse(Some(Err(e))) => {
tracing::warn!("sse stream error: {e}");

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think it deserve a error level log.

Copy link
Copy Markdown
ContributorAuthor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

As long as it's still trying to reconnect, it haven't reached an unrecoverable error, that's what I think.

Comment threadcrates/rmcp/src/transport/sse_client.rs Outdated
Comment threadcrates/rmcp/src/transport/worker.rs
let next_sse = match event {
Some(Ok(next_sse)) => next_sse,
Some(Err(e)) => {
tracing::warn!("sse stream error: {e}");

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

error level log

Comment threadcrates/rmcp/src/transport/streamable_http_client.rs Outdated
Self {
uri: "localhost".into(),
retry_config: SseRetryConfig::default(),
channel_buffer_capacity: 16,

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If it too small?

Comment threadcrates/rmcp/src/transport/streamable_http_server/axum.rs Outdated
Comment threadcrates/rmcp/src/transport/worker.rs Outdated
@jokemanfire

jokemanfire commented May 17, 2025

Copy link
Copy Markdown
Member

Actually, for me, things like sse_client, sse_client streamable_tttp_client streamable_tttp_derver require a higher level of work async_rw, sink_stream, and so on are more low level. As they are called, I tend to separate these two parts in the code organization of transport for a fresher code structure. Thank you for completing such a large workload.

@4t145
4t145force-pushed the streamable-http-client branch from 0576a27 to cc141b3CompareMay 17, 2025 10:32
@4t145

4t145 commented May 17, 2025

Copy link
Copy Markdown
ContributorAuthor

@jokemanfire Now it's ready for review again

  1. I move the sse stream loop to common::client_side_sse
  2. Still remain the older interface for sse client, instead of use WorkerTransport
  3. For worker channel buffer size, I think 16 is enough for most case. Sending to transport and handling by handler both won't block the service loop both now, so there are little chance to really pending on the await point.

@4t145
4t145 requested a review from jokemanfireMay 17, 2025 13:43
@4t145
4t145 merged commit 9a771fb into modelcontextprotocol:mainMay 17, 2025
@gau-nernstgau-nernst mentioned this pull request May 19, 2025
3 tasks
@github-actionsgithub-actionsBot mentioned this pull request Jul 2, 2025
takumi-earth pushed a commit to earthlings-dev/rmcp that referenced this pull request Jan 27, 2026
…lient with those new features. (modelcontextprotocol#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
Sign up for freeto join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

3 participants

@4t145@jokemanfire
, 'i'); if (__m === '*' || __re.test(location.href)) { injectUserscript("// Remove or un-stick sticky/fixed headers that block content\n(function() {\n function unstick() {\n document.querySelectorAll('header, nav, [role=\"banner\"], .header, .navbar, .sticky, .fixed-top, [style*=\"position: fixed\"], [style*=\"position:sticky\"]').forEach(function(el) {\n if (el.style.position === 'fixed' || el.style.position === 'sticky' || \n getComputedStyle(el).position === 'fixed' || getComputedStyle(el).position === 'sticky') {\n el.style.position = 'static';\n el.style.top = 'auto';\n el.style.zIndex = 'auto';\n }\n });\n }\n \n unstick();\n \n var observer = new MutationObserver(unstick);\n observer.observe(document.body, { childList: true, subtree: true, attributes: true, attributeFilter: ['style', 'class'] });\n})();", "Kill Sticky Headers"); } } catch(__e) { console.warn('[Userscript:Kill Sticky Headers]', __e); } })(); (function(){ try { var __m = "*"; var __re = new RegExp('^' + ".*" + '
Skip to content

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

Merged
4t145 merged 5 commits into
modelcontextprotocol:mainfrom
4t145:streamable-http-client
May 17, 2025
Merged

refactor: Transport trait and worker transport, and streamable http client with those new features.#167
4t145 merged 5 commits into
modelcontextprotocol:mainfrom
4t145:streamable-http-client

Conversation

@4t145

@4t1454t145 commented May 9, 2025

Copy link
Copy Markdown
Contributor

related issue: #170, #55

1. Refactor transport trait

Now we have a real transport trait:

pubtraitTransport<R>:SendwhereR:ServiceRole,{typeError;fnsend(&mutself,item:TxJsonRpcMessage<R>,) -> implFuture<Output = Result<(),Self::Error>> + Send + 'static;fnreceive(&mutself) -> implFuture<Output = Option<RxJsonRpcMessage<R>>> + Send + '_;fnclose(&mutself) -> implFuture<Output = Result<(),Self::Error>> + Send;}

and this new transport trait is compatible with old Sink + Stream constrainment, with a thread shared sink.

pubstructSinkStreamTransport<Si,St>{stream:St,sink:Arc<Mutex<Si>>,}impl<Si,St>SinkStreamTransport<Si,St>{pubfnnew(sink:Si,stream:St) -> Self{Self{
stream,sink:Arc::new(Mutex::new(sink)),}}}impl<Role:ServiceRole,Si,St>Transport<Role>forSinkStreamTransport<Si,St>whereSt:Send + Stream<Item = RxJsonRpcMessage<Role>> + Unpin,Si:Send + Sink<TxJsonRpcMessage<Role>> + Unpin + 'static,{typeError = Si::Error;fnsend(&mutself,item:TxJsonRpcMessage<Role>,) -> implFuture<Output = Result<(),Self::Error>> + Send + 'static{use futures::SinkExt;let lock = self.sink.clone();asyncmove{letmut write = lock.lock().await;
write.send(item).await}}fnreceive(&mutself) -> implFuture<Output = Option<RxJsonRpcMessage<Role>>> + '_{use futures::StreamExt;self.stream.next()}asyncfnclose(&mutself) -> Result<(),Self::Error>{Ok(())}}

Based on those changes, we can make serve_inner loop not blocked by sending request.

  1. New worker transport type:
    For the case those we need to create a new worker task to run the transport, there's a new trait Worker, you can implement Worker and hand it over to WorkerTransport
pubtraitWorker:Sized + Send + 'static{typeError: std::error::Error + Send + Sync + 'static;typeRole:ServiceRole;fnerr_closed() -> Self::Error;fnerr_join(e: tokio::task::JoinError) -> Self::Error;fnrun(self,context:WorkerContext<Self>,) -> implFuture<Output = Result<(),WorkerQuitReason>> + Send;fnconfig(&self) -> WorkerConfig{WorkerConfig::default()}}

And with this new trait, I refactor many existed implementation based on it.

  1. Streamable http client
    Implemented streamable http client transport based on new worker.

  2. Auth client
    Trying to make auth client reusable between different http client.

  3. Add cfg-features for document

Motivation and Context

#170

How Has This Been Tested?

Breaking Changes

  1. Rename some feature:
    sse -> sse_client (it's more concret)

  2. Refactor the sse client transport, so do the construct methods.

Types of changes

  • Bug fix (non-breaking change which fixes an issue)
  • New feature (non-breaking change which adds functionality)
  • Breaking change (fix or feature that would cause existing functionality to change)
  • Documentation update

Checklist

  • I have read the MCP Documentation
  • My code follows the repository's style guidelines
  • New and existing tests pass locally
  • I have added appropriate error handling
  • I have added or updated documentation as needed

Additional context

@4t145

4t145 commented May 9, 2025

Copy link
Copy Markdown
ContributorAuthor

@jokemanfire I'am planing to extract the common part of reqwest client, so there may have the necessary to rewrite some part of current sse client, and then, both of streamable client and sse client can share the logic of Oauth2.

@jokemanfire

Copy link
Copy Markdown
Member

Thanks, That's well.

@4t145

4t145 commented May 12, 2025

Copy link
Copy Markdown
ContributorAuthor

I am going to refactor sse client in those aspects:

Cleint trait:

  1. move error from generic parameter to associated type (usually the error type should be a associated type)
  2. use impl Future as return type
pubtraitNeoSseClient:Clone + Send + Sync + 'static{typeError: std::error::Error + Send + Sync + 'static;fnpost_message(&self,uri:Arc<str>,message:ClientJsonRpcMessage,) -> implFuture<Output = Result<(),SseTransportError<Self::Error>>>
+ Send
+ '_;fnget_stream(&self,uri:Arc<str>,last_event_id:Option<String>,) -> implFuture<Output = Result<BoxedSseResponse,SseTransportError<Self::Error>,>,> + Send
+ '_;}

Move the connection logic to a worker

This can make code more maintainable and easier to understand.

Just like what I've done with this streamable http client in this PR.

Implement client trait directly for raw reqwest client

  1. The reason we have a wrapper at the first is we don't have a trait. And now we have a trait so we can remove the wrapper. (I guess)
  2. User can share the same configured client between diffent http based client transport types.

Make OAuth2 a common wrapper for different client.

@jokemanfire

Copy link
Copy Markdown
Member

No problem.

@4t145

4t145 commented May 14, 2025

Copy link
Copy Markdown
ContributorAuthor

As it‘s messioned in #170, There are a lot of refactor. But the changes in examples is minimal. So it should not be impact to users.

@4t145
4t145 requested a review from CopilotMay 14, 2025 09:29

CopilotAI left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pull Request Overview

This PR introduces a new streamable HTTP client feature while phasing out legacy SSE and IO transport implementations and unifying the transport interface. Key changes include:

  • Removal of the legacy SSE transport (sse.rs) and IO transport code.
  • Addition of a new sink_stream transport module and updates to common reqwest-based streamable HTTP client and SSE client modules.
  • Modifications in service and transport modules to use a unified Transport trait.

Reviewed Changes

Copilot reviewed 31 out of 31 changed files in this pull request and generated 1 comment.

Show a summary per file
FileDescription
crates/rmcp/src/transport/sse.rsRemoved outdated SSE transport implementation.
crates/rmcp/src/transport/sink_stream.rsIntroduced new sink_stream transport implementation.
crates/rmcp/src/transport/io.rsRemoved legacy IO transport code.
crates/rmcp/src/transport/common/reqwest/streamable_http_client.rsAdded streamable HTTP client support using reqwest with SSE and JSON responses.
crates/rmcp/Cargo.tomlUpdated dependency features and configuration for streamable HTTP client and related modules.
(Other transport and service modules)Updated to adopt the new unified Transport trait and auth wrappers.
Comments suppressed due to low confidence (1)

crates/rmcp/src/transport/common/reqwest/streamable_http_client.rs:92

  • Setting the ACCEPT header twice may result in the first value being overwritten. Consider combining the MIME types into a single header using a comma-separated string.
let mut request = request.header(ACCEPT, EVENT_STREAM_MIME_TYPE).header(ACCEPT, JSON_MIME_TYPE);

Comment threadcrates/rmcp/Cargo.toml Outdated
@4t1454t145 changed the title feat: Streamable http clientrefactor: Transport trait and worker transport, and streamable http client with those new features.May 14, 2025
@4t145

4t145 commented May 14, 2025

Copy link
Copy Markdown
ContributorAuthor

It'a big refactor, please check it again @jokemanfire. And I just updated the introduce of this PR.

@4t145
4t145 requested a review from CopilotMay 14, 2025 11:06

CopilotAI left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pull Request Overview

This PR refactors the core transport abstraction to a new Transport trait, adds worker-based transports, and implements streamable HTTP and SSE clients (with optional OAuth2 via AuthClient).

  • Define and implement the new Transport<R> trait in place of raw Sink/Stream.
  • Add SinkStreamTransport, AsyncRwTransport, and WorkerTransport adapters.
  • Introduce streamable HTTP/SSE client modules and an AuthClient wrapper.

Reviewed Changes

Copilot reviewed 31 out of 31 changed files in this pull request and generated no comments.

Show a summary per file
FileDescription
crates/rmcp/src/transport/sink_stream.rsNew SinkStreamTransport adapter implementing Transport
crates/rmcp/src/transport/async_rw.rsNew AsyncRwTransport for AsyncRead/AsyncWrite under Transport
crates/rmcp/src/transport/common/reqwest/streamable_http_client.rsNew StreamableHttpClient impl for reqwest::Client
crates/rmcp/src/transport/common/reqwest/sse_client.rsNew SSE client impl for reqwest::Client
crates/rmcp/src/transport/auth.rsAdded AuthClient wrapper for OAuth2
crates/rmcp/src/transport.rsUpdated Transport/IntoTransport traits and module exports
crates/rmcp/src/service/server.rsRefactored server to use Transport trait
crates/rmcp/src/service/client.rsRefactored client to use Transport trait
Comments suppressed due to low confidence (3)

crates/rmcp/src/transport/sink_stream.rs:1

  • The file defines methods returning impl Future<...> but does not import std::future::Future. Add use std::future::Future; to bring Future into scope.
use std::sync::Arc;

crates/rmcp/src/transport/auth.rs:16

  • The RwLock import is unused in this file. Consider removing it to avoid warnings and improve clarity.
use tokio::sync::{Mutex, RwLock};

crates/rmcp/src/transport/sink_stream.rs:66

  • [nitpick] The TransportAdapterAsyncCombinedRW enum is declared but never used. Remove or repurpose it to avoid dead code.
pub enum TransportAdapterAsyncCombinedRW {}

@jokemanfire

Copy link
Copy Markdown
Member

I will check ,after this PR merger, I plan to include the example in the integration testing and further improve the documentation. I think we can release the first release version, but we need to agree that minor modifications should not include interface destructive changes, and only consider introducing it in the larger version. We also need to consider forward compatibility issues. What do you think?

@4t145

Copy link
Copy Markdown
ContributorAuthor

@jokemanfire
I guess there could be possible breaking changes in transport part like web server support. And we still have a broken macro support, which could also be unstable for a while.

@4t145
4t145force-pushed the streamable-http-client branch from 880ff03 to 89fa930CompareMay 15, 2025 09:40
@jokemanfire

Copy link
Copy Markdown
Member

I will watch it after work tomorrow.

@4t145
4t145 marked this pull request as ready for review May 15, 2025 11:10
@4t145

Copy link
Copy Markdown
ContributorAuthor

I think it's ready for review now. But still need to add an example to show how to use streamable http client.

@4t1454t145 mentioned this pull request May 16, 2025

@jokemanfirejokemanfire left a comment

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'll have to keep watching tomorrow.

Comment threadcrates/rmcp/src/service.rs Outdated
let next_sse = match event {
Event::Sse(Some(Ok(next_sse))) => next_sse,
Event::Sse(Some(Err(e))) => {
tracing::warn!("sse stream error: {e}");

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think it deserve a error level log.

Copy link
Copy Markdown
ContributorAuthor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

As long as it's still trying to reconnect, it haven't reached an unrecoverable error, that's what I think.

Comment threadcrates/rmcp/src/transport/sse_client.rs Outdated
Comment threadcrates/rmcp/src/transport/worker.rs
let next_sse = match event {
Some(Ok(next_sse)) => next_sse,
Some(Err(e)) => {
tracing::warn!("sse stream error: {e}");

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

error level log

Comment threadcrates/rmcp/src/transport/streamable_http_client.rs Outdated
Self {
uri: "localhost".into(),
retry_config: SseRetryConfig::default(),
channel_buffer_capacity: 16,

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If it too small?

Comment threadcrates/rmcp/src/transport/streamable_http_server/axum.rs Outdated
Comment threadcrates/rmcp/src/transport/worker.rs Outdated
@jokemanfire

jokemanfire commented May 17, 2025

Copy link
Copy Markdown
Member

Actually, for me, things like sse_client, sse_client streamable_tttp_client streamable_tttp_derver require a higher level of work async_rw, sink_stream, and so on are more low level. As they are called, I tend to separate these two parts in the code organization of transport for a fresher code structure. Thank you for completing such a large workload.

@4t145
4t145force-pushed the streamable-http-client branch from 0576a27 to cc141b3CompareMay 17, 2025 10:32
@4t145

4t145 commented May 17, 2025

Copy link
Copy Markdown
ContributorAuthor

@jokemanfire Now it's ready for review again

  1. I move the sse stream loop to common::client_side_sse
  2. Still remain the older interface for sse client, instead of use WorkerTransport
  3. For worker channel buffer size, I think 16 is enough for most case. Sending to transport and handling by handler both won't block the service loop both now, so there are little chance to really pending on the await point.

@4t145
4t145 requested a review from jokemanfireMay 17, 2025 13:43
@4t145
4t145 merged commit 9a771fb into modelcontextprotocol:mainMay 17, 2025
@gau-nernstgau-nernst mentioned this pull request May 19, 2025
3 tasks
@github-actionsgithub-actionsBot mentioned this pull request Jul 2, 2025
takumi-earth pushed a commit to earthlings-dev/rmcp that referenced this pull request Jan 27, 2026
…lient with those new features. (modelcontextprotocol#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
Sign up for freeto join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

3 participants

@4t145@jokemanfire
, 'i'); if (__m === '*' || __re.test(location.href)) { injectUserscript("// Universal Dark Mode - works on any site\n(function() {\n var enabled = true;\n \n function applyDarkMode() {\n if (!enabled) return;\n \n // Create style element if it doesn't exist\n var style = document.getElementById('universal-dark-mode-style');\n if (!style) {\n style = document.createElement('style');\n style.id = 'universal-dark-mode-style';\n document.head.appendChild(style);\n }\n \n // Dark mode CSS - inverts colors but preserves images/video\n style.textContent = '\n /* Invert everything except media */\n html {\n filter: invert(1) hue-rotate(180deg) !important;\n background: #1a1a2e !important;\n }\n \n /* Restore images, videos, iframes, canvas */\n img, video, iframe, canvas, svg, picture, [style*=\"background-image\"] {\n filter: invert(1) hue-rotate(180deg) !important;\n }\n \n /* Preserve specific elements that should not be inverted */\n .no-dark-mode, .no-dark-mode *,\n [data-theme=\"light\"], [data-theme=\"light\"],\n .ace_editor, .ace_editor *,\n .CodeMirror, .CodeMirror *,\n .monaco-editor, .monaco-editor *,\n .markdown-body pre, .markdown-body pre *,\n .highlight, .highlight *,\n pre code, pre code * {\n filter: none !important;\n }\n \n /* Fix common UI elements */\n .modal, .popup, .dropdown-menu, .tooltip, .popover {\n filter: invert(1) hue-rotate(180deg) !important;\n background: #2d2d44 !important;\n border-color: #444 !important;\n }\n \n /* Scrollbars */\n ::-webkit-scrollbar { background: #1a1a2e !important; }\n ::-webkit-scrollbar-thumb { background: #444 !important; }\n ::-webkit-scrollbar-thumb:hover { background: #555 !important; }\n \n /* Selection */\n ::selection { background: #4ecdc4 !important; color: #1a1a2e !important; }\n ::-moz-selection { background: #4ecdc4 !important; color: #1a1a2e !important; }\n ';\n }\n \n function removeDarkMode() {\n var style = document.getElementById('universal-dark-mode-style');\n if (style) style.remove();\n }\n \n // Toggle with Alt+Shift+D\n document.addEventListener('keydown', function(e) {\n if (e.altKey && e.shiftKey && e.key === 'D') {\n e.preventDefault();\n enabled = !enabled;\n if (enabled) {\n applyDarkMode();\n console.log('[Universal Dark Mode] Enabled');\n } else {\n removeDarkMode();\n console.log('[Universal Dark Mode] Disabled');\n }\n }\n });\n \n // Apply on load\n applyDarkMode();\n \n // Re-apply on dynamic content\n var observer = new MutationObserver(function(mutations) {\n if (enabled && !document.getElementById('universal-dark-mode-style')) {\n applyDarkMode();\n }\n });\n observer.observe(document.head, { childList: true });\n \n console.log('[Universal Dark Mode] Loaded - Press Alt+Shift+D to toggle');\n})();", "Universal Dark Mode"); } } catch(__e) { console.warn('[Userscript:Universal Dark Mode]', __e); } })(); })();
Skip to content

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

Merged
4t145 merged 5 commits into
modelcontextprotocol:mainfrom
4t145:streamable-http-client
May 17, 2025
Merged

refactor: Transport trait and worker transport, and streamable http client with those new features.#167
4t145 merged 5 commits into
modelcontextprotocol:mainfrom
4t145:streamable-http-client

Conversation

@4t145

@4t1454t145 commented May 9, 2025

Copy link
Copy Markdown
Contributor

related issue: #170, #55

1. Refactor transport trait

Now we have a real transport trait:

pubtraitTransport<R>:SendwhereR:ServiceRole,{typeError;fnsend(&mutself,item:TxJsonRpcMessage<R>,) -> implFuture<Output = Result<(),Self::Error>> + Send + 'static;fnreceive(&mutself) -> implFuture<Output = Option<RxJsonRpcMessage<R>>> + Send + '_;fnclose(&mutself) -> implFuture<Output = Result<(),Self::Error>> + Send;}

and this new transport trait is compatible with old Sink + Stream constrainment, with a thread shared sink.

pubstructSinkStreamTransport<Si,St>{stream:St,sink:Arc<Mutex<Si>>,}impl<Si,St>SinkStreamTransport<Si,St>{pubfnnew(sink:Si,stream:St) -> Self{Self{
stream,sink:Arc::new(Mutex::new(sink)),}}}impl<Role:ServiceRole,Si,St>Transport<Role>forSinkStreamTransport<Si,St>whereSt:Send + Stream<Item = RxJsonRpcMessage<Role>> + Unpin,Si:Send + Sink<TxJsonRpcMessage<Role>> + Unpin + 'static,{typeError = Si::Error;fnsend(&mutself,item:TxJsonRpcMessage<Role>,) -> implFuture<Output = Result<(),Self::Error>> + Send + 'static{use futures::SinkExt;let lock = self.sink.clone();asyncmove{letmut write = lock.lock().await;
write.send(item).await}}fnreceive(&mutself) -> implFuture<Output = Option<RxJsonRpcMessage<Role>>> + '_{use futures::StreamExt;self.stream.next()}asyncfnclose(&mutself) -> Result<(),Self::Error>{Ok(())}}

Based on those changes, we can make serve_inner loop not blocked by sending request.

  1. New worker transport type:
    For the case those we need to create a new worker task to run the transport, there's a new trait Worker, you can implement Worker and hand it over to WorkerTransport
pubtraitWorker:Sized + Send + 'static{typeError: std::error::Error + Send + Sync + 'static;typeRole:ServiceRole;fnerr_closed() -> Self::Error;fnerr_join(e: tokio::task::JoinError) -> Self::Error;fnrun(self,context:WorkerContext<Self>,) -> implFuture<Output = Result<(),WorkerQuitReason>> + Send;fnconfig(&self) -> WorkerConfig{WorkerConfig::default()}}

And with this new trait, I refactor many existed implementation based on it.

  1. Streamable http client
    Implemented streamable http client transport based on new worker.

  2. Auth client
    Trying to make auth client reusable between different http client.

  3. Add cfg-features for document

Motivation and Context

#170

How Has This Been Tested?

Breaking Changes

  1. Rename some feature:
    sse -> sse_client (it's more concret)

  2. Refactor the sse client transport, so do the construct methods.

Types of changes

  • Bug fix (non-breaking change which fixes an issue)
  • New feature (non-breaking change which adds functionality)
  • Breaking change (fix or feature that would cause existing functionality to change)
  • Documentation update

Checklist

  • I have read the MCP Documentation
  • My code follows the repository's style guidelines
  • New and existing tests pass locally
  • I have added appropriate error handling
  • I have added or updated documentation as needed

Additional context

@4t145

4t145 commented May 9, 2025

Copy link
Copy Markdown
ContributorAuthor

@jokemanfire I'am planing to extract the common part of reqwest client, so there may have the necessary to rewrite some part of current sse client, and then, both of streamable client and sse client can share the logic of Oauth2.

@jokemanfire

Copy link
Copy Markdown
Member

Thanks, That's well.

@4t145

4t145 commented May 12, 2025

Copy link
Copy Markdown
ContributorAuthor

I am going to refactor sse client in those aspects:

Cleint trait:

  1. move error from generic parameter to associated type (usually the error type should be a associated type)
  2. use impl Future as return type
pubtraitNeoSseClient:Clone + Send + Sync + 'static{typeError: std::error::Error + Send + Sync + 'static;fnpost_message(&self,uri:Arc<str>,message:ClientJsonRpcMessage,) -> implFuture<Output = Result<(),SseTransportError<Self::Error>>>
+ Send
+ '_;fnget_stream(&self,uri:Arc<str>,last_event_id:Option<String>,) -> implFuture<Output = Result<BoxedSseResponse,SseTransportError<Self::Error>,>,> + Send
+ '_;}

Move the connection logic to a worker

This can make code more maintainable and easier to understand.

Just like what I've done with this streamable http client in this PR.

Implement client trait directly for raw reqwest client

  1. The reason we have a wrapper at the first is we don't have a trait. And now we have a trait so we can remove the wrapper. (I guess)
  2. User can share the same configured client between diffent http based client transport types.

Make OAuth2 a common wrapper for different client.

@jokemanfire

Copy link
Copy Markdown
Member

No problem.

@4t145

4t145 commented May 14, 2025

Copy link
Copy Markdown
ContributorAuthor

As it‘s messioned in #170, There are a lot of refactor. But the changes in examples is minimal. So it should not be impact to users.

@4t145
4t145 requested a review from CopilotMay 14, 2025 09:29

CopilotAI left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pull Request Overview

This PR introduces a new streamable HTTP client feature while phasing out legacy SSE and IO transport implementations and unifying the transport interface. Key changes include:

  • Removal of the legacy SSE transport (sse.rs) and IO transport code.
  • Addition of a new sink_stream transport module and updates to common reqwest-based streamable HTTP client and SSE client modules.
  • Modifications in service and transport modules to use a unified Transport trait.

Reviewed Changes

Copilot reviewed 31 out of 31 changed files in this pull request and generated 1 comment.

Show a summary per file
FileDescription
crates/rmcp/src/transport/sse.rsRemoved outdated SSE transport implementation.
crates/rmcp/src/transport/sink_stream.rsIntroduced new sink_stream transport implementation.
crates/rmcp/src/transport/io.rsRemoved legacy IO transport code.
crates/rmcp/src/transport/common/reqwest/streamable_http_client.rsAdded streamable HTTP client support using reqwest with SSE and JSON responses.
crates/rmcp/Cargo.tomlUpdated dependency features and configuration for streamable HTTP client and related modules.
(Other transport and service modules)Updated to adopt the new unified Transport trait and auth wrappers.
Comments suppressed due to low confidence (1)

crates/rmcp/src/transport/common/reqwest/streamable_http_client.rs:92

  • Setting the ACCEPT header twice may result in the first value being overwritten. Consider combining the MIME types into a single header using a comma-separated string.
let mut request = request.header(ACCEPT, EVENT_STREAM_MIME_TYPE).header(ACCEPT, JSON_MIME_TYPE);

Comment threadcrates/rmcp/Cargo.toml Outdated
@4t1454t145 changed the title feat: Streamable http clientrefactor: Transport trait and worker transport, and streamable http client with those new features.May 14, 2025
@4t145

4t145 commented May 14, 2025

Copy link
Copy Markdown
ContributorAuthor

It'a big refactor, please check it again @jokemanfire. And I just updated the introduce of this PR.

@4t145
4t145 requested a review from CopilotMay 14, 2025 11:06

CopilotAI left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pull Request Overview

This PR refactors the core transport abstraction to a new Transport trait, adds worker-based transports, and implements streamable HTTP and SSE clients (with optional OAuth2 via AuthClient).

  • Define and implement the new Transport<R> trait in place of raw Sink/Stream.
  • Add SinkStreamTransport, AsyncRwTransport, and WorkerTransport adapters.
  • Introduce streamable HTTP/SSE client modules and an AuthClient wrapper.

Reviewed Changes

Copilot reviewed 31 out of 31 changed files in this pull request and generated no comments.

Show a summary per file
FileDescription
crates/rmcp/src/transport/sink_stream.rsNew SinkStreamTransport adapter implementing Transport
crates/rmcp/src/transport/async_rw.rsNew AsyncRwTransport for AsyncRead/AsyncWrite under Transport
crates/rmcp/src/transport/common/reqwest/streamable_http_client.rsNew StreamableHttpClient impl for reqwest::Client
crates/rmcp/src/transport/common/reqwest/sse_client.rsNew SSE client impl for reqwest::Client
crates/rmcp/src/transport/auth.rsAdded AuthClient wrapper for OAuth2
crates/rmcp/src/transport.rsUpdated Transport/IntoTransport traits and module exports
crates/rmcp/src/service/server.rsRefactored server to use Transport trait
crates/rmcp/src/service/client.rsRefactored client to use Transport trait
Comments suppressed due to low confidence (3)

crates/rmcp/src/transport/sink_stream.rs:1

  • The file defines methods returning impl Future<...> but does not import std::future::Future. Add use std::future::Future; to bring Future into scope.
use std::sync::Arc;

crates/rmcp/src/transport/auth.rs:16

  • The RwLock import is unused in this file. Consider removing it to avoid warnings and improve clarity.
use tokio::sync::{Mutex, RwLock};

crates/rmcp/src/transport/sink_stream.rs:66

  • [nitpick] The TransportAdapterAsyncCombinedRW enum is declared but never used. Remove or repurpose it to avoid dead code.
pub enum TransportAdapterAsyncCombinedRW {}

@jokemanfire

Copy link
Copy Markdown
Member

I will check ,after this PR merger, I plan to include the example in the integration testing and further improve the documentation. I think we can release the first release version, but we need to agree that minor modifications should not include interface destructive changes, and only consider introducing it in the larger version. We also need to consider forward compatibility issues. What do you think?

@4t145

Copy link
Copy Markdown
ContributorAuthor

@jokemanfire
I guess there could be possible breaking changes in transport part like web server support. And we still have a broken macro support, which could also be unstable for a while.

@4t145
4t145force-pushed the streamable-http-client branch from 880ff03 to 89fa930CompareMay 15, 2025 09:40
@jokemanfire

Copy link
Copy Markdown
Member

I will watch it after work tomorrow.

@4t145
4t145 marked this pull request as ready for review May 15, 2025 11:10
@4t145

Copy link
Copy Markdown
ContributorAuthor

I think it's ready for review now. But still need to add an example to show how to use streamable http client.

@4t1454t145 mentioned this pull request May 16, 2025

@jokemanfirejokemanfire left a comment

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'll have to keep watching tomorrow.

Comment threadcrates/rmcp/src/service.rs Outdated
let next_sse = match event {
Event::Sse(Some(Ok(next_sse))) => next_sse,
Event::Sse(Some(Err(e))) => {
tracing::warn!("sse stream error: {e}");

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think it deserve a error level log.

Copy link
Copy Markdown
ContributorAuthor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

As long as it's still trying to reconnect, it haven't reached an unrecoverable error, that's what I think.

Comment threadcrates/rmcp/src/transport/sse_client.rs Outdated
Comment threadcrates/rmcp/src/transport/worker.rs
let next_sse = match event {
Some(Ok(next_sse)) => next_sse,
Some(Err(e)) => {
tracing::warn!("sse stream error: {e}");

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

error level log

Comment threadcrates/rmcp/src/transport/streamable_http_client.rs Outdated
Self {
uri: "localhost".into(),
retry_config: SseRetryConfig::default(),
channel_buffer_capacity: 16,

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If it too small?

Comment threadcrates/rmcp/src/transport/streamable_http_server/axum.rs Outdated
Comment threadcrates/rmcp/src/transport/worker.rs Outdated
@jokemanfire

jokemanfire commented May 17, 2025

Copy link
Copy Markdown
Member

Actually, for me, things like sse_client, sse_client streamable_tttp_client streamable_tttp_derver require a higher level of work async_rw, sink_stream, and so on are more low level. As they are called, I tend to separate these two parts in the code organization of transport for a fresher code structure. Thank you for completing such a large workload.

@4t145
4t145force-pushed the streamable-http-client branch from 0576a27 to cc141b3CompareMay 17, 2025 10:32
@4t145

4t145 commented May 17, 2025

Copy link
Copy Markdown
ContributorAuthor

@jokemanfire Now it's ready for review again

  1. I move the sse stream loop to common::client_side_sse
  2. Still remain the older interface for sse client, instead of use WorkerTransport
  3. For worker channel buffer size, I think 16 is enough for most case. Sending to transport and handling by handler both won't block the service loop both now, so there are little chance to really pending on the await point.

@4t145
4t145 requested a review from jokemanfireMay 17, 2025 13:43
@4t145
4t145 merged commit 9a771fb into modelcontextprotocol:mainMay 17, 2025
@gau-nernstgau-nernst mentioned this pull request May 19, 2025
3 tasks
@github-actionsgithub-actionsBot mentioned this pull request Jul 2, 2025
takumi-earth pushed a commit to earthlings-dev/rmcp that referenced this pull request Jan 27, 2026
…lient with those new features. (modelcontextprotocol#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
Sign up for freeto join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

3 participants

@4t145@jokemanfire