diff --git a/src/main/java/rife/bld/operations/McpOperation.java b/src/main/java/rife/bld/operations/McpOperation.java index 96841b9..35a0c4a 100644 --- a/src/main/java/rife/bld/operations/McpOperation.java +++ b/src/main/java/rife/bld/operations/McpOperation.java @@ -91,6 +91,11 @@ public class McpOperation extends AbstractOperation { */ 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 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 { private final Set excludedCommands_ = new LinkedHashSet<>(List.of("mcp")); private final Set confirmationCommands_ = new LinkedHashSet<>(List.of("publish")); private final Set 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 { */ 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. + *

+ * 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 { 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 { .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 { .set("instructions", instructions_ == null ? defaultInstructions() : instructions_); } + /** + * Part of the {@link #execute} operation, handles the MCP + * {@code logging/setLevel} request. + *

+ * 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 { * @since 2.4.0 */ protected JsonObject executeToolCall(String command, List 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 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 { 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 { private JsonObject runToolCallProcess(String command, Process process, Set 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 { 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 { 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); diff --git a/src/test/java/rife/bld/operations/TestMcpOperation.java b/src/test/java/rife/bld/operations/TestMcpOperation.java index 9b3dcb0..242bcf0 100644 --- a/src/test/java/rife/bld/operations/TestMcpOperation.java +++ b/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(); + 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(); + 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()); }