Swap MCP client implementations to rmcp (#3634)
This commit is contained in:
Jack Amadeo
+240
-386
@@ -1,97 +1,30 @@
|
||||
use mcp_core::protocol::{
|
||||
CallToolResult, Implementation, InitializeResult, ListPromptsResult, ListResourcesResult,
|
||||
ListToolsResult, ReadResourceResult, ServerCapabilities, METHOD_NOT_FOUND,
|
||||
use rmcp::{
|
||||
model::{
|
||||
CallToolRequest, CallToolRequestParam, CallToolResult, ClientCapabilities, ClientInfo,
|
||||
ClientRequest, GetPromptRequest, GetPromptRequestParam, GetPromptResult, Implementation,
|
||||
InitializeResult, ListPromptsRequest, ListPromptsResult, ListResourcesRequest,
|
||||
ListResourcesResult, ListToolsRequest, ListToolsResult, LoggingMessageNotification,
|
||||
LoggingMessageNotificationMethod, PaginatedRequestParam, ProgressNotification,
|
||||
ProgressNotificationMethod, ProtocolVersion, ReadResourceRequest, ReadResourceRequestParam,
|
||||
ReadResourceResult, ServerNotification, ServerResult,
|
||||
},
|
||||
service::{ClientInitializeError, PeerRequestOptions, RunningService},
|
||||
transport::IntoTransport,
|
||||
ClientHandler, RoleClient, ServiceError, ServiceExt,
|
||||
};
|
||||
use rmcp::model::{
|
||||
GetPromptResult, JsonRpcError, JsonRpcMessage, JsonRpcNotification, JsonRpcRequest,
|
||||
JsonRpcResponse, JsonRpcVersion2_0, Notification, NumberOrString, Request, RequestId,
|
||||
ServerNotification,
|
||||
use serde_json::Value;
|
||||
use std::sync::Arc;
|
||||
use tokio::sync::{
|
||||
mpsc::{self, Sender},
|
||||
Mutex,
|
||||
};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use serde_json::{json, Value};
|
||||
use std::sync::{
|
||||
atomic::{AtomicU64, Ordering},
|
||||
Arc,
|
||||
};
|
||||
use thiserror::Error;
|
||||
use tokio::sync::{mpsc, Mutex};
|
||||
use tower::{timeout::TimeoutLayer, Layer, Service, ServiceExt};
|
||||
|
||||
use crate::{McpService, TransportHandle};
|
||||
|
||||
pub type BoxError = Box<dyn std::error::Error + Sync + Send>;
|
||||
|
||||
/// Error type for MCP client operations.
|
||||
#[derive(Debug, Error)]
|
||||
pub enum Error {
|
||||
#[error("Transport error: {0}")]
|
||||
Transport(#[from] super::transport::Error),
|
||||
|
||||
#[error("RPC error: code={code}, message={message}")]
|
||||
RpcError { code: i32, message: String },
|
||||
|
||||
#[error("Serialization error: {0}")]
|
||||
Serialization(#[from] serde_json::Error),
|
||||
|
||||
#[error("Unexpected response from server: {0}")]
|
||||
UnexpectedResponse(String),
|
||||
|
||||
#[error("Not initialized")]
|
||||
NotInitialized,
|
||||
|
||||
#[error("Timeout or service not ready")]
|
||||
NotReady,
|
||||
|
||||
#[error("Request timed out")]
|
||||
Timeout(#[from] tower::timeout::error::Elapsed),
|
||||
|
||||
#[error("Error from mcp-server: {0}")]
|
||||
ServerBoxError(BoxError),
|
||||
|
||||
#[error("Call to '{server}' failed for '{method}'. {source}")]
|
||||
McpServerError {
|
||||
method: String,
|
||||
server: String,
|
||||
#[source]
|
||||
source: BoxError,
|
||||
},
|
||||
}
|
||||
|
||||
// BoxError from mcp-server gets converted to our Error type
|
||||
impl From<BoxError> for Error {
|
||||
fn from(err: BoxError) -> Self {
|
||||
Error::ServerBoxError(err)
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Serialize, Deserialize)]
|
||||
pub struct ClientInfo {
|
||||
pub name: String,
|
||||
pub version: String,
|
||||
}
|
||||
|
||||
#[derive(Serialize, Deserialize, Default)]
|
||||
pub struct ClientCapabilities {
|
||||
// Add fields as needed. For now, empty capabilities are fine.
|
||||
}
|
||||
|
||||
#[derive(Serialize, Deserialize)]
|
||||
pub struct InitializeParams {
|
||||
#[serde(rename = "protocolVersion")]
|
||||
pub protocol_version: String,
|
||||
pub capabilities: ClientCapabilities,
|
||||
#[serde(rename = "clientInfo")]
|
||||
pub client_info: ClientInfo,
|
||||
}
|
||||
pub type Error = rmcp::ServiceError;
|
||||
|
||||
#[async_trait::async_trait]
|
||||
pub trait McpClientTrait: Send + Sync {
|
||||
async fn initialize(
|
||||
&mut self,
|
||||
info: ClientInfo,
|
||||
capabilities: ClientCapabilities,
|
||||
) -> Result<InitializeResult, Error>;
|
||||
|
||||
async fn list_resources(
|
||||
&self,
|
||||
next_cursor: Option<String>,
|
||||
@@ -108,347 +41,268 @@ pub trait McpClientTrait: Send + Sync {
|
||||
async fn get_prompt(&self, name: &str, arguments: Value) -> Result<GetPromptResult, Error>;
|
||||
|
||||
async fn subscribe(&self) -> mpsc::Receiver<ServerNotification>;
|
||||
|
||||
fn get_info(&self) -> Option<&InitializeResult>;
|
||||
}
|
||||
|
||||
pub struct GooseClient {
|
||||
notification_handlers: Arc<Mutex<Vec<Sender<ServerNotification>>>>,
|
||||
}
|
||||
|
||||
impl GooseClient {
|
||||
pub fn new(handlers: Arc<Mutex<Vec<Sender<ServerNotification>>>>) -> Self {
|
||||
GooseClient {
|
||||
notification_handlers: handlers,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl ClientHandler for GooseClient {
|
||||
async fn on_progress(
|
||||
&self,
|
||||
params: rmcp::model::ProgressNotificationParam,
|
||||
context: rmcp::service::NotificationContext<rmcp::RoleClient>,
|
||||
) -> () {
|
||||
self.notification_handlers
|
||||
.lock()
|
||||
.await
|
||||
.iter()
|
||||
.for_each(|handler| {
|
||||
let _ = handler.try_send(ServerNotification::ProgressNotification(
|
||||
ProgressNotification {
|
||||
params: params.clone(),
|
||||
method: ProgressNotificationMethod,
|
||||
extensions: context.extensions.clone(),
|
||||
},
|
||||
));
|
||||
});
|
||||
}
|
||||
|
||||
async fn on_logging_message(
|
||||
&self,
|
||||
params: rmcp::model::LoggingMessageNotificationParam,
|
||||
context: rmcp::service::NotificationContext<rmcp::RoleClient>,
|
||||
) -> () {
|
||||
self.notification_handlers
|
||||
.lock()
|
||||
.await
|
||||
.iter()
|
||||
.for_each(|handler| {
|
||||
let _ = handler.try_send(ServerNotification::LoggingMessageNotification(
|
||||
LoggingMessageNotification {
|
||||
params: params.clone(),
|
||||
method: LoggingMessageNotificationMethod,
|
||||
extensions: context.extensions.clone(),
|
||||
},
|
||||
));
|
||||
});
|
||||
}
|
||||
|
||||
fn get_info(&self) -> ClientInfo {
|
||||
ClientInfo {
|
||||
protocol_version: ProtocolVersion::V_2025_03_26,
|
||||
capabilities: ClientCapabilities::builder().build(),
|
||||
client_info: Implementation {
|
||||
name: "goose".to_string(),
|
||||
version: env!("CARGO_PKG_VERSION").to_owned(),
|
||||
},
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// The MCP client is the interface for MCP operations.
|
||||
pub struct McpClient<T>
|
||||
where
|
||||
T: TransportHandle + Send + Sync + 'static,
|
||||
{
|
||||
service: Mutex<tower::timeout::Timeout<McpService<T>>>,
|
||||
next_id_counter: AtomicU64, // Added for atomic ID generation
|
||||
server_capabilities: Option<ServerCapabilities>,
|
||||
server_info: Option<Implementation>,
|
||||
pub struct McpClient {
|
||||
client: Mutex<RunningService<RoleClient, GooseClient>>,
|
||||
notification_subscribers: Arc<Mutex<Vec<mpsc::Sender<ServerNotification>>>>,
|
||||
server_info: Option<InitializeResult>,
|
||||
timeout: std::time::Duration,
|
||||
}
|
||||
|
||||
impl<T> McpClient<T>
|
||||
where
|
||||
T: TransportHandle + Send + Sync + 'static,
|
||||
{
|
||||
pub async fn connect(transport: T, timeout: std::time::Duration) -> Result<Self, Error> {
|
||||
let service = McpService::new(transport.clone());
|
||||
let service_ptr = service.clone();
|
||||
impl McpClient {
|
||||
pub async fn connect<T, E, A>(
|
||||
transport: T,
|
||||
timeout: std::time::Duration,
|
||||
) -> Result<Self, ClientInitializeError>
|
||||
where
|
||||
T: IntoTransport<RoleClient, E, A>,
|
||||
E: std::error::Error + From<std::io::Error> + Send + Sync + 'static,
|
||||
{
|
||||
let notification_subscribers =
|
||||
Arc::new(Mutex::new(Vec::<mpsc::Sender<ServerNotification>>::new()));
|
||||
let subscribers_ptr = notification_subscribers.clone();
|
||||
|
||||
tokio::spawn(async move {
|
||||
loop {
|
||||
match transport.receive().await {
|
||||
Ok(message) => {
|
||||
tracing::info!("Received message: {:?}", message);
|
||||
match message {
|
||||
JsonRpcMessage::Response(JsonRpcResponse {
|
||||
id: NumberOrString::Number(id),
|
||||
..
|
||||
})
|
||||
| JsonRpcMessage::Error(JsonRpcError {
|
||||
id: NumberOrString::Number(id),
|
||||
..
|
||||
}) => {
|
||||
service_ptr.respond(&id.to_string(), Ok(message)).await;
|
||||
}
|
||||
JsonRpcMessage::Notification(JsonRpcNotification {
|
||||
notification,
|
||||
..
|
||||
}) => {
|
||||
let mut subs = subscribers_ptr.lock().await;
|
||||
if let Some(server_notification) = notification.into() {
|
||||
subs.retain(|sub| {
|
||||
sub.try_send(server_notification.clone()).is_ok()
|
||||
});
|
||||
}
|
||||
}
|
||||
_ => {
|
||||
tracing::warn!(
|
||||
"Received unexpected received message type: {:?}",
|
||||
message
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
service_ptr.hangup(e).await;
|
||||
subscribers_ptr.lock().await.clear();
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
let middleware = TimeoutLayer::new(timeout);
|
||||
let client = GooseClient::new(notification_subscribers.clone());
|
||||
let client: rmcp::service::RunningService<rmcp::RoleClient, GooseClient> =
|
||||
client.serve(transport).await?;
|
||||
let server_info = client.peer_info().cloned();
|
||||
|
||||
Ok(Self {
|
||||
service: Mutex::new(middleware.layer(service)),
|
||||
next_id_counter: AtomicU64::new(1),
|
||||
server_capabilities: None,
|
||||
server_info: None,
|
||||
client: Mutex::new(client),
|
||||
notification_subscribers,
|
||||
server_info,
|
||||
timeout,
|
||||
})
|
||||
}
|
||||
|
||||
/// Send a JSON-RPC request and check we don't get an error response.
|
||||
async fn send_request<R>(&self, method: &str, params: Value) -> Result<R, Error>
|
||||
where
|
||||
R: for<'de> Deserialize<'de>,
|
||||
{
|
||||
let mut service = self.service.lock().await;
|
||||
service.ready().await.map_err(|_| Error::NotReady)?;
|
||||
let id_num = self.next_id_counter.fetch_add(1, Ordering::SeqCst);
|
||||
let id = RequestId::Number(id_num as u32);
|
||||
|
||||
let mut params = params.clone();
|
||||
params["_meta"] = json!({
|
||||
"progressToken": format!("prog-{}", id),
|
||||
});
|
||||
|
||||
let request = JsonRpcMessage::Request(JsonRpcRequest {
|
||||
jsonrpc: JsonRpcVersion2_0,
|
||||
id,
|
||||
request: Request {
|
||||
method: method.to_string(),
|
||||
params: params.as_object().unwrap().clone(),
|
||||
extensions: Default::default(),
|
||||
},
|
||||
});
|
||||
|
||||
let response_msg = service
|
||||
.call(request)
|
||||
.await
|
||||
.map_err(|e| Error::McpServerError {
|
||||
server: self
|
||||
.server_info
|
||||
.as_ref()
|
||||
.map(|s| s.name.clone())
|
||||
.unwrap_or("".to_string()),
|
||||
method: method.to_string(),
|
||||
// we don't need include params because it can be really large
|
||||
source: Box::<Error>::new(e.into()),
|
||||
})?;
|
||||
|
||||
match response_msg {
|
||||
JsonRpcMessage::Response(JsonRpcResponse { id, result, .. }) => {
|
||||
// Verify id matches - convert current id to match expected format
|
||||
let expected_id = RequestId::Number((id_num) as u32);
|
||||
if id != expected_id {
|
||||
return Err(Error::UnexpectedResponse(
|
||||
"id mismatch for JsonRpcResponse".to_string(),
|
||||
));
|
||||
}
|
||||
Ok(serde_json::from_value(serde_json::to_value(result)?)?)
|
||||
}
|
||||
JsonRpcMessage::Error(JsonRpcError { id, error, .. }) => {
|
||||
let expected_id = RequestId::Number((id_num) as u32);
|
||||
if id != expected_id {
|
||||
return Err(Error::UnexpectedResponse(
|
||||
"id mismatch for JsonRpcError".to_string(),
|
||||
));
|
||||
}
|
||||
Err(Error::RpcError {
|
||||
code: error.code.0, // Extract the i32 from ErrorCode
|
||||
message: error.message.to_string(), // Convert Cow to String
|
||||
})
|
||||
}
|
||||
_ => {
|
||||
// Requests/notifications not expected as a response
|
||||
Err(Error::UnexpectedResponse(
|
||||
"unexpected message type".to_string(),
|
||||
))
|
||||
}
|
||||
fn get_request_options(&self) -> PeerRequestOptions {
|
||||
PeerRequestOptions {
|
||||
timeout: Some(self.timeout),
|
||||
meta: None,
|
||||
}
|
||||
}
|
||||
|
||||
/// Send a JSON-RPC notification.
|
||||
async fn send_notification(&self, method: &str, params: Value) -> Result<(), Error> {
|
||||
let mut service = self.service.lock().await;
|
||||
service.ready().await.map_err(|_| Error::NotReady)?;
|
||||
|
||||
let notification = JsonRpcMessage::Notification(JsonRpcNotification {
|
||||
jsonrpc: JsonRpcVersion2_0,
|
||||
notification: Notification {
|
||||
method: method.to_string(),
|
||||
params: params.as_object().unwrap().clone(),
|
||||
extensions: Default::default(),
|
||||
},
|
||||
});
|
||||
|
||||
service
|
||||
.call(notification)
|
||||
.await
|
||||
.map_err(|e| Error::McpServerError {
|
||||
server: self
|
||||
.server_info
|
||||
.as_ref()
|
||||
.map(|s| s.name.clone())
|
||||
.unwrap_or("".to_string()),
|
||||
method: method.to_string(),
|
||||
// we don't need include params because it can be really large
|
||||
source: Box::<Error>::new(e.into()),
|
||||
})?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
// Check if the client has completed initialization
|
||||
fn completed_initialization(&self) -> bool {
|
||||
self.server_capabilities.is_some()
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait::async_trait]
|
||||
impl<T> McpClientTrait for McpClient<T>
|
||||
where
|
||||
T: TransportHandle + Send + Sync + 'static,
|
||||
{
|
||||
async fn initialize(
|
||||
&mut self,
|
||||
info: ClientInfo,
|
||||
capabilities: ClientCapabilities,
|
||||
) -> Result<InitializeResult, Error> {
|
||||
let params = InitializeParams {
|
||||
protocol_version: "2025-03-26".to_string(),
|
||||
client_info: info,
|
||||
capabilities,
|
||||
};
|
||||
let result: InitializeResult = self
|
||||
.send_request("initialize", serde_json::to_value(params)?)
|
||||
.await?;
|
||||
|
||||
self.send_notification("notifications/initialized", serde_json::json!({}))
|
||||
.await?;
|
||||
|
||||
self.server_capabilities = Some(result.capabilities.clone());
|
||||
|
||||
self.server_info = Some(result.server_info.clone());
|
||||
|
||||
Ok(result)
|
||||
impl McpClientTrait for McpClient {
|
||||
fn get_info(&self) -> Option<&InitializeResult> {
|
||||
self.server_info.as_ref()
|
||||
}
|
||||
|
||||
async fn list_resources(
|
||||
&self,
|
||||
next_cursor: Option<String>,
|
||||
) -> Result<ListResourcesResult, Error> {
|
||||
if !self.completed_initialization() {
|
||||
return Err(Error::NotInitialized);
|
||||
async fn list_resources(&self, cursor: Option<String>) -> Result<ListResourcesResult, Error> {
|
||||
let res = self
|
||||
.client
|
||||
.lock()
|
||||
.await
|
||||
.send_request_with_option(
|
||||
ClientRequest::ListResourcesRequest(ListResourcesRequest {
|
||||
params: Some(PaginatedRequestParam { cursor }),
|
||||
method: Default::default(),
|
||||
extensions: Default::default(),
|
||||
}),
|
||||
self.get_request_options(),
|
||||
)
|
||||
.await?
|
||||
.await_response()
|
||||
.await?;
|
||||
match res {
|
||||
ServerResult::ListResourcesResult(result) => Ok(result),
|
||||
_ => Err(ServiceError::UnexpectedResponse),
|
||||
}
|
||||
// If resources is not supported, return an empty list
|
||||
if self
|
||||
.server_capabilities
|
||||
.as_ref()
|
||||
.unwrap()
|
||||
.resources
|
||||
.is_none()
|
||||
{
|
||||
return Ok(ListResourcesResult {
|
||||
resources: vec![],
|
||||
next_cursor: None,
|
||||
});
|
||||
}
|
||||
|
||||
let payload = next_cursor
|
||||
.map(|cursor| serde_json::json!({"cursor": cursor}))
|
||||
.unwrap_or_else(|| serde_json::json!({}));
|
||||
|
||||
self.send_request("resources/list", payload).await
|
||||
}
|
||||
|
||||
async fn read_resource(&self, uri: &str) -> Result<ReadResourceResult, Error> {
|
||||
if !self.completed_initialization() {
|
||||
return Err(Error::NotInitialized);
|
||||
let res = self
|
||||
.client
|
||||
.lock()
|
||||
.await
|
||||
.send_request_with_option(
|
||||
ClientRequest::ReadResourceRequest(ReadResourceRequest {
|
||||
params: ReadResourceRequestParam {
|
||||
uri: uri.to_string(),
|
||||
},
|
||||
method: Default::default(),
|
||||
extensions: Default::default(),
|
||||
}),
|
||||
self.get_request_options(),
|
||||
)
|
||||
.await?
|
||||
.await_response()
|
||||
.await?;
|
||||
match res {
|
||||
ServerResult::ReadResourceResult(result) => Ok(result),
|
||||
_ => Err(ServiceError::UnexpectedResponse),
|
||||
}
|
||||
// If resources is not supported, return an error
|
||||
if self
|
||||
.server_capabilities
|
||||
.as_ref()
|
||||
.unwrap()
|
||||
.resources
|
||||
.is_none()
|
||||
{
|
||||
return Err(Error::RpcError {
|
||||
code: METHOD_NOT_FOUND,
|
||||
message: "Server does not support 'resources' capability".to_string(),
|
||||
});
|
||||
}
|
||||
|
||||
let params = serde_json::json!({ "uri": uri });
|
||||
self.send_request("resources/read", params).await
|
||||
}
|
||||
|
||||
async fn list_tools(&self, next_cursor: Option<String>) -> Result<ListToolsResult, Error> {
|
||||
if !self.completed_initialization() {
|
||||
return Err(Error::NotInitialized);
|
||||
async fn list_tools(&self, cursor: Option<String>) -> Result<ListToolsResult, Error> {
|
||||
let res = self
|
||||
.client
|
||||
.lock()
|
||||
.await
|
||||
.send_request_with_option(
|
||||
ClientRequest::ListToolsRequest(ListToolsRequest {
|
||||
params: Some(PaginatedRequestParam { cursor }),
|
||||
method: Default::default(),
|
||||
extensions: Default::default(),
|
||||
}),
|
||||
self.get_request_options(),
|
||||
)
|
||||
.await?
|
||||
.await_response()
|
||||
.await?;
|
||||
match res {
|
||||
ServerResult::ListToolsResult(result) => Ok(result),
|
||||
_ => Err(ServiceError::UnexpectedResponse),
|
||||
}
|
||||
// If tools is not supported, return an empty list
|
||||
if self.server_capabilities.as_ref().unwrap().tools.is_none() {
|
||||
return Ok(ListToolsResult {
|
||||
tools: vec![],
|
||||
next_cursor: None,
|
||||
});
|
||||
}
|
||||
|
||||
let payload = next_cursor
|
||||
.map(|cursor| serde_json::json!({"cursor": cursor}))
|
||||
.unwrap_or_else(|| serde_json::json!({}));
|
||||
|
||||
self.send_request("tools/list", payload).await
|
||||
}
|
||||
|
||||
async fn call_tool(&self, name: &str, arguments: Value) -> Result<CallToolResult, Error> {
|
||||
if !self.completed_initialization() {
|
||||
return Err(Error::NotInitialized);
|
||||
let arguments = match arguments {
|
||||
Value::Object(map) => Some(map),
|
||||
_ => None,
|
||||
};
|
||||
let res = self
|
||||
.client
|
||||
.lock()
|
||||
.await
|
||||
.send_request_with_option(
|
||||
ClientRequest::CallToolRequest(CallToolRequest {
|
||||
params: CallToolRequestParam {
|
||||
name: name.to_string().into(),
|
||||
arguments,
|
||||
},
|
||||
method: Default::default(),
|
||||
extensions: Default::default(),
|
||||
}),
|
||||
self.get_request_options(),
|
||||
)
|
||||
.await?
|
||||
.await_response()
|
||||
.await?;
|
||||
match res {
|
||||
ServerResult::CallToolResult(result) => Ok(result),
|
||||
_ => Err(ServiceError::UnexpectedResponse),
|
||||
}
|
||||
// If tools is not supported, return an error
|
||||
if self.server_capabilities.as_ref().unwrap().tools.is_none() {
|
||||
return Err(Error::RpcError {
|
||||
code: METHOD_NOT_FOUND,
|
||||
message: "Server does not support 'tools' capability".to_string(),
|
||||
});
|
||||
}
|
||||
|
||||
let params = serde_json::json!({ "name": name, "arguments": arguments });
|
||||
|
||||
// TODO ERROR: check that if there is an error, we send back is_error: true with msg
|
||||
// https://modelcontextprotocol.io/docs/concepts/tools#error-handling-2
|
||||
self.send_request("tools/call", params).await
|
||||
}
|
||||
|
||||
async fn list_prompts(&self, next_cursor: Option<String>) -> Result<ListPromptsResult, Error> {
|
||||
if !self.completed_initialization() {
|
||||
return Err(Error::NotInitialized);
|
||||
async fn list_prompts(&self, cursor: Option<String>) -> Result<ListPromptsResult, Error> {
|
||||
let res = self
|
||||
.client
|
||||
.lock()
|
||||
.await
|
||||
.send_request_with_option(
|
||||
ClientRequest::ListPromptsRequest(ListPromptsRequest {
|
||||
params: Some(PaginatedRequestParam { cursor }),
|
||||
method: Default::default(),
|
||||
extensions: Default::default(),
|
||||
}),
|
||||
self.get_request_options(),
|
||||
)
|
||||
.await?
|
||||
.await_response()
|
||||
.await?;
|
||||
match res {
|
||||
ServerResult::ListPromptsResult(result) => Ok(result),
|
||||
_ => Err(ServiceError::UnexpectedResponse),
|
||||
}
|
||||
|
||||
// If prompts is not supported, return an error
|
||||
if self.server_capabilities.as_ref().unwrap().prompts.is_none() {
|
||||
return Err(Error::RpcError {
|
||||
code: METHOD_NOT_FOUND,
|
||||
message: "Server does not support 'prompts' capability".to_string(),
|
||||
});
|
||||
}
|
||||
|
||||
let payload = next_cursor
|
||||
.map(|cursor| serde_json::json!({"cursor": cursor}))
|
||||
.unwrap_or_else(|| serde_json::json!({}));
|
||||
|
||||
self.send_request("prompts/list", payload).await
|
||||
}
|
||||
|
||||
async fn get_prompt(&self, name: &str, arguments: Value) -> Result<GetPromptResult, Error> {
|
||||
if !self.completed_initialization() {
|
||||
return Err(Error::NotInitialized);
|
||||
let arguments = match arguments {
|
||||
Value::Object(map) => Some(map),
|
||||
_ => None,
|
||||
};
|
||||
let res = self
|
||||
.client
|
||||
.lock()
|
||||
.await
|
||||
.send_request_with_option(
|
||||
ClientRequest::GetPromptRequest(GetPromptRequest {
|
||||
params: GetPromptRequestParam {
|
||||
name: name.to_string(),
|
||||
arguments,
|
||||
},
|
||||
method: Default::default(),
|
||||
extensions: Default::default(),
|
||||
}),
|
||||
self.get_request_options(),
|
||||
)
|
||||
.await?
|
||||
.await_response()
|
||||
.await?;
|
||||
match res {
|
||||
ServerResult::GetPromptResult(result) => Ok(result),
|
||||
_ => Err(ServiceError::UnexpectedResponse),
|
||||
}
|
||||
|
||||
// If prompts is not supported, return an error
|
||||
if self.server_capabilities.as_ref().unwrap().prompts.is_none() {
|
||||
return Err(Error::RpcError {
|
||||
code: METHOD_NOT_FOUND,
|
||||
message: "Server does not support 'prompts' capability".to_string(),
|
||||
});
|
||||
}
|
||||
|
||||
let params = serde_json::json!({ "name": name, "arguments": arguments });
|
||||
|
||||
self.send_request("prompts/get", params).await
|
||||
}
|
||||
|
||||
async fn subscribe(&self) -> mpsc::Receiver<ServerNotification> {
|
||||
|
||||
@@ -1,14 +1,8 @@
|
||||
pub mod client;
|
||||
pub mod oauth;
|
||||
pub mod service;
|
||||
pub mod transport;
|
||||
|
||||
#[cfg(test)]
|
||||
mod oauth_tests;
|
||||
|
||||
pub use client::{ClientCapabilities, ClientInfo, Error, McpClient, McpClientTrait};
|
||||
pub use client::{Error, McpClient, McpClientTrait};
|
||||
pub use oauth::{authenticate_service, ServiceConfig};
|
||||
pub use service::McpService;
|
||||
pub use transport::{
|
||||
SseTransport, StdioTransport, StreamableHttpTransport, Transport, TransportHandle,
|
||||
};
|
||||
|
||||
@@ -1,145 +0,0 @@
|
||||
use futures::future::BoxFuture;
|
||||
use rmcp::model::{JsonRpcMessage, JsonRpcRequest};
|
||||
use std::collections::HashMap;
|
||||
use std::sync::Arc;
|
||||
use std::task::{Context, Poll};
|
||||
use tokio::sync::{oneshot, RwLock};
|
||||
use tower::{timeout::Timeout, Service, ServiceBuilder};
|
||||
|
||||
use crate::transport::{Error, TransportHandle, TransportMessageRecv};
|
||||
|
||||
/// A wrapper service that implements Tower's Service trait for MCP transport
|
||||
#[derive(Clone)]
|
||||
pub struct McpService<T: TransportHandle> {
|
||||
inner: Arc<T>,
|
||||
pending_requests: Arc<PendingRequests>,
|
||||
}
|
||||
|
||||
impl<T: TransportHandle> McpService<T> {
|
||||
pub fn new(transport: T) -> Self {
|
||||
Self {
|
||||
inner: Arc::new(transport),
|
||||
pending_requests: Arc::new(PendingRequests::default()),
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn respond(&self, id: &str, response: Result<TransportMessageRecv, Error>) {
|
||||
self.pending_requests.respond(id, response).await
|
||||
}
|
||||
|
||||
pub async fn hangup(&self, error: Error) {
|
||||
self.pending_requests.broadcast_close(error).await
|
||||
}
|
||||
}
|
||||
|
||||
impl<T> Service<JsonRpcMessage> for McpService<T>
|
||||
where
|
||||
T: TransportHandle + Send + Sync + 'static,
|
||||
{
|
||||
type Response = TransportMessageRecv;
|
||||
type Error = Error;
|
||||
type Future = BoxFuture<'static, Result<Self::Response, Self::Error>>;
|
||||
|
||||
fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
|
||||
// Most transports are always ready, but this could be customized if needed
|
||||
Poll::Ready(Ok(()))
|
||||
}
|
||||
|
||||
fn call(&mut self, request: JsonRpcMessage) -> Self::Future {
|
||||
let transport = self.inner.clone();
|
||||
let pending_requests = self.pending_requests.clone();
|
||||
|
||||
Box::pin(async move {
|
||||
match &request {
|
||||
JsonRpcMessage::Request(JsonRpcRequest { id, .. }) => {
|
||||
// Create a channel to receive the response
|
||||
let (sender, receiver) = oneshot::channel();
|
||||
pending_requests.insert(id.to_string(), sender).await;
|
||||
|
||||
transport.send(request).await?;
|
||||
receiver.await.map_err(|_| Error::ChannelClosed)?
|
||||
}
|
||||
JsonRpcMessage::Notification(_) => {
|
||||
// Handle notifications without waiting for a response
|
||||
transport.send(request).await?;
|
||||
// Return a dummy response for notifications
|
||||
let dummy_response: Self::Response =
|
||||
JsonRpcMessage::Response(rmcp::model::JsonRpcResponse {
|
||||
jsonrpc: rmcp::model::JsonRpcVersion2_0,
|
||||
id: rmcp::model::RequestId::Number(0),
|
||||
result: serde_json::Map::new(),
|
||||
});
|
||||
Ok(dummy_response)
|
||||
}
|
||||
_ => Err(Error::UnsupportedMessage),
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// Add a convenience constructor for creating a service with timeout
|
||||
impl<T> McpService<T>
|
||||
where
|
||||
T: TransportHandle,
|
||||
{
|
||||
pub fn with_timeout(transport: T, timeout: std::time::Duration) -> Timeout<McpService<T>> {
|
||||
ServiceBuilder::new()
|
||||
.timeout(timeout)
|
||||
.service(McpService::new(transport))
|
||||
}
|
||||
}
|
||||
|
||||
// A data structure to store pending requests and their response channels
|
||||
pub struct PendingRequests {
|
||||
requests: RwLock<HashMap<String, oneshot::Sender<Result<TransportMessageRecv, Error>>>>,
|
||||
}
|
||||
|
||||
impl Default for PendingRequests {
|
||||
fn default() -> Self {
|
||||
Self::new()
|
||||
}
|
||||
}
|
||||
|
||||
impl PendingRequests {
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
requests: RwLock::new(HashMap::new()),
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn insert(
|
||||
&self,
|
||||
id: String,
|
||||
sender: oneshot::Sender<Result<TransportMessageRecv, Error>>,
|
||||
) {
|
||||
self.requests.write().await.insert(id, sender);
|
||||
}
|
||||
|
||||
pub async fn respond(&self, id: &str, response: Result<TransportMessageRecv, Error>) {
|
||||
if let Some(tx) = self.requests.write().await.remove(id) {
|
||||
let _ = tx.send(response);
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn broadcast_close(&self, error: Error) {
|
||||
for (_, tx) in self.requests.write().await.drain() {
|
||||
let err = match &error {
|
||||
Error::StdioProcessError(s) => Error::StdioProcessError(s.clone()),
|
||||
_ => Error::ChannelClosed,
|
||||
};
|
||||
let _ = tx.send(Err(err));
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn clear(&self) {
|
||||
self.requests.write().await.clear();
|
||||
}
|
||||
|
||||
pub async fn len(&self) -> usize {
|
||||
self.requests.read().await.len()
|
||||
}
|
||||
|
||||
pub async fn is_empty(&self) -> bool {
|
||||
self.len().await == 0
|
||||
}
|
||||
}
|
||||
@@ -1,82 +0,0 @@
|
||||
use async_trait::async_trait;
|
||||
use rmcp::model::{JsonObject, JsonRpcMessage, Request, ServerNotification};
|
||||
use thiserror::Error;
|
||||
use tokio::sync::mpsc;
|
||||
|
||||
pub type BoxError = Box<dyn std::error::Error + Sync + Send>;
|
||||
/// A generic error type for transport operations.
|
||||
#[derive(Debug, Error)]
|
||||
pub enum Error {
|
||||
#[error("I/O error: {0}")]
|
||||
Io(#[from] std::io::Error),
|
||||
|
||||
#[error("Transport was not connected or is already closed")]
|
||||
NotConnected,
|
||||
|
||||
#[error("Channel closed")]
|
||||
ChannelClosed,
|
||||
|
||||
#[error("Serialization error: {0}")]
|
||||
Serialization(#[from] serde_json::Error),
|
||||
|
||||
#[error("Unsupported message type. JsonRpcMessage can only be Request or Notification.")]
|
||||
UnsupportedMessage,
|
||||
|
||||
#[error("Stdio process error: {0}")]
|
||||
StdioProcessError(String),
|
||||
|
||||
#[error("SSE connection error: {0}")]
|
||||
SseConnection(String),
|
||||
|
||||
#[error("HTTP error: {status} - {message}")]
|
||||
HttpError { status: u16, message: String },
|
||||
|
||||
#[error("Streamable HTTP error: {0}")]
|
||||
StreamableHttpError(String),
|
||||
|
||||
#[error("Session error: {0}")]
|
||||
SessionError(String),
|
||||
}
|
||||
|
||||
/// A generic asynchronous transport trait with channel-based communication
|
||||
#[async_trait]
|
||||
pub trait Transport {
|
||||
type Handle: TransportHandle;
|
||||
|
||||
/// Start the transport and establish the underlying connection.
|
||||
/// Returns the transport handle for sending messages.
|
||||
async fn start(&self) -> Result<Self::Handle, Error>;
|
||||
|
||||
/// Close the transport and free any resources.
|
||||
async fn close(&self) -> Result<(), Error>;
|
||||
}
|
||||
|
||||
pub type TransportMessageRecv = JsonRpcMessage<Request, JsonObject, ServerNotification>;
|
||||
|
||||
#[async_trait]
|
||||
pub trait TransportHandle: Send + Sync + Clone + 'static {
|
||||
async fn send(&self, message: JsonRpcMessage) -> Result<(), Error>;
|
||||
async fn receive(&self) -> Result<TransportMessageRecv, Error>;
|
||||
}
|
||||
|
||||
pub async fn serialize_and_send(
|
||||
sender: &mpsc::Sender<String>,
|
||||
message: JsonRpcMessage,
|
||||
) -> Result<(), Error> {
|
||||
match serde_json::to_string(&message).map_err(Error::Serialization) {
|
||||
Ok(msg) => sender.send(msg).await.map_err(|_| Error::ChannelClosed),
|
||||
Err(e) => {
|
||||
tracing::error!(error = ?e, "Error serializing message");
|
||||
Err(e)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub mod stdio;
|
||||
pub use stdio::StdioTransport;
|
||||
|
||||
pub mod sse;
|
||||
pub use sse::SseTransport;
|
||||
|
||||
pub mod streamable_http;
|
||||
pub use streamable_http::StreamableHttpTransport;
|
||||
@@ -1,280 +0,0 @@
|
||||
use crate::transport::{Error, TransportMessageRecv};
|
||||
use async_trait::async_trait;
|
||||
use eventsource_client::{Client, SSE};
|
||||
use futures::TryStreamExt;
|
||||
use reqwest::Client as HttpClient;
|
||||
use rmcp::model::JsonRpcMessage;
|
||||
use std::collections::HashMap;
|
||||
use std::sync::Arc;
|
||||
use tokio::sync::{mpsc, Mutex, RwLock};
|
||||
use tokio::time::{timeout, Duration};
|
||||
use tracing::warn;
|
||||
use url::Url;
|
||||
|
||||
use super::{serialize_and_send, Transport, TransportHandle};
|
||||
|
||||
// Timeout for the endpoint discovery
|
||||
const ENDPOINT_TIMEOUT_SECS: u64 = 5;
|
||||
|
||||
/// The SSE-based actor that continuously:
|
||||
/// - Reads incoming events from the SSE stream.
|
||||
/// - Sends outgoing messages via HTTP POST (once the post endpoint is known).
|
||||
pub struct SseActor {
|
||||
/// Receives messages (requests/notifications) from the handle
|
||||
receiver: mpsc::Receiver<String>,
|
||||
/// Sends messages (responses) back to the handle
|
||||
sender: mpsc::Sender<TransportMessageRecv>,
|
||||
/// Base SSE URL
|
||||
sse_url: String,
|
||||
/// For sending HTTP POST requests
|
||||
http_client: HttpClient,
|
||||
/// The discovered endpoint for POST requests (once "endpoint" SSE event arrives)
|
||||
post_endpoint: Arc<RwLock<Option<String>>>,
|
||||
}
|
||||
|
||||
impl SseActor {
|
||||
pub fn new(
|
||||
receiver: mpsc::Receiver<String>,
|
||||
sender: mpsc::Sender<TransportMessageRecv>,
|
||||
sse_url: String,
|
||||
post_endpoint: Arc<RwLock<Option<String>>>,
|
||||
) -> Self {
|
||||
Self {
|
||||
receiver,
|
||||
sender,
|
||||
sse_url,
|
||||
post_endpoint,
|
||||
http_client: HttpClient::new(),
|
||||
}
|
||||
}
|
||||
|
||||
/// The main entry point for the actor. Spawns two concurrent loops:
|
||||
/// 1) handle_incoming_messages (SSE events)
|
||||
/// 2) handle_outgoing_messages (sending messages via POST)
|
||||
pub async fn run(self) {
|
||||
tokio::join!(
|
||||
Self::handle_incoming_messages(
|
||||
self.sender,
|
||||
self.sse_url.clone(),
|
||||
Arc::clone(&self.post_endpoint)
|
||||
),
|
||||
Self::handle_outgoing_messages(
|
||||
self.receiver,
|
||||
self.http_client.clone(),
|
||||
Arc::clone(&self.post_endpoint),
|
||||
)
|
||||
);
|
||||
}
|
||||
|
||||
/// Continuously reads SSE events from `sse_url`.
|
||||
/// - If an `endpoint` event is received, store it in `post_endpoint`.
|
||||
/// - If a `message` event is received, parse it as `JsonRpcMessage`
|
||||
/// and respond to pending requests if it's a `Response`.
|
||||
async fn handle_incoming_messages(
|
||||
sender: mpsc::Sender<TransportMessageRecv>,
|
||||
sse_url: String,
|
||||
post_endpoint: Arc<RwLock<Option<String>>>,
|
||||
) {
|
||||
let client = match eventsource_client::ClientBuilder::for_url(&sse_url) {
|
||||
Ok(builder) => builder.build(),
|
||||
Err(e) => {
|
||||
warn!("Failed to connect SSE client: {}", e);
|
||||
return;
|
||||
}
|
||||
};
|
||||
let mut stream = client.stream();
|
||||
|
||||
// First, wait for the "endpoint" event
|
||||
while let Ok(Some(event)) = stream.try_next().await {
|
||||
match event {
|
||||
SSE::Event(e) if e.event_type == "endpoint" => {
|
||||
// SSE server uses the "endpoint" event to tell us the POST URL
|
||||
let base_url = Url::parse(&sse_url).expect("Invalid base URL");
|
||||
let post_url = base_url
|
||||
.join(&e.data)
|
||||
.expect("Failed to resolve endpoint URL");
|
||||
|
||||
tracing::debug!("Discovered SSE POST endpoint: {}", post_url);
|
||||
*post_endpoint.write().await = Some(post_url.to_string());
|
||||
break;
|
||||
}
|
||||
_ => continue,
|
||||
}
|
||||
}
|
||||
|
||||
// Now handle subsequent events
|
||||
loop {
|
||||
match stream.try_next().await {
|
||||
Ok(Some(event)) => {
|
||||
match event {
|
||||
SSE::Event(e) if e.event_type == "message" => {
|
||||
// Attempt to parse the SSE data as a JsonRpcMessage
|
||||
match serde_json::from_str::<TransportMessageRecv>(&e.data) {
|
||||
Ok(message) => {
|
||||
let _ = sender.send(message).await;
|
||||
}
|
||||
Err(err) => {
|
||||
warn!("Failed to parse SSE message: {err}");
|
||||
}
|
||||
}
|
||||
}
|
||||
_ => { /* ignore other events */ }
|
||||
}
|
||||
}
|
||||
Ok(None) => {
|
||||
// Stream ended
|
||||
tracing::info!("SSE stream ended.");
|
||||
break;
|
||||
}
|
||||
Err(e) => {
|
||||
warn!("Error reading SSE stream: {e}");
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
tracing::error!("SSE stream ended or encountered an error.");
|
||||
}
|
||||
|
||||
async fn handle_outgoing_messages(
|
||||
mut receiver: mpsc::Receiver<String>,
|
||||
http_client: HttpClient,
|
||||
post_endpoint: Arc<RwLock<Option<String>>>,
|
||||
) {
|
||||
while let Some(message_str) = receiver.recv().await {
|
||||
let post_url = match post_endpoint.read().await.as_ref() {
|
||||
Some(url) => url.clone(),
|
||||
None => {
|
||||
// TODO: the endpoint isn't discovered yet. This shouldn't happen -- we only return the handle
|
||||
// after the endpoint is set.
|
||||
continue;
|
||||
}
|
||||
};
|
||||
|
||||
// Perform the HTTP POST
|
||||
match http_client
|
||||
.post(&post_url)
|
||||
.header("Content-Type", "application/json")
|
||||
.body(message_str)
|
||||
.send()
|
||||
.await
|
||||
{
|
||||
Ok(resp) => {
|
||||
if !resp.status().is_success() {
|
||||
let err = Error::HttpError {
|
||||
status: resp.status().as_u16(),
|
||||
message: resp.status().to_string(),
|
||||
};
|
||||
warn!("HTTP request returned error: {err}");
|
||||
// This doesn't directly fail the request,
|
||||
// because we rely on SSE to deliver the error response
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
warn!("HTTP POST failed: {e}");
|
||||
// Similarly, SSE might eventually reveal the error
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
tracing::info!("SseActor shut down.");
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct SseTransportHandle {
|
||||
sender: mpsc::Sender<String>,
|
||||
receiver: Arc<Mutex<mpsc::Receiver<TransportMessageRecv>>>,
|
||||
}
|
||||
|
||||
#[async_trait::async_trait]
|
||||
impl TransportHandle for SseTransportHandle {
|
||||
async fn send(&self, message: JsonRpcMessage) -> Result<(), Error> {
|
||||
serialize_and_send(&self.sender, message).await
|
||||
}
|
||||
|
||||
async fn receive(&self) -> Result<TransportMessageRecv, Error> {
|
||||
let mut receiver = self.receiver.lock().await;
|
||||
receiver.recv().await.ok_or(Error::ChannelClosed)
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct SseTransport {
|
||||
sse_url: String,
|
||||
env: HashMap<String, String>,
|
||||
}
|
||||
|
||||
/// The SSE transport spawns an `SseActor` on `start()`.
|
||||
impl SseTransport {
|
||||
pub fn new<S: Into<String>>(sse_url: S, env: HashMap<String, String>) -> Self {
|
||||
Self {
|
||||
sse_url: sse_url.into(),
|
||||
env,
|
||||
}
|
||||
}
|
||||
|
||||
/// Waits for the endpoint to be set, up to 10 attempts.
|
||||
async fn wait_for_endpoint(
|
||||
post_endpoint: Arc<RwLock<Option<String>>>,
|
||||
) -> Result<String, Error> {
|
||||
// Check every 100ms for the endpoint, for up to 10 attempts
|
||||
let check_interval = Duration::from_millis(100);
|
||||
let mut attempts = 0;
|
||||
let max_attempts = 10;
|
||||
|
||||
while attempts < max_attempts {
|
||||
if let Some(url) = post_endpoint.read().await.clone() {
|
||||
return Ok(url);
|
||||
}
|
||||
tokio::time::sleep(check_interval).await;
|
||||
attempts += 1;
|
||||
}
|
||||
Err(Error::SseConnection("No endpoint discovered".to_string()))
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl Transport for SseTransport {
|
||||
type Handle = SseTransportHandle;
|
||||
|
||||
async fn start(&self) -> Result<Self::Handle, Error> {
|
||||
// Set environment variables
|
||||
for (key, value) in &self.env {
|
||||
std::env::set_var(key, value);
|
||||
}
|
||||
|
||||
// Create a channel for outgoing TransportMessages
|
||||
let (tx, rx) = mpsc::channel(32);
|
||||
let (otx, orx) = mpsc::channel(32);
|
||||
|
||||
let post_endpoint: Arc<RwLock<Option<String>>> = Arc::new(RwLock::new(None));
|
||||
let post_endpoint_clone = Arc::clone(&post_endpoint);
|
||||
|
||||
// Build the actor
|
||||
let actor = SseActor::new(rx, otx, self.sse_url.clone(), post_endpoint);
|
||||
|
||||
// Spawn the actor task
|
||||
tokio::spawn(actor.run());
|
||||
|
||||
// Wait for the endpoint to be discovered before returning the handle
|
||||
match timeout(
|
||||
Duration::from_secs(ENDPOINT_TIMEOUT_SECS),
|
||||
Self::wait_for_endpoint(post_endpoint_clone),
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(_) => Ok(SseTransportHandle {
|
||||
sender: tx,
|
||||
receiver: Arc::new(Mutex::new(orx)),
|
||||
}),
|
||||
Err(e) => Err(Error::SseConnection(e.to_string())),
|
||||
}
|
||||
}
|
||||
|
||||
async fn close(&self) -> Result<(), Error> {
|
||||
// For SSE, you might close the stream or send a shutdown signal to the actor.
|
||||
// Here, we do nothing special.
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
@@ -1,319 +0,0 @@
|
||||
use std::collections::HashMap;
|
||||
use std::sync::atomic::{AtomicI32, Ordering};
|
||||
use std::sync::Arc;
|
||||
use tokio::process::{Child, ChildStderr, ChildStdin, ChildStdout, Command};
|
||||
|
||||
use async_trait::async_trait;
|
||||
use rmcp::model::JsonRpcMessage;
|
||||
use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader};
|
||||
use tokio::sync::{mpsc, Mutex};
|
||||
|
||||
// Import nix crate components instead of libc
|
||||
#[cfg(unix)]
|
||||
use nix::sys::signal::{kill, Signal};
|
||||
#[cfg(unix)]
|
||||
use nix::unistd::{getpgid, Pid};
|
||||
|
||||
use crate::transport::TransportMessageRecv;
|
||||
|
||||
use super::{serialize_and_send, Error, Transport, TransportHandle};
|
||||
|
||||
// Global to track process groups we've created
|
||||
static PROCESS_GROUP: AtomicI32 = AtomicI32::new(-1);
|
||||
|
||||
/// A `StdioTransport` uses a child process's stdin/stdout as a communication channel.
|
||||
///
|
||||
/// It uses channels for message passing and handles responses asynchronously through a background task.
|
||||
pub struct StdioActor {
|
||||
receiver: Option<mpsc::Receiver<String>>,
|
||||
sender: Option<mpsc::Sender<TransportMessageRecv>>,
|
||||
process: Child, // we store the process to keep it alive
|
||||
error_sender: mpsc::Sender<Error>,
|
||||
stdin: Option<ChildStdin>,
|
||||
stdout: Option<ChildStdout>,
|
||||
stderr: Option<ChildStderr>,
|
||||
}
|
||||
|
||||
impl Drop for StdioActor {
|
||||
fn drop(&mut self) {
|
||||
// Get the process group ID before attempting cleanup
|
||||
#[cfg(unix)]
|
||||
if let Some(pid) = self.process.id() {
|
||||
if let Ok(pgid) = getpgid(Some(Pid::from_raw(pid as i32))) {
|
||||
// Send SIGTERM to the entire process group
|
||||
let _ = kill(Pid::from_raw(-pgid.as_raw()), Signal::SIGTERM);
|
||||
// Give processes a moment to cleanup
|
||||
std::thread::sleep(std::time::Duration::from_millis(100));
|
||||
// Force kill if still running
|
||||
let _ = kill(Pid::from_raw(-pgid.as_raw()), Signal::SIGKILL);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl StdioActor {
|
||||
pub async fn run(mut self) {
|
||||
use tokio::pin;
|
||||
|
||||
let stdout = self.stdout.take().expect("stdout should be available");
|
||||
let stdin = self.stdin.take().expect("stdin should be available");
|
||||
let msg_inbox = self.receiver.take().expect("receiver should be available");
|
||||
let msg_outbox = self.sender.take().expect("sender should be available");
|
||||
|
||||
let incoming = Self::handle_proc_output(stdout, msg_outbox);
|
||||
let outgoing = Self::handle_proc_input(stdin, msg_inbox);
|
||||
|
||||
// take ownership of futures for tokio::select
|
||||
pin!(incoming);
|
||||
pin!(outgoing);
|
||||
|
||||
// Use select! to wait for either I/O completion or process exit
|
||||
tokio::select! {
|
||||
result = &mut incoming => {
|
||||
tracing::debug!("Stdin handler completed: {:?}", result);
|
||||
}
|
||||
result = &mut outgoing => {
|
||||
tracing::debug!("Stdout handler completed: {:?}", result);
|
||||
}
|
||||
// capture the status so we don't need to wait for a timeout
|
||||
status = self.process.wait() => {
|
||||
tracing::debug!("Process exited with status: {:?}", status);
|
||||
}
|
||||
}
|
||||
|
||||
// Then always try to read stderr before cleaning up
|
||||
let mut stderr_buffer = Vec::new();
|
||||
if let Some(mut stderr) = self.stderr.take() {
|
||||
if let Ok(bytes) = stderr.read_to_end(&mut stderr_buffer).await {
|
||||
let err_msg = if bytes > 0 {
|
||||
String::from_utf8_lossy(&stderr_buffer).to_string()
|
||||
} else {
|
||||
"Process ended unexpectedly".to_string()
|
||||
};
|
||||
|
||||
tracing::info!("Process stderr: {}", err_msg);
|
||||
let _ = self
|
||||
.error_sender
|
||||
.send(Error::StdioProcessError(err_msg))
|
||||
.await;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn handle_proc_output(stdout: ChildStdout, sender: mpsc::Sender<TransportMessageRecv>) {
|
||||
let mut reader = BufReader::new(stdout);
|
||||
let mut line = String::new();
|
||||
loop {
|
||||
match reader.read_line(&mut line).await {
|
||||
Ok(0) => {
|
||||
tracing::error!("Child process ended (EOF on stdout)");
|
||||
break;
|
||||
} // EOF
|
||||
Ok(_) => {
|
||||
if let Ok(message) = serde_json::from_str::<TransportMessageRecv>(&line) {
|
||||
tracing::debug!(
|
||||
message = ?message,
|
||||
"Received incoming message"
|
||||
);
|
||||
let _ = sender.send(message).await;
|
||||
} else {
|
||||
tracing::warn!(
|
||||
message = ?line,
|
||||
"Failed to parse incoming message"
|
||||
);
|
||||
}
|
||||
line.clear();
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::error!(error = ?e, "Error reading line");
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn handle_proc_input(mut stdin: ChildStdin, mut receiver: mpsc::Receiver<String>) {
|
||||
while let Some(message_str) = receiver.recv().await {
|
||||
tracing::debug!(message = ?message_str, "Sending outgoing message");
|
||||
|
||||
if let Err(e) = stdin.write_all(format!("{message_str}\n").as_bytes()).await {
|
||||
tracing::error!(error = ?e, "Error writing message to child process");
|
||||
break;
|
||||
}
|
||||
|
||||
if let Err(e) = stdin.flush().await {
|
||||
tracing::error!(error = ?e, "Error flushing message to child process");
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct StdioTransportHandle {
|
||||
sender: mpsc::Sender<String>, // to process
|
||||
receiver: Arc<Mutex<mpsc::Receiver<TransportMessageRecv>>>, // from process
|
||||
error_receiver: Arc<Mutex<mpsc::Receiver<Error>>>,
|
||||
}
|
||||
|
||||
#[async_trait::async_trait]
|
||||
impl TransportHandle for StdioTransportHandle {
|
||||
async fn send(&self, message: JsonRpcMessage) -> Result<(), Error> {
|
||||
let result = serialize_and_send(&self.sender, message).await;
|
||||
// Check for any pending errors even if send is successful
|
||||
self.check_for_errors().await?;
|
||||
result
|
||||
}
|
||||
|
||||
async fn receive(&self) -> Result<TransportMessageRecv, Error> {
|
||||
let mut receiver = self.receiver.lock().await;
|
||||
match receiver.recv().await {
|
||||
Some(message) => Ok(message),
|
||||
None => {
|
||||
self.check_for_errors().await?;
|
||||
Err(Error::ChannelClosed)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl StdioTransportHandle {
|
||||
/// Check if there are any process errors
|
||||
pub async fn check_for_errors(&self) -> Result<(), Error> {
|
||||
match self.error_receiver.lock().await.try_recv() {
|
||||
Ok(error) => {
|
||||
tracing::debug!("Found error: {:?}", error);
|
||||
Err(error)
|
||||
}
|
||||
Err(_) => Ok(()),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub struct StdioTransport {
|
||||
command: String,
|
||||
args: Vec<String>,
|
||||
env: HashMap<String, String>,
|
||||
}
|
||||
|
||||
impl StdioTransport {
|
||||
pub fn new<S: Into<String>>(
|
||||
command: S,
|
||||
args: Vec<String>,
|
||||
env: HashMap<String, String>,
|
||||
) -> Self {
|
||||
Self {
|
||||
command: command.into(),
|
||||
args,
|
||||
env,
|
||||
}
|
||||
}
|
||||
|
||||
async fn spawn_process(&self) -> Result<(Child, ChildStdin, ChildStdout, ChildStderr), Error> {
|
||||
let mut command = Command::new(&self.command);
|
||||
command
|
||||
.envs(&self.env)
|
||||
.args(&self.args)
|
||||
.stdin(std::process::Stdio::piped())
|
||||
.stdout(std::process::Stdio::piped())
|
||||
.stderr(std::process::Stdio::piped())
|
||||
.kill_on_drop(true);
|
||||
|
||||
// Set process group and ensure signal handling on Unix systems
|
||||
#[cfg(unix)]
|
||||
command.process_group(0);
|
||||
|
||||
// Hide console window on Windows
|
||||
#[cfg(windows)]
|
||||
command.creation_flags(0x08000000); // CREATE_NO_WINDOW flag
|
||||
|
||||
let mut process = command.spawn().map_err(|e| {
|
||||
let command = command.into_std();
|
||||
Error::StdioProcessError(format!(
|
||||
"Could not run extension command (`{} {}`): {}",
|
||||
command
|
||||
.get_program()
|
||||
.to_str()
|
||||
.unwrap_or("[invalid command]"),
|
||||
command
|
||||
.get_args()
|
||||
.map(|arg| arg.to_str().unwrap_or("[invalid arg]"))
|
||||
.collect::<Vec<_>>()
|
||||
.join(" "),
|
||||
e
|
||||
))
|
||||
})?;
|
||||
|
||||
let stdin = process
|
||||
.stdin
|
||||
.take()
|
||||
.ok_or_else(|| Error::StdioProcessError("Failed to get stdin".into()))?;
|
||||
|
||||
let stdout = process
|
||||
.stdout
|
||||
.take()
|
||||
.ok_or_else(|| Error::StdioProcessError("Failed to get stdout".into()))?;
|
||||
|
||||
let stderr = process
|
||||
.stderr
|
||||
.take()
|
||||
.ok_or_else(|| Error::StdioProcessError("Failed to get stderr".into()))?;
|
||||
|
||||
// Store the process group ID for cleanup
|
||||
#[cfg(unix)]
|
||||
if let Some(pid) = process.id() {
|
||||
// Use nix instead of unsafe libc calls
|
||||
if let Ok(pgid) = getpgid(Some(Pid::from_raw(pid as i32))) {
|
||||
PROCESS_GROUP.store(pgid.as_raw(), Ordering::SeqCst);
|
||||
}
|
||||
}
|
||||
|
||||
Ok((process, stdin, stdout, stderr))
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl Transport for StdioTransport {
|
||||
type Handle = StdioTransportHandle;
|
||||
|
||||
async fn start(&self) -> Result<Self::Handle, Error> {
|
||||
let (process, stdin, stdout, stderr) = self.spawn_process().await?;
|
||||
let (outbox_tx, outbox_rx) = mpsc::channel(32);
|
||||
let (inbox_tx, inbox_rx) = mpsc::channel(32);
|
||||
let (error_tx, error_rx) = mpsc::channel(1);
|
||||
|
||||
let actor = StdioActor {
|
||||
receiver: Some(outbox_rx), // client to process
|
||||
sender: Some(inbox_tx), // process to client
|
||||
process,
|
||||
error_sender: error_tx,
|
||||
stdin: Some(stdin),
|
||||
stdout: Some(stdout),
|
||||
stderr: Some(stderr),
|
||||
};
|
||||
|
||||
tokio::spawn(actor.run());
|
||||
|
||||
let handle = StdioTransportHandle {
|
||||
sender: outbox_tx, // client to process
|
||||
receiver: Arc::new(Mutex::new(inbox_rx)), // process to client
|
||||
error_receiver: Arc::new(Mutex::new(error_rx)),
|
||||
};
|
||||
Ok(handle)
|
||||
}
|
||||
|
||||
async fn close(&self) -> Result<(), Error> {
|
||||
// Attempt to clean up the process group on close
|
||||
#[cfg(unix)]
|
||||
if let Some(pgid) = PROCESS_GROUP.load(Ordering::SeqCst).checked_abs() {
|
||||
// Use nix instead of unsafe libc calls
|
||||
// Try SIGTERM first
|
||||
let _ = kill(Pid::from_raw(-pgid), Signal::SIGTERM);
|
||||
// Give processes a moment to cleanup
|
||||
tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
|
||||
// Force kill if still running
|
||||
let _ = kill(Pid::from_raw(-pgid), Signal::SIGKILL);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user