From e615300feae62a01f90f40fc7711ab15c3f0a88c Mon Sep 17 00:00:00 2001 From: 4t145 Date: Wed, 9 Jul 2025 11:23:39 +0800 Subject: [PATCH] feat: unified error type (#308) * feat: unified error type * fix: use type id to compare transport type --- crates/rmcp-macros/README.md | 4 +- crates/rmcp-macros/src/lib.rs | 4 +- crates/rmcp-macros/src/tool_handler.rs | 4 +- crates/rmcp/README.md | 2 +- crates/rmcp/src/error.rs | 47 ++++++++++- crates/rmcp/src/handler/client.rs | 2 +- crates/rmcp/src/handler/server.rs | 2 +- crates/rmcp/src/handler/server/router.rs | 4 +- crates/rmcp/src/handler/server/router/tool.rs | 6 +- crates/rmcp/src/handler/server/tool.rs | 84 ++++++++++++------- crates/rmcp/src/lib.rs | 5 +- crates/rmcp/src/model/content.rs | 6 +- crates/rmcp/src/service.rs | 48 +++++------ crates/rmcp/src/service/client.rs | 83 ++++++++++-------- crates/rmcp/src/service/server.rs | 74 +++++++++------- crates/rmcp/src/service/tower.rs | 6 +- crates/rmcp/src/transport.rs | 38 ++++++++- crates/rmcp/tests/common/handlers.rs | 2 +- crates/rmcp/tests/test_complex_schema.rs | 2 +- crates/rmcp/tests/test_notification.rs | 2 +- crates/rmcp/tests/test_progress_subscriber.rs | 7 +- examples/clients/src/git_stdio.rs | 13 +-- examples/servers/src/common/counter.rs | 2 +- 23 files changed, 286 insertions(+), 161 deletions(-) diff --git a/crates/rmcp-macros/README.md b/crates/rmcp-macros/README.md index 03ef93c..fd7222c 100644 --- a/crates/rmcp-macros/README.md +++ b/crates/rmcp-macros/README.md @@ -134,7 +134,7 @@ impl ServerHandler for MyToolHandler { &self, request: CallToolRequestParam, context: RequestContext, - ) -> Result { + ) -> Result { let tcc = ToolCallContext::new(self, request, context); self.tool_router.call(tcc).await } @@ -143,7 +143,7 @@ impl ServerHandler for MyToolHandler { &self, _request: Option, _context: RequestContext, - ) -> Result { + ) -> Result { let items = self.tool_router.list_all(); Ok(ListToolsResult::with_all_items(items)) } diff --git a/crates/rmcp-macros/src/lib.rs b/crates/rmcp-macros/src/lib.rs index ddd1e7d..57e175b 100644 --- a/crates/rmcp-macros/src/lib.rs +++ b/crates/rmcp-macros/src/lib.rs @@ -139,7 +139,7 @@ pub fn tool_router(attr: TokenStream, input: TokenStream) -> TokenStream { /// &self, /// request: CallToolRequestParam, /// context: RequestContext, -/// ) -> Result { +/// ) -> Result { /// let tcc = ToolCallContext::new(self, request, context); /// self.tool_router.call(tcc).await /// } @@ -148,7 +148,7 @@ pub fn tool_router(attr: TokenStream, input: TokenStream) -> TokenStream { /// &self, /// _request: Option, /// _context: RequestContext, -/// ) -> Result { +/// ) -> Result { /// let items = self.tool_router.list_all(); /// Ok(ListToolsResult::with_all_items(items)) /// } diff --git a/crates/rmcp-macros/src/tool_handler.rs b/crates/rmcp-macros/src/tool_handler.rs index c6b168e..4e56946 100644 --- a/crates/rmcp-macros/src/tool_handler.rs +++ b/crates/rmcp-macros/src/tool_handler.rs @@ -29,7 +29,7 @@ pub fn tool_handler(attr: TokenStream, input: TokenStream) -> syn::Result, - ) -> Result { + ) -> Result { let tcc = rmcp::handler::server::tool::ToolCallContext::new(self, request, context); #router.call(tcc).await } @@ -39,7 +39,7 @@ pub fn tool_handler(attr: TokenStream, input: TokenStream) -> syn::Result, _context: rmcp::service::RequestContext, - ) -> Result { + ) -> Result { Ok(rmcp::model::ListToolsResult::with_all_items(#router.list_all())) } }; diff --git a/crates/rmcp/README.md b/crates/rmcp/README.md index eb93c2d..578a228 100644 --- a/crates/rmcp/README.md +++ b/crates/rmcp/README.md @@ -15,7 +15,7 @@ wait for the first release. Creating a server with tools is simple using the `#[tool]` macro: ```rust, ignore -use rmcp::{Error as McpError, ServiceExt, model::*, tool, tool_handler, tool_router, transport::stdio, handler::server::router::tool::ToolRouter}; +use rmcp::{ErrorData as McpError, ServiceExt, model::*, tool, tool_router, transport::stdio, handler::server::tool::ToolCallContext, handler::server::router::tool::ToolRouter}; use std::sync::Arc; use tokio::sync::Mutex; diff --git a/crates/rmcp/src/error.rs b/crates/rmcp/src/error.rs index 76d9f80..e0da2b3 100644 --- a/crates/rmcp/src/error.rs +++ b/crates/rmcp/src/error.rs @@ -1,9 +1,11 @@ -use std::fmt::Display; - -use crate::model::ErrorData; +use std::{borrow::Cow, fmt::Display}; +use crate::ServiceError; +pub use crate::model::ErrorData; +#[deprecated( + note = "Use `rmcp::ErrorData` instead, `rmcp::ErrorData` could become `RmcpError` in the future." +)] pub type Error = ErrorData; - impl Display for ErrorData { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { write!(f, "{}: {}", self.code.0, self.message)?; @@ -15,3 +17,40 @@ impl Display for ErrorData { } impl std::error::Error for ErrorData {} + +/// This is an unified error type for the errors could be returned by the service. +#[derive(Debug, thiserror::Error)] +pub enum RmcpError { + #[error("Service error: {0}")] + Service(#[from] ServiceError), + #[cfg(feature = "client")] + #[error("Client initialization error: {0}")] + ClientInitialize(#[from] crate::service::ClientInitializeError), + #[cfg(feature = "server")] + #[error("Server initialization error: {0}")] + ServerInitialize(#[from] crate::service::ServerInitializeError), + #[error("Runtime error: {0}")] + Runtime(#[from] tokio::task::JoinError), + #[error("Transport creation error: {error}")] + // TODO: Maybe we can introduce something like `TryIntoTransport` to auto wrap transport type, + // but it could be an breaking change, so we could do it in the future. + TransportCreation { + into_transport_type_name: Cow<'static, str>, + into_transport_type_id: std::any::TypeId, + #[source] + error: Box, + }, + // and cancellation shouldn't be an error? +} + +impl RmcpError { + pub fn transport_creation( + error: impl Into>, + ) -> Self { + RmcpError::TransportCreation { + into_transport_type_id: std::any::TypeId::of::(), + into_transport_type_name: std::any::type_name::().into(), + error: error.into(), + } + } +} diff --git a/crates/rmcp/src/handler/client.rs b/crates/rmcp/src/handler/client.rs index 37203a6..40daaf3 100644 --- a/crates/rmcp/src/handler/client.rs +++ b/crates/rmcp/src/handler/client.rs @@ -1,6 +1,6 @@ pub mod progress; use crate::{ - error::Error as McpError, + error::ErrorData as McpError, model::*, service::{NotificationContext, RequestContext, RoleClient, Service, ServiceRole}, }; diff --git a/crates/rmcp/src/handler/server.rs b/crates/rmcp/src/handler/server.rs index ed74a80..13bb69f 100644 --- a/crates/rmcp/src/handler/server.rs +++ b/crates/rmcp/src/handler/server.rs @@ -1,5 +1,5 @@ use crate::{ - error::Error as McpError, + error::ErrorData as McpError, model::*, service::{NotificationContext, RequestContext, RoleServer, Service, ServiceRole}, }; diff --git a/crates/rmcp/src/handler/server/router.rs b/crates/rmcp/src/handler/server/router.rs index ea972f5..7401b39 100644 --- a/crates/rmcp/src/handler/server/router.rs +++ b/crates/rmcp/src/handler/server/router.rs @@ -51,7 +51,7 @@ where &self, notification: ::PeerNot, context: NotificationContext, - ) -> Result<(), crate::Error> { + ) -> Result<(), crate::ErrorData> { self.service .handle_notification(notification, context) .await @@ -60,7 +60,7 @@ where &self, request: ::PeerReq, context: crate::service::RequestContext, - ) -> Result<::Resp, crate::Error> { + ) -> Result<::Resp, crate::ErrorData> { match request { ClientRequest::CallToolRequest(request) => { if self.tool_router.has_route(request.params.name.as_ref()) diff --git a/crates/rmcp/src/handler/server/router/tool.rs b/crates/rmcp/src/handler/server/router/tool.rs index bb5fda2..9735d28 100644 --- a/crates/rmcp/src/handler/server/router/tool.rs +++ b/crates/rmcp/src/handler/server/router/tool.rs @@ -52,7 +52,7 @@ impl ToolRoute { where C: for<'a> Fn( ToolCallContext<'a, S>, - ) -> BoxFuture<'a, Result> + ) -> BoxFuture<'a, Result> + Send + Sync + 'static, @@ -237,11 +237,11 @@ where pub async fn call( &self, context: ToolCallContext<'_, S>, - ) -> Result { + ) -> Result { let item = self .map .get(context.name()) - .ok_or_else(|| crate::Error::invalid_params("tool not found", None))?; + .ok_or_else(|| crate::ErrorData::invalid_params("tool not found", None))?; (item.call)(context).await } diff --git a/crates/rmcp/src/handler/server/tool.rs b/crates/rmcp/src/handler/server/tool.rs index 5f33a74..3eefb06 100644 --- a/crates/rmcp/src/handler/server/tool.rs +++ b/crates/rmcp/src/handler/server/tool.rs @@ -56,9 +56,9 @@ pub fn cached_schema_for_type() -> Arc(input: JsonObject) -> Result { +pub fn parse_json_object(input: JsonObject) -> Result { serde_json::from_value(serde_json::Value::Object(input)).map_err(|e| { - crate::Error::invalid_params( + crate::ErrorData::invalid_params( format!("failed to deserialize parameters: {error}", error = e), None, ) @@ -93,21 +93,23 @@ impl<'s, S> ToolCallContext<'s, S> { } pub trait FromToolCallContextPart: Sized { - fn from_tool_call_context_part(context: &mut ToolCallContext) -> Result; + fn from_tool_call_context_part( + context: &mut ToolCallContext, + ) -> Result; } pub trait IntoCallToolResult { - fn into_call_tool_result(self) -> Result; + fn into_call_tool_result(self) -> Result; } impl IntoCallToolResult for T { - fn into_call_tool_result(self) -> Result { + fn into_call_tool_result(self) -> Result { Ok(CallToolResult::success(self.into_contents())) } } impl IntoCallToolResult for Result { - fn into_call_tool_result(self) -> Result { + fn into_call_tool_result(self) -> Result { match self { Ok(value) => Ok(CallToolResult::success(value.into_contents())), Err(error) => Ok(CallToolResult::error(error.into_contents())), @@ -115,8 +117,8 @@ impl IntoCallToolResult for Result { } } -impl IntoCallToolResult for Result { - fn into_call_tool_result(self) -> Result { +impl IntoCallToolResult for Result { + fn into_call_tool_result(self) -> Result { match self { Ok(value) => value.into_call_tool_result(), Err(error) => Err(error), @@ -134,7 +136,7 @@ pin_project_lite::pin_project! { }, Ready { #[pin] - result: Ready>, + result: Ready>, } } } @@ -144,7 +146,7 @@ where F: Future, R: IntoCallToolResult, { - type Output = Result; + type Output = Result; fn poll( self: std::pin::Pin<&mut Self>, @@ -159,8 +161,8 @@ where } } -impl IntoCallToolResult for Result { - fn into_call_tool_result(self) -> Result { +impl IntoCallToolResult for Result { + fn into_call_tool_result(self) -> Result { self } } @@ -169,10 +171,10 @@ pub trait CallToolHandler { fn call( self, context: ToolCallContext<'_, S>, - ) -> BoxFuture<'_, Result>; + ) -> BoxFuture<'_, Result>; } -pub type DynCallToolHandler = dyn for<'s> Fn(ToolCallContext<'s, S>) -> BoxFuture<'s, Result> +pub type DynCallToolHandler = dyn for<'s> Fn(ToolCallContext<'s, S>) -> BoxFuture<'s, Result> + Send + Sync; @@ -193,7 +195,9 @@ impl JsonSchema for Parameters

{ } impl FromToolCallContextPart for CancellationToken { - fn from_tool_call_context_part(context: &mut ToolCallContext) -> Result { + fn from_tool_call_context_part( + context: &mut ToolCallContext, + ) -> Result { Ok(context.request_context.ct.clone()) } } @@ -201,7 +205,9 @@ impl FromToolCallContextPart for CancellationToken { pub struct ToolName(pub Cow<'static, str>); impl FromToolCallContextPart for ToolName { - fn from_tool_call_context_part(context: &mut ToolCallContext) -> Result { + fn from_tool_call_context_part( + context: &mut ToolCallContext, + ) -> Result { Ok(Self(context.name.clone())) } } @@ -210,11 +216,13 @@ impl FromToolCallContextPart for Parameters

where P: DeserializeOwned, { - fn from_tool_call_context_part(context: &mut ToolCallContext) -> Result { + fn from_tool_call_context_part( + context: &mut ToolCallContext, + ) -> Result { let arguments = context.arguments.take().unwrap_or_default(); let value: P = serde_json::from_value(serde_json::Value::Object(arguments)).map_err(|e| { - crate::Error::invalid_params( + crate::ErrorData::invalid_params( format!("failed to deserialize parameters: {error}", error = e), None, ) @@ -224,14 +232,18 @@ where } impl FromToolCallContextPart for JsonObject { - fn from_tool_call_context_part(context: &mut ToolCallContext) -> Result { + fn from_tool_call_context_part( + context: &mut ToolCallContext, + ) -> Result { let object = context.arguments.take().unwrap_or_default(); Ok(object) } } impl FromToolCallContextPart for crate::model::Extensions { - fn from_tool_call_context_part(context: &mut ToolCallContext) -> Result { + fn from_tool_call_context_part( + context: &mut ToolCallContext, + ) -> Result { let extensions = context.request_context.extensions.clone(); Ok(extensions) } @@ -243,14 +255,16 @@ impl FromToolCallContextPart for Extension where T: Send + Sync + 'static + Clone, { - fn from_tool_call_context_part(context: &mut ToolCallContext) -> Result { + fn from_tool_call_context_part( + context: &mut ToolCallContext, + ) -> Result { let extension = context .request_context .extensions .get::() .cloned() .ok_or_else(|| { - crate::Error::invalid_params( + crate::ErrorData::invalid_params( format!("missing extension {}", std::any::type_name::()), None, ) @@ -260,14 +274,18 @@ where } impl FromToolCallContextPart for crate::Peer { - fn from_tool_call_context_part(context: &mut ToolCallContext) -> Result { + fn from_tool_call_context_part( + context: &mut ToolCallContext, + ) -> Result { let peer = context.request_context.peer.clone(); Ok(peer) } } impl FromToolCallContextPart for crate::model::Meta { - fn from_tool_call_context_part(context: &mut ToolCallContext) -> Result { + fn from_tool_call_context_part( + context: &mut ToolCallContext, + ) -> Result { let mut meta = crate::model::Meta::default(); std::mem::swap(&mut meta, &mut context.request_context.meta); Ok(meta) @@ -276,19 +294,23 @@ impl FromToolCallContextPart for crate::model::Meta { pub struct RequestId(pub crate::model::RequestId); impl FromToolCallContextPart for RequestId { - fn from_tool_call_context_part(context: &mut ToolCallContext) -> Result { + fn from_tool_call_context_part( + context: &mut ToolCallContext, + ) -> Result { Ok(RequestId(context.request_context.id.clone())) } } impl FromToolCallContextPart for RequestContext { - fn from_tool_call_context_part(context: &mut ToolCallContext) -> Result { + fn from_tool_call_context_part( + context: &mut ToolCallContext, + ) -> Result { Ok(context.request_context.clone()) } } impl<'s, S> ToolCallContext<'s, S> { - pub fn invoke(self, h: H) -> BoxFuture<'s, Result> + pub fn invoke(self, h: H) -> BoxFuture<'s, Result> where H: CallToolHandler, { @@ -331,7 +353,7 @@ macro_rules! impl_for { fn call( self, mut context: ToolCallContext<'_, S>, - ) -> BoxFuture<'_, Result>{ + ) -> BoxFuture<'_, Result>{ $( let result = $Tn::from_tool_call_context_part(&mut context); let $Tn = match result { @@ -362,7 +384,7 @@ macro_rules! impl_for { fn call( self, mut context: ToolCallContext, - ) -> BoxFuture<'static, Result>{ + ) -> BoxFuture<'static, Result>{ $( let result = $Tn::from_tool_call_context_part(&mut context); let $Tn = match result { @@ -391,7 +413,7 @@ macro_rules! impl_for { fn call( self, mut context: ToolCallContext, - ) -> BoxFuture<'static, Result> { + ) -> BoxFuture<'static, Result> { $( let result = $Tn::from_tool_call_context_part(&mut context); let $Tn = match result { @@ -416,7 +438,7 @@ macro_rules! impl_for { fn call( self, mut context: ToolCallContext, - ) -> BoxFuture<'static, Result> { + ) -> BoxFuture<'static, Result> { $( let result = $Tn::from_tool_call_context_part(&mut context); let $Tn = match result { diff --git a/crates/rmcp/src/lib.rs b/crates/rmcp/src/lib.rs index a3d1a41..d5dbaad 100644 --- a/crates/rmcp/src/lib.rs +++ b/crates/rmcp/src/lib.rs @@ -19,7 +19,7 @@ //! //! ```rust //! use std::sync::Arc; -//! use rmcp::{Error as McpError, model::*, tool, tool_router, handler::server::tool::ToolRouter}; +//! use rmcp::{ErrorData as McpError, model::*, tool, tool_router, handler::server::tool::ToolRouter}; //! use tokio::sync::Mutex; //! //! #[derive(Clone)] @@ -90,7 +90,8 @@ //! } //! ``` mod error; -pub use error::Error; +#[allow(deprecated)] +pub use error::{Error, ErrorData, RmcpError}; /// Basic data types in MCP specification pub mod model; diff --git a/crates/rmcp/src/model/content.rs b/crates/rmcp/src/model/content.rs index eab5f6e..7c75381 100644 --- a/crates/rmcp/src/model/content.rs +++ b/crates/rmcp/src/model/content.rs @@ -63,9 +63,9 @@ pub enum RawContent { pub type Content = Annotated; impl RawContent { - pub fn json(json: S) -> Result { + pub fn json(json: S) -> Result { let json = serde_json::to_string(&json).map_err(|e| { - crate::Error::internal_error( + crate::ErrorData::internal_error( "fail to serialize response to json", Some(json!( {"reason": e.to_string()} @@ -142,7 +142,7 @@ impl Content { RawContent::embedded_text(uri, content).no_annotation() } - pub fn json(json: S) -> Result { + pub fn json(json: S) -> Result { RawContent::json(json).map(|c| c.no_annotation()) } } diff --git a/crates/rmcp/src/service.rs b/crates/rmcp/src/service.rs index 306c4c9..e1e72c9 100644 --- a/crates/rmcp/src/service.rs +++ b/crates/rmcp/src/service.rs @@ -2,14 +2,14 @@ use futures::{FutureExt, future::BoxFuture}; use thiserror::Error; use crate::{ - error::Error as McpError, + error::ErrorData as McpError, model::{ CancelledNotification, CancelledNotificationParam, Extensions, GetExtensions, GetMeta, JsonRpcBatchRequestItem, JsonRpcBatchResponseItem, JsonRpcError, JsonRpcMessage, JsonRpcNotification, JsonRpcRequest, JsonRpcResponse, Meta, NumberOrString, ProgressToken, RequestId, ServerJsonRpcMessage, }, - transport::{IntoTransport, Transport}, + transport::{DynamicTransportError, IntoTransport, Transport}, }; #[cfg(feature = "client")] #[cfg_attr(docsrs, doc(cfg(feature = "client")))] @@ -37,7 +37,7 @@ pub enum ServiceError { #[error("Mcp error: {0}")] McpError(McpError), #[error("Transport send error: {0}")] - TransportSend(Box), + TransportSend(DynamicTransportError), #[error("Transport closed")] TransportClosed, #[error("Unexpected response type")] @@ -48,7 +48,6 @@ pub enum ServiceError { Timeout { timeout: Duration }, } -impl ServiceError {} trait TransferObject: std::fmt::Debug + Clone + serde::Serialize + serde::de::DeserializeOwned + Send + Sync + 'static { @@ -79,7 +78,7 @@ pub trait ServiceRole: std::fmt::Debug + Send + Sync + 'static + Copy + Clone { + TransferObject + GetMeta + GetExtensions; - type InitializeError; + type InitializeError; const IS_CLIENT: bool; type Info: TransferObject; type PeerInfo: TransferObject; @@ -117,10 +116,10 @@ pub trait ServiceExt: Service + Sized { fn serve( self, transport: T, - ) -> impl Future, R::InitializeError>> + Send + ) -> impl Future, R::InitializeError>> + Send where T: IntoTransport, - E: std::error::Error + From + Send + Sync + 'static, + E: std::error::Error + Send + Sync + 'static, Self: Sized, { Self::serve_with_ct(self, transport, Default::default()) @@ -129,10 +128,10 @@ pub trait ServiceExt: Service + Sized { self, transport: T, ct: CancellationToken, - ) -> impl Future, R::InitializeError>> + Send + ) -> impl Future, R::InitializeError>> + Send where T: IntoTransport, - E: std::error::Error + From + Send + Sync + 'static, + E: std::error::Error + Send + Sync + 'static, Self: Sized; } @@ -537,11 +536,11 @@ where E: std::error::Error + Send + Sync + 'static, { let (peer, peer_rx) = Peer::new(Arc::new(AtomicU32RequestIdProvider::default()), peer_info); - serve_inner(service, transport, peer, peer_rx, ct) + serve_inner(service, transport.into_transport(), peer, peer_rx, ct) } #[instrument(skip_all)] -fn serve_inner( +fn serve_inner( service: S, transport: T, peer: Peer, @@ -551,8 +550,7 @@ fn serve_inner( where R: ServiceRole, S: Service, - T: IntoTransport, - E: std::error::Error + Send + Sync + 'static, + T: Transport + 'static, { const SINK_PROXY_BUFFER_SIZE: usize = 64; let (sink_proxy_tx, mut sink_proxy_rx) = @@ -578,25 +576,25 @@ where let handle = tokio::spawn(async move { let mut transport = transport.into_transport(); let mut batch_messages = VecDeque::>::new(); - let mut send_task_set = tokio::task::JoinSet::>::new(); + let mut send_task_set = tokio::task::JoinSet::::new(); #[derive(Debug)] - enum SendTaskResult { + enum SendTaskResult { Request { id: RequestId, - result: Result<(), E>, + result: Result<(), DynamicTransportError>, }, Notification { responder: Responder>, cancellation_param: Option, - result: Result<(), E>, + result: Result<(), DynamicTransportError>, }, } #[derive(Debug)] - enum Event { + enum Event { ProxyMessage(PeerSinkMessage), PeerMessage(RxJsonRpcMessage), ToSink(TxJsonRpcMessage), - SendTaskResult(SendTaskResult), + SendTaskResult(SendTaskResult), } let quit_reason = loop { @@ -654,7 +652,7 @@ where Event::SendTaskResult(SendTaskResult::Request { id, result }) => { if let Err(e) = result { if let Some(responder) = local_responder_pool.remove(&id) { - let _ = responder.send(Err(ServiceError::TransportSend(Box::new(e)))); + let _ = responder.send(Err(ServiceError::TransportSend(e))); } } } @@ -664,7 +662,7 @@ where cancellation_param, }) => { let response = if let Err(e) = result { - Err(ServiceError::TransportSend(Box::new(e))) + Err(ServiceError::TransportSend(e)) } else { Ok(()) }; @@ -706,8 +704,10 @@ where let send = transport.send(JsonRpcMessage::request(request, id.clone())); { let id = id.clone(); - send_task_set - .spawn(send.map(move |r| SendTaskResult::Request { id, result: r })); + send_task_set.spawn(send.map(move |r| SendTaskResult::Request { + id, + result: r.map_err(DynamicTransportError::new::), + })); } } Event::ProxyMessage(PeerSinkMessage::Notification { @@ -727,7 +727,7 @@ where send_task_set.spawn(send.map(move |result| SendTaskResult::Notification { responder, cancellation_param, - result, + result: result.map_err(DynamicTransportError::new::), })); } Event::PeerMessage(JsonRpcMessage::Request(JsonRpcRequest { diff --git a/crates/rmcp/src/service/client.rs b/crates/rmcp/src/service/client.rs index 3de4609..dba7d04 100644 --- a/crates/rmcp/src/service/client.rs +++ b/crates/rmcp/src/service/client.rs @@ -3,25 +3,29 @@ use std::borrow::Cow; use thiserror::Error; use super::*; -use crate::model::{ - CallToolRequest, CallToolRequestParam, CallToolResult, CancelledNotification, - CancelledNotificationParam, ClientInfo, ClientJsonRpcMessage, ClientNotification, - ClientRequest, ClientResult, CompleteRequest, CompleteRequestParam, CompleteResult, - GetPromptRequest, GetPromptRequestParam, GetPromptResult, InitializeRequest, - InitializedNotification, JsonRpcResponse, ListPromptsRequest, ListPromptsResult, - ListResourceTemplatesRequest, ListResourceTemplatesResult, ListResourcesRequest, - ListResourcesResult, ListToolsRequest, ListToolsResult, PaginatedRequestParam, - ProgressNotification, ProgressNotificationParam, ReadResourceRequest, ReadResourceRequestParam, - ReadResourceResult, RequestId, RootsListChangedNotification, ServerInfo, ServerJsonRpcMessage, - ServerNotification, ServerRequest, ServerResult, SetLevelRequest, SetLevelRequestParam, - SubscribeRequest, SubscribeRequestParam, UnsubscribeRequest, UnsubscribeRequestParam, +use crate::{ + model::{ + CallToolRequest, CallToolRequestParam, CallToolResult, CancelledNotification, + CancelledNotificationParam, ClientInfo, ClientJsonRpcMessage, ClientNotification, + ClientRequest, ClientResult, CompleteRequest, CompleteRequestParam, CompleteResult, + GetPromptRequest, GetPromptRequestParam, GetPromptResult, InitializeRequest, + InitializedNotification, JsonRpcResponse, ListPromptsRequest, ListPromptsResult, + ListResourceTemplatesRequest, ListResourceTemplatesResult, ListResourcesRequest, + ListResourcesResult, ListToolsRequest, ListToolsResult, PaginatedRequestParam, + ProgressNotification, ProgressNotificationParam, ReadResourceRequest, + ReadResourceRequestParam, ReadResourceResult, RequestId, RootsListChangedNotification, + ServerInfo, ServerJsonRpcMessage, ServerNotification, ServerRequest, ServerResult, + SetLevelRequest, SetLevelRequestParam, SubscribeRequest, SubscribeRequestParam, + UnsubscribeRequest, UnsubscribeRequestParam, + }, + transport::DynamicTransportError, }; /// It represents the error that may occur when serving the client. /// /// if you want to handle the error, you can use `serve_client_with_ct` or `serve_client` with `Result, ClientError>` #[derive(Error, Debug)] -pub enum ClientInitializeError { +pub enum ClientInitializeError { #[error("expect initialized response, but received: {0:?}")] ExpectedInitResponse(Option), @@ -36,7 +40,7 @@ pub enum ClientInitializeError { #[error("Send message error {error}, when {context}")] TransportError { - error: E, + error: DynamicTransportError, context: Cow<'static, str>, }, @@ -44,11 +48,23 @@ pub enum ClientInitializeError { Cancelled, } +impl ClientInitializeError { + pub fn transport + 'static>( + error: T::Error, + context: impl Into>, + ) -> Self { + Self::TransportError { + error: DynamicTransportError::new::(error), + context: context.into(), + } + } +} + /// Helper function to get the next message from the stream -async fn expect_next_message( +async fn expect_next_message( transport: &mut T, context: &str, -) -> Result> +) -> Result where T: Transport, { @@ -59,10 +75,10 @@ where } /// Helper function to expect a response from the stream -async fn expect_response( +async fn expect_response( transport: &mut T, context: &str, -) -> Result<(ServerResult, RequestId), ClientInitializeError> +) -> Result<(ServerResult, RequestId), ClientInitializeError> where T: Transport, { @@ -86,7 +102,7 @@ impl ServiceRole for RoleClient { type PeerNot = ServerNotification; type Info = ClientInfo; type PeerInfo = ServerInfo; - type InitializeError = ClientInitializeError; + type InitializeError = ClientInitializeError; const IS_CLIENT: bool = true; } @@ -97,10 +113,10 @@ impl> ServiceExt for S { self, transport: T, ct: CancellationToken, - ) -> impl Future, ClientInitializeError>> + Send + ) -> impl Future, ClientInitializeError>> + Send where T: IntoTransport, - E: std::error::Error + From + Send + Sync + 'static, + E: std::error::Error + Send + Sync + 'static, Self: Sized, { serve_client_with_ct(self, transport, ct) @@ -110,7 +126,7 @@ impl> ServiceExt for S { pub async fn serve_client( service: S, transport: T, -) -> Result, ClientInitializeError> +) -> Result, ClientInitializeError> where S: Service, T: IntoTransport, @@ -123,29 +139,28 @@ pub async fn serve_client_with_ct( service: S, transport: T, ct: CancellationToken, -) -> Result, ClientInitializeError> +) -> Result, ClientInitializeError> where S: Service, T: IntoTransport, E: std::error::Error + Send + Sync + 'static, { tokio::select! { - result = serve_client_with_ct_inner(service, transport, ct.clone()) => { result } + result = serve_client_with_ct_inner(service, transport.into_transport(), ct.clone()) => { result } _ = ct.cancelled() => { Err(ClientInitializeError::Cancelled) } } } -async fn serve_client_with_ct_inner( +async fn serve_client_with_ct_inner( service: S, transport: T, ct: CancellationToken, -) -> Result, ClientInitializeError> +) -> Result, ClientInitializeError> where S: Service, - T: IntoTransport, - E: std::error::Error + Send + Sync + 'static, + T: Transport + 'static, { let mut transport = transport.into_transport(); let id_provider = >::default(); @@ -164,7 +179,7 @@ where )) .await .map_err(|error| ClientInitializeError::TransportError { - error, + error: DynamicTransportError::new::(error), context: "send initialize request".into(), })?; @@ -188,13 +203,9 @@ where extensions: Default::default(), }), ); - transport - .send(notification) - .await - .map_err(|error| ClientInitializeError::TransportError { - error, - context: "send initialized notification".into(), - })?; + transport.send(notification).await.map_err(|error| { + ClientInitializeError::transport::(error, "send initialized notification") + })?; let (peer, peer_rx) = Peer::new(id_provider, Some(initialize_result)); Ok(serve_inner(service, transport, peer, peer_rx, ct)) } diff --git a/crates/rmcp/src/service/server.rs b/crates/rmcp/src/service/server.rs index b43bdb9..05f6d38 100644 --- a/crates/rmcp/src/service/server.rs +++ b/crates/rmcp/src/service/server.rs @@ -3,14 +3,18 @@ use std::borrow::Cow; use thiserror::Error; use super::*; -use crate::model::{ - CancelledNotification, CancelledNotificationParam, ClientInfo, ClientJsonRpcMessage, - ClientNotification, ClientRequest, ClientResult, CreateMessageRequest, - CreateMessageRequestParam, CreateMessageResult, ErrorData, ListRootsRequest, ListRootsResult, - LoggingMessageNotification, LoggingMessageNotificationParam, ProgressNotification, - ProgressNotificationParam, PromptListChangedNotification, ProtocolVersion, - ResourceListChangedNotification, ResourceUpdatedNotification, ResourceUpdatedNotificationParam, - ServerInfo, ServerNotification, ServerRequest, ServerResult, ToolListChangedNotification, +use crate::{ + model::{ + CancelledNotification, CancelledNotificationParam, ClientInfo, ClientJsonRpcMessage, + ClientNotification, ClientRequest, ClientResult, CreateMessageRequest, + CreateMessageRequestParam, CreateMessageResult, ErrorData, ListRootsRequest, + ListRootsResult, LoggingMessageNotification, LoggingMessageNotificationParam, + ProgressNotification, ProgressNotificationParam, PromptListChangedNotification, + ProtocolVersion, ResourceListChangedNotification, ResourceUpdatedNotification, + ResourceUpdatedNotificationParam, ServerInfo, ServerNotification, ServerRequest, + ServerResult, ToolListChangedNotification, + }, + transport::DynamicTransportError, }; #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] @@ -26,7 +30,7 @@ impl ServiceRole for RoleServer { type Info = ServerInfo; type PeerInfo = ClientInfo; - type InitializeError = ServerInitializeError; + type InitializeError = ServerInitializeError; const IS_CLIENT: bool = false; } @@ -34,7 +38,7 @@ impl ServiceRole for RoleServer { /// /// if you want to handle the error, you can use `serve_server_with_ct` or `serve_server` with `Result, ServerError>` #[derive(Error, Debug)] -pub enum ServerInitializeError { +pub enum ServerInitializeError { #[error("expect initialized request, but received: {0:?}")] ExpectedInitializeRequest(Option), @@ -55,7 +59,7 @@ pub enum ServerInitializeError { #[error("Send message error {error}, when {context}")] TransportError { - error: E, + error: DynamicTransportError, context: Cow<'static, str>, }, @@ -63,6 +67,17 @@ pub enum ServerInitializeError { Cancelled, } +impl ServerInitializeError { + pub fn transport + 'static>( + error: T::Error, + context: impl Into>, + ) -> Self { + Self::TransportError { + error: DynamicTransportError::new::(error), + context: context.into(), + } + } +} pub type ClientSink = Peer; impl> ServiceExt for S { @@ -70,7 +85,7 @@ impl> ServiceExt for S { self, transport: T, ct: CancellationToken, - ) -> impl Future, ServerInitializeError>> + Send + ) -> impl Future, ServerInitializeError>> + Send where T: IntoTransport, E: std::error::Error + Send + Sync + 'static, @@ -83,7 +98,7 @@ impl> ServiceExt for S { pub async fn serve_server( service: S, transport: T, -) -> Result, ServerInitializeError> +) -> Result, ServerInitializeError> where S: Service, T: IntoTransport, @@ -93,10 +108,10 @@ where } /// Helper function to get the next message from the stream -async fn expect_next_message( +async fn expect_next_message( transport: &mut T, context: &str, -) -> Result> +) -> Result where T: Transport, { @@ -107,10 +122,10 @@ where } /// Helper function to expect a request from the stream -async fn expect_request( +async fn expect_request( transport: &mut T, context: &str, -) -> Result<(ClientRequest, RequestId), ServerInitializeError> +) -> Result<(ClientRequest, RequestId), ServerInitializeError> where T: Transport, { @@ -123,10 +138,10 @@ where } /// Helper function to expect a notification from the stream -async fn expect_notification( +async fn expect_notification( transport: &mut T, context: &str, -) -> Result> +) -> Result where T: Transport, { @@ -142,29 +157,28 @@ pub async fn serve_server_with_ct( service: S, transport: T, ct: CancellationToken, -) -> Result, ServerInitializeError> +) -> Result, ServerInitializeError> where S: Service, T: IntoTransport, E: std::error::Error + Send + Sync + 'static, { tokio::select! { - result = serve_server_with_ct_inner(service, transport, ct.clone()) => { result } + result = serve_server_with_ct_inner(service, transport.into_transport(), ct.clone()) => { result } _ = ct.cancelled() => { Err(ServerInitializeError::Cancelled) } } } -async fn serve_server_with_ct_inner( +async fn serve_server_with_ct_inner( service: S, transport: T, ct: CancellationToken, -) -> Result, ServerInitializeError> +) -> Result, ServerInitializeError> where S: Service, - T: IntoTransport, - E: std::error::Error + Send + Sync + 'static, + T: Transport + 'static, { let mut transport = transport.into_transport(); let id_provider = >::default(); @@ -196,9 +210,8 @@ where transport .send(ServerJsonRpcMessage::error(e.clone(), id)) .await - .map_err(|error| ServerInitializeError::TransportError { - error, - context: "sending error response".into(), + .map_err(|error| { + ServerInitializeError::transport::(error, "sending error response") })?; return Err(ServerInitializeError::InitializeFailed(e)); } @@ -219,9 +232,8 @@ where id, )) .await - .map_err(|error| ServerInitializeError::TransportError { - error, - context: "sending initialize response".into(), + .map_err(|error| { + ServerInitializeError::transport::(error, "sending initialize response") })?; // Wait for initialize notification diff --git a/crates/rmcp/src/service/tower.rs b/crates/rmcp/src/service/tower.rs index 454a977..ac4a66f 100644 --- a/crates/rmcp/src/service/tower.rs +++ b/crates/rmcp/src/service/tower.rs @@ -24,14 +24,14 @@ impl TowerHandler { impl Service for TowerHandler where S: TowerService + Sync + Send + Clone + 'static, - S::Error: Into, + S::Error: Into, S::Future: Send, { async fn handle_request( &self, request: R::PeerReq, _context: RequestContext, - ) -> Result { + ) -> Result { let mut service = self.service.clone(); poll_fn(|cx| service.poll_ready(cx)) .await @@ -44,7 +44,7 @@ where &self, _notification: R::PeerNot, _context: NotificationContext, - ) -> impl Future> + Send + '_ { + ) -> impl Future> + Send + '_ { std::future::ready(Ok(())) } diff --git a/crates/rmcp/src/transport.rs b/crates/rmcp/src/transport.rs index 20e6ce7..4cc8476 100644 --- a/crates/rmcp/src/transport.rs +++ b/crates/rmcp/src/transport.rs @@ -64,7 +64,7 @@ //! } //! ``` -use std::sync::Arc; +use std::{borrow::Cow, sync::Arc}; use crate::service::{RxJsonRpcMessage, ServiceRole, TxJsonRpcMessage}; @@ -141,6 +141,9 @@ where R: ServiceRole, { type Error: std::error::Error + Send + Sync + 'static; + fn name() -> Cow<'static, str> { + std::any::type_name::().into() + } /// Send a message to the transport /// /// Notice that the future returned by this function should be `Send` and `'static`. @@ -241,3 +244,36 @@ where std::future::ready(Ok(())) } } + +#[derive(Debug, thiserror::Error)] +#[error("Transport [{transport_name}] error: {error}")] +pub struct DynamicTransportError { + pub transport_name: Cow<'static, str>, + pub transport_type_id: std::any::TypeId, + #[source] + pub error: Box, +} + +impl DynamicTransportError { + pub fn new + 'static, R: ServiceRole>(e: T::Error) -> Self { + Self { + transport_name: T::name(), + transport_type_id: std::any::TypeId::of::(), + error: Box::new(e), + } + } + pub fn downcast + 'static, R: ServiceRole>(self) -> Result { + if !self.is::() { + Err(self) + } else { + Ok(self + .error + .downcast::() + .map(|e| *e) + .expect("type is checked")) + } + } + pub fn is + 'static, R: ServiceRole>(&self) -> bool { + self.error.is::() && self.transport_type_id == std::any::TypeId::of::() + } +} diff --git a/crates/rmcp/tests/common/handlers.rs b/crates/rmcp/tests/common/handlers.rs index 23bb511..db82d47 100644 --- a/crates/rmcp/tests/common/handlers.rs +++ b/crates/rmcp/tests/common/handlers.rs @@ -4,7 +4,7 @@ use std::{ }; use rmcp::{ - ClientHandler, Error as McpError, RoleClient, RoleServer, ServerHandler, + ClientHandler, ErrorData as McpError, RoleClient, RoleServer, ServerHandler, model::*, service::{NotificationContext, RequestContext}, }; diff --git a/crates/rmcp/tests/test_complex_schema.rs b/crates/rmcp/tests/test_complex_schema.rs index c199e7e..1234c83 100644 --- a/crates/rmcp/tests/test_complex_schema.rs +++ b/crates/rmcp/tests/test_complex_schema.rs @@ -1,5 +1,5 @@ use rmcp::{ - Error as McpError, handler::server::tool::Parameters, model::*, schemars, tool, tool_router, + ErrorData as McpError, handler::server::tool::Parameters, model::*, schemars, tool, tool_router, }; use serde::{Deserialize, Serialize}; diff --git a/crates/rmcp/tests/test_notification.rs b/crates/rmcp/tests/test_notification.rs index 0950181..a46ac2f 100644 --- a/crates/rmcp/tests/test_notification.rs +++ b/crates/rmcp/tests/test_notification.rs @@ -27,7 +27,7 @@ impl ServerHandler for Server { &self, request: rmcp::model::SubscribeRequestParam, context: rmcp::service::RequestContext, - ) -> Result<(), rmcp::Error> { + ) -> Result<(), rmcp::ErrorData> { let uri = request.uri; let peer = context.peer; diff --git a/crates/rmcp/tests/test_progress_subscriber.rs b/crates/rmcp/tests/test_progress_subscriber.rs index 566667d..24576c0 100644 --- a/crates/rmcp/tests/test_progress_subscriber.rs +++ b/crates/rmcp/tests/test_progress_subscriber.rs @@ -55,10 +55,13 @@ impl MyServer { } } #[tool] - pub async fn some_progress(meta: Meta, client: Peer) -> Result<(), rmcp::Error> { + pub async fn some_progress( + meta: Meta, + client: Peer, + ) -> Result<(), rmcp::ErrorData> { let progress_token = meta .get_progress_token() - .ok_or(rmcp::Error::invalid_params( + .ok_or(rmcp::ErrorData::invalid_params( "Progress token is required for this tool", None, ))?; diff --git a/examples/clients/src/git_stdio.rs b/examples/clients/src/git_stdio.rs index bda978d..65dbf84 100644 --- a/examples/clients/src/git_stdio.rs +++ b/examples/clients/src/git_stdio.rs @@ -1,5 +1,5 @@ -use anyhow::Result; use rmcp::{ + RmcpError, model::CallToolRequestParam, service::ServiceExt, transport::{ConfigureCommandExt, TokioChildProcess}, @@ -8,7 +8,7 @@ use tokio::process::Command; use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt}; #[tokio::main] -async fn main() -> Result<()> { +async fn main() -> Result<(), RmcpError> { // Initialize logging tracing_subscriber::registry() .with( @@ -18,11 +18,12 @@ async fn main() -> Result<()> { .with(tracing_subscriber::fmt::layer()) .init(); let client = () - .serve(TokioChildProcess::new(Command::new("uvx").configure( - |cmd| { + .serve( + TokioChildProcess::new(Command::new("uvx").configure(|cmd| { cmd.arg("mcp-server-git"); - }, - ))?) + })) + .map_err(RmcpError::transport_creation::)?, + ) .await?; // or serve_client((), TokioChildProcess::new(cmd)?).await?; diff --git a/examples/servers/src/common/counter.rs b/examples/servers/src/common/counter.rs index c205e29..4acacb8 100644 --- a/examples/servers/src/common/counter.rs +++ b/examples/servers/src/common/counter.rs @@ -2,7 +2,7 @@ use std::sync::Arc; use rmcp::{ - Error as McpError, RoleServer, ServerHandler, + ErrorData as McpError, RoleServer, ServerHandler, handler::server::{router::tool::ToolRouter, tool::Parameters}, model::*, schemars,