opensource/mateclaw
2
0
Fork 0
mirror of https://gitee.com/mateos/mateclaw.git synced 2026-09-13 11:13:43 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

fix(mcp): resilient inbound processing for non-JSON stdout from MCP servers

Browse Source 
The upstream StdioClientTransport breaks out of its inbound read loop on
any JSON parse error, permanently killing the reader thread. Some MCP
servers write non-JSON debug output to stdout (e.g.
"=== Document parser messages ==="), which triggers this and causes all
subsequent valid JSON-RPC responses to be lost → 30s timeout → agent stuck.

Override connect() in CwdAwareStdioClientTransport to skip non-JSON lines
(log at DEBUG) instead of breaking, keeping the inbound thread alive.

Closes #226
This commit is contained in:
倪程伟 2026-05-26 16:41:33 +08:00 committed by matevip
parent 9488634372
commit 7f4d3e3ab9
1 changed files with 193 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

193
mateclaw-server/src/main/java/vip/mate/tool/mcp/runtime/CwdAwareStdioClientTransport.java
View File

@ -3,12 +3,22 @@ package vip.mate.tool.mcp.runtime;
import io.modelcontextprotocol.client.transport.ServerParameters;
import io.modelcontextprotocol.client.transport.StdioClientTransport;
import io.modelcontextprotocol.json.McpJsonMapper;
import io.modelcontextprotocol.spec.McpSchema;
import lombok.extern.slf4j.Slf4j;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Sinks;
import reactor.core.scheduler.Schedulers;
import java.io.BufferedReader;
import java.io.File;
import java.io.InputStreamReader;
import java.lang.reflect.Field;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.function.Function;
/**
* Enhanced stdio MCP transport with:
@ -16,6 +26,9 @@ import java.util.Map;
* <li>Working directory (cwd) support</li>
* <li>Automatic PATH enrichment for Desktop app environments where
* Node.js/npx may not be in the JRE process's PATH</li>
* <li>Resilient inbound processing: non-JSON lines from the server
* (e.g. debug output) are skipped instead of killing the
* inbound reader thread</li>
* </ul>
*/
@Slf4j
@ -52,6 +65,186 @@ public class CwdAwareStdioClientTransport extends StdioClientTransport {
this.cwd = cwd;
}
/**
* Override parent {@code connect()} to add resilient inbound processing.
*
* <p>The upstream {@link StdioClientTransport} breaks out of its inbound
* read loop on any JSON parse error, killing the reader thread permanently.
* Some MCP servers write non-JSON debug output to stdout (e.g.
* {@code "=== Document parser messages ==="}), which triggers this and
* causes subsequent valid JSON-RPC responses to be lost → 30 s timeout.
*
* <p>This override replaces the inbound processing with a version that
* {@link Log#debug logs} and {@code continue}s past non-JSON lines
* instead of breaking. Outbound and error processing remain unchanged.
*/
@Override
@SuppressWarnings("unchecked")
public Mono<Void> connect(
Function<Mono<McpSchema.JSONRPCMessage>, Mono<McpSchema.JSONRPCMessage>> handler) {
return Mono.<Void>fromRunnable(() -> {
log.info("MCP server starting (resilient mode).");
// Wire up the parent's inbound + error sinks so downstream
// McpSyncClient message dispatch works unchanged.
Sinks.Many<McpSchema.JSONRPCMessage> inboundSink = getPrivateField("inboundSink");
Sinks.Many<String> errorSink = getPrivateField("errorSink");
if (inboundSink != null) {
inboundSink.asFlux()
.flatMap(msg -> Mono.just(msg).transform(handler))
.subscribe();
}
if (errorSink != null) {
errorSink.asFlux().subscribe(line -> log.info("MCP stdio stderr: {}", line));
}
// Build and start the server process (reuses parent's ProcessBuilder hook)
List<String> fullCommand = new ArrayList<>();
fullCommand.add(params().getCommand());
fullCommand.addAll(params().getArgs());
ProcessBuilder processBuilder = getProcessBuilder();
processBuilder.command(fullCommand);
processBuilder.environment().putAll(params().getEnv());
Process process;
try {
process = processBuilder.start();
} catch (Exception e) {
throw new RuntimeException("Failed to start MCP process: " + fullCommand, e);
}
if (process.getInputStream() == null || process.getOutputStream() == null) {
process.destroy();
throw new RuntimeException("MCP process input or output stream is null");
}
// --- Resilient inbound reader (the key fix) ---
Thread inboundThread = new Thread(() -> {
try (BufferedReader reader = new BufferedReader(
new InputStreamReader(process.getInputStream()))) {
String line;
while ((line = reader.readLine()) != null) {
try {
McpSchema.JSONRPCMessage message =
McpSchema.deserializeJsonRpcMessage(jsonMapper(), line);
if (inboundSink != null && !inboundSink.tryEmitNext(message).isSuccess()) {
log.error("Failed to enqueue inbound MCP message");
break;
}
} catch (Exception e) {
// Non-JSON line from server (debug output, etc.) — skip it.
log.debug("MCP server stdout non-JSON line (skipped): {}", line);
}
}
} catch (Exception e) {
log.error("MCP inbound reader error", e);
} finally {
if (inboundSink != null) inboundSink.tryEmitComplete();
if (errorSink != null) errorSink.tryEmitComplete();
}
}, "mcp-inbound-resilient");
inboundThread.setDaemon(true);
inboundThread.start();
// Outbound writer — delegate to parent infrastructure
startOutboundFromConnect(process);
// Stderr reader
Thread errThread = new Thread(() -> {
try (BufferedReader errReader = new BufferedReader(
new InputStreamReader(process.getErrorStream()))) {
String line;
while ((line = errReader.readLine()) != null) {
if (errorSink != null) errorSink.tryEmitNext(line);
}
} catch (Exception e) {
log.error("MCP stderr reader error", e);
}
}, "mcp-stderr");
errThread.setDaemon(true);
errThread.start();
log.info("MCP server started (resilient mode).");
}).subscribeOn(Schedulers.boundedElastic());
}
/** Read a private field from the parent {@link StdioClientTransport}. */
@SuppressWarnings("unchecked")
private <T> T getPrivateField(String name) {
try {
Field field = StdioClientTransport.class.getDeclaredField(name);
field.setAccessible(true);
return (T) field.get(this);
} catch (Exception e) {
log.debug("Cannot access parent field {}: {}", name, e.getMessage());
return null;
}
}
/** Accessor for the jsonMapper (needed by the resilient inbound reader). */
private io.modelcontextprotocol.json.McpJsonMapper jsonMapper() {
try {
Field f = StdioClientTransport.class.getDeclaredField("jsonMapper");
f.setAccessible(true);
return (io.modelcontextprotocol.json.McpJsonMapper) f.get(this);
} catch (Exception e) {
throw new RuntimeException("Cannot access jsonMapper", e);
}
}
/** Accessor for the ServerParameters (needed by connect). */
private io.modelcontextprotocol.client.transport.ServerParameters params() {
try {
Field f = StdioClientTransport.class.getDeclaredField("params");
f.setAccessible(true);
return (io.modelcontextprotocol.client.transport.ServerParameters) f.get(this);
} catch (Exception e) {
throw new RuntimeException("Cannot access params", e);
}
}
/**
* Wire up outbound message writing from the parent's outboundSink.
* Reads from the sink's flux and writes JSON to the process stdout.
*/
private void startOutboundFromConnect(Process process) {
try {
Field f = StdioClientTransport.class.getDeclaredField("outboundSink");
f.setAccessible(true);
@SuppressWarnings("unchecked")
Sinks.Many<McpSchema.JSONRPCMessage> outboundSink =
(Sinks.Many<McpSchema.JSONRPCMessage>) f.get(this);
if (outboundSink == null) return;
Thread outThread = new Thread(() -> {
try (var writer = new java.io.BufferedWriter(
new java.io.OutputStreamWriter(process.getOutputStream()))) {
outboundSink.asFlux().subscribe(message -> {
try {
String json = jsonMapper().writeValueAsString(message);
json = json.replace("\r\n", "\\n").replace("\n", "\\n").replace("\r", "\\n");
writer.write(json);
writer.newLine();
writer.flush();
} catch (Exception e) {
log.error("MCP outbound write error", e);
}
});
// Keep thread alive while process runs
process.waitFor();
} catch (Exception e) {
log.error("MCP outbound writer error", e);
}
}, "mcp-outbound-resilient");
outThread.setDaemon(true);
outThread.start();
} catch (Exception e) {
log.warn("Cannot wire outbound sink: {}", e.getMessage());
}
}
@Override
protected ProcessBuilder getProcessBuilder() {
ProcessBuilder builder = super.getProcessBuilder();