rmcp/
transport.rs

1//! # Transport
2//! The transport type must implemented [`Transport`] trait, which allow it send message concurrently and receive message sequentially.
3//！
4//! ## Standard Transport Types
5//! There are 2 pairs of standard transport types:
6//!
7//! | transport         | client                                                    | server                                                |
8//! |:-:                |:-:                                                        |:-:                                                    |
9//! | std IO            | [`child_process::TokioChildProcess`]                      | [`io::stdio`]                                         |
10//! | streamable http   | [`streamable_http_client::StreamableHttpClientTransport`] | `streamable_http_server::StreamableHttpService`     |
11//!
12//！## Helper Transport Types
13//! Thers are several helper transport types that can help you to create transport quickly.
14//!
15//! ### [Worker Transport](`worker::WorkerTransport`)
16//! Which allows you to run a worker and process messages in another tokio task.
17//!
18//! ### [Async Read/Write Transport](`async_rw::AsyncRwTransport`)
19//! You need to enable `transport-async-rw` feature to use this transport.
20//!
21//! This transport is used to create a transport from a byte stream which implemented [`tokio::io::AsyncRead`] and [`tokio::io::AsyncWrite`].
22//!
23//! This could be very helpful when you want to create a transport from a byte stream, such as a file or a tcp connection.
24//!
25//! ### [Sink/Stream Transport](`sink_stream::SinkStreamTransport`)
26//! This transport is used to create a transport from a sink and a stream.
27//!
28//! This could be very helpful when you want to create a transport from a duplex object stream, such as a websocket connection.
29//!
30//! ## [IntoTransport](`IntoTransport`) trait
31//! [`IntoTransport`] is a helper trait that implicitly convert a type into a transport type.
32//!
33//! ### These types is automatically implemented [`IntoTransport`] trait
34//! 1. A type that already implement both [`futures::Sink`] and [`futures::Stream`] trait, or a tuple `(Tx, Rx)`  where `Tx` is [`futures::Sink`] and `Rx` is [`futures::Stream`].
35//! 2. A type that implement both [`tokio::io::AsyncRead`] and [`tokio::io::AsyncWrite`] trait. or a tuple `(R, W)` where `R` is [`tokio::io::AsyncRead`] and `W` is [`tokio::io::AsyncWrite`].
36//! 3. A type that implement [Worker](`worker::Worker`) trait.
37//! 4. A type that implement [`Transport`] trait.
38//!
39//! ## Examples
40//!
41//! ```rust
42//! # use rmcp::{
43//! #     ServiceExt, serve_server,
44//! # };
45//! #[cfg(feature = "client")]
46//! # use rmcp::serve_client;
47//!
48//! // create transport from tcp stream
49//! #[cfg(feature = "client")]
50//! async fn client() -> Result<(), Box<dyn std::error::Error>> {
51//!     let stream = tokio::net::TcpSocket::new_v4()?
52//!         .connect("127.0.0.1:8001".parse()?)
53//!         .await?;
54//!     let client = ().serve(stream).await?;
55//!     let tools = client.peer().list_tools(Default::default()).await?;
56//!     println!("{:?}", tools);
57//!     Ok(())
58//! }
59//!
60//! // create transport from std io
61//! #[cfg(feature = "client")]
62//! async fn io()  -> Result<(), Box<dyn std::error::Error>> {
63//!     let client = ().serve((tokio::io::stdin(), tokio::io::stdout())).await?;
64//!     let tools = client.peer().list_tools(Default::default()).await?;
65//!     println!("{:?}", tools);
66//!     Ok(())
67//! }
68//! ```
69
70use std::{borrow::Cow, sync::Arc};
71
72use crate::service::{RxJsonRpcMessage, ServiceRole, TxJsonRpcMessage};
73
74pub mod sink_stream;
75
76#[cfg(feature = "transport-async-rw")]
77pub mod async_rw;
78
79#[cfg(feature = "transport-worker")]
80pub mod worker;
81#[cfg(feature = "transport-worker")]
82pub use worker::WorkerTransport;
83
84#[cfg(feature = "transport-child-process")]
85pub mod child_process;
86#[cfg(feature = "which-command")]
87pub use child_process::which_command;
88#[cfg(feature = "transport-child-process")]
89pub use child_process::{ConfigureCommandExt, TokioChildProcess};
90
91#[cfg(feature = "transport-io")]
92pub mod io;
93#[cfg(feature = "transport-io")]
94pub use io::stdio;
95
96#[cfg(feature = "auth")]
97pub mod auth;
98#[cfg(feature = "auth-client-credentials-jwt")]
99pub use auth::JwtSigningAlgorithm;
100#[cfg(feature = "auth")]
101pub use auth::{
102    AuthClient, AuthError, AuthorizationManager, AuthorizationRequest, AuthorizationSession,
103    AuthorizedHttpClient, ClientCredentialsConfig, CredentialRefreshGuard, CredentialStore,
104    EXTENSION_OAUTH_CLIENT_CREDENTIALS, InMemoryCredentialStore, InMemoryStateStore,
105    OAuthHttpClient, OAuthHttpClientError, OAuthHttpClientFuture, OAuthHttpRedirectPolicy,
106    OAuthHttpRequest, ScopeUpgradeConfig, StateStore, StoredAuthorizationState, StoredCredentials,
107    WWWAuthenticateParams,
108};
109
110// #[cfg(feature = "transport-ws")]
111// pub mod ws;
112#[cfg(feature = "transport-streamable-http-server-session")]
113pub mod streamable_http_server;
114#[cfg(all(feature = "transport-streamable-http-server", not(feature = "local")))]
115pub use streamable_http_server::tower::{StreamableHttpServerConfig, StreamableHttpService};
116
117#[cfg(feature = "transport-streamable-http-client")]
118pub mod streamable_http_client;
119#[cfg(all(unix, feature = "transport-streamable-http-client-unix-socket"))]
120pub use common::unix_socket::UnixSocketHttpClient;
121#[cfg(feature = "transport-streamable-http-client")]
122pub use streamable_http_client::StreamableHttpClientTransport;
123
124/// Common use codes
125pub mod common;
126
127pub trait Transport<R>: Send
128where
129    R: ServiceRole,
130{
131    type Error: std::error::Error + Send + Sync + 'static;
132    fn name() -> Cow<'static, str> {
133        std::any::type_name::<Self>().into()
134    }
135    /// Send a message to the transport
136    ///
137    /// Notice that the future returned by this function should be `Send` and `'static`.
138    /// It's because the sending message could be executed concurrently.
139    ///
140    fn send(
141        &mut self,
142        item: TxJsonRpcMessage<R>,
143    ) -> impl Future<Output = Result<(), Self::Error>> + Send + 'static;
144
145    /// Receive a message from the transport, this operation is sequential.
146    fn receive(&mut self) -> impl Future<Output = Option<RxJsonRpcMessage<R>>> + Send;
147
148    /// Close the transport
149    fn close(&mut self) -> impl Future<Output = Result<(), Self::Error>> + Send;
150}
151
152pub trait IntoTransport<R, E, A>: Send + 'static
153where
154    R: ServiceRole,
155    E: std::error::Error + Send + 'static,
156{
157    fn into_transport(self) -> impl Transport<R, Error = E> + 'static;
158}
159
160#[non_exhaustive]
161pub enum TransportAdapterIdentity {}
162impl<R, T, E> IntoTransport<R, E, TransportAdapterIdentity> for T
163where
164    T: Transport<R, Error = E> + Send + 'static,
165    R: ServiceRole,
166    E: std::error::Error + Send + Sync + 'static,
167{
168    fn into_transport(self) -> impl Transport<R, Error = E> + 'static {
169        self
170    }
171}
172
173/// A transport that can send a single message and then close itself
174pub struct OneshotTransport<R>
175where
176    R: ServiceRole,
177{
178    message: Option<RxJsonRpcMessage<R>>,
179    sender: tokio::sync::mpsc::Sender<TxJsonRpcMessage<R>>,
180    termination: Arc<tokio::sync::Semaphore>,
181}
182
183impl<R> OneshotTransport<R>
184where
185    R: ServiceRole,
186{
187    pub fn new(
188        message: RxJsonRpcMessage<R>,
189    ) -> (Self, tokio::sync::mpsc::Receiver<TxJsonRpcMessage<R>>) {
190        let (sender, receiver) = tokio::sync::mpsc::channel(16);
191        (
192            Self {
193                message: Some(message),
194                sender,
195                termination: Arc::new(tokio::sync::Semaphore::new(0)),
196            },
197            receiver,
198        )
199    }
200}
201
202impl<R> Transport<R> for OneshotTransport<R>
203where
204    R: ServiceRole,
205{
206    type Error = tokio::sync::mpsc::error::SendError<TxJsonRpcMessage<R>>;
207
208    fn send(
209        &mut self,
210        item: TxJsonRpcMessage<R>,
211    ) -> impl Future<Output = Result<(), Self::Error>> + Send + 'static {
212        let sender = self.sender.clone();
213        let terminate = matches!(item, TxJsonRpcMessage::<R>::Response(_))
214            || matches!(item, TxJsonRpcMessage::<R>::Error(_));
215        let termination = self.termination.clone();
216        async move {
217            sender.send(item).await?;
218            if terminate {
219                termination.add_permits(1);
220            }
221            Ok(())
222        }
223    }
224
225    async fn receive(&mut self) -> Option<RxJsonRpcMessage<R>> {
226        if let Some(msg) = self.message.take() {
227            return Some(msg);
228        }
229        let _ = self.termination.acquire().await;
230        None
231    }
232
233    fn close(&mut self) -> impl Future<Output = Result<(), Self::Error>> + Send {
234        self.message.take();
235        std::future::ready(Ok(()))
236    }
237}
238
239#[derive(Debug, thiserror::Error)]
240#[error("Transport [{transport_name}] error: {error}")]
241#[non_exhaustive]
242pub struct DynamicTransportError {
243    pub transport_name: Cow<'static, str>,
244    pub transport_type_id: std::any::TypeId,
245    #[source]
246    pub error: Box<dyn std::error::Error + Send + Sync>,
247}
248
249impl DynamicTransportError {
250    pub fn new<T: Transport<R> + 'static, R: ServiceRole>(e: T::Error) -> Self {
251        Self {
252            transport_name: T::name(),
253            transport_type_id: std::any::TypeId::of::<T>(),
254            error: Box::new(e),
255        }
256    }
257
258    /// Create a `DynamicTransportError` from raw parts.
259    ///
260    /// Unlike [`new`](Self::new), this does not require a concrete [`Transport`] type,
261    /// making it usable in test fixtures and other contexts where a real transport
262    /// implementation is not available.
263    pub fn from_parts(
264        transport_name: impl Into<Cow<'static, str>>,
265        transport_type_id: std::any::TypeId,
266        error: Box<dyn std::error::Error + Send + Sync>,
267    ) -> Self {
268        Self {
269            transport_name: transport_name.into(),
270            transport_type_id,
271            error,
272        }
273    }
274
275    pub(crate) fn is_authorization_required(&self) -> bool {
276        let mut error = Some(self.error.as_ref() as &(dyn std::error::Error + 'static));
277        while let Some(current) = error {
278            #[cfg(feature = "auth")]
279            if matches!(
280                current.downcast_ref::<auth::AuthError>(),
281                Some(auth::AuthError::AuthorizationRequired)
282            ) {
283                return true;
284            }
285
286            #[cfg(feature = "transport-streamable-http-client")]
287            if current.is::<streamable_http_client::AuthRequiredError>() {
288                return true;
289            }
290
291            error = current.source();
292        }
293        false
294    }
295
296    pub fn downcast<T: Transport<R> + 'static, R: ServiceRole>(self) -> Result<T::Error, Self> {
297        if !self.is::<T, R>() {
298            Err(self)
299        } else {
300            Ok(self
301                .error
302                .downcast::<T::Error>()
303                .map(|e| *e)
304                .expect("type is checked"))
305        }
306    }
307    pub fn is<T: Transport<R> + 'static, R: ServiceRole>(&self) -> bool {
308        self.error.is::<T::Error>() && self.transport_type_id == std::any::TypeId::of::<T>()
309    }
310}
rmcp/transport.rs

rmcp/
transport.rs
