use serde::{Deserialize, Serialize}; use serde_json::{json, Value}; use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader}; #[derive(Serialize, Deserialize, Debug, Clone)] pub struct JsonRpcRequest { pub jsonrpc: String, pub id: Option, pub method: String, pub params: Option, } #[derive(Serialize, Debug, Clone)] pub struct JsonRpcResponse { pub jsonrpc: String, pub id: Value, #[serde(skip_serializing_if = "Option::is_none")] pub result: Option, #[serde(skip_serializing_if = "Option::is_none")] pub error: Option, } pub async fn read_message( stdin: &mut BufReader, ) -> Option { let mut length = 0; loop { let mut line = String::new(); if stdin.read_line(&mut line).await.unwrap_or(0) == 0 { return None; } if line.starts_with('{') { return match serde_json::from_str::(line.trim_end()) { Ok(req) => Some(req), Err(e) => { tracing::error!( "Failed to parse JSON-RPC request from JSONL: {}. Payload: {}", e, line ); None } }; } let line = line.trim_end(); if line.is_empty() { break; } let lower_line = line.to_lowercase(); if let Some(len_str) = lower_line.strip_prefix("content-length:") { length = len_str.trim().parse().unwrap_or(0); } } if length == 0 { return None; } let mut buffer = vec![0; length]; stdin.read_exact(&mut buffer).await.unwrap_or(0); serde_json::from_slice(&buffer).ok() } pub async fn send_response(response: JsonRpcResponse) { let msg = serde_json::to_string(&response).unwrap(); tracing::info!( "Sending JSON-RPC response (id: {:?}): {}", response.id, if msg.len() > 500 { format!("{}...", &msg[..500]) } else { msg.clone() } ); // CRITICAL ARCHITECTURAL DECISION: // The MCP StdioTransport MUST use Newline-Delimited JSON (NDJSON). // Do NOT use LSP-style HTTP headers (e.g. Content-Length). // See MCP protocol spec (SEP-2575) and mcp-go-sdk bufio.Scanner implementation. let payload = format!("{}\n", msg); let mut stdout = tokio::io::stdout(); let _ = stdout.write_all(payload.as_bytes()).await; let _ = stdout.flush().await; } pub async fn send_error(id: Value, code: i32, message: &str) { send_response(JsonRpcResponse { jsonrpc: "2.0".to_string(), id, result: None, error: Some(serde_json::json!({"code": code, "message": message})), }) .await; } #[cfg(windows)] async fn get_socket_path() -> Result { let profile = std::env::var("USERPROFILE").unwrap_or_else(|_| "C:\\Users\\reazul.ashraf".into()); let path = format!("{}\\.gemini\\active_nvim.txt", profile); if let Ok(content) = tokio::fs::read_to_string(&path).await { let p = content.trim().to_string(); if !p.is_empty() { if p.starts_with(r"\\.\pipe\") { return Ok(p); } else if p.starts_with("nvim.") { return Ok(format!(r"\\.\pipe\{}", p)); } else { return Ok(p); } } } tracing::warn!("active_nvim.txt missing or invalid, falling back to pipe discovery"); if let Ok(mut dir) = tokio::fs::read_dir(r"\\.\pipe\").await { while let Ok(Some(entry)) = dir.next_entry().await { let name = entry.file_name(); let name_str = name.to_string_lossy(); if name_str.starts_with("nvim.") { return Ok(format!(r"\\.\pipe\{}", name_str)); } } } Err("Could not find active Windows Neovim named pipe".to_string()) } #[cfg(unix)] async fn get_socket_path() -> Result { if let Ok(home) = std::env::var("HOME") { let path = format!("{}/.gemini/active_nvim.txt", home); if let Ok(content) = tokio::fs::read_to_string(&path).await { let p = content.trim().to_string(); if !p.is_empty() { return Ok(p); } } } if let Ok(mut entries) = tokio::fs::read_dir("/tmp").await { while let Ok(Some(entry)) = entries.next_entry().await { if let Ok(name) = entry.file_name().into_string() { if name.starts_with("nvim.") { let path = entry.path().join("0"); if tokio::fs::try_exists(&path).await.unwrap_or(false) { return Ok(path.to_string_lossy().to_string()); } } } } } Err("Could not find Neovim socket".to_string()) } #[cfg(windows)] async fn call_nvim(req: rmpv::Value) -> Result { use tokio::net::windows::named_pipe::ClientOptions; let msgid = if let rmpv::Value::Array(ref arr) = req { if arr.len() > 1 { arr[1].clone() } else { rmpv::Value::Nil } } else { rmpv::Value::Nil }; tracing::info!("Connecting to neovim pipe"); let socket_path = get_socket_path().await?; let mut client = ClientOptions::new() .open(&socket_path) .map_err(|e| e.to_string())?; let mut buf = Vec::new(); rmpv::encode::write_value(&mut buf, &req).map_err(|e| e.to_string())?; tracing::info!("Sending RPC request to neovim (msgid: {})", msgid); client.write_all(&buf).await.map_err(|e| e.to_string())?; let mut resp_buf = Vec::new(); let mut chunk = vec![0u8; 8192]; let mut offset = 0; loop { let mut cursor = std::io::Cursor::new(&resp_buf[offset..]); match rmpv::decode::read_value(&mut cursor) { Ok(val) => { offset += cursor.position() as usize; if let rmpv::Value::Array(ref arr) = val { if arr.len() >= 4 && arr[0] == rmpv::Value::Integer(1.into()) && arr[1] == msgid { tracing::info!("Received RPC response from neovim (msgid: {})", msgid); return Ok(val); } } continue; } Err(_) => { let read_future = client.read(&mut chunk); match tokio::time::timeout(tokio::time::Duration::from_secs(5), read_future).await { Ok(Ok(n)) => { if n == 0 { return Err("Connection closed".into()); } resp_buf.extend_from_slice(&chunk[..n]); } Ok(Err(e)) => return Err(e.to_string()), Err(_) => { tracing::error!("Timeout waiting for Neovim response (msgid: {})", msgid); return Err("Timeout waiting for Neovim response".into()); } } } } } } #[cfg(unix)] async fn call_nvim(req: rmpv::Value) -> Result { use tokio::net::UnixStream; let msgid = if let rmpv::Value::Array(ref arr) = req { if arr.len() > 1 { arr[1].clone() } else { rmpv::Value::Nil } } else { rmpv::Value::Nil }; tracing::info!("Connecting to neovim socket"); let socket_path = get_socket_path().await?; let mut stream = UnixStream::connect(socket_path) .await .map_err(|e| e.to_string())?; let mut buf = Vec::new(); rmpv::encode::write_value(&mut buf, &req).map_err(|e| e.to_string())?; tracing::info!("Sending RPC request to neovim (msgid: {})", msgid); stream.write_all(&buf).await.map_err(|e| e.to_string())?; let mut resp_buf = Vec::new(); let mut chunk = vec![0u8; 8192]; let mut offset = 0; loop { let mut cursor = std::io::Cursor::new(&resp_buf[offset..]); match rmpv::decode::read_value(&mut cursor) { Ok(val) => { offset += cursor.position() as usize; if let rmpv::Value::Array(ref arr) = val { if arr.len() >= 4 && arr[0] == rmpv::Value::Integer(1.into()) && arr[1] == msgid { tracing::info!("Received RPC response from neovim (msgid: {})", msgid); return Ok(val); } } continue; } Err(_) => { let read_future = stream.read(&mut chunk); match tokio::time::timeout(tokio::time::Duration::from_secs(5), read_future).await { Ok(Ok(n)) => { if n == 0 { return Err("Connection closed".into()); } resp_buf.extend_from_slice(&chunk[..n]); } Ok(Err(e)) => return Err(e.to_string()), Err(_) => { tracing::error!("Timeout waiting for Neovim response (msgid: {})", msgid); return Err("Timeout waiting for Neovim response".into()); } } } } } } async fn send_nvim_command(cmd: &str) -> Result<(), String> { use rmpv::Value as RmpValue; let req = RmpValue::Array(vec![ RmpValue::Integer(0.into()), RmpValue::Integer(1.into()), // msgid RmpValue::String("nvim_command".into()), RmpValue::Array(vec![RmpValue::String(cmd.into())]), ]); let resp = call_nvim(req).await?; if let RmpValue::Array(arr) = resp { if !arr[2].is_nil() { return Err(format!("Neovim error: {:?}", arr[2])); } return Ok(()); } Err("Invalid response".to_string()) } async fn get_nvim_active_buffer() -> Result { use rmpv::Value as RmpValue; let req = RmpValue::Array(vec![ RmpValue::Integer(0.into()), RmpValue::Integer(2.into()), // msgid RmpValue::String("nvim_buf_get_lines".into()), RmpValue::Array(vec![ RmpValue::Integer(0.into()), RmpValue::Integer(0.into()), RmpValue::Integer((-1).into()), RmpValue::Boolean(true), ]), ]); let resp = call_nvim(req).await?; if let RmpValue::Array(arr) = resp { if !arr[2].is_nil() { return Err(format!("Neovim error: {:?}", arr[2])); } if let RmpValue::Array(lines) = &arr[3] { let mut text = String::new(); for line in lines { if let RmpValue::String(s) = line { if let Some(s) = s.as_str() { text.push_str(s); text.push('\n'); } } } return Ok(text); } } Err("Invalid response".to_string()) } async fn get_nvim_cursor() -> Result { use rmpv::Value as RmpValue; let req = RmpValue::Array(vec![ RmpValue::Integer(0.into()), RmpValue::Integer(3.into()), // msgid RmpValue::String("nvim_win_get_cursor".into()), RmpValue::Array(vec![RmpValue::Integer(0.into())]), ]); let resp = call_nvim(req).await?; if let RmpValue::Array(arr) = resp { if !arr[2].is_nil() { return Err(format!("Neovim error: {:?}", arr[2])); } if let RmpValue::Array(pos) = &arr[3] { if pos.len() == 2 { if let (RmpValue::Integer(row), RmpValue::Integer(col)) = (&pos[0], &pos[1]) { return Ok(format!("Line: {}, Column: {}", row, col)); } } } } Err("Invalid response".to_string()) } async fn get_nvim_visual_selection() -> Result { let lua_script = r#" local _, csrow, cscol, _ = unpack(vim.fn.getpos("'<")) local _, cerow, cecol, _ = unpack(vim.fn.getpos("'>")) local lines = vim.fn.getline(csrow, cerow) if type(lines) == "table" then return table.concat(lines, "\n") else return lines end "#; use rmpv::Value as RmpValue; let req = RmpValue::Array(vec![ RmpValue::Integer(0.into()), RmpValue::Integer(4.into()), // msgid RmpValue::String("nvim_exec_lua".into()), RmpValue::Array(vec![ RmpValue::String(lua_script.into()), RmpValue::Array(vec![]), ]), ]); let resp = call_nvim(req).await?; if let RmpValue::Array(arr) = resp { if !arr[2].is_nil() { return Err(format!("Neovim error: {:?}", arr[2])); } if let RmpValue::String(s) = &arr[3] { if let Some(text) = s.as_str() { return Ok(text.to_string()); } } } Err("Invalid response".to_string()) } async fn set_nvim_diagnostics(line: i64, message: &str) -> Result<(), String> { let escaped_message = message.replace("\\", "\\\\").replace("\"", "\\\""); let lua_script = format!( r#" local ns = vim.api.nvim_create_namespace("gemini_diagnostics") local diagnostics = {{{{ lnum = {} - 1, col = 0, severity = vim.diagnostic.severity.WARN, message = "{}", }}}} vim.diagnostic.set(ns, 0, diagnostics, {{}}) "#, line, escaped_message ); use rmpv::Value as RmpValue; let req = RmpValue::Array(vec![ RmpValue::Integer(0.into()), RmpValue::Integer(5.into()), // msgid RmpValue::String("nvim_exec_lua".into()), RmpValue::Array(vec![ RmpValue::String(lua_script.into()), RmpValue::Array(vec![]), ]), ]); let resp = call_nvim(req).await?; if let RmpValue::Array(arr) = resp { if !arr[2].is_nil() { return Err(format!("Neovim error: {:?}", arr[2])); } return Ok(()); } Err("Invalid response".to_string()) } fn rmpv_to_json(val: &rmpv::Value) -> serde_json::Value { match val { rmpv::Value::Nil => serde_json::Value::Null, rmpv::Value::Boolean(b) => serde_json::json!(*b), rmpv::Value::Integer(i) => { if let Some(n) = i.as_i64() { serde_json::json!(n) } else if let Some(n) = i.as_u64() { serde_json::json!(n) } else { serde_json::Value::Null } } rmpv::Value::F32(f) => serde_json::json!(f), rmpv::Value::F64(f) => serde_json::json!(f), rmpv::Value::String(s) => { if let Some(str_val) = s.as_str() { serde_json::json!(str_val) } else { serde_json::Value::Null } } rmpv::Value::Array(arr) => { let vec: Vec = arr.iter().map(rmpv_to_json).collect(); serde_json::Value::Array(vec) } rmpv::Value::Map(map) => { let mut obj = serde_json::Map::new(); for (k, v) in map { let key_str = if let rmpv::Value::String(s) = k { s.as_str().unwrap_or("").to_string() } else { format!("{:?}", k) }; obj.insert(key_str, rmpv_to_json(v)); } serde_json::Value::Object(obj) } _ => serde_json::json!(format!("{:?}", val)), } } async fn execute_nvim_lua(code: &str) -> Result { use rmpv::Value as RmpValue; let req = RmpValue::Array(vec![ RmpValue::Integer(0.into()), RmpValue::Integer(6.into()), // msgid RmpValue::String("nvim_exec_lua".into()), RmpValue::Array(vec![RmpValue::String(code.into()), RmpValue::Array(vec![])]), ]); let resp = call_nvim(req).await?; if let RmpValue::Array(arr) = resp { if !arr[2].is_nil() { return Err(format!("Neovim error: {:?}", arr[2])); } if arr.len() > 3 { return Ok(serde_json::to_string_pretty(&rmpv_to_json(&arr[3])).unwrap_or_default()); } return Ok("".to_string()); } Err("Invalid response".to_string()) } pub async fn run_mcp_loop(app_name: &str, app_version: &str) { if std::env::args().any(|arg| arg == "--version") { println!( "{} {} ({})", app_name, app_version, std::env::var("GIT_HASH").unwrap_or_else(|_| "unknown".to_string()) ); return; } let _guard = init_logging(app_name); tracing::info!("{} MCP server started", app_name); let mut stdin = tokio::io::BufReader::new(tokio::io::stdin()); loop { let msg = match read_message(&mut stdin).await { Some(m) => { tracing::info!("Received message method: {}", m.method); m } None => { tracing::info!("Stdin closed, exiting loop"); break; } }; let app_name = app_name.to_string(); let app_version = app_version.to_string(); tokio::spawn(async move { let id = msg.id.clone().unwrap_or(json!(null)); let id_clone = id.clone(); let start_time = std::time::Instant::now(); let method_clone = if msg.method == "tools/call" { let tool_name = msg.params.as_ref().and_then(|p| p.get("name")).and_then(|n| n.as_str()).unwrap_or("unknown"); format!("ToolCall[{}]", tool_name) } else { msg.method.clone() }; match msg.method.as_str() { "initialize" => { let init = rmcp::model::InitializeResult::new( rmcp::model::ServerCapabilities::builder() .enable_tools() .build(), ) .with_server_info(rmcp::model::Implementation::new( app_name.clone(), app_version.clone(), )) .with_protocol_version(rmcp::model::ProtocolVersion::V_2024_11_05); send_response(JsonRpcResponse { jsonrpc: "2.0".to_string(), id, result: Some(serde_json::to_value(init).unwrap()), error: None, }) .await; } "notifications/initialized" => {} "tools/list" => { send_response(JsonRpcResponse { jsonrpc: "2.0".to_string(), id, result: Some(json!({ "tools": [ { "name": "nvim_goto_line", "description": "Open a file and jump to a specific line", "inputSchema": { "type": "object", "properties": { "file": { "type": "string" }, "line": { "type": "integer" } }, "required": ["file", "line"] } }, { "name": "nvim_get_active_buffer", "description": "Get the contents of the currently active Neovim buffer", "inputSchema": { "type": "object", "properties": {} } }, { "name": "nvim_get_cursor", "description": "Get the current cursor position (line and column) in the active Neovim buffer", "inputSchema": { "type": "object", "properties": {} } }, { "name": "nvim_get_visual_selection", "description": "Get the text that is currently highlighted or was last highlighted in Visual mode", "inputSchema": { "type": "object", "properties": {} } }, { "name": "nvim_set_diagnostics", "description": "Push a diagnostic message (like an LSP warning) to a specific line in the active buffer", "inputSchema": { "type": "object", "properties": { "line": { "type": "integer" }, "message": { "type": "string" } }, "required": ["line", "message"] } }, { "name": "nvim_list_buffers", "description": "List all open buffers in Neovim", "inputSchema": { "type": "object", "properties": {} } }, { "name": "nvim_get_diagnostics", "description": "Get all diagnostics for the current active buffer", "inputSchema": { "type": "object", "properties": {} } }, { "name": "nvim_execute_lua", "description": "Execute arbitrary Lua code in Neovim and return the result (JSON serialized).", "inputSchema": { "type": "object", "properties": { "code": { "type": "string" } }, "required": ["code"] } } ] })), error: None, }).await; } "tools/call" => { let params = msg.params.unwrap_or(json!({})); let name = params.get("name").and_then(|n| n.as_str()).unwrap_or(""); let default_args = json!({}); let args = params.get("arguments").unwrap_or(&default_args); match name { "nvim_goto_line" => { if let (Some(file), Some(line)) = ( args.get("file").and_then(|v| v.as_str()), args.get("line").and_then(|v| v.as_i64()), ) { let escaped_file = file.replace("\\", "\\\\"); let cmd = format!("e {} | {} | normal! zz", escaped_file, line); match send_nvim_command(&cmd).await { Ok(_) => { send_response(JsonRpcResponse { jsonrpc: "2.0".to_string(), id, result: Some(json!({ "content": [{"type": "text", "text": format!("Successfully jumped to {} line {}", file, line)}] })), error: None, }).await; } Err(e) => send_error(id, -32603, &e).await, } } else { send_error(id, -32602, "Missing 'file' or 'line'").await; } } "nvim_get_active_buffer" => match get_nvim_active_buffer().await { Ok(content) => { send_response(JsonRpcResponse { jsonrpc: "2.0".to_string(), id, result: Some(json!({ "content": [{"type": "text", "text": content}] })), error: None, }) .await; } Err(e) => send_error(id, -32603, &e).await, }, "nvim_get_cursor" => match get_nvim_cursor().await { Ok(content) => { send_response(JsonRpcResponse { jsonrpc: "2.0".to_string(), id, result: Some(json!({ "content": [{"type": "text", "text": content}] })), error: None, }) .await; } Err(e) => send_error(id, -32603, &e).await, }, "nvim_get_visual_selection" => match get_nvim_visual_selection().await { Ok(content) => { send_response(JsonRpcResponse { jsonrpc: "2.0".to_string(), id, result: Some(json!({ "content": [{"type": "text", "text": content}] })), error: None, }) .await; } Err(e) => send_error(id, -32603, &e).await, }, "nvim_set_diagnostics" => { if let (Some(line), Some(message)) = ( args.get("line").and_then(|v| v.as_i64()), args.get("message").and_then(|v| v.as_str()), ) { match set_nvim_diagnostics(line, message).await { Ok(_) => { send_response(JsonRpcResponse { jsonrpc: "2.0".to_string(), id, result: Some(json!({ "content": [{"type": "text", "text": format!("Successfully set diagnostic on line {}", line)}] })), error: None, }).await; } Err(e) => send_error(id, -32603, &e).await, } } else { send_error(id, -32602, "Missing 'line' or 'message'").await; } } "nvim_list_buffers" => { let lua_code = r#" local bufs = vim.api.nvim_list_bufs() local result = {} for _, buf in ipairs(bufs) do if vim.api.nvim_buf_is_loaded(buf) then local name = vim.api.nvim_buf_get_name(buf) table.insert(result, { id = buf, name = name }) end end return vim.fn.json_encode(result) "#; match execute_nvim_lua(lua_code).await { Ok(result) => { send_response(JsonRpcResponse { jsonrpc: "2.0".to_string(), id, result: Some(json!({ "content": [{"type": "text", "text": result}] })), error: None, }) .await; } Err(e) => send_error(id, -32603, &e).await, } } "nvim_get_diagnostics" => { let lua_code = r#" local bufnr = vim.api.nvim_get_current_buf() local diagnostics = vim.diagnostic.get(bufnr) local result = {} for _, d in ipairs(diagnostics) do table.insert(result, { lnum = d.lnum, col = d.col, severity = d.severity, message = d.message, source = d.source }) end return vim.fn.json_encode(result) "#; match execute_nvim_lua(lua_code).await { Ok(result) => { send_response(JsonRpcResponse { jsonrpc: "2.0".to_string(), id, result: Some(json!({ "content": [{"type": "text", "text": result}] })), error: None, }) .await; } Err(e) => send_error(id, -32603, &e).await, } } "nvim_execute_lua" => { if let Some(code) = args.get("code").and_then(|v| v.as_str()) { match execute_nvim_lua(code).await { Ok(result) => { send_response(JsonRpcResponse { jsonrpc: "2.0".to_string(), id, result: Some(json!({ "content": [{"type": "text", "text": result}] })), error: None, }) .await; } Err(e) => send_error(id, -32603, &e).await, } } else { send_error(id, -32602, "Missing 'code'").await; } } _ => send_error(id, -32601, "Method not found").await, } } _ => { if !id.is_null() { send_error(id, -32601, "Method not found").await; } else { // Ignore notifications silently } } } let elapsed = start_time.elapsed(); tracing::info!("<<< [Nvim] {} (id: {}) completed in {:?}", method_clone, id_clone, elapsed); }); } } fn init_logging(app_name: &str) -> tracing_appender::non_blocking::WorkerGuard { let log_dir = dirs::home_dir() .unwrap_or_default() .join(".gemini/mcp_memory/logs"); std::fs::create_dir_all(&log_dir).unwrap_or_default(); let file_appender = tracing_appender::rolling::daily(log_dir, format!("{}.log", app_name)); let (non_blocking, guard) = tracing_appender::non_blocking(file_appender); let _ = tracing_subscriber::fmt() .with_writer(non_blocking) .with_ansi(false) .with_max_level(tracing::Level::INFO) .try_init(); guard } #[cfg(test)] mod tests { use super::*; #[test] fn test_rmpv_to_json_primitives() { assert_eq!(rmpv_to_json(&rmpv::Value::Nil), serde_json::Value::Null); assert_eq!(rmpv_to_json(&rmpv::Value::Boolean(true)), json!(true)); assert_eq!(rmpv_to_json(&rmpv::Value::Integer(42.into())), json!(42)); assert_eq!( rmpv_to_json(&rmpv::Value::String("hello".into())), json!("hello") ); } #[test] fn test_rmpv_to_json_array() { let arr = rmpv::Value::Array(vec![ rmpv::Value::Integer(1.into()), rmpv::Value::String("test".into()), ]); assert_eq!(rmpv_to_json(&arr), json!([1, "test"])); } #[test] fn test_rmpv_to_json_map() { let mut map = vec![]; map.push(( rmpv::Value::String("key1".into()), rmpv::Value::Integer(100.into()), )); let rmp_map = rmpv::Value::Map(map); let json_map = rmpv_to_json(&rmp_map); assert_eq!(json_map, json!({ "key1": 100 })); } #[tokio::test] async fn test_read_message_jsonl() { let input = "{\"jsonrpc\": \"2.0\", \"id\": 1, \"method\": \"test\"}\n"; let mut reader = BufReader::new(input.as_bytes()); let req = read_message(&mut reader).await.unwrap(); assert_eq!(req.method, "test"); } #[tokio::test] async fn test_read_message_http_headers() { let payload = "{\"jsonrpc\": \"2.0\", \"id\": 2, \"method\": \"test2\"}"; let input = format!("Content-Length: {}\r\n\r\n{}", payload.len(), payload); let mut reader = BufReader::new(input.as_bytes()); let req = read_message(&mut reader).await.unwrap(); assert_eq!(req.method, "test2"); } #[tokio::test] async fn test_read_message_malformed() { let input = "Content-Length: abc\r\n\r\n{}"; let mut reader = BufReader::new(input.as_bytes()); let req = read_message(&mut reader).await; assert!(req.is_none()); } }