rmcp_soddygo/
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, AuthorizationSession, AuthorizedHttpClient,
103    ClientCredentialsConfig, CredentialStore, EXTENSION_OAUTH_CLIENT_CREDENTIALS,
104    InMemoryCredentialStore, InMemoryStateStore, OAuthHttpClient, OAuthHttpClientError,
105    OAuthHttpClientFuture, OAuthHttpRedirectPolicy, OAuthHttpRequest, ScopeUpgradeConfig,
106    StateStore, StoredAuthorizationState, StoredCredentials, WWWAuthenticateParams,
107};
108
109// #[cfg(feature = "transport-ws")]
110// pub mod ws;
111#[cfg(feature = "transport-streamable-http-server-session")]
112pub mod streamable_http_server;
113#[cfg(all(feature = "transport-streamable-http-server", not(feature = "local")))]
114pub use streamable_http_server::tower::{StreamableHttpServerConfig, StreamableHttpService};
115
116#[cfg(feature = "transport-streamable-http-client")]
117pub mod streamable_http_client;
118#[cfg(all(unix, feature = "transport-streamable-http-client-unix-socket"))]
119pub use common::unix_socket::UnixSocketHttpClient;
120#[cfg(feature = "transport-streamable-http-client")]
121pub use streamable_http_client::StreamableHttpClientTransport;
122
123/// Common use codes
124pub mod common;
125
126pub trait Transport<R>: Send
127where
128    R: ServiceRole,
129{
130    type Error: std::error::Error + Send + Sync + 'static;
131    fn name() -> Cow<'static, str> {
132        std::any::type_name::<Self>().into()
133    }
134    /// Send a message to the transport
135    ///
136    /// Notice that the future returned by this function should be `Send` and `'static`.
137    /// It's because the sending message could be executed concurrently.
138    ///
139    fn send(
140        &mut self,
141        item: TxJsonRpcMessage<R>,
142    ) -> impl Future<Output = Result<(), Self::Error>> + Send + 'static;
143
144    /// Receive a message from the transport, this operation is sequential.
145    fn receive(&mut self) -> impl Future<Output = Option<RxJsonRpcMessage<R>>> + Send;
146
147    /// Close the transport
148    fn close(&mut self) -> impl Future<Output = Result<(), Self::Error>> + Send;
149}
150
151pub trait IntoTransport<R, E, A>: Send + 'static
152where
153    R: ServiceRole,
154    E: std::error::Error + Send + 'static,
155{
156    fn into_transport(self) -> impl Transport<R, Error = E> + 'static;
157}
158
159#[non_exhaustive]
160pub enum TransportAdapterIdentity {}
161impl<R, T, E> IntoTransport<R, E, TransportAdapterIdentity> for T
162where
163    T: Transport<R, Error = E> + Send + 'static,
164    R: ServiceRole,
165    E: std::error::Error + Send + Sync + 'static,
166{
167    fn into_transport(self) -> impl Transport<R, Error = E> + 'static {
168        self
169    }
170}
171
172/// A transport that can send a single message and then close itself
173pub struct OneshotTransport<R>
174where
175    R: ServiceRole,
176{
177    message: Option<RxJsonRpcMessage<R>>,
178    sender: tokio::sync::mpsc::Sender<TxJsonRpcMessage<R>>,
179    termination: Arc<tokio::sync::Semaphore>,
180}
181
182impl<R> OneshotTransport<R>
183where
184    R: ServiceRole,
185{
186    pub fn new(
187        message: RxJsonRpcMessage<R>,
188    ) -> (Self, tokio::sync::mpsc::Receiver<TxJsonRpcMessage<R>>) {
189        let (sender, receiver) = tokio::sync::mpsc::channel(16);
190        (
191            Self {
192                message: Some(message),
193                sender,
194                termination: Arc::new(tokio::sync::Semaphore::new(0)),
195            },
196            receiver,
197        )
198    }
199}
200
201impl<R> Transport<R> for OneshotTransport<R>
202where
203    R: ServiceRole,
204{
205    type Error = tokio::sync::mpsc::error::SendError<TxJsonRpcMessage<R>>;
206
207    fn send(
208        &mut self,
209        item: TxJsonRpcMessage<R>,
210    ) -> impl Future<Output = Result<(), Self::Error>> + Send + 'static {
211        let sender = self.sender.clone();
212        let terminate = matches!(item, TxJsonRpcMessage::<R>::Response(_))
213            || matches!(item, TxJsonRpcMessage::<R>::Error(_));
214        let termination = self.termination.clone();
215        async move {
216            sender.send(item).await?;
217            if terminate {
218                termination.add_permits(1);
219            }
220            Ok(())
221        }
222    }
223
224    async fn receive(&mut self) -> Option<RxJsonRpcMessage<R>> {
225        if let Some(msg) = self.message.take() {
226            return Some(msg);
227        }
228        let _ = self.termination.acquire().await;
229        None
230    }
231
232    fn close(&mut self) -> impl Future<Output = Result<(), Self::Error>> + Send {
233        self.message.take();
234        std::future::ready(Ok(()))
235    }
236}
237
238#[derive(Debug, thiserror::Error)]
239#[error("Transport [{transport_name}] error: {error}")]
240#[non_exhaustive]
241pub struct DynamicTransportError {
242    pub transport_name: Cow<'static, str>,
243    pub transport_type_id: std::any::TypeId,
244    #[source]
245    pub error: Box<dyn std::error::Error + Send + Sync>,
246}
247
248impl DynamicTransportError {
249    pub fn new<T: Transport<R> + 'static, R: ServiceRole>(e: T::Error) -> Self {
250        Self {
251            transport_name: T::name(),
252            transport_type_id: std::any::TypeId::of::<T>(),
253            error: Box::new(e),
254        }
255    }
256
257    /// Create a `DynamicTransportError` from raw parts.
258    ///
259    /// Unlike [`new`](Self::new), this does not require a concrete [`Transport`] type,
260    /// making it usable in test fixtures and other contexts where a real transport
261    /// implementation is not available.
262    pub fn from_parts(
263        transport_name: impl Into<Cow<'static, str>>,
264        transport_type_id: std::any::TypeId,
265        error: Box<dyn std::error::Error + Send + Sync>,
266    ) -> Self {
267        Self {
268            transport_name: transport_name.into(),
269            transport_type_id,
270            error,
271        }
272    }
273
274    pub fn downcast<T: Transport<R> + 'static, R: ServiceRole>(self) -> Result<T::Error, Self> {
275        if !self.is::<T, R>() {
276            Err(self)
277        } else {
278            Ok(self
279                .error
280                .downcast::<T::Error>()
281                .map(|e| *e)
282                .expect("type is checked"))
283        }
284    }
285    pub fn is<T: Transport<R> + 'static, R: ServiceRole>(&self) -> bool {
286        self.error.is::<T::Error>() && self.transport_type_id == std::any::TypeId::of::<T>()
287    }
288}
rmcp_soddygo/transport.rs

rmcp_soddygo/
transport.rs
