merge PR #11 from shikokuchuo/dev: Support multiple proxies for mcp_server() instances

Author: simonpcouch

DESCRIPTION

@@ -4,9 +4,11 @@ Version: 0.0.0.9000
 Authors@R: c(
     person("Simon", "Couch", , "[email protected]", role = c("aut", "cre"),
            comment = c(ORCID = "0000-0001-5676-5107")),
-    person("Winston", "Chang", , "[email protected]", role = "aut",
-           comment = c(ORCID = "0000-0001-5676-5107")),
-    person("Posit Software, PBC", role = c("cph", "fnd"))
+    person("Winston", "Chang", , "[email protected]", role = "aut"),
+    person("Charlie", "Gao", , "[email protected]", role = "aut",
+           comment = c(ORCID = "0000-0002-0750-061X")),
+    person("Posit Software, PBC", role = c("cph", "fnd"),
+           comment = c(ROR = "03wc8by49"))
   )
 Description: The goal of acquaint is to enable LLM-enabled tools like Claude Code to 
     learn about the R packages you have installed using the 
@@ -19,14 +21,14 @@ Suggests:
 Config/testthat/edition: 3
 Encoding: UTF-8
 Roxygen: list(markdown = TRUE)
-RoxygenNote: 7.3.2
+RoxygenNote: 7.3.2.9000
 Imports: 
     btw (>= 0.0.1.9000),
     cli,
     ellmer,
     jsonlite,
     later,
-    nanonext (>= 1.5.2.9009),
+    nanonext (>= 1.5.2.9012),
     promises,
     rlang
 Depends: R (>= 4.1.0)

R/proxy.R

@@ -7,7 +7,8 @@ mcp_proxy <- function() {
   # TODO: should this actually be a check for being called within Rscript or not?
   check_not_interactive()
 
-  the$proxy_socket <- nanonext::socket("pair", dial = acquaint_socket)
+  the$proxy_socket <- nanonext::socket("poly")
+  nanonext::dial(the$proxy_socket, url = sprintf("%s%d", acquaint_socket, 1L))
 
   # Note that we're using file("stdin") instead of stdin(), which are not the
   # same.
@@ -103,6 +104,10 @@ schedule_handle_message_from_client <- function() {
 }
 
 handle_message_from_server <- function(data) {
+  if (!is.character(data)) {
+    return()
+  }
+
   schedule_handle_message_from_server()
 
   logcat("FROM SERVER: ", data)
@@ -119,7 +124,7 @@ schedule_handle_message_from_server <- function() {
 forward_request <- function(data) {
   logcat("TO SERVER: ", data)
 
-  the$saio <- nanonext::send_aio(the$proxy_socket, data, mode = "raw")
+  nanonext::send_aio(the$proxy_socket, data, mode = "raw")
 }
 
 # This process will be launched by the MCP client, so stdout/stderr aren't
@@ -201,3 +206,29 @@ check_not_interactive <- function(call = caller_env()) {
     )
   }
 }
+
+mcp_discover <- function() {
+  sock <- nanonext::socket("poly")
+  on.exit(nanonext:::reap(sock))
+  cv <- nanonext::cv()
+  monitor <- nanonext::monitor(sock, cv)
+  suppressWarnings(
+    for (i in seq_len(1024L)) {
+      nanonext::dial(sock, url = sprintf("%s%d", acquaint_socket, i), autostart = NA) &&
+        break
+    }
+  )
+  pipes <- nanonext::read_monitor(monitor)
+  res <- lapply(seq_along(pipes), function(x) nanonext::recv_aio(sock))
+  lapply(pipes, function(x) nanonext::send_aio(sock, "", mode = "raw", pipe = x))
+  nanonext::collect_aio_(res)
+}
+
+select_server <- function(i) {
+  lapply(the$proxy_socket[["dialer"]], nanonext::reap)
+  attr(the$proxy_socket, "dialer") <- NULL
+  nanonext::dial(
+    the$proxy_socket,
+    url = sprintf("%s%d", acquaint_socket, as.integer(i))
+  )
+}

R/server.R

@@ -53,14 +53,26 @@ mcp_serve <- function() {
     return(invisible())
   }
 
-  the$server_socket <- nanonext::socket("pair", listen = acquaint_socket)
+  the$server_socket <- nanonext::socket("poly")
+  i <- 1L
+  suppressWarnings(
+    while (i < 1024L) { # prevent indefinite loop
+      nanonext::listen(the$server_socket, url = sprintf("%s%d", acquaint_socket, i)) || break
+      i <- i + 1L
+    }
+  )
+
   schedule_handle_message_from_proxy()
 }
 
 handle_message_from_proxy <- function(msg) {
+  pipe <- nanonext::pipe_id(the$raio)
   schedule_handle_message_from_proxy()
 
   # cat("RECV :", msg, "\n", sep = "", file = stderr())
+  if (!nzchar(msg)) {
+    return(nanonext::send_aio(the$server_socket, commandArgs(), pipe = pipe))
+  }
   data <- jsonlite::parse_json(msg)
 
   if (data$method == "tools/call") {
@@ -99,13 +111,12 @@ handle_message_from_proxy <- function(msg) {
   }
   # cat("SEND:", to_json(body), "\n", sep = "", file = stderr())
 
-  # TODO: consider if better / more robust using synchronous sends
-  the$saio <- nanonext::send_aio(the$server_socket, to_json(body), mode = "raw")
+  nanonext::send_aio(the$server_socket, to_json(body), mode = "raw", pipe = pipe)
 }
 
 schedule_handle_message_from_proxy <- function() {
-  r <- nanonext::recv_aio(the$server_socket, mode = "string")
-  promises::as.promise(r)$then(handle_message_from_proxy)$catch(function(e) {
+  the$raio <- nanonext::recv_aio(the$server_socket, mode = "string")
+  promises::as.promise(the$raio)$then(handle_message_from_proxy)$catch(function(e) {
     print(e)
   })
 }