Stream MCP tool call output as log message notifications

src/main/java/rife/bld/operations/McpOperation.java

@@ -91,6 +91,11 @@ public class McpOperation extends AbstractOperation<McpOperation> {
      */
     public static final String RESOURCE_DEPENDENCY_TREE = "bld://dependency-tree";
 
+    // the RFC 5424 syslog severities that MCP logging uses, from the least
+    // to the most severe, the streamed build output is sent at info
+    private static final List<String> LOG_LEVELS = List.of(
+        "debug", "info", "notice", "warning", "error", "critical", "alert", "emergency");
+
     private BuildExecutor executor_;
     private BaseProject project_ = null;
     private String serverTitle_ = null;
@@ -101,6 +106,11 @@ public class McpOperation extends AbstractOperation<McpOperation> {
     private final Set<String> excludedCommands_ = new LinkedHashSet<>(List.of("mcp"));
     private final Set<String> confirmationCommands_ = new LinkedHashSet<>(List.of("publish"));
     private final Set<Process> activeProcesses_ = ConcurrentHashMap.newKeySet();
+    // the writer of the running server loop, log notifications from the
+    // output reader threads are interleaved with the responses through it
+    private volatile PrintWriter notificationWriter_ = null;
+    private volatile String logLevel_ = "info";
+    private final Object writeLock_ = new Object();
 
     private static final class ResourceNotFoundException extends RuntimeException {
         ResourceNotFoundException(String uri) {
@@ -171,20 +181,65 @@ public class McpOperation extends AbstractOperation<McpOperation> {
      */
     protected void executeServerLoop(BufferedReader reader, PrintWriter writer)
     throws IOException {
-        String line;
-        while ((line = reader.readLine()) != null) {
-            if (line.isBlank()) {
-                continue;
-            }
-            var response = processMessage(line);
-            if (response != null) {
-                writer.print(response);
-                writer.print('\n');
-                writer.flush();
+        // the writer is shared with the output streaming of tool calls
+        // while the loop runs
+        notificationWriter_ = writer;
+        try {
+            String line;
+            while ((line = reader.readLine()) != null) {
+                if (line.isBlank()) {
+                    continue;
+                }
+                var response = processMessage(line);
+                if (response != null) {
+                    writeProtocolMessage(writer, response);
+                }
             }
+        } finally {
+            notificationWriter_ = null;
         }
     }
 
+    // responses and log notifications are written as complete lines under
+    // a shared lock, so that concurrent streaming can never interleave
+    // inside a message
+    private void writeProtocolMessage(PrintWriter writer, String message) {
+        synchronized (writeLock_) {
+            writer.print(message);
+            writer.print('\n');
+            writer.flush();
+        }
+    }
+
+    /**
+     * Part of the {@link #execute} operation, sends the console output of
+     * a running tool call as an MCP log message notification.
+     * <p>
+     * The output is sent at the {@code info} level with the command name
+     * as the logger, when the client set a more severe minimum level with
+     * {@code logging/setLevel} the output isn't streamed. Nothing is sent
+     * when no server loop is running.
+     *
+     * @param command the name of the build command that produced the output
+     * @param data    the chunk of console output to send
+     * @since 2.4.0
+     */
+    protected void sendToolCallOutput(String command, String data) {
+        var writer = notificationWriter_;
+        if (writer == null || LOG_LEVELS.indexOf(logLevel_) > LOG_LEVELS.indexOf("info")) {
+            return;
+        }
+        var notification = new JsonObject()
+            .set("jsonrpc", "2.0")
+            .set("method", "notifications/message")
+            .object("params", p -> p
+                .set("level", "info")
+                .set("logger", command)
+                .set("data", data))
+            .toString();
+        writeProtocolMessage(writer, notification);
+    }
+
     /**
      * Part of the {@link #execute} operation, processes a single JSON-RPC
      * message and returns the response.
@@ -252,6 +307,7 @@ public class McpOperation extends AbstractOperation<McpOperation> {
                     yield processResourcesList();
                 }
                 case "resources/read" -> processResourcesRead(params);
+                case "logging/setLevel" -> processLoggingSetLevel(params);
                 default -> null;
             };
         } catch (InvalidParamsException e) {
@@ -334,7 +390,8 @@ public class McpOperation extends AbstractOperation<McpOperation> {
             .set("protocolVersion", version)
             .object("capabilities", c -> c
                 .object("tools", t -> {})
-                .object("resources", r -> {}))
+                .object("resources", r -> {})
+                .object("logging", l -> {}))
             .object("serverInfo", s -> {
                 s.set("name", "bld");
                 if (serverTitle_ != null) {
@@ -345,6 +402,28 @@ public class McpOperation extends AbstractOperation<McpOperation> {
             .set("instructions", instructions_ == null ? defaultInstructions() : instructions_);
     }
 
+    /**
+     * Part of the {@link #execute} operation, handles the MCP
+     * {@code logging/setLevel} request.
+     * <p>
+     * The minimum level determines whether the console output of running
+     * tool calls is streamed as log message notifications, the output is
+     * sent at the {@code info} level.
+     *
+     * @param params the parameters of the set level request
+     * @return the empty set level result
+     * @since 2.4.0
+     */
+    protected JsonObject processLoggingSetLevel(JsonObject params) {
+        if (params == null ||
+            !(params.get("level") instanceof String level) ||
+            !LOG_LEVELS.contains(level)) {
+            throw new InvalidParamsException("Invalid logging level");
+        }
+        logLevel_ = level;
+        return new JsonObject();
+    }
+
     /**
      * Part of the {@link #execute} operation, generates the default
      * instructions that the server reports during initialization when
@@ -518,6 +597,25 @@ public class McpOperation extends AbstractOperation<McpOperation> {
      * @since 2.4.0
      */
     protected JsonObject executeToolCall(String command, List<String> arguments, boolean enforceExclusions) {
+        return executeToolCall(command, arguments, enforceExclusions, true);
+    }
+
+    /**
+     * Part of the {@link #execute} operation, executes a single build
+     * command with the provided arguments as a separate build process,
+     * capturing its console output.
+     *
+     * @param command           the name of the build command to execute
+     * @param arguments          the arguments to pass to the build command
+     * @param enforceExclusions  whether the excluded commands are refused
+     * @param streamOutput       whether the console output is streamed as
+     *                           log message notifications while the
+     *                           command runs, the internal resource
+     *                           generation doesn't stream
+     * @return the tool call result with the captured console output
+     * @since 2.4.0
+     */
+    protected JsonObject executeToolCall(String command, List<String> arguments, boolean enforceExclusions, boolean streamOutput) {
         var output = new StringBuilder();
         var truncated = new boolean[]{false};
         // the descendants are collected while the process is alive, so that
@@ -562,7 +660,7 @@ public class McpOperation extends AbstractOperation<McpOperation> {
         var final_status_file = status_file;
         var final_control_file = control_file;
         try {
-            return runToolCallProcess(command, process, descendants, output, truncated, final_status_file);
+            return runToolCallProcess(command, process, descendants, output, truncated, final_status_file, streamOutput);
         } finally {
             deleteQuietly(final_control_file);
             deleteQuietly(final_status_file);
@@ -577,7 +675,7 @@ public class McpOperation extends AbstractOperation<McpOperation> {
 
 
     private JsonObject runToolCallProcess(String command, Process process, Set<ProcessHandle> descendants,
-                                          StringBuilder output, boolean[] truncated, File status_file) {
+                                          StringBuilder output, boolean[] truncated, File status_file, boolean stream_output) {
 
         var error = false;
         var timed_out = false;
@@ -612,15 +710,25 @@ public class McpOperation extends AbstractOperation<McpOperation> {
                     var buffer = new char[8192];
                     int read;
                     while ((read = stream.read(buffer)) != -1) {
+                        String streamed = null;
                         synchronized (output) {
                             var remaining = outputLimit_ - output.length();
                             if (remaining > 0) {
-                                output.append(buffer, 0, Math.min(read, remaining));
+                                var appended = Math.min(read, remaining);
+                                output.append(buffer, 0, appended);
+                                if (stream_output) {
+                                    streamed = new String(buffer, 0, appended);
+                                }
                             }
                             if (read > remaining) {
                                 truncated[0] = true;
                             }
                         }
+                        // the captured chunk is also streamed live as a log
+                        // message notification, outside of the output lock
+                        if (streamed != null) {
+                            sendToolCallOutput(command, streamed);
+                        }
                     }
                 } catch (IOException e) {
                     // the stream ended, the process is gone
@@ -976,8 +1084,9 @@ public class McpOperation extends AbstractOperation<McpOperation> {
     protected String readDependencyTreeResource() {
         // the resource generation runs the command internally and doesn't
         // enforce the tool exclusions, so that excluding the tool doesn't
-        // break the still advertised resource
-        var result = executeToolCall("dependency-tree", List.of(), false);
+        // break the still advertised resource, its output is the resource
+        // itself and isn't streamed
+        var result = executeToolCall("dependency-tree", List.of(), false, false);
         var text = result.getArray("content").getObject(0).getString("text");
         if (result.getBoolean("isError")) {
             throw new RuntimeException("Unable to generate the dependency tree:\n" + text);

src/test/java/rife/bld/operations/TestMcpOperation.java

@@ -24,6 +24,7 @@ import java.io.PrintWriter;
 import java.io.StringReader;
 import java.io.StringWriter;
 import java.nio.file.Files;
+import java.util.ArrayList;
 import java.util.List;
 
 import static org.junit.jupiter.api.Assertions.*;
@@ -174,6 +175,7 @@ public class TestMcpOperation {
         // a supported requested protocol version is echoed
         assertEquals("2025-06-18", result.getString("protocolVersion"));
         assertNotNull(result.getObject("capabilities").getObject("tools"));
+        assertNotNull(result.getObject("capabilities").getObject("logging"));
         assertEquals("bld", result.getObject("serverInfo").getString("name"));
         assertEquals("mcp_test", result.getObject("serverInfo").getString("title"));
         assertFalse(result.getObject("serverInfo").getString("version").isEmpty());
@@ -950,14 +952,21 @@ public class TestMcpOperation {
             var output = new StringWriter();
             operation.executeServerLoop(new BufferedReader(new StringReader(input)), new PrintWriter(output));
 
-            var responses = output.toString().trim().split("\n");
-            assertEquals(4, responses.length);
+            // the streamed log notifications are interleaved with the
+            // responses, only the responses are counted here
+            var responses = new ArrayList<JsonObject>();
+            for (var line : output.toString().trim().split("\n")) {
+                var message = Json.parseObject(line);
+                if (message.getString("method") == null) {
+                    responses.add(message);
+                }
+            }
+            assertEquals(4, responses.size());
 
             // both compile tool calls executed and reported their output,
             // repeated calls behave like separate command line invocations
             for (var i = 1; i <= 2; ++i) {
-                var compile = Json.parseObject(responses[i]);
-                var compile_result = compile.getObject("result");
+                var compile_result = responses.get(i).getObject("result");
                 assertFalse(compile_result.getBoolean("isError"));
                 assertTrue(compile_result.getArray("content").getObject(0).getString("text").contains("Compilation finished successfully"));
             }
@@ -967,7 +976,7 @@ public class TestMcpOperation {
             assertTrue(listing.contains(".class"), listing);
 
             // the project resource reflects the created project
-            var read_project = Json.parseObject(responses[3]);
+            var read_project = responses.get(3);
             var description = Json.parseObject(read_project.getObject("result").getArray("contents").getObject(0).getString("text"));
             assertEquals("myapp", description.getString("name"));
         } finally {
@@ -975,6 +984,86 @@ public class TestMcpOperation {
         }
     }
 
+    @Test
+    void testToolCallOutputIsStreamed()
+    throws Exception {
+        var operation = createOperation();
+        var input = """
+            {"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-06-18","capabilities":{},"clientInfo":{"name":"test","version":"1.0"}}}
+            {"jsonrpc":"2.0","method":"notifications/initialized"}
+            {"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"hello","arguments":{"arguments":["stream"]}}}
+            """;
+        var output = new StringWriter();
+        operation.executeServerLoop(new BufferedReader(new StringReader(input)), new PrintWriter(output));
+
+        var lines = output.toString().trim().split("\n");
+        // the console output arrives live as log message notifications
+        // before the tool call response
+        var streamed = new StringBuilder();
+        var notifications = 0;
+        for (var line : lines) {
+            var message = Json.parseObject(line);
+            if ("notifications/message".equals(message.getString("method"))) {
+                ++notifications;
+                var params = message.getObject("params");
+                assertEquals("info", params.getString("level"));
+                assertEquals("hello", params.getString("logger"));
+                streamed.append(params.getString("data"));
+            }
+        }
+        assertTrue(notifications >= 1);
+        assertTrue(streamed.toString().contains("hello stream"), streamed.toString());
+
+        // the tool call response still carries the full output and comes
+        // after the streamed notifications
+        var last = Json.parseObject(lines[lines.length - 1]);
+        assertEquals(2, last.getInt("id"));
+        assertTrue(last.getObject("result").getArray("content").getObject(0).getString("text").contains("hello stream"));
+    }
+
+    @Test
+    void testLoggingSetLevelSuppressesStreaming()
+    throws Exception {
+        var operation = createOperation();
+        var input = """
+            {"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-06-18","capabilities":{},"clientInfo":{"name":"test","version":"1.0"}}}
+            {"jsonrpc":"2.0","method":"notifications/initialized"}
+            {"jsonrpc":"2.0","id":2,"method":"logging/setLevel","params":{"level":"warning"}}
+            {"jsonrpc":"2.0","id":3,"method":"tools/call","params":{"name":"hello","arguments":{"arguments":["quiet"]}}}
+            """;
+        var output = new StringWriter();
+        operation.executeServerLoop(new BufferedReader(new StringReader(input)), new PrintWriter(output));
+
+        var lines = output.toString().trim().split("\n");
+        // a more severe minimum level suppresses the output streaming,
+        // only the three responses are written
+        assertEquals(3, lines.length);
+        for (var line : lines) {
+            assertNull(Json.parseObject(line).getString("method"));
+        }
+        // the tool call response still carries the full output
+        var last = Json.parseObject(lines[2]);
+        assertTrue(last.getObject("result").getArray("content").getObject(0).getString("text").contains("hello quiet"));
+    }
+
+    @Test
+    void testLoggingSetLevelValidation() {
+        var operation = createOperation();
+
+        // the params and a valid RFC 5424 level are required
+        var missing = process(operation, """
+            {"jsonrpc":"2.0","id":1,"method":"logging/setLevel"}""");
+        assertEquals(-32602, missing.getObject("error").getInt("code"));
+        var bogus = process(operation, """
+            {"jsonrpc":"2.0","id":2,"method":"logging/setLevel","params":{"level":"loud"}}""");
+        assertEquals(-32602, bogus.getObject("error").getInt("code"));
+
+        // a valid level is acknowledged with an empty result
+        var valid = process(operation, """
+            {"jsonrpc":"2.0","id":3,"method":"logging/setLevel","params":{"level":"debug"}}""");
+        assertTrue(valid.getObject("result").isEmpty());
+    }
+
     @Test
     void testServerLoop()
     throws Exception {
@@ -988,10 +1077,18 @@ public class TestMcpOperation {
         var output = new StringWriter();
         operation.executeServerLoop(new BufferedReader(new StringReader(input)), new PrintWriter(output));
 
-        var responses = output.toString().trim().split("\n");
-        assertEquals(2, responses.length);
-        assertEquals(1, Json.parseObject(responses[0]).getInt("id"));
-        var call = Json.parseObject(responses[1]);
+        // the streamed log notifications are interleaved with the
+        // responses, only the responses are counted here
+        var responses = new ArrayList<JsonObject>();
+        for (var line : output.toString().trim().split("\n")) {
+            var message = Json.parseObject(line);
+            if (message.getString("method") == null) {
+                responses.add(message);
+            }
+        }
+        assertEquals(2, responses.size());
+        assertEquals(1, responses.get(0).getInt("id"));
+        var call = responses.get(1);
         assertEquals(2, call.getInt("id"));
         assertEquals("hello loop", call.getObject("result").getArray("content").getObject(0).getString("text").trim());
     }