Stash

ext/mcp/Cargo.toml

@@ -3,9 +3,21 @@ name = "mcp"
 version = "0.1.0"
 edition = "2021"
 
+[[bin]]
+name = "internal-test"
+path = "src/internal-test.rs"
+
 [lib]
 crate-type = ["cdylib"]
 
 [dependencies]
+jsonrpsee = { version = "0.24.8", features = ["client", "tracing"] }
 magnus = "0.7"
 mcp-sdk = "0.0.3"
+tokio = { version = "1.44.1", features = ["full"] }
+anyhow = "1.0.97"
+tracing = "0.1.41"
+tokio-stdin-stdout = "0.1.5"
+serde = { version = "1.0.219", features = ["derive"] }
+serde_json = "1.0.140"
+tracing-subscriber = { version = "0.3.19", features = ["env-filter"] }

ext/mcp/src/internal-test.rs

@@ -0,0 +1,152 @@
+use std::collections::HashMap;
+use std::process::Stdio;
+use std::sync::Arc;
+use jsonrpsee::core::async_trait;
+use jsonrpsee::core::client::{
+    Client, ClientBuilder, ClientT, ReceivedMessage, TransportReceiverT, TransportSenderT,
+};
+use jsonrpsee::core::traits::ToRpcParams;
+use jsonrpsee::rpc_params;
+use tokio::process::{Child, ChildStdin, ChildStdout};
+use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader, Stdin, Stdout};
+use serde::{Deserialize, Serialize};
+use serde_json::Error;
+use serde_json::value::RawValue;
+mod types;
+
+#[derive(Debug, Clone)]
+struct StdioTransport {
+    stdin: Arc<tokio::sync::Mutex<ChildStdin>>,
+    stdout: Arc<tokio::sync::Mutex<BufReader<ChildStdout>>>,
+}
+
+impl StdioTransport {
+    fn new(mut child: Child) -> Self {
+        let stdin = Arc::new(tokio::sync::Mutex::new(child.stdin.take().unwrap()));
+        let stdout = Arc::new(tokio::sync::Mutex::new(BufReader::new(child.stdout.take().unwrap())));
+        Self { stdin, stdout }
+    }
+}
+
+#[async_trait]
+impl TransportSenderT for StdioTransport {
+    type Error = tokio::io::Error;
+
+    #[tracing::instrument(skip(self), level = "trace")]
+    async fn send(&mut self, msg: String) -> Result<(), Self::Error> {
+        tracing::debug!("Sending message: {}", msg);
+        let mut stdin = self.stdin.lock().await;
+        tracing::debug!("Locked stdin");
+        stdin.write_all(msg.as_bytes()).await?;
+        stdin.write_all(b"\n").await?;
+        tracing::debug!("Wrote to stdin");
+        Ok(())
+    }
+}
+
+#[async_trait]
+impl TransportReceiverT for StdioTransport {
+    type Error = tokio::io::Error;
+
+    #[tracing::instrument(skip(self), level = "trace")]
+    async fn receive(&mut self) -> Result<ReceivedMessage, Self::Error> {
+        tracing::debug!("Receiving message");
+        let mut stdout = self.stdout.lock().await;
+        tracing::debug!("Locked stdout");
+        let mut str = String::new();
+        tracing::debug!("Reading from stdout");
+        stdout.read_line(&mut str).await?;
+        tracing::debug!("Read from stdout: {:?}", str);
+        Ok(ReceivedMessage::Text(str))
+    }
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+struct InitializeRequest {
+    protocol_version: String,
+    capabilities: Capabilities,
+    client_info: ClientInfo,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+struct Capabilities {
+    roots: Roots,
+    sampling: HashMap<String, String>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+struct Sampling {
+    sampling_interval: u64,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+struct Roots {
+    list_changed: bool,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+struct ClientInfo {
+    name: String,
+    version: String,
+}
+
+impl ToRpcParams for InitializeRequest {
+    fn to_rpc_params(self) -> Result<Option<Box<RawValue>>, serde_json::Error> {
+        let s = String::from_utf8(serde_json::to_vec(&self)?).expect("Valid UTF8 format");
+        RawValue::from_string(s).map(Some)
+    }
+}
+
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    tracing_subscriber::fmt()
+        .with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
+        .with_file(true)
+        .with_line_number(true)
+        .with_thread_ids(true)
+        .with_thread_names(true)
+        .with_target(true)
+        .with_level(true)
+        .init();
+
+    tracing::debug!("Hello");
+
+    types::InitializeRequest {
+        method: "".to_string(),
+        params: types::InitializeRequestParams {
+            capabilities: Default::default(),
+            client_info: types::Implementation { name: "".to_string(), version: "".to_string() },
+            protocol_version: "".to_string(),
+        },
+    }
+
+    let cmd = tokio::process::Command::new("/Users/joshuacoles/.local/bin/mcp-server-fetch")
+        .stdin(Stdio::piped())
+        .stdout(Stdio::piped())
+        .spawn()?;
+
+    let transport = StdioTransport::new(cmd);
+
+    let client: Client = ClientBuilder::default().build_with_tokio(
+        transport.clone(),
+        transport.clone(),
+    );
+
+    let response: serde_json::Value = client.request("initialize", InitializeRequest {
+        protocol_version: "2024-11-05".to_string(),
+        capabilities: Capabilities {
+            roots: Roots { list_changed: true },
+            sampling: HashMap::default(),
+        },
+        client_info: ClientInfo { name: "ExampleClient".to_string(), version: "1.0.0".to_string() },
+    }).await?;
+
+    println!("response: {:?}", response);
+
+    Ok(())
+}

ext/mcp/src/lib.rs

@@ -1,3 +1,5 @@
+mod mcp_client;
+
 use magnus::{function, Error, Ruby};
 
 fn distance(a: (f64, f64), b: (f64, f64)) -> f64 {

ext/mcp/src/mcp_client.rs

@@ -0,0 +1,30 @@
+use std::time::Duration;
+use mcp_sdk::client::ClientBuilder;
+use mcp_sdk::protocol::RequestOptions;
+use mcp_sdk::transport::Transport;
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn test_a() {
+    let transport = mcp_sdk::transport::ClientStdioTransport::new(
+        "/Users/joshuacoles/.local/bin/mcp-server-fetch",
+        &[],
+    ).unwrap();
+
+    transport.open().unwrap();
+
+    let client = ClientBuilder::new(transport).build();
+    let client_clone = client.clone();
+
+    let a = tokio::spawn(async move { client_clone.start().await });
+    let response = client
+        .request(
+            "echo",
+            None,
+            RequestOptions::default().timeout(Duration::from_secs(1)),
+        )
+        .await.unwrap();
+
+    println!("{:?}", response);
+
+    a.abort();
+}