entry : extraOptions.entrySet()) {
+ args.add("--" + entry.getKey()); args.add(entry.getValue());
+ }
+ for (String flag : extraFlags) { args.add("--" + flag); }
return args;
}
+
+ private static String optionName(String name) {
+ Objects.requireNonNull(name, "name");
+ String normalized = name.startsWith("--") ? name.substring(2) : name;
+ if (!normalized.matches("[A-Za-z0-9][A-Za-z0-9_-]*")) {
+ throw new IllegalArgumentException("invalid option name: " + name);
+ }
+ return normalized;
+ }
}
}
diff --git a/src/main/java/io/github/easy4j/comfy/cli/ComfyCliExecutor.java b/src/main/java/io/github/easy4j/comfy/cli/ComfyCliExecutor.java
index 2ef76b3..eda3d13 100644
--- a/src/main/java/io/github/easy4j/comfy/cli/ComfyCliExecutor.java
+++ b/src/main/java/io/github/easy4j/comfy/cli/ComfyCliExecutor.java
@@ -2,22 +2,13 @@
* Copyright (c) 2018-present, easy-4-java (https://github.com/easy-4-java).
*
* Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
*/
package io.github.easy4j.comfy.cli;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
+import java.io.OutputStream;
import java.nio.charset.StandardCharsets;
import java.util.LinkedHashMap;
import java.util.Map;
@@ -34,34 +25,11 @@
import io.github.easy4j.comfy.ComfyClientConfig;
/**
- * Thin wrapper around Apache Commons {@code exec} that launches the local
- * {@code comfy} CLI as a child process.
- *
- * Every call to {@link #execute(String...)} performs the following steps:
- *
- * - Build a {@link CommandLine} rooted at {@link ComfyClientConfig#getLocalExecutable()}.
- * - Append each non-{@code null} argument via
- * {@link CommandLine#addArgument(String, boolean)} with
- * {@code handleQuoting=false} — the child is spawned through
- * {@code exec(argv)}, not a shell, so quoting would embed literal double
- * quotes inside multi-word arguments (prompts, paths) and corrupt them
- * on arrival.
- * - Capture stdout and stderr into in-memory buffers; the child always
- * receives a (possibly empty) stdin pipe that closes right after the
- * payload, so consumers reading to EOF cannot race the input pump.
- * - Run the process under an {@link ExecuteWatchdog} whose timeout is
- * derived from {@link ComfyClientConfig#getLocalTimeoutSeconds()}.
- * - Return a {@link ComfyCliResult} preserving the real exit code and both
- * captured streams.
- *
+ * Synchronous, thread-safe subprocess executor for the local {@code comfy} CLI.
*
- * The class is intentionally synchronous and stateless (apart from the
- * injected configuration) so it can be safely shared between threads and
- * pooled by higher-level components.
- *
- * @author Loong Wan
- * @since 1.0.0
- * @see ComfyCliResult
+ * Arguments are passed as argv entries without a shell, stdout/stderr are
+ * drained concurrently, output capture is bounded, and the real exit code is
+ * preserved for non-zero exits.
*/
public class ComfyCliExecutor {
@@ -69,143 +37,143 @@ public class ComfyCliExecutor {
private final ComfyClientConfig config;
- /**
- * Creates a new executor bound to the given configuration.
- *
- * @param config the runtime configuration providing the executable path,
- * timeout, and probe-timeout settings; must not be {@code null}.
- */
public ComfyCliExecutor(ComfyClientConfig config) {
this.config = Objects.requireNonNull(config, "config");
}
- /**
- * Runs the {@code kimi} executable with the given CLI arguments.
- *
- * {@code null} entries in {@code args} are skipped silently to make
- * varargs usage easier. Failure modes:
- *
- * - Process timeout — {@link ComfyCliResult#isTimeout()} returns
- * {@code true}; exit code is {@code -1}; stderr starts with the
- * timeout notice.
- * - Non-zero process exit — the real exit code is preserved in
- * {@link ComfyCliResult#getExitCode()}, and both captured streams are
- * returned as-is.
- * - IOException (missing executable, permission denied, etc.) —
- * the {@link IOException#getMessage()} is captured in
- * {@link ComfyCliResult#getStderr()} and the exit code is {@code -1}.
- *
- *
- * @param args CLI arguments to pass to the {@code kimi} binary.
- * @return a {@link ComfyCliResult} describing the outcome; never {@code null}.
- */
public ComfyCliResult execute(String... args) {
- return runProcess(null, args);
+ return runProcess(null, secondsToMillis(config.getLocalTimeoutSeconds()), args);
}
- /**
- * Runs the {@code kimi} executable with the given CLI arguments, feeding
- * {@code stdin} to the child process.
- *
- * Used by commands that read their payload from standard input. A
- * {@code null} or empty {@code stdin} behaves exactly like
- * {@link #execute(String...)} — the child receives an immediately
- * closing pipe. Failure modes are identical to the varargs overload.
- *
- * @param stdin optional text piped to the child process's standard input.
- * @param args CLI arguments to pass to the {@code kimi} binary.
- * @return a {@link ComfyCliResult} describing the outcome; never {@code null}.
- */
public ComfyCliResult executeWithStdin(String stdin, String... args) {
- return runProcess(stdin, args);
+ return runProcess(stdin, secondsToMillis(config.getLocalTimeoutSeconds()), args);
+ }
+
+ ComfyCliResult executeWithTimeoutSeconds(int timeoutSeconds, String... args) {
+ return runProcess(null, secondsToMillis(timeoutSeconds), args);
}
- /**
- * Lightweight reachability probe used by {@code KimiClient#isAvailable()}.
- *
- * Runs {@code kimi --version} with the configured timeout and returns
- * {@code true} only if the process exits with status {@code 0}. Any
- * exception (missing executable, non-zero exit, timeout) is swallowed and
- * reported as {@code false} so callers can use the result without a
- * try/catch block.
- *
- * @return {@code true} if the local CLI is reachable and reports a version,
- * {@code false} otherwise.
- */
public boolean probe() {
try {
- ComfyCliResult result = execute("--version");
- return result.isSuccess();
- } catch (Exception e) {
+ return executeWithTimeoutSeconds(config.getLocalProbeTimeoutSeconds(), "--version").isSuccess();
+ } catch (RuntimeException e) {
return false;
}
}
- private ComfyCliResult runProcess(String stdin, String... args) {
- CommandLine cmd = CommandLine.parse(config.getLocalExecutable());
- for (String arg : args) {
- if (arg != null) {
- // handleQuoting=false: the child is spawned via exec(argv), not
- // a shell — commons-exec's default quoting would embed literal
- // double quotes inside arguments containing spaces (prompts,
- // paths), corrupting them on arrival.
- cmd.addArgument(arg, false);
+ private ComfyCliResult runProcess(String stdin, long timeoutMs, String... args) {
+ CommandLine cmd = new CommandLine(config.getLocalExecutable());
+ if (args != null) {
+ for (String arg : args) {
+ if (arg != null) {
+ cmd.addArgument(arg, false);
+ }
}
}
DefaultExecutor executor = new DefaultExecutor();
- // Credentials (COMFY_API_KEY) and routing hints must travel via the
- // environment — argv is world-readable in `ps` output. commons-exec's
- // environment parameter REPLACES the parent environment, so merge the
- // overrides over System.getenv() to keep PATH (bare `comfy` lookup
- // depends on it).
Map childEnv = null;
if (config.getEnvironment() != null && !config.getEnvironment().isEmpty()) {
childEnv = new LinkedHashMap(System.getenv());
childEnv.putAll(config.getEnvironment());
}
- ByteArrayOutputStream stdout = new ByteArrayOutputStream();
- ByteArrayOutputStream stderr = new ByteArrayOutputStream();
- // Always hand the child a (possibly empty) stdin pipe that closes
- // right after the payload: consumers reading to EOF finish instantly,
- // and a closed pipe cannot race the input pump.
+
+ BoundedOutput stdout = new BoundedOutput(config.getMaxOutputBytes());
+ BoundedOutput stderr = new BoundedOutput(config.getMaxOutputBytes());
byte[] stdinBytes = stdin == null ? new byte[0] : stdin.getBytes(StandardCharsets.UTF_8);
- executor.setStreamHandler(new PumpStreamHandler(stdout, stderr,
- new ByteArrayInputStream(stdinBytes)));
+ executor.setStreamHandler(new PumpStreamHandler(stdout, stderr, new ByteArrayInputStream(stdinBytes)));
- long timeoutMs = config.getLocalTimeoutSeconds() * 1000L;
ExecuteWatchdog watchdog = new ExecuteWatchdog(timeoutMs);
executor.setWatchdog(watchdog);
-
long startNanos = System.nanoTime();
+
try {
int exitCode = childEnv == null ? executor.execute(cmd) : executor.execute(cmd, childEnv);
- String out = stdout.toString(StandardCharsets.UTF_8).trim();
- String err = stderr.toString(StandardCharsets.UTF_8).trim();
- log.debug("comfy CLI executed: exitCode={}, stdout.len={}", exitCode, out.length());
+ String out = stdout.asUtf8().trim();
+ String err = stderr.asUtf8().trim();
+ log.debug("comfy CLI executed: exitCode={}, stdout.len={}, stdout.truncated={}, stderr.truncated={}",
+ exitCode, out.length(), stdout.isTruncated(), stderr.isTruncated());
if (watchdog.killedProcess()) {
- return new ComfyCliResult(-1, out, "comfy CLI timed out after " + timeoutMs + " ms\n" + err);
+ return result(-1, out, timeoutMessage(timeoutMs, err), stdout, stderr);
}
- return new ComfyCliResult(exitCode, out, err);
+ return result(exitCode, out, err, stdout, stderr);
} catch (ExecuteException e) {
- // commons-exec throws ExecuteException for EVERY non-zero exit
- // (and for watchdog kills). The stream pumps are joined before it
- // is thrown, so both buffers are complete — surface them together
- // with the real exit code instead of discarding the output. The
- // deadline check makes the timeout verdict race-free even when
- // {@code watchdog.killedProcess()} has not observed the kill yet.
- String out = stdout.toString(StandardCharsets.UTF_8).trim();
- String err = stderr.toString(StandardCharsets.UTF_8).trim();
+ String out = stdout.asUtf8().trim();
+ String err = stderr.asUtf8().trim();
boolean timedOut = watchdog.killedProcess()
|| System.nanoTime() - startNanos >= timeoutMs * 1_000_000L;
if (timedOut) {
- return new ComfyCliResult(-1, out, "comfy CLI timed out after " + timeoutMs + " ms\n" + err);
+ return result(-1, out, timeoutMessage(timeoutMs, err), stdout, stderr);
}
- log.debug("comfy CLI failed: exitCode={}, stdout.len={}, stderr.len={}",
- e.getExitValue(), out.length(), err.length());
- return new ComfyCliResult(e.getExitValue(), out, err);
+ return result(e.getExitValue(), out, err, stdout, stderr);
} catch (IOException e) {
- return new ComfyCliResult(-1, "", e.getMessage());
+ return result(-1, "", safeMessage(e), stdout, stderr);
+ }
+ }
+
+ private static ComfyCliResult result(int exitCode, String out, String err,
+ BoundedOutput stdout, BoundedOutput stderr) {
+ return new ComfyCliResult(exitCode, out, err, stdout.isTruncated(), stderr.isTruncated());
+ }
+
+ private static String timeoutMessage(long timeoutMs, String stderr) {
+ String prefix = "comfy CLI timed out after " + timeoutMs + " ms";
+ return stderr == null || stderr.isEmpty() ? prefix : prefix + "\n" + stderr;
+ }
+
+ private static String safeMessage(IOException error) {
+ return error.getMessage() == null ? error.getClass().getSimpleName() : error.getMessage();
+ }
+
+ private static long secondsToMillis(int seconds) {
+ return Math.max(1L, (long) seconds) * 1000L;
+ }
+
+ private static final class BoundedOutput extends OutputStream {
+ private final ByteArrayOutputStream delegate = new ByteArrayOutputStream();
+ private final int maxBytes;
+ private boolean truncated;
+
+ private BoundedOutput(int maxBytes) {
+ this.maxBytes = maxBytes;
+ }
+
+ @Override
+ public synchronized void write(int b) {
+ if (maxBytes < 0 || delegate.size() < maxBytes) {
+ delegate.write(b);
+ } else {
+ truncated = true;
+ }
+ }
+
+ @Override
+ public synchronized void write(byte[] bytes, int off, int len) {
+ if (len <= 0) {
+ return;
+ }
+ if (maxBytes < 0) {
+ delegate.write(bytes, off, len);
+ return;
+ }
+ int remaining = maxBytes - delegate.size();
+ if (remaining > 0) {
+ int accepted = Math.min(remaining, len);
+ delegate.write(bytes, off, accepted);
+ if (accepted < len) {
+ truncated = true;
+ }
+ } else {
+ truncated = true;
+ }
+ }
+
+ private synchronized String asUtf8() {
+ return new String(delegate.toByteArray(), StandardCharsets.UTF_8);
+ }
+
+ private synchronized boolean isTruncated() {
+ return truncated;
}
}
}
diff --git a/src/main/java/io/github/easy4j/comfy/cli/ComfyCliResult.java b/src/main/java/io/github/easy4j/comfy/cli/ComfyCliResult.java
index babcf3d..915dc2b 100644
--- a/src/main/java/io/github/easy4j/comfy/cli/ComfyCliResult.java
+++ b/src/main/java/io/github/easy4j/comfy/cli/ComfyCliResult.java
@@ -2,16 +2,6 @@
* Copyright (c) 2018-present, easy-4-java (https://github.com/easy-4-java).
*
* Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
*/
package io.github.easy4j.comfy.cli;
@@ -19,45 +9,40 @@
import lombok.Data;
-/**
- * Outcome of one {@code comfy} CLI invocation: exit code plus both captured
- * streams. Non-zero exits preserve the real exit code and both buffers —
- * nothing is collapsed into a sentinel value.
- *
- * @author Loong Wan
- * @since 1.0.0
- * @see ComfyCliExecutor
- */
+/** Outcome of one {@code comfy} CLI invocation. */
@Data
public class ComfyCliResult {
private static final String TIMEOUT_PREFIX = "comfy CLI timed out after ";
- /** Process exit code; {@code -1} when the executable was missing or the run timed out. */
private final int exitCode;
-
- /** Trimmed standard output of the child process. */
private final String stdout;
-
- /** Trimmed standard error of the child process. */
private final String stderr;
+ private final boolean stdoutTruncated;
+ private final boolean stderrTruncated;
+
+ public ComfyCliResult(int exitCode, String stdout, String stderr) {
+ this(exitCode, stdout, stderr, false, false);
+ }
+
+ public ComfyCliResult(int exitCode, String stdout, String stderr,
+ boolean stdoutTruncated, boolean stderrTruncated) {
+ this.exitCode = exitCode;
+ this.stdout = stdout;
+ this.stderr = stderr;
+ this.stdoutTruncated = stdoutTruncated;
+ this.stderrTruncated = stderrTruncated;
+ }
- /**
- * Returns whether the invocation exited with status zero.
- *
- * @return {@code true} when {@code exitCode == 0}.
- */
public boolean isSuccess() {
return exitCode == 0;
}
- /**
- * Returns whether the run was terminated by the watchdog timeout.
- *
- * @return {@code true} when the exit code is {@code -1} and stderr carries
- * the timeout notice.
- */
public boolean isTimeout() {
return exitCode == -1 && Objects.nonNull(stderr) && stderr.startsWith(TIMEOUT_PREFIX);
}
+
+ public boolean isTruncated() {
+ return stdoutTruncated || stderrTruncated;
+ }
}
diff --git a/src/main/java/io/github/easy4j/comfy/mcp/ComfyMcpCallResult.java b/src/main/java/io/github/easy4j/comfy/mcp/ComfyMcpCallResult.java
index 1991b4e..ba84e26 100644
--- a/src/main/java/io/github/easy4j/comfy/mcp/ComfyMcpCallResult.java
+++ b/src/main/java/io/github/easy4j/comfy/mcp/ComfyMcpCallResult.java
@@ -2,42 +2,37 @@
* Copyright (c) 2018-present, easy-4-java (https://github.com/easy-4-java).
*
* Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
*/
package io.github.easy4j.comfy.mcp;
-import lombok.Data;
+import java.util.Collections;
+import java.util.List;
+import lombok.Data;
import com.fasterxml.jackson.databind.JsonNode;
-/**
- * Outcome of one {@code tools/call}.
- *
- * @author Loong Wan
- * @since 1.0.0
- * @see ComfyMcpClient#callTool(String, java.util.Map)
- */
+/** Outcome of one MCP {@code tools/call}. */
@Data
public class ComfyMcpCallResult {
-
- /**
- * Concatenated text of every {@code text} content item the tool returned,
- * subject to {@code maxContentChars} truncation.
- */
private final String text;
-
- /** Whether the tool reported a business-level failure ({@code isError}). */
private final boolean isError;
-
- /** The raw JSON-RPC result node, for callers needing non-text content. */
private final JsonNode raw;
+ private final List contents;
+ private final boolean textTruncated;
+
+ /** Backward-compatible constructor retained for existing SDK callers. */
+ public ComfyMcpCallResult(String text, boolean isError, JsonNode raw) {
+ this(text, isError, raw, Collections.emptyList(), false);
+ }
+
+ public ComfyMcpCallResult(String text, boolean isError, JsonNode raw,
+ List contents, boolean textTruncated) {
+ this.text = text;
+ this.isError = isError;
+ this.raw = raw;
+ this.contents = contents == null
+ ? Collections.emptyList()
+ : Collections.unmodifiableList(contents);
+ this.textTruncated = textTruncated;
+ }
}
diff --git a/src/main/java/io/github/easy4j/comfy/mcp/ComfyMcpClient.java b/src/main/java/io/github/easy4j/comfy/mcp/ComfyMcpClient.java
index 4c599ec..63f705c 100644
--- a/src/main/java/io/github/easy4j/comfy/mcp/ComfyMcpClient.java
+++ b/src/main/java/io/github/easy4j/comfy/mcp/ComfyMcpClient.java
@@ -2,35 +2,28 @@
* Copyright (c) 2018-present, easy-4-java (https://github.com/easy-4-java).
*
* Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
*/
package io.github.easy4j.comfy.mcp;
-import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.io.PrintWriter;
+import java.io.Reader;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
-import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
+import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
@@ -39,31 +32,18 @@
import org.slf4j.LoggerFactory;
import io.github.easy4j.comfy.ComfyException;
+import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.json.JsonMapper;
/**
- * Client for the comfy-mcp route: spawns the MCP server as a child process
- * and drives it over JSON-RPC on stdio (newline-delimited JSON, the MCP
- * stdio transport).
- *
- * Lifecycle: {@link #connect()} performs the MCP {@code initialize}
- * handshake and completes the {@code notifications/initialized} sequence;
- * {@link #listTools()} discovers the tool catalog ({@code server_info},
- * {@code run_workflow}, {@code search_templates}, {@code launch_comfyui},
- * …); {@link #callTool(String, Map)} executes one tool and returns the
- * concatenated text content. {@link #close()} terminates the server.
+ * Local Comfy MCP client using newline-delimited JSON-RPC over stdio.
*
- * The client is thread-safe: concurrent {@code tools/call} requests are
- * correlated by JSON-RPC id. Each client owns exactly one server child
- * process; {@link #close()} destroys it.
- *
- * @author Loong Wan
- * @since 1.0.0
- * @see ComfyMcpConfig
- * @see ComfyMcpTool
- * @see ComfyMcpCallResult
+ * One client owns exactly one child process. Requests may be concurrent and
+ * are correlated by JSON-RPC id. Both stdout and stderr are continuously
+ * drained; pending requests are removed on response, timeout, write failure,
+ * process exit and close.
*/
public class ComfyMcpClient implements AutoCloseable {
@@ -71,8 +51,7 @@ public class ComfyMcpClient implements AutoCloseable {
private final ComfyMcpConfig config;
private final ObjectMapper mapper =
- JsonMapper.builder().disable(com.fasterxml.jackson.databind.DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES)
- .build();
+ JsonMapper.builder().disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES).build();
private final Map> pendingRpcs =
new ConcurrentHashMap>();
private final AtomicLong rpcIds = new AtomicLong();
@@ -83,78 +62,88 @@ public class ComfyMcpClient implements AutoCloseable {
thread.setDaemon(true);
return thread;
});
+ private final Object writeLock = new Object();
private volatile Process process;
private volatile PrintWriter stdin;
+ private volatile Thread stdoutThread;
+ private volatile Thread stderrThread;
private volatile String serverName;
private volatile String serverVersion;
- /**
- * Creates a new client bound to the given configuration.
- *
- * @param config runtime configuration; must not be {@code null}.
- */
public ComfyMcpClient(ComfyMcpConfig config) {
this.config = Objects.requireNonNull(config, "config");
this.config.validate();
}
/**
- * Spawns the MCP server child process and performs the MCP
- * {@code initialize} handshake plus {@code notifications/initialized}.
+ * Starts {@code comfy-mcp} and performs MCP initialize/initialized.
*
- * @return the server-reported version string, may be {@code null}.
- * @throws ComfyException when the process fails to start or the handshake
- * fails or times out.
+ * @throws IllegalStateException if already connected or closed.
*/
- public String connect() {
+ public synchronized String connect() {
+ if (closed.get()) {
+ throw new IllegalStateException("comfy mcp client is closed");
+ }
+ if (initialized.get() || process != null) {
+ throw new IllegalStateException("comfy mcp client is already connected");
+ }
+
List command = new ArrayList();
command.add(config.getLocalExecutable());
if (config.getMcpArgs() != null) {
- for (String arg : config.getMcpArgs()) {
- command.add(arg);
- }
+ Collections.addAll(command, config.getMcpArgs());
}
+
ProcessBuilder builder = new ProcessBuilder(command);
builder.redirectErrorStream(false);
if (config.getEnvironment() != null && !config.getEnvironment().isEmpty()) {
builder.environment().putAll(config.getEnvironment());
}
+
try {
- process = builder.start();
+ Process child = builder.start();
+ process = child;
+ stdin = new PrintWriter(new OutputStreamWriter(child.getOutputStream(), StandardCharsets.UTF_8), true);
+ stdoutThread = daemon("comfy-mcp-reader", () -> readLoop(child));
+ stderrThread = daemon("comfy-mcp-stderr", () -> drainStderr(child));
+ stdoutThread.start();
+ stderrThread.start();
+
+ Map clientInfo = new LinkedHashMap();
+ clientInfo.put("name", config.getClientName());
+ clientInfo.put("version", config.getClientVersion());
+ Map params = new LinkedHashMap();
+ params.put("protocolVersion", config.getProtocolVersion());
+ params.put("capabilities", new LinkedHashMap());
+ params.put("clientInfo", clientInfo);
+
+ JsonNode result = await(
+ request("initialize", params, config.getConnectTimeoutMillis(), "initialize"),
+ "initialize");
+ notify("notifications/initialized", new LinkedHashMap());
+ if (result.hasNonNull("serverInfo")) {
+ serverName = result.path("serverInfo").path("name").asText(null);
+ serverVersion = result.path("serverInfo").path("version").asText(null);
+ }
+ initialized.set(true);
+ return serverVersion;
} catch (IOException e) {
+ cleanupTransport();
throw new ComfyException("Failed to spawn comfy-mcp: " + config.getLocalExecutable(), e);
+ } catch (RuntimeException e) {
+ failAllPending(new ComfyException("comfy mcp connect failed", e));
+ cleanupTransport();
+ throw e;
}
- stdin = new PrintWriter(new OutputStreamWriter(process.getOutputStream(), StandardCharsets.UTF_8), true);
- Thread reader = new Thread(this::readLoop, "comfy-mcp-reader");
- reader.setDaemon(true);
- reader.start();
-
- Map clientInfo = new LinkedHashMap();
- clientInfo.put("name", config.getClientName());
- clientInfo.put("version", config.getClientVersion());
- Map params = new LinkedHashMap();
- params.put("protocolVersion", config.getProtocolVersion());
- params.put("capabilities", new LinkedHashMap());
- params.put("clientInfo", clientInfo);
- JsonNode result = await(request("initialize", params), config.getConnectTimeoutMillis(), "initialize");
- notify("notifications/initialized", new LinkedHashMap());
- if (result.hasNonNull("serverInfo")) {
- serverName = result.path("serverInfo").path("name").asText(null);
- serverVersion = result.path("serverInfo").path("version").asText(null);
- }
- initialized.set(true);
- return serverVersion;
}
- /**
- * Lists the tools the server advertises.
- *
- * @return the tool catalog; never {@code null}.
- */
public List listTools() {
- JsonNode result = await(request("tools/list", new LinkedHashMap()),
- config.getConnectTimeoutMillis(), "tools/list");
+ ensureConnected();
+ JsonNode result = await(
+ request("tools/list", new LinkedHashMap(),
+ config.getConnectTimeoutMillis(), "tools/list"),
+ "tools/list");
List tools = new ArrayList();
for (JsonNode tool : result.path("tools")) {
tools.add(new ComfyMcpTool(
@@ -165,166 +154,349 @@ public List listTools() {
return tools;
}
- /**
- * Calls one tool asynchronously.
- *
- * @param name the tool name.
- * @param arguments tool arguments (JSON-Schema-shaped); may be {@code null}.
- * @return a future completed with the call result, or completed
- * exceptionally with a {@link ComfyException}.
- */
- public CompletableFuture callToolAsync(String name, Map arguments) {
+ public CompletableFuture callToolAsync(
+ String name, Map arguments) {
Objects.requireNonNull(name, "name");
- if (closed.get()) {
- throw new IllegalStateException("comfy mcp client is closed");
- }
- if (!initialized.get()) {
- throw new IllegalStateException("comfy mcp client is not connected");
- }
+ ensureConnected();
Map params = new LinkedHashMap();
params.put("name", name);
- params.put("arguments", arguments == null ? new LinkedHashMap() : arguments);
- CompletableFuture response = request("tools/call", params);
- CompletableFuture future = response.thenApply(this::toCallResult);
- scheduleTimeout(future, config.getReadTimeoutMillis(), "tools/call " + name);
- return future;
+ params.put("arguments", arguments == null
+ ? new LinkedHashMap() : new LinkedHashMap(arguments));
+ return request("tools/call", params, config.getReadTimeoutMillis(), "tools/call " + name)
+ .thenApply(this::toCallResult);
}
- /**
- * Calls one tool, blocking until it completes.
- *
- * @param name the tool name.
- * @param arguments tool arguments; may be {@code null}.
- * @return the call result; never {@code null}.
- * @throws ComfyException when the call fails or times out.
- */
public ComfyMcpCallResult callTool(String name, Map arguments) {
- try {
- return callToolAsync(name, arguments).get();
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- throw new ComfyException("comfy mcp tools/call interrupted", e);
- } catch (ExecutionException e) {
- Throwable cause = e.getCause() == null ? e : e.getCause();
- if (cause instanceof ComfyException) {
- throw (ComfyException) cause;
- }
- if (cause instanceof RuntimeException) {
- throw (RuntimeException) cause;
- }
- throw new ComfyException("comfy mcp tools/call failed", cause);
- }
+ return awaitCall(callToolAsync(name, arguments), "tools/call " + name);
}
- /**
- * Convenience wrapper for the {@code server_info} tool ("call first" per
- * the Comfy docs — verifies the local ComfyUI is up).
- *
- * @return the raw JSON-RPC result node; never {@code null}.
- */
- public JsonNode serverInfo() {
- return rawCall("server_info");
+ // ---- current comfy-mcp typed conveniences ------------------------------
+
+ public JsonNode serverInfo() { return callTool("server_info", null).getRaw(); }
+ public ComfyMcpCallResult authStatus() { return callTool("auth_status", null); }
+ public ComfyMcpCallResult billingStatus() { return callTool("billing_status", null); }
+ public ComfyMcpCallResult authLogin() { return callTool("auth_login", null); }
+
+ public ComfyMcpCallResult runWorkflow(String workflowPath, boolean wait,
+ double timeoutSeconds, boolean confirmSpend) {
+ return callTool("run_workflow", params(
+ "workflow_path", Objects.requireNonNull(workflowPath, "workflowPath"),
+ "wait", Boolean.valueOf(wait),
+ "timeout_seconds", Double.valueOf(timeoutSeconds),
+ "confirm_spend", Boolean.valueOf(confirmSpend)));
}
- /**
- * Returns the server name reported by {@code initialize}, or {@code null}
- * before {@link #connect()}.
- *
- * @return the server name, may be {@code null}.
- */
- public String getServerName() {
- return serverName;
+ public ComfyMcpCallResult generateImage(String prompt, String checkpoint,
+ boolean wait, double timeoutSeconds) {
+ Map args = params(
+ "prompt", Objects.requireNonNull(prompt, "prompt"),
+ "wait", Boolean.valueOf(wait),
+ "timeout_seconds", Double.valueOf(timeoutSeconds));
+ putIfNotNull(args, "checkpoint", checkpoint);
+ return callTool("generate_image", args);
}
- /**
- * Returns the server version reported by {@code initialize}, or
- * {@code null} before {@link #connect()}.
- *
- * @return the server version, may be {@code null}.
- */
- public String getServerVersion() {
- return serverVersion;
+ public ComfyMcpCallResult listPartnerModels(String style, String partner,
+ String query, int limit, int offset) {
+ return callTool("list_partner_models", params(
+ "style", nullToEmpty(style),
+ "partner", nullToEmpty(partner),
+ "query", nullToEmpty(query),
+ "limit", Integer.valueOf(limit),
+ "offset", Integer.valueOf(offset)));
}
- /**
- * Returns whether the client has been closed.
- *
- * @return {@code true} after {@link #close()}.
- */
- public boolean isClosed() {
- return closed.get();
+ public ComfyMcpCallResult partnerModelSchema(String model) {
+ return callTool("partner_model_schema",
+ params("model", Objects.requireNonNull(model, "model")));
}
- /**
- * Terminates the MCP server child process and releases the timer.
- * Idempotent.
- */
- @Override
- public void close() {
- if (!closed.compareAndSet(false, true)) {
- return;
- }
- timer.shutdownNow();
- Process current = process;
- if (current != null) {
- current.destroy();
- }
- failAllPending(new ComfyException("comfy mcp client closed"));
+ public ComfyMcpCallResult partnerGenerate(String model, Map modelParams,
+ boolean confirmSpend, String outPath, double timeoutSeconds) {
+ Map args = params(
+ "model", Objects.requireNonNull(model, "model"),
+ "params", modelParams == null ? new LinkedHashMap() : modelParams,
+ "confirm_spend", Boolean.valueOf(confirmSpend),
+ "timeout_seconds", Double.valueOf(timeoutSeconds));
+ putIfNotNull(args, "out_path", outPath);
+ return callTool("partner_generate", args);
}
- // ============================================================
- // transport internals
- // ============================================================
+ public ComfyMcpCallResult emitPartnerWorkflow(String model, String outPath,
+ Map modelParams) {
+ return callTool("emit_partner_workflow", params(
+ "model", Objects.requireNonNull(model, "model"),
+ "out_path", Objects.requireNonNull(outPath, "outPath"),
+ "params", modelParams == null ? new LinkedHashMap() : modelParams));
+ }
- private JsonNode rawCall(String tool) {
- return await(request("tools/call", paramsFor(tool, null)),
- config.getReadTimeoutMillis(), "tools/call " + tool);
+ public ComfyMcpCallResult runTemplate(String name, Map templateParams,
+ boolean confirmSpend, boolean wait, double timeoutSeconds) {
+ return callTool("run_template", params(
+ "name", Objects.requireNonNull(name, "name"),
+ "params", templateParams == null ? new LinkedHashMap() : templateParams,
+ "confirm_spend", Boolean.valueOf(confirmSpend),
+ "wait", Boolean.valueOf(wait),
+ "timeout_seconds", Double.valueOf(timeoutSeconds)));
}
- private Map paramsFor(String name, Map arguments) {
- Map params = new LinkedHashMap();
- params.put("name", name);
- params.put("arguments", arguments == null ? new LinkedHashMap() : arguments);
- return params;
+ public ComfyMcpCallResult job(String action, String promptId, Double timeoutSeconds) {
+ Map args = params("action", action == null ? "status" : action);
+ putIfNotNull(args, "prompt_id", promptId);
+ putIfNotNull(args, "timeout_seconds", timeoutSeconds);
+ return callTool("job", args);
}
- private ComfyMcpCallResult toCallResult(JsonNode result) {
- StringBuilder text = new StringBuilder();
- int cap = config.getMaxContentChars() <= 0 ? Integer.MAX_VALUE : config.getMaxContentChars();
- boolean truncated = false;
- for (JsonNode content : result.path("content")) {
- if (!"text".equals(content.path("type").asText(""))) {
- continue;
- }
- String piece = content.path("text").asText("");
- if (text.length() >= cap) {
- truncated = true;
- break;
- }
- if (text.length() + piece.length() > cap) {
- text.append(piece, 0, cap - text.length());
- truncated = true;
- break;
- }
- text.append(piece);
- }
- if (truncated) {
- log.warn("comfy mcp tool content truncated at maxContentChars={}", config.getMaxContentChars());
+ public ComfyMcpCallResult jobStatus(String promptId) { return job("status", promptId, null); }
+ public ComfyMcpCallResult waitForJob(String promptId, double timeoutSeconds) {
+ return job("wait", promptId, Double.valueOf(timeoutSeconds));
+ }
+ public ComfyMcpCallResult watchJob(String promptId, double timeoutSeconds) {
+ return job("watch", promptId, Double.valueOf(timeoutSeconds));
+ }
+ public ComfyMcpCallResult cancelJob(String promptId) { return job("cancel", promptId, null); }
+ public ComfyMcpCallResult getQueue() { return job("queue", null, null); }
+
+ public ComfyMcpCallResult systemStats() { return callTool("system_stats", null); }
+ public ComfyMcpCallResult freeMemory() { return freeMemory(true, null); }
+ public ComfyMcpCallResult freeMemory(boolean unloadModels, Boolean freeMemory) {
+ Map args = params("unload_models", Boolean.valueOf(unloadModels));
+ putIfNotNull(args, "free_memory", freeMemory);
+ return callTool("free_memory", args);
+ }
+
+ public ComfyMcpCallResult fetchOutputs(String promptId, String outDir,
+ boolean urlOnly, boolean inlineImages) {
+ return callTool("fetch_outputs", params(
+ "prompt_id", Objects.requireNonNull(promptId, "promptId"),
+ "out_dir", Objects.requireNonNull(outDir, "outDir"),
+ "url_only", Boolean.valueOf(urlOnly),
+ "inline_images", Boolean.valueOf(inlineImages)));
+ }
+
+ public ComfyMcpCallResult launchComfyUi(List extraArgs, boolean confirmNetworkExposure) {
+ return callTool("launch_comfyui", params(
+ "extra_args", extraArgs == null ? Collections.emptyList() : extraArgs,
+ "confirm_network_exposure", Boolean.valueOf(confirmNetworkExposure)));
+ }
+
+ public ComfyMcpCallResult stopComfyUi() { return callTool("stop_comfyui", null); }
+
+ public ComfyMcpCallResult restartComfyUi(List extraArgs,
+ boolean confirmNetworkExposure, boolean confirmKillUntracked) {
+ return callTool("restart_comfyui", params(
+ "extra_args", extraArgs == null ? Collections.emptyList() : extraArgs,
+ "confirm_network_exposure", Boolean.valueOf(confirmNetworkExposure),
+ "confirm_kill_untracked", Boolean.valueOf(confirmKillUntracked)));
+ }
+
+ public ComfyMcpCallResult updateComfyUi(String target, boolean confirmUpdateAll) {
+ return callTool("update_comfyui", params(
+ "target", target == null ? "comfy" : target,
+ "confirm_update_all", Boolean.valueOf(confirmUpdateAll)));
+ }
+
+ public ComfyMcpCallResult switchComfyUiVersion(String version, boolean confirmSwitch) {
+ return callTool("switch_comfyui_version", params(
+ "version", Objects.requireNonNull(version, "version"),
+ "confirm_switch", Boolean.valueOf(confirmSwitch)));
+ }
+
+ public ComfyMcpCallResult installNode(List names, boolean confirmInstall) {
+ return callTool("install_node", params(
+ "names", Objects.requireNonNull(names, "names"),
+ "confirm_install", Boolean.valueOf(confirmInstall)));
+ }
+
+ public ComfyMcpCallResult getLogs(int tail, Integer port) {
+ Map args = params("tail", Integer.valueOf(tail));
+ putIfNotNull(args, "port", port);
+ return callTool("get_logs", args);
+ }
+
+ public ComfyMcpCallResult discover() { return discover(true, ""); }
+ public ComfyMcpCallResult discover(boolean schemasOnly, String command) {
+ return callTool("discover", params(
+ "schemas_only", Boolean.valueOf(schemasOnly),
+ "command", nullToEmpty(command)));
+ }
+ public ComfyMcpCallResult which() { return callTool("which", null); }
+
+ public ComfyMcpCallResult project(String action) {
+ return callTool("project", params("action", action == null ? "status" : action));
+ }
+
+ public ComfyMcpCallResult searchTemplates(String query, int limit, int offset,
+ String tag, String type, String model, String provider, boolean excludeApi) {
+ return callTool("search_templates", params(
+ "query", nullToEmpty(query),
+ "limit", Integer.valueOf(limit),
+ "offset", Integer.valueOf(offset),
+ "tag", nullToEmpty(tag),
+ "type", nullToEmpty(type),
+ "model", nullToEmpty(model),
+ "provider", nullToEmpty(provider),
+ "exclude_api", Boolean.valueOf(excludeApi)));
+ }
+
+ public ComfyMcpCallResult getTemplate(String name) {
+ return getTemplate(name, true);
+ }
+
+ public ComfyMcpCallResult getTemplate(String name, boolean checkLocal) {
+ return callTool("get_template", params(
+ "name", Objects.requireNonNull(name, "name"),
+ "check_local", Boolean.valueOf(checkLocal)));
+ }
+
+ public ComfyMcpCallResult fetchTemplate(String name, String outPath, boolean checkLocal) {
+ return callTool("fetch_template", params(
+ "name", Objects.requireNonNull(name, "name"),
+ "out_path", Objects.requireNonNull(outPath, "outPath"),
+ "check_local", Boolean.valueOf(checkLocal)));
+ }
+
+ public ComfyMcpCallResult nodes(String action, Map options) {
+ Map args = options == null
+ ? new LinkedHashMap()
+ : new LinkedHashMap(options);
+ args.put("action", action == null ? "search" : action);
+ return callTool("nodes", args);
+ }
+
+ public ComfyMcpCallResult nodeDependencies(String pack, String registryId) {
+ return callTool("node_dependencies", params(
+ "pack", nullToEmpty(pack),
+ "registry_id", nullToEmpty(registryId)));
+ }
+
+ public ComfyMcpCallResult workflowDeps(String workflowPath) {
+ return callTool("workflow_deps",
+ params("workflow_path", Objects.requireNonNull(workflowPath, "workflowPath")));
+ }
+
+ public ComfyMcpCallResult searchModels(String query, String folder) {
+ return callTool("search_models", params(
+ "query", nullToEmpty(query),
+ "folder", nullToEmpty(folder)));
+ }
+
+ public ComfyMcpCallResult downloadModel(String url, String relativePath,
+ String filename, boolean wait, double timeoutSeconds) {
+ Map args = params(
+ "url", Objects.requireNonNull(url, "url"),
+ "wait", Boolean.valueOf(wait),
+ "timeout_seconds", Double.valueOf(timeoutSeconds));
+ putIfNotNull(args, "relative_path", relativePath);
+ putIfNotNull(args, "filename", filename);
+ return callTool("download_model", args);
+ }
+
+ public ComfyMcpCallResult download(String action, String downloadId, Double timeoutSeconds) {
+ Map args = params("action", action == null ? "status" : action);
+ putIfNotNull(args, "download_id", downloadId);
+ putIfNotNull(args, "timeout_seconds", timeoutSeconds);
+ return callTool("download", args);
+ }
+
+ public ComfyMcpCallResult uploadFile(List paths, boolean overwrite) {
+ return callTool("upload_file", params(
+ "paths", Objects.requireNonNull(paths, "paths"),
+ "overwrite", Boolean.valueOf(overwrite)));
+ }
+
+ public ComfyMcpCallResult validateWorkflow(String workflowPath) {
+ return callTool("validate_workflow",
+ params("workflow_path", Objects.requireNonNull(workflowPath, "workflowPath")));
+ }
+
+ public ComfyMcpCallResult listWorkflowSlots(String workflowPath) {
+ return callTool("list_workflow_slots",
+ params("workflow_path", Objects.requireNonNull(workflowPath, "workflowPath")));
+ }
+
+ public ComfyMcpCallResult listWorkflowNotes(String workflowPath) {
+ return callTool("list_workflow_notes",
+ params("workflow_path", Objects.requireNonNull(workflowPath, "workflowPath")));
+ }
+
+ public ComfyMcpCallResult setWorkflowSlot(String workflowPath, List overrides, boolean stdout) {
+ return callTool("set_workflow_slot", params(
+ "workflow_path", Objects.requireNonNull(workflowPath, "workflowPath"),
+ "overrides", Objects.requireNonNull(overrides, "overrides"),
+ "stdout", Boolean.valueOf(stdout)));
+ }
+
+ public ComfyMcpCallResult varyWorkflow(String workflowPath, List slots, String outDir) {
+ Map args = params(
+ "workflow_path", Objects.requireNonNull(workflowPath, "workflowPath"),
+ "slots", Objects.requireNonNull(slots, "slots"));
+ putIfNotNull(args, "out_dir", outDir);
+ return callTool("vary_workflow", args);
+ }
+
+ // ---- state/lifecycle ---------------------------------------------------
+
+ public String getServerName() { return serverName; }
+ public String getServerVersion() { return serverVersion; }
+ public boolean isClosed() { return closed.get(); }
+ public boolean isConnected() { return initialized.get() && !closed.get(); }
+ int pendingRequestCount() { return pendingRpcs.size(); }
+
+ @Override
+ public void close() {
+ if (!closed.compareAndSet(false, true)) {
+ return;
}
- boolean isError = result.path("isError").asBoolean(false);
- return new ComfyMcpCallResult(text.toString(), isError, result);
+ initialized.set(false);
+ failAllPending(new ComfyException("comfy mcp client closed"));
+ cleanupTransport();
+ timer.shutdownNow();
}
- private CompletableFuture request(String method, Map params) {
- long id = rpcIds.incrementAndGet();
+ // ---- transport internals ----------------------------------------------
+
+ private CompletableFuture request(String method, Map params,
+ long timeoutMillis, String what) {
+ ensureTransportOpen();
+ final long id = rpcIds.incrementAndGet();
Map payload = new LinkedHashMap();
payload.put("jsonrpc", "2.0");
payload.put("id", Long.valueOf(id));
payload.put("method", method);
payload.put("params", params);
- CompletableFuture future = new CompletableFuture();
+
+ final CompletableFuture future = new CompletableFuture();
pendingRpcs.put(Long.valueOf(id), future);
- writeJson(payload);
+
+ try {
+ writeJson(payload);
+ } catch (RuntimeException e) {
+ pendingRpcs.remove(Long.valueOf(id), future);
+ future.completeExceptionally(e);
+ throw e;
+ }
+
+ final ScheduledFuture guard;
+ try {
+ guard = timeoutMillis <= 0 ? null : timer.schedule(() -> {
+ if (pendingRpcs.remove(Long.valueOf(id), future)) {
+ future.completeExceptionally(new ComfyException(
+ "comfy mcp " + what + " timed out after " + timeoutMillis + " ms"));
+ }
+ }, timeoutMillis, TimeUnit.MILLISECONDS);
+ } catch (RejectedExecutionException e) {
+ pendingRpcs.remove(Long.valueOf(id), future);
+ future.completeExceptionally(new ComfyException("comfy mcp timer is closed", e));
+ return future;
+ }
+
+ future.whenComplete((value, error) -> {
+ pendingRpcs.remove(Long.valueOf(id), future);
+ if (guard != null) {
+ guard.cancel(false);
+ }
+ });
return future;
}
@@ -337,20 +509,19 @@ private void notify(String method, Map params) {
}
private void writeJson(Map payload) {
- if (closed.get()) {
- throw new ComfyException("comfy mcp client is closed");
- }
+ ensureTransportOpen();
String line;
try {
line = mapper.writeValueAsString(payload);
} catch (Exception e) {
throw new ComfyException("comfy mcp RPC serialization failed", e);
}
- PrintWriter writer = stdin;
- if (writer == null) {
- throw new ComfyException("comfy mcp client is not connected");
- }
- synchronized (this) {
+
+ synchronized (writeLock) {
+ PrintWriter writer = stdin;
+ if (writer == null) {
+ throw new ComfyException("comfy mcp client is not connected");
+ }
writer.println(line);
if (writer.checkError()) {
throw new ComfyException("comfy mcp stdin write failed (server exited?)");
@@ -358,44 +529,82 @@ private void writeJson(Map payload) {
}
}
- private void readLoop() {
- try (BufferedReader reader =
- new BufferedReader(new InputStreamReader(process.getInputStream(), StandardCharsets.UTF_8))) {
- String line;
- while ((line = reader.readLine()) != null) {
+ private void readLoop(Process child) {
+ try (Reader reader = new InputStreamReader(child.getInputStream(), StandardCharsets.UTF_8)) {
+ String frame;
+ while ((frame = readBoundedLine(reader)) != null) {
if (closed.get()) {
return;
}
- if (line.trim().isEmpty()) {
+ if (frame.trim().isEmpty()) {
continue;
}
- if (line.length() > effectiveMaxFrameChars()) {
- log.warn("comfy mcp frame over cap, tearing transport down");
- failAllPending(new ComfyException(
- "comfy mcp frame exceeded maxFrameChars=" + config.getMaxFrameChars()));
- process.destroy();
- return;
- }
- handleFrame(line);
+ handleFrame(frame);
}
- failAllPending(new ComfyException("comfy mcp stdout closed (server exited)"));
+ if (!closed.get() && process == child) {
+ failAllPending(new ComfyException("comfy mcp stdout closed (server exited)"));
+ }
+ } catch (FrameTooLargeException e) {
+ failAllPending(new ComfyException(
+ "comfy mcp frame exceeded maxFrameChars=" + config.getMaxFrameChars(), e));
+ child.destroy();
} catch (IOException e) {
- if (!closed.get()) {
+ if (!closed.get() && process == child) {
failAllPending(new ComfyException("comfy mcp stdout read failed", e));
}
}
}
+ /** Drains stderr without logging content, because child logs may contain secrets/paths. */
+ private void drainStderr(Process child) {
+ long chars = 0L;
+ try (Reader reader = new InputStreamReader(child.getErrorStream(), StandardCharsets.UTF_8)) {
+ char[] buffer = new char[4096];
+ int n;
+ while ((n = reader.read(buffer)) >= 0) {
+ chars += n;
+ }
+ } catch (IOException e) {
+ if (!closed.get() && process == child) {
+ log.debug("comfy-mcp stderr drain ended with {}", e.getClass().getSimpleName());
+ }
+ }
+ log.debug("comfy-mcp stderr drained chars={}", chars);
+ }
+
+ private String readBoundedLine(Reader reader) throws IOException {
+ StringBuilder line = new StringBuilder();
+ int max = config.getMaxFrameChars() <= 0 ? Integer.MAX_VALUE : config.getMaxFrameChars();
+ for (;;) {
+ int ch = reader.read();
+ if (ch < 0) {
+ return line.length() == 0 ? null : line.toString();
+ }
+ if (ch == '\n') {
+ return line.toString();
+ }
+ if (ch == '\r') {
+ continue;
+ }
+ if (line.length() >= max) {
+ throw new FrameTooLargeException();
+ }
+ line.append((char) ch);
+ }
+ }
+
private void handleFrame(String frame) {
JsonNode node;
try {
node = mapper.readTree(frame);
- } catch (Exception ex) {
- log.warn("Ignored non-JSON frame from comfy mcp");
+ } catch (Exception e) {
+ log.warn("Ignored non-JSON frame from comfy-mcp");
return;
}
+
if (node.hasNonNull("id")) {
- CompletableFuture pending = pendingRpcs.remove(Long.valueOf(node.get("id").asLong()));
+ Long id = Long.valueOf(node.get("id").asLong());
+ CompletableFuture pending = pendingRpcs.remove(id);
if (pending == null) {
return;
}
@@ -407,48 +616,181 @@ private void handleFrame(String frame) {
}
return;
}
- log.debug("Ignored comfy mcp notification: method={}", node.path("method").asText(""));
+ log.debug("Ignored comfy-mcp notification: method={}", node.path("method").asText(""));
}
- private void scheduleTimeout(CompletableFuture future, long timeoutMillis, String what) {
- if (timeoutMillis <= 0) {
- return;
+ private ComfyMcpCallResult toCallResult(JsonNode result) {
+ StringBuilder text = new StringBuilder();
+ List contents = new ArrayList();
+ int cap = config.getMaxContentChars() <= 0 ? Integer.MAX_VALUE : config.getMaxContentChars();
+ boolean truncated = false;
+
+ for (JsonNode content : result.path("content")) {
+ String type = content.path("type").asText("");
+ String piece = content.hasNonNull("text") ? content.path("text").asText() : null;
+ contents.add(new ComfyMcpContent(
+ type,
+ piece,
+ content.hasNonNull("mimeType") ? content.path("mimeType").asText() : null,
+ content.hasNonNull("data") ? content.path("data").asText() : null,
+ content.hasNonNull("uri") ? content.path("uri").asText() : null,
+ content));
+
+ if (piece == null || !"text".equals(type)) {
+ continue;
+ }
+ int remaining = cap - text.length();
+ if (remaining <= 0) {
+ truncated = true;
+ continue;
+ }
+ if (piece.length() > remaining) {
+ text.append(piece, 0, remaining);
+ truncated = true;
+ } else {
+ text.append(piece);
+ }
}
- java.util.concurrent.ScheduledFuture guard = timer.schedule(() -> future
- .completeExceptionally(new ComfyException("comfy mcp " + what + " timed out after "
- + timeoutMillis + " ms")), timeoutMillis, TimeUnit.MILLISECONDS);
- future.whenComplete((r, error) -> guard.cancel(false));
+
+ return new ComfyMcpCallResult(
+ text.toString(),
+ result.path("isError").asBoolean(false),
+ result,
+ contents,
+ truncated);
}
- private JsonNode await(CompletableFuture future, long timeoutMillis, String what) {
- scheduleTimeout(future, timeoutMillis, what);
+ private JsonNode await(CompletableFuture future, String what) {
try {
return future.get();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new ComfyException("comfy mcp " + what + " interrupted", e);
} catch (ExecutionException e) {
- Throwable cause = e.getCause() == null ? e : e.getCause();
- if (cause instanceof ComfyException) {
- throw (ComfyException) cause;
- }
- if (cause instanceof RuntimeException) {
- throw (RuntimeException) cause;
- }
- throw new ComfyException("comfy mcp " + what + " failed", cause);
+ throw propagate("comfy mcp " + what + " failed", e.getCause());
+ }
+ }
+
+ private ComfyMcpCallResult awaitCall(CompletableFuture future, String what) {
+ try {
+ return future.get();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new ComfyException("comfy mcp " + what + " interrupted", e);
+ } catch (ExecutionException e) {
+ throw propagate("comfy mcp " + what + " failed", e.getCause());
}
}
+ private RuntimeException propagate(String message, Throwable cause) {
+ Throwable actual = cause == null ? new ComfyException(message) : cause;
+ if (actual instanceof ComfyException) {
+ return (ComfyException) actual;
+ }
+ if (actual instanceof RuntimeException) {
+ return (RuntimeException) actual;
+ }
+ return new ComfyException(message, actual);
+ }
+
private void failAllPending(ComfyException error) {
for (Map.Entry> entry : pendingRpcs.entrySet()) {
- CompletableFuture future = pendingRpcs.remove(entry.getKey());
- if (future != null) {
- future.completeExceptionally(error);
+ if (pendingRpcs.remove(entry.getKey(), entry.getValue())) {
+ entry.getValue().completeExceptionally(error);
+ }
+ }
+ }
+
+ private synchronized void cleanupTransport() {
+ initialized.set(false);
+
+ PrintWriter writer = stdin;
+ stdin = null;
+ if (writer != null) {
+ writer.close();
+ }
+
+ Process child = process;
+ process = null;
+ if (child != null) {
+ child.destroy();
+ try {
+ if (config.getShutdownTimeoutMillis() > 0
+ && !child.waitFor(config.getShutdownTimeoutMillis(), TimeUnit.MILLISECONDS)) {
+ child.destroyForcibly();
+ child.waitFor(config.getShutdownTimeoutMillis(), TimeUnit.MILLISECONDS);
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ child.destroyForcibly();
}
}
+
+ Thread out = stdoutThread;
+ Thread err = stderrThread;
+ stdoutThread = null;
+ stderrThread = null;
+ interruptAndJoin(out);
+ interruptAndJoin(err);
+ }
+
+ private void interruptAndJoin(Thread thread) {
+ if (thread == null || thread == Thread.currentThread()) {
+ return;
+ }
+ thread.interrupt();
+ try {
+ thread.join(Math.max(100L, config.getShutdownTimeoutMillis()));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ private void ensureConnected() {
+ if (closed.get()) {
+ throw new IllegalStateException("comfy mcp client is closed");
+ }
+ if (!initialized.get()) {
+ throw new IllegalStateException("comfy mcp client is not connected");
+ }
+ ensureTransportOpen();
+ }
+
+ private void ensureTransportOpen() {
+ if (closed.get()) {
+ throw new ComfyException("comfy mcp client is closed");
+ }
+ Process child = process;
+ if (child == null || stdin == null) {
+ throw new ComfyException("comfy mcp client is not connected");
+ }
+ }
+
+ private static Thread daemon(String name, Runnable task) {
+ Thread thread = new Thread(task, name);
+ thread.setDaemon(true);
+ return thread;
+ }
+
+ private static Map params(Object... keyValues) {
+ Map result = new LinkedHashMap();
+ for (int i = 0; i + 1 < keyValues.length; i += 2) {
+ result.put(String.valueOf(keyValues[i]), keyValues[i + 1]);
+ }
+ return result;
+ }
+
+ private static void putIfNotNull(Map target, String key, Object value) {
+ if (value != null) {
+ target.put(key, value);
+ }
+ }
+
+ private static String nullToEmpty(String value) {
+ return value == null ? "" : value;
}
- private int effectiveMaxFrameChars() {
- return config.getMaxFrameChars() <= 0 ? Integer.MAX_VALUE : config.getMaxFrameChars();
+ private static final class FrameTooLargeException extends IOException {
+ private static final long serialVersionUID = 1L;
}
}
diff --git a/src/main/java/io/github/easy4j/comfy/mcp/ComfyMcpConfig.java b/src/main/java/io/github/easy4j/comfy/mcp/ComfyMcpConfig.java
index 9d4499b..62919b5 100644
--- a/src/main/java/io/github/easy4j/comfy/mcp/ComfyMcpConfig.java
+++ b/src/main/java/io/github/easy4j/comfy/mcp/ComfyMcpConfig.java
@@ -2,16 +2,6 @@
* Copyright (c) 2018-present, easy-4-java (https://github.com/easy-4-java).
*
* Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
*/
package io.github.easy4j.comfy.mcp;
@@ -20,74 +10,45 @@
import lombok.Data;
-/**
- * Configuration for the comfy-mcp route.
- *
- * {@code comfy-mcp} speaks MCP over stdio (newline-delimited JSON-RPC).
- * The SDK spawns it as a child process and drives {@code tools/list} /
- * {@code tools/call} through that channel — no sockets, so this route
- * works on every JDK line the SDK supports.
- *
- * @author Loong Wan
- * @since 1.0.0
- * @see ComfyMcpClient
- */
+/** Configuration for the local {@code comfy-mcp} stdio route. */
@Data
public class ComfyMcpConfig {
- /**
- * Name or absolute path of the MCP server executable
- * (e.g. {@code comfy-mcp} from PyPI, or an absolute venv path).
- */
private String localExecutable = "comfy-mcp";
-
- /** Extra arguments forwarded to the server executable. */
private String[] mcpArgs;
-
- /**
- * Extra environment variables for the server process (e.g.
- * {@code COMFY_BIN} pointing at the workspace's {@code comfy} binary);
- * merged over the parent environment. Required when {@code comfy} is not
- * on the PATH the SDK's process sees.
- */
private Map environment;
-
- /** MCP protocol version advertised in {@code initialize}. */
private String protocolVersion = "2024-11-05";
-
- /** Client name advertised in {@code initialize}. */
private String clientName = "comfy-java-sdk";
-
- /** Client version advertised in {@code initialize}. */
private String clientVersion = "1.0.0";
-
- /** Timeout in milliseconds for process startup plus the MCP {@code initialize} handshake. */
private int connectTimeoutMillis = 10_000;
-
- /** Upper bound in milliseconds for one {@code tools/call} (generation tools can run for minutes). */
private int readTimeoutMillis = 900_000;
+ private int shutdownTimeoutMillis = 2_000;
/**
- * Hard cap in characters for one JSON-RPC frame on the wire; a frame
- * exceeding it tears the transport down. {@code <= 0} means unbounded.
- * Defaults to 1 MiB characters.
+ * Hard frame limit before JSON parsing. 16 MiB allows normal inline image
+ * responses while still bounding a malicious/broken single-line frame.
+ * {@code <= 0} means unbounded.
*/
- private int maxFrameChars = 1_048_576;
+ private int maxFrameChars = 16 * 1024 * 1024;
- /**
- * Hard cap in characters for the text content accumulated per
- * {@code tools/call}; excess content items are truncated with a warning.
- * {@code <= 0} means unbounded. Defaults to 1 MiB characters.
- */
- private int maxContentChars = 1_048_576;
+ /** Text aggregation cap for one tools/call. {@code <= 0} means unbounded. */
+ private int maxContentChars = 4 * 1024 * 1024;
- /**
- * Validates the configuration.
- *
- * @throws NullPointerException when the executable or client name is {@code null}.
- */
public void validate() {
Objects.requireNonNull(localExecutable, "localExecutable");
Objects.requireNonNull(clientName, "clientName");
+ Objects.requireNonNull(protocolVersion, "protocolVersion");
+ if (localExecutable.trim().isEmpty()) {
+ throw new IllegalStateException("localExecutable must not be blank");
+ }
+ if (connectTimeoutMillis <= 0) {
+ throw new IllegalStateException("connectTimeoutMillis must be > 0");
+ }
+ if (readTimeoutMillis <= 0) {
+ throw new IllegalStateException("readTimeoutMillis must be > 0");
+ }
+ if (shutdownTimeoutMillis < 0) {
+ throw new IllegalStateException("shutdownTimeoutMillis must be >= 0");
+ }
}
}
diff --git a/src/main/java/io/github/easy4j/comfy/mcp/ComfyMcpContent.java b/src/main/java/io/github/easy4j/comfy/mcp/ComfyMcpContent.java
new file mode 100644
index 0000000..479d889
--- /dev/null
+++ b/src/main/java/io/github/easy4j/comfy/mcp/ComfyMcpContent.java
@@ -0,0 +1,23 @@
+/*
+ * Copyright (c) 2018-present, easy-4-java (https://github.com/easy-4-java).
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ */
+package io.github.easy4j.comfy.mcp;
+
+import lombok.Data;
+import com.fasterxml.jackson.databind.JsonNode;
+
+/**
+ * One MCP content item. Known common fields are projected for convenience and
+ * {@link #raw} preserves new MCP content variants without SDK upgrades.
+ */
+@Data
+public class ComfyMcpContent {
+ private final String type;
+ private final String text;
+ private final String mimeType;
+ private final String data;
+ private final String uri;
+ private final JsonNode raw;
+}
diff --git a/src/main/java/io/github/easy4j/comfy/model/ComfyCliEnvelope.java b/src/main/java/io/github/easy4j/comfy/model/ComfyCliEnvelope.java
new file mode 100644
index 0000000..94ebacd
--- /dev/null
+++ b/src/main/java/io/github/easy4j/comfy/model/ComfyCliEnvelope.java
@@ -0,0 +1,26 @@
+/*
+ * Copyright (c) 2018-present, easy-4-java (https://github.com/easy-4-java).
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ */
+package io.github.easy4j.comfy.model;
+
+import lombok.Data;
+import com.fasterxml.jackson.databind.JsonNode;
+
+/**
+ * Uniform JSON envelope emitted by {@code comfy --json }.
+ *
+ * The payload intentionally keeps {@code data} and {@code error} as
+ * {@link JsonNode}: the command tree is self-describing via
+ * {@code comfy --json discover} and evolves independently of this SDK.
+ */
+@Data
+public class ComfyCliEnvelope {
+ private boolean ok;
+ private String command;
+ private String version;
+ private String where;
+ private JsonNode data;
+ private JsonNode error;
+}
diff --git a/src/test/java/io/github/easy4j/comfy/ComfyClientConfigTest.java b/src/test/java/io/github/easy4j/comfy/ComfyClientConfigTest.java
index 2d976bf..12e3c0e 100644
--- a/src/test/java/io/github/easy4j/comfy/ComfyClientConfigTest.java
+++ b/src/test/java/io/github/easy4j/comfy/ComfyClientConfigTest.java
@@ -1,17 +1,5 @@
/*
- * Copyright (c) 2018-present, easy-4-java (https://github.com/easy-4-java).
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
+ * Copyright (c) 2018-present, easy-4-java.
*/
package io.github.easy4j.comfy;
@@ -20,39 +8,29 @@
import org.junit.jupiter.api.Test;
-/** Unit tests for {@link ComfyClientConfig} defaults and validation. @since 1.0.0 */
class ComfyClientConfigTest {
@Test
- void shouldExposeSensibleDefaults() {
+ void shouldExposeProductionSafeDefaults() {
ComfyClientConfig config = new ComfyClientConfig();
assertEquals("comfy", config.getLocalExecutable());
assertEquals(600, config.getLocalTimeoutSeconds());
assertEquals(5, config.getLocalProbeTimeoutSeconds());
+ assertEquals(16 * 1024 * 1024, config.getMaxOutputBytes());
}
@Test
- void shouldAcceptValidWhereValues() {
- ComfyClientConfig config = new ComfyClientConfig();
- config.setDefaultWhere("local");
- config.validate();
- config.setDefaultWhere("cloud");
- config.validate();
- }
-
- @Test
- void shouldRejectInvalidWhereValues() {
+ void shouldValidateRoutingAndTimeouts() {
ComfyClientConfig config = new ComfyClientConfig();
config.setDefaultWhere("bogus");
+ assertThrows(IllegalArgumentException.class, config::validate);
+
+ config = new ComfyClientConfig();
+ config.setLocalTimeoutSeconds(0);
assertThrows(IllegalStateException.class, config::validate);
- }
- @Test
- void shouldAcceptEnvironmentOverrides() {
- ComfyClientConfig config = new ComfyClientConfig();
- config.getEnvironment(); // null by default
- config.setEnvironment(java.util.Collections.singletonMap("COMFY_API_KEY", "k"));
- config.validate();
- assertEquals("k", config.getEnvironment().get("COMFY_API_KEY"));
+ config = new ComfyClientConfig();
+ config.setLocalProbeTimeoutSeconds(0);
+ assertThrows(IllegalStateException.class, config::validate);
}
}
diff --git a/src/test/java/io/github/easy4j/comfy/ComfyClientTest.java b/src/test/java/io/github/easy4j/comfy/ComfyClientTest.java
index e2ad1d1..23f43b1 100644
--- a/src/test/java/io/github/easy4j/comfy/ComfyClientTest.java
+++ b/src/test/java/io/github/easy4j/comfy/ComfyClientTest.java
@@ -1,17 +1,5 @@
/*
- * Copyright (c) 2018-present, easy-4-java (https://github.com/easy-4-java).
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
+ * Copyright (c) 2018-present, easy-4-java.
*/
package io.github.easy4j.comfy;
@@ -21,77 +9,75 @@
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import java.nio.file.Paths;
+
import org.junit.jupiter.api.Test;
import io.github.easy4j.comfy.cli.ComfyCli;
+import io.github.easy4j.comfy.model.ComfyCliEnvelope;
import com.fasterxml.jackson.databind.JsonNode;
-import io.github.easy4j.comfy.cli.ComfyCliExecutor;
-/**
- * Unit tests for {@link ComfyClient} validation and delegation.
- *
- * @since 1.0.0
- */
class ComfyClientTest {
- private static ComfyClientConfig echoConfig() {
+ private static String resource(String name) {
+ return Paths.get("src", "test", "resources", name).toAbsolutePath().toString();
+ }
+
+ private static ComfyClientConfig config(String executable) {
ComfyClientConfig config = new ComfyClientConfig();
- config.setLocalExecutable(
- java.nio.file.Paths.get("src", "test", "resources", "comfy-echo.sh").toAbsolutePath().toString());
+ config.setLocalExecutable(executable);
config.setLocalTimeoutSeconds(2);
return config;
}
- @Test
- void shouldRejectNullConfig() {
- assertThrows(NullPointerException.class, () -> new ComfyClient(null));
- }
-
- @Test
- void shouldRejectInvalidConfig() {
- ComfyClientConfig config = echoConfig();
- config.setDefaultWhere("bogus");
- assertThrows(IllegalStateException.class, () -> new ComfyClient(config));
- }
-
@Test
void shouldDelegateBasics() {
- try (ComfyClient client = new ComfyClient(echoConfig())) {
+ try (ComfyClient client = new ComfyClient(config(resource("comfy-echo.sh")))) {
assertTrue(client.version().getStdout().contains("--version"));
assertTrue(client.isAvailable());
- assertTrue(client.cloudLogin().getStdout().contains("cloud login"));
- assertTrue(client.setup().getStdout().contains("-y"));
- assertTrue(client.skillsInstall().getStdout().contains("skills install"));
- assertNotNull(client.getConfig());
assertNotNull(client.cli());
+ assertNotNull(client.getConfig());
+ }
+ }
+
+ @Test
+ void shouldParseGenerateJsonWithoutMutatingOptions() {
+ ComfyCli.GenerateOptions options = new ComfyCli.GenerateOptions().prompt("hi");
+ try (ComfyClient jsonClient = new ComfyClient(config(resource("comfy-json.sh")))) {
+ JsonNode json = jsonClient.generateJson("flux-pro", options);
+ assertEquals("https://example/asset.png", json.path("data").get(0).path("url").asText());
+ }
+
+ try (ComfyClient echoClient = new ComfyClient(config(resource("comfy-echo.sh")))) {
+ String args = echoClient.cli().generate("flux-pro", options).getStdout();
+ assertFalse(args.contains("--json"), "generateJson must not mutate caller options");
}
}
@Test
- void shouldRaiseWhenGenerateJsonPrintsNonJson() {
- // The echo fixture prints its argument list, which is not JSON —
- // generateJson must surface that as ComfyException.
- try (ComfyClient client = new ComfyClient(echoConfig())) {
- assertThrows(ComfyException.class,
- () -> client.generateJson("flux-pro", new ComfyCli.GenerateOptions().prompt("hi")));
+ void shouldParseUniformCliEnvelope() {
+ try (ComfyClient client = new ComfyClient(config(resource("comfy-envelope.sh")))) {
+ ComfyCliEnvelope envelope = client.environment();
+ assertTrue(envelope.isOk());
+ assertEquals("env", envelope.getCommand());
+ assertEquals("local", envelope.getWhere());
+ assertTrue(envelope.getData().path("running").asBoolean());
}
}
@Test
- void shouldParseGenerateJsonOutput() throws Exception {
- ComfyClientConfig config = echoConfig();
- config.setLocalExecutable(
- java.nio.file.Paths.get("src", "test", "resources", "comfy-json.sh").toAbsolutePath().toString());
+ void shouldRejectTruncatedJsonEnvelope() {
+ ComfyClientConfig config = config(resource("comfy-envelope.sh"));
+ config.setMaxOutputBytes(12);
try (ComfyClient client = new ComfyClient(config)) {
- JsonNode json = client.generateJson("flux-pro",
- new ComfyCli.GenerateOptions().prompt("hi").json(true));
- assertEquals("https://example/asset.png", json.path("data").get(0).path("url").asText());
+ assertThrows(ComfyException.class, client::environment);
}
}
@Test
- void shouldCloseWithoutError() {
- ComfyClient client = new ComfyClient(echoConfig());
- client.close();
+ void shouldRejectInvalidConfig() {
+ ComfyClientConfig config = config(resource("comfy-echo.sh"));
+ config.setDefaultWhere("bogus");
+ assertThrows(IllegalArgumentException.class, () -> new ComfyClient(config));
}
}
diff --git a/src/test/java/io/github/easy4j/comfy/cli/ComfyCliExecutorTest.java b/src/test/java/io/github/easy4j/comfy/cli/ComfyCliExecutorTest.java
index da8885a..3ded163 100644
--- a/src/test/java/io/github/easy4j/comfy/cli/ComfyCliExecutorTest.java
+++ b/src/test/java/io/github/easy4j/comfy/cli/ComfyCliExecutorTest.java
@@ -1,17 +1,5 @@
/*
- * Copyright (c) 2018-present, easy-4-java (https://github.com/easy-4-java).
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
+ * Copyright (c) 2018-present, easy-4-java.
*/
package io.github.easy4j.comfy.cli;
@@ -19,161 +7,117 @@
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import java.nio.file.Paths;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
import org.junit.jupiter.api.Test;
import io.github.easy4j.comfy.ComfyClientConfig;
-/**
- * Unit tests for {@link ComfyCliExecutor}: raw argv passing, exit-code
- * preservation, stream capture, stdin piping and the watchdog timeout.
- *
- * @since 1.0.0
- */
class ComfyCliExecutorTest {
- /** Absolute path of the argument-echoing fixture script (surefire runs from the module base dir). */
private static final String ECHO =
- java.nio.file.Paths.get("src", "test", "resources", "comfy-echo.sh").toAbsolutePath().toString();
+ Paths.get("src", "test", "resources", "comfy-echo.sh").toAbsolutePath().toString();
+ private static final String SLOW =
+ Paths.get("src", "test", "resources", "comfy-slow.sh").toAbsolutePath().toString();
private ComfyClientConfig configFor(String executable) {
ComfyClientConfig config = new ComfyClientConfig();
config.setLocalExecutable(executable);
- // Short timeouts so failing tests stay fast.
config.setLocalTimeoutSeconds(2);
- config.setLocalProbeTimeoutSeconds(2);
+ config.setLocalProbeTimeoutSeconds(1);
return config;
}
@Test
void shouldExecuteSuccessfullyWithCapturedStdout() {
- ComfyCliExecutor executor = new ComfyCliExecutor(configFor(ECHO));
-
- ComfyCliResult result = executor.execute("hello", "world");
-
+ ComfyCliResult result = new ComfyCliExecutor(configFor(ECHO)).execute("hello", "world");
assertEquals(0, result.getExitCode());
assertTrue(result.isSuccess());
assertEquals("hello world", result.getStdout());
}
@Test
- void shouldPassArgumentsRawWithoutEmbeddedQuotes() {
- ComfyCliExecutor executor = new ComfyCliExecutor(configFor(ECHO));
-
- ComfyCliResult result = executor.execute("Write a failing test", "--model", "kimi k2");
-
- assertEquals("Write a failing test --model kimi k2", result.getStdout(),
- "multi-word arguments must arrive without embedded literal quotes");
+ void shouldPassArgumentsRawWithoutShellInterpretation() {
+ ComfyCliResult result = new ComfyCliExecutor(configFor(ECHO))
+ .execute("a b", ";rm -rf /", "$(whoami)");
+ assertEquals("a b ;rm -rf / $(whoami)", result.getStdout());
}
@Test
void shouldPreserveRealExitCodeAndStreamsOnNonZeroExit() {
- ComfyCliExecutor executor = new ComfyCliExecutor(configFor("/bin/sh"));
-
- ComfyCliResult result = executor.execute("-c", "echo out-marker; echo err-marker 1>&2; exit 7");
-
+ ComfyCliResult result = new ComfyCliExecutor(configFor("/bin/sh"))
+ .execute("-c", "echo out-marker; echo err-marker 1>&2; exit 7");
assertEquals(7, result.getExitCode());
assertFalse(result.isSuccess());
- assertTrue(result.getStdout().contains("out-marker"), "stdout must survive a non-zero exit");
- assertTrue(result.getStderr().contains("err-marker"), "stderr must survive a non-zero exit");
- }
-
- @Test
- void shouldReturnIoExceptionMessageWhenExecutableMissing() {
- ComfyCliExecutor executor = new ComfyCliExecutor(configFor("/nonexistent/path/to/comfy"));
-
- ComfyCliResult result = executor.execute("--version");
-
- assertEquals(-1, result.getExitCode());
- assertFalse(result.isSuccess());
- assertTrue(result.getStderr() != null && !result.getStderr().isEmpty());
+ assertTrue(result.getStdout().contains("out-marker"));
+ assertTrue(result.getStderr().contains("err-marker"));
}
@Test
- void shouldDecodeUtf8OutputRegardlessOfPlatformCharset() {
- // POSIX printf octal escapes emit 你好 as raw UTF-8 bytes; with a
- // platform-default-charset decode this corrupts on C-locale JVMs.
- // NOTE: the backslashes are doubled in Java source so the shell
- // receives single ones — an octal escape like \344 must be written
- // \\344 here or the compiler eats it at compile time.
- ComfyCliExecutor executor = new ComfyCliExecutor(configFor("/bin/sh"));
-
- ComfyCliResult result = executor.execute("-c", "printf '\\344\\275\\240\\345\\245\\275'");
-
+ void shouldDecodeUtf8Output() {
+ ComfyCliResult result = new ComfyCliExecutor(configFor("/bin/sh"))
+ .execute("-c", "printf '\\344\\275\\240\\345\\245\\275'");
assertEquals("你好", result.getStdout());
}
@Test
- void shouldIgnoreNullArguments() {
- ComfyCliExecutor executor = new ComfyCliExecutor(configFor(ECHO));
-
- ComfyCliResult result = executor.execute("hello", null, "world");
-
- assertEquals(0, result.getExitCode());
- assertEquals("hello world", result.getStdout());
- }
-
- @Test
- void shouldFeedStdinToChildProcess() {
- // `cat` with no file arguments echoes its standard input verbatim,
- // which is how stdin-consuming CLI forms receive their payload.
- ComfyCliExecutor executor = new ComfyCliExecutor(configFor("/bin/cat"));
-
- ComfyCliResult result = executor.executeWithStdin("secret-api-key");
-
- assertEquals(0, result.getExitCode());
- assertEquals("secret-api-key", result.getStdout());
- }
-
- @Test
- void shouldExecuteWithoutStdinAsBefore() {
- ComfyCliExecutor executor = new ComfyCliExecutor(configFor(ECHO));
-
- assertEquals("plain", executor.executeWithStdin(null, "plain").getStdout());
- assertEquals("plain", executor.executeWithStdin("", "plain").getStdout());
+ void shouldFeedStdinAndCloseIt() {
+ ComfyCliResult result = new ComfyCliExecutor(configFor("/bin/cat"))
+ .executeWithStdin("payload");
+ assertEquals("payload", result.getStdout());
}
@Test
void shouldInjectEnvironmentIntoChildProcess() {
ComfyClientConfig config = configFor("/usr/bin/env");
- java.util.Map env = new java.util.LinkedHashMap();
+ Map env = new LinkedHashMap();
env.put("COMFY_PROBE_MARKER", "injected-ok");
config.setEnvironment(env);
-
- ComfyCliResult result = executor(config).execute();
-
- assertTrue(result.getStdout().contains("COMFY_PROBE_MARKER=injected-ok"),
- "child environment must carry the injected variables");
- }
-
- private ComfyCliExecutor executor(ComfyClientConfig config) {
- return new ComfyCliExecutor(config);
+ assertTrue(new ComfyCliExecutor(config).execute().getStdout()
+ .contains("COMFY_PROBE_MARKER=injected-ok"));
}
@Test
- void shouldReportSuccessFromProbeWhenExecutableWorks() {
- ComfyCliExecutor executor = new ComfyCliExecutor(configFor(ECHO));
-
- assertTrue(executor.probe());
+ void shouldUseDedicatedProbeTimeout() {
+ ComfyClientConfig config = configFor(SLOW);
+ config.setLocalTimeoutSeconds(30);
+ config.setLocalProbeTimeoutSeconds(1);
+ long started = System.nanoTime();
+ assertFalse(new ComfyCliExecutor(config).probe());
+ long millis = (System.nanoTime() - started) / 1_000_000L;
+ assertTrue(millis < 4_000L, "probe must not inherit the 30 second execution timeout");
}
@Test
- void shouldReportFailureFromProbeWhenExecutableMissing() {
- ComfyCliExecutor executor = new ComfyCliExecutor(configFor("/nonexistent/path/to/comfy"));
-
- assertFalse(executor.probe());
+ void shouldBoundCapturedOutput() {
+ ComfyClientConfig config = configFor("/bin/sh");
+ config.setMaxOutputBytes(8);
+ ComfyCliResult result = new ComfyCliExecutor(config)
+ .execute("-c", "printf '12345678901234567890'; printf 'abcdefghijklmnop' 1>&2");
+ assertEquals("12345678", result.getStdout());
+ assertEquals("abcdefgh", result.getStderr());
+ assertTrue(result.isStdoutTruncated());
+ assertTrue(result.isStderrTruncated());
+ assertTrue(result.isTruncated());
}
@Test
void shouldTimeoutOnHangingProcess() {
- // Use a short timeout and a command that sleeps for a long time.
ComfyClientConfig config = configFor("/bin/sh");
config.setLocalTimeoutSeconds(1);
- ComfyCliExecutor executor = new ComfyCliExecutor(config);
-
- ComfyCliResult result = executor.execute("-c", "sleep 30");
+ ComfyCliResult result = new ComfyCliExecutor(config).execute("-c", "sleep 30");
+ assertEquals(-1, result.getExitCode());
+ assertTrue(result.isTimeout());
+ }
+ @Test
+ void shouldReturnMessageWhenExecutableMissing() {
+ ComfyCliResult result = new ComfyCliExecutor(configFor("/nonexistent/path/to/comfy"))
+ .execute("--version");
assertEquals(-1, result.getExitCode());
assertFalse(result.isSuccess());
- assertTrue(result.isTimeout(), "stderr must carry the timeout notice");
+ assertTrue(result.getStderr() != null && !result.getStderr().isEmpty());
}
}
diff --git a/src/test/java/io/github/easy4j/comfy/cli/ComfyCliTest.java b/src/test/java/io/github/easy4j/comfy/cli/ComfyCliTest.java
index 4900d95..5fad610 100644
--- a/src/test/java/io/github/easy4j/comfy/cli/ComfyCliTest.java
+++ b/src/test/java/io/github/easy4j/comfy/cli/ComfyCliTest.java
@@ -1,162 +1,146 @@
/*
- * Copyright (c) 2018-present, easy-4-java (https://github.com/easy-4-java).
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
+ * Copyright (c) 2018-present, easy-4-java.
*/
package io.github.easy4j.comfy.cli;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import java.nio.file.Paths;
+
import org.junit.jupiter.api.Test;
import io.github.easy4j.comfy.ComfyClientConfig;
-/**
- * Contract tests for {@link ComfyCli} driven without a socket: outgoing
- * argument lists are verified through the echo fixture.
- *
- * @since 1.0.0
- */
class ComfyCliTest {
- /** Absolute path of the argument-echoing fixture script (surefire runs from the module base dir). */
private static final String ECHO =
- java.nio.file.Paths.get("src", "test", "resources", "comfy-echo.sh").toAbsolutePath().toString();
+ Paths.get("src", "test", "resources", "comfy-echo.sh").toAbsolutePath().toString();
- private static ComfyClientConfig echoConfig() {
+ private static ComfyCli cli() {
ComfyClientConfig config = new ComfyClientConfig();
config.setLocalExecutable(ECHO);
config.setLocalTimeoutSeconds(2);
- return config;
- }
-
- private static ComfyCli echoCli() {
- return new ComfyCli(echoConfig(), new ComfyCliExecutor(echoConfig()));
+ return new ComfyCli(config, new ComfyCliExecutor(config));
}
@Test
- void shouldExposeExecutor() {
- assertNotNull(echoCli().executor());
+ void shouldMapGlobalAndEnvironmentCommands() {
+ assertTrue(cli().discoverJson().getStdout().contains("--json discover"));
+ assertTrue(cli().executeJsonStream("jobs", "watch", "p1").getStdout()
+ .contains("--json-stream jobs watch p1"));
+ assertTrue(cli().which().getStdout().contains("which"));
+ assertTrue(cli().env().getStdout().contains("env"));
+ assertTrue(cli().outdated().getStdout().contains("outdated"));
+ assertTrue(cli().systemStats().getStdout().contains("system-stats"));
+ assertTrue(cli().freeMemory().getStdout().contains("free-memory"));
}
@Test
- void shouldDelegateVersionAndHelp() {
- assertTrue(echoCli().version().getStdout().contains("--version"));
- assertTrue(echoCli().help().getStdout().contains("--help"));
- assertTrue(echoCli().installCompletion().getStdout().contains("--install-completion"));
- assertTrue(echoCli().discoverJson().getStdout().contains("--json discover"));
+ void shouldMapSetupCloudAndRouting() {
+ assertTrue(cli().setupYes().getStdout().contains("setup -y"));
+ assertTrue(cli().cloudLoginNoBrowser().getStdout().contains("cloud login --no-browser"));
+ assertTrue(cli().cloudWhoami().getStdout().contains("cloud whoami"));
+ assertTrue(cli().cloudLogout().getStdout().contains("cloud logout"));
+ assertTrue(cli().cloudStatus().getStdout().contains("cloud status"));
+ assertTrue(cli().cloudSetBaseUrl("https://example.test").getStdout()
+ .contains("cloud set-base-url https://example.test"));
+ assertTrue(cli().setDefaultWhere("local").getStdout().contains("set-default --where local"));
+ assertThrows(IllegalArgumentException.class, () -> cli().setDefaultWhere("bogus"));
}
@Test
- void shouldDelegateSetupAndCloudAuth() {
- assertTrue(echoCli().setup().getStdout().contains("setup"));
- assertTrue(echoCli().setupYes().getStdout().contains("-y"));
- assertTrue(echoCli().cloudLogin().getStdout().contains("cloud login"));
- assertTrue(echoCli().cloudWhoami().getStdout().contains("cloud whoami"));
+ void shouldMapLifecycle() {
+ assertTrue(cli().install("--here").getStdout().contains("install --here"));
+ assertTrue(cli().launchBackground("--port", "8188").getStdout()
+ .contains("launch --background --port 8188"));
+ assertTrue(cli().stop().getStdout().contains("stop"));
+ assertTrue(cli().update("comfy").getStdout().contains("update comfy"));
+ assertTrue(cli().logs("--tail", "20").getStdout().contains("logs --tail 20"));
}
@Test
- void shouldDelegateSetDefaultWhere() {
- assertTrue(echoCli().setDefaultWhere("cloud").getStdout().contains("--where cloud"));
- assertThrows(IllegalArgumentException.class, () -> echoCli().setDefaultWhere("bogus"));
- }
-
- @Test
- void shouldDelegateComfyUiLifecycle() {
- assertTrue(echoCli().install("--here").getStdout().contains("install --here"));
- assertTrue(echoCli().launch("--port", "8188").getStdout().contains("launch --port 8188"));
- assertTrue(echoCli().stop().getStdout().contains("stop"));
- assertTrue(echoCli().update().getStdout().contains("update"));
- }
-
- @Test
- void shouldBuildGenerateWithAllFlags() {
+ void shouldBuildGenerationAndValidateDynamicOptions() {
ComfyCli.GenerateOptions options = new ComfyCli.GenerateOptions()
- .prompt("a cat on the moon")
- .width(1024).height(1024)
- .download("cat.png")
- .image("in.png").mask("mask.png")
- .resolution("1080p").duration(5).aspectRatio("16:9")
- .renderingSpeed("quality")
- .async(true).json(true)
- .where("cloud");
-
- ComfyCliResult result = echoCli().generate("flux-pro", options);
- String out = result.getStdout();
+ .prompt("a cat")
+ .width(1024).height(768)
+ .duration(5)
+ .timeoutSeconds(90)
+ .where("cloud")
+ .option("seed", 42)
+ .flag("enhance-prompt");
+
+ String out = cli().generate("flux-pro", options).getStdout();
assertTrue(out.contains("generate flux-pro"));
- assertTrue(out.contains("--prompt a cat on the moon"));
+ assertTrue(out.contains("--prompt a cat"));
assertTrue(out.contains("--width 1024"));
- assertTrue(out.contains("--height 1024"));
- assertTrue(out.contains("--download cat.png"));
- assertTrue(out.contains("--image in.png"));
- assertTrue(out.contains("--mask mask.png"));
- assertTrue(out.contains("--resolution 1080p"));
- assertTrue(out.contains("--duration 5"));
- assertTrue(out.contains("--aspect_ratio 16:9"));
- assertTrue(out.contains("--rendering_speed quality"));
- assertTrue(out.contains("--async"));
- assertTrue(out.contains("--json"));
- assertTrue(out.contains("--where cloud"), "explicit where must be appended");
- }
+ assertTrue(out.contains("--timeout 90"));
+ assertTrue(out.contains("--seed 42"));
+ assertTrue(out.contains("--enhance-prompt"));
+ assertTrue(out.contains("--where cloud"));
- @Test
- void shouldPropagateConfigDefaultWhere() {
- ComfyClientConfig config = echoConfig();
- config.setDefaultWhere("local");
- ComfyCliResult result = new ComfyCli(config, new ComfyCliExecutor(config))
- .generate("seedance", new ComfyCli.GenerateOptions().prompt("hi").where(null));
- assertTrue(result.getStdout().contains("--where local"));
- }
-
- @Test
- void shouldRejectNullModelAndOptions() {
- assertThrows(NullPointerException.class, () -> echoCli().generate(null, new ComfyCli.GenerateOptions()));
- assertThrows(NullPointerException.class, () -> echoCli().generate("flux-pro", null));
+ assertThrows(IllegalArgumentException.class, () -> new ComfyCli.GenerateOptions().where("x"));
+ assertThrows(IllegalArgumentException.class, () -> new ComfyCli.GenerateOptions().option("bad flag", 1));
}
@Test
- void shouldDelegateGenerateVariants() {
- assertTrue(echoCli().generateList("text-to-video", "kling").getStdout()
- .contains("--category text-to-video --partner kling"));
- assertTrue(echoCli().generateSchema("flux-kontext").getStdout().contains("schema flux-kontext"));
- assertTrue(echoCli().generateUpload("in.png").getStdout().contains("upload in.png"));
- String resume = echoCli().generateResume("luma", "job-1", "out.mp4").getStdout();
- assertTrue(resume.contains("resume luma job-1 --download out.mp4"));
+ void shouldMapWorkflowJobAndTemplateCommands() {
+ assertTrue(cli().runWorkflow("wf.json", true).getStdout()
+ .contains("run --workflow wf.json --wait"));
+ assertTrue(cli().jobsList().getStdout().contains("jobs ls"));
+ assertTrue(cli().jobStatus("p1").getStdout().contains("jobs status p1"));
+ assertTrue(cli().jobsWait("p1", "p2").getStdout().contains("jobs wait p1 p2"));
+ assertTrue(cli().jobCancel("p1").getStdout().contains("jobs cancel p1"));
+ assertTrue(cli().validateWorkflow("wf.json").getStdout()
+ .contains("validate --workflow wf.json"));
+ assertTrue(cli().templatesList("image", "Text to Image").getStdout()
+ .contains("templates ls --type image --tag Text to Image"));
+ assertTrue(cli().templateFetch("basic", "out.json").getStdout()
+ .contains("templates fetch basic --out out.json"));
}
@Test
- void shouldDelegateWorkflowAndDiscovery() {
- assertTrue(echoCli().run("--workflow", "a.json").getStdout().contains("run --workflow a.json"));
- assertTrue(echoCli().jobs().getStdout().contains("jobs"));
- assertTrue(echoCli().validate("wf.json").getStdout().contains("validate wf.json"));
- assertTrue(echoCli().workflow("list").getStdout().contains("workflow list"));
- assertTrue(echoCli().templates().getStdout().contains("templates"));
- assertTrue(echoCli().nodes("search").getStdout().contains("nodes search"));
- assertTrue(echoCli().models("list").getStdout().contains("models list"));
+ void shouldMapWorkflowEditing() {
+ assertTrue(cli().workflowSlots("wf.json").getStdout().contains("workflow slots wf.json"));
+ assertTrue(cli().workflowSetSlot("wf.json", "6.text=a fox").getStdout()
+ .contains("workflow set-slot wf.json 6.text=a fox"));
+ assertTrue(cli().workflowVary("wf.json", "variants", "6.seed=[1,2]").getStdout()
+ .contains("workflow vary wf.json --slot 6.seed=[1,2] --out-dir variants"));
+ assertTrue(cli().workflowList().getStdout().contains("workflow list"));
+ assertTrue(cli().workflowGet("id1", "wf.json").getStdout()
+ .contains("workflow get id1 --out wf.json"));
+ assertTrue(cli().workflowSave("wf.json", "My Flow").getStdout()
+ .contains("workflow save wf.json --name My Flow"));
+ assertTrue(cli().workflowDelete("id1").getStdout().contains("workflow delete id1"));
+ assertTrue(cli().workflowCompose("pipe.yaml", "wf.json").getStdout()
+ .contains("workflow compose pipe.yaml -o wf.json"));
+ assertTrue(cli().workflowDecompose("wf.json").getStdout()
+ .contains("workflow decompose wf.json"));
}
@Test
- void shouldDelegateSkills() {
- assertTrue(echoCli().skillsInstall().getStdout().contains("skills install"));
- assertTrue(echoCli().skillsList().getStdout().contains("skills list"));
- assertTrue(echoCli().skillsStatus().getStdout().contains("skills status"));
+ void shouldMapDiscoveryAssetsAndManagementFamilies() {
+ assertTrue(cli().nodesSearch("checkpoint").getStdout().contains("nodes search checkpoint"));
+ assertTrue(cli().nodeShow("KSampler").getStdout().contains("nodes show KSampler"));
+ assertTrue(cli().modelFolders().getStdout().contains("models list-folders"));
+ assertTrue(cli().modelsSearch("wan", "lora").getStdout()
+ .contains("models search --text wan --type lora"));
+ assertTrue(cli().nodeInstall("pack").getStdout().contains("node install pack"));
+ assertTrue(cli().modelDownload("https://example/model", "models/checkpoints").getStdout()
+ .contains("model download --url https://example/model --relative-path models/checkpoints"));
+ assertTrue(cli().upload("a.png", "b.png").getStdout().contains("upload a.png b.png"));
+ assertTrue(cli().download("p1", "-o", "outputs").getStdout()
+ .contains("download p1 -o outputs"));
}
@Test
- void shouldDelegateRawExecute() {
- assertTrue(echoCli().execute("--version").getStdout().contains("--version"));
+ void shouldMapSkillsAndTrackingAndKeepEscapeHatch() {
+ assertTrue(cli().skillsInstall().getStdout().contains("skills install"));
+ assertTrue(cli().skillsList().getStdout().contains("skills list"));
+ assertTrue(cli().skillsStatus().getStdout().contains("skills status"));
+ assertTrue(cli().trackingDisable().getStdout().contains("tracking disable"));
+ assertTrue(cli().trackingEnable().getStdout().contains("tracking enable"));
+ assertTrue(cli().execute("future-command", "--x").getStdout()
+ .contains("future-command --x"));
}
}
diff --git a/src/test/java/io/github/easy4j/comfy/mcp/ComfyMcpClientE2ETest.java b/src/test/java/io/github/easy4j/comfy/mcp/ComfyMcpClientE2ETest.java
index 9102532..95a7ce3 100644
--- a/src/test/java/io/github/easy4j/comfy/mcp/ComfyMcpClientE2ETest.java
+++ b/src/test/java/io/github/easy4j/comfy/mcp/ComfyMcpClientE2ETest.java
@@ -1,17 +1,5 @@
/*
- * Copyright (c) 2018-present, easy-4-java (https://github.com/easy-4-java).
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
+ * Copyright (c) 2018-present, easy-4-java.
*/
package io.github.easy4j.comfy.mcp;
@@ -22,6 +10,7 @@
import static org.junit.jupiter.api.Assertions.assertTrue;
import java.nio.file.Paths;
+import java.util.Arrays;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
@@ -30,59 +19,80 @@
import io.github.easy4j.comfy.ComfyException;
-/**
- * End-to-end tests running {@link ComfyMcpClient} against a fake MCP server
- * process (python3, NDJSON JSON-RPC on stdio) — the same wire contract as
- * {@code comfy-mcp}.
- *
- * @since 1.0.0
- */
class ComfyMcpClientE2ETest {
- private static final String FAKE_SERVER = Paths.get("src", "test", "resources", "fake-mcp-server.py")
+ private static final String FAKE_SERVER = Paths
+ .get("src", "test", "resources", "fake-mcp-server.py")
.toAbsolutePath().toString();
private static ComfyMcpConfig config() {
ComfyMcpConfig config = new ComfyMcpConfig();
config.setLocalExecutable("python3");
config.setMcpArgs(new String[] {FAKE_SERVER});
- config.setConnectTimeoutMillis(10_000);
- config.setReadTimeoutMillis(10_000);
+ config.setConnectTimeoutMillis(5_000);
+ config.setReadTimeoutMillis(5_000);
+ config.setShutdownTimeoutMillis(1_000);
return config;
}
@Test
- void shouldConnectAndInitialize() {
+ void shouldConnectListToolsAndRejectDuplicateConnect() {
try (ComfyMcpClient client = new ComfyMcpClient(config())) {
- String version = client.connect();
-
- assertEquals("0.0.0-test", version);
+ assertEquals("0.0.0-test", client.connect());
assertEquals("FakeComfyMcp", client.getServerName());
+ assertTrue(client.isConnected());
+ List tools = client.listTools();
+ assertEquals(3, tools.size());
+ assertThrows(IllegalStateException.class, client::connect);
}
}
@Test
- void shouldListToolsAndCallThem() {
+ void shouldCallTypedWorkflowWrapper() {
try (ComfyMcpClient client = new ComfyMcpClient(config())) {
client.connect();
+ ComfyMcpCallResult result = client.runWorkflow("wf.json", false, 12.5, false);
+ assertFalse(result.isError());
+ assertTrue(result.getText().contains("\"workflow_path\": \"wf.json\""));
+ assertTrue(result.getText().contains("\"wait\": false"));
+ }
+ }
- List tools = client.listTools();
- assertEquals(2, tools.size());
- assertEquals("server_info", tools.get(0).getName());
- assertNotNull(tools.get(1).getInputSchema());
-
- ComfyMcpCallResult info = client.callTool("server_info", null);
- assertFalse(info.isError());
- assertEquals("comfyui up", info.getText());
-
- Map args = new LinkedHashMap();
- args.put("workflow_path", "wf.json");
- ComfyMcpCallResult run = client.callTool("run_workflow", args);
- assertFalse(run.isError());
- assertEquals("queued wf.json", run.getText());
-
- ComfyMcpCallResult unknown = client.callTool("nope", null);
- assertTrue(unknown.isError());
+ @Test
+ void shouldPreserveNonTextContent() {
+ try (ComfyMcpClient client = new ComfyMcpClient(config())) {
+ client.connect();
+ ComfyMcpCallResult result = client.callTool("mixed_content", null);
+ assertEquals("hello", result.getText());
+ assertEquals(3, result.getContents().size());
+ assertEquals("image", result.getContents().get(1).getType());
+ assertEquals("image/png", result.getContents().get(1).getMimeType());
+ assertEquals("aGVsbG8=", result.getContents().get(1).getData());
+ assertEquals("file:///tmp/out.png", result.getContents().get(2).getUri());
+ }
+ }
+
+ @Test
+ void shouldDrainLargeStderrWithoutDeadlock() {
+ ComfyMcpConfig config = config();
+ Map env = new LinkedHashMap();
+ env.put("FAKE_MCP_STDERR_BYTES", "262144");
+ config.setEnvironment(env);
+ try (ComfyMcpClient client = new ComfyMcpClient(config)) {
+ client.connect();
+ assertEquals("comfyui up", client.callTool("server_info", null).getText());
+ }
+ }
+
+ @Test
+ void shouldRemoveTimedOutRpcFromPendingMap() throws Exception {
+ ComfyMcpConfig config = config();
+ config.setReadTimeoutMillis(100);
+ try (ComfyMcpClient client = new ComfyMcpClient(config)) {
+ client.connect();
+ assertThrows(ComfyException.class, () -> client.callTool("slow", null));
+ Thread.sleep(50L);
+ assertEquals(0, client.pendingRequestCount());
}
}
@@ -92,8 +102,8 @@ void shouldRejectCallsBeforeConnectAndAfterClose() {
assertThrows(IllegalStateException.class, () -> client.callToolAsync("server_info", null));
client.close();
assertTrue(client.isClosed());
+ assertFalse(client.isConnected());
assertThrows(IllegalStateException.class, () -> client.callToolAsync("server_info", null));
- assertThrows(ComfyException.class, () -> client.listTools());
client.close();
}
@@ -101,34 +111,32 @@ void shouldRejectCallsBeforeConnectAndAfterClose() {
void shouldFailConnectWhenServerExitsPrematurely() {
ComfyMcpConfig config = config();
config.setLocalExecutable("/bin/echo");
- ComfyMcpClient client = new ComfyMcpClient(config);
- assertThrows(ComfyException.class, client::connect);
- client.close();
+ try (ComfyMcpClient client = new ComfyMcpClient(config)) {
+ assertThrows(ComfyException.class, client::connect);
+ }
}
@Test
- void shouldTimeoutWhenServerNeverAnswers() {
- ComfyMcpConfig config = new ComfyMcpConfig();
- // `sleep` produces no stdout: the initialize request is never answered.
- config.setLocalExecutable("/bin/sleep");
- config.setMcpArgs(new String[] {"30"});
- config.setConnectTimeoutMillis(1_000);
- ComfyMcpClient client = new ComfyMcpClient(config);
- assertThrows(ComfyException.class, client::connect);
- client.close();
+ void shouldPassEnvironmentAndExerciseCurrentToolConveniences() {
+ ComfyMcpConfig config = config();
+ Map env = new LinkedHashMap();
+ env.put("COMFY_BIN", "/opt/venv/bin/comfy");
+ config.setEnvironment(env);
+ try (ComfyMcpClient client = new ComfyMcpClient(config)) {
+ client.connect();
+ assertNotNull(client.serverInfo());
+ assertTrue(client.fetchOutputs("p1", "/tmp/out", true, false).getText().contains("p1"));
+ assertTrue(client.searchModels("wan", "loras").getText().contains("wan"));
+ assertTrue(client.launchComfyUi(Arrays.asList("--port", "8188"), false)
+ .getText().contains("8188"));
+ }
}
@Test
- void shouldPassEnvironmentToServer() {
- ComfyMcpConfig config = config();
- Map args = new LinkedHashMap();
- // COMFY_BIN 场景由 env 注入承载——这里以 env 透传间接验证（fake server 不读 env，
- // 但 spawn 不因额外 env 失败即视为通过；真实验证在 executor 测试覆盖）。
- config.setEnvironment(new LinkedHashMap());
- config.getEnvironment().put("COMFY_BIN", "/opt/venv/bin/comfy");
- try (ComfyMcpClient client = new ComfyMcpClient(config)) {
+ void businessLevelMcpErrorsShouldRemainResults() {
+ try (ComfyMcpClient client = new ComfyMcpClient(config())) {
client.connect();
- assertFalse(client.listTools().isEmpty());
+ assertTrue(client.callTool("nope", null).isError());
}
}
}
diff --git a/src/test/java/io/github/easy4j/comfy/mcp/ComfyMcpConfigTest.java b/src/test/java/io/github/easy4j/comfy/mcp/ComfyMcpConfigTest.java
index c98e2a2..026f711 100644
--- a/src/test/java/io/github/easy4j/comfy/mcp/ComfyMcpConfigTest.java
+++ b/src/test/java/io/github/easy4j/comfy/mcp/ComfyMcpConfigTest.java
@@ -1,17 +1,5 @@
/*
- * Copyright (c) 2018-present, easy-4-java (https://github.com/easy-4-java).
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
+ * Copyright (c) 2018-present, easy-4-java.
*/
package io.github.easy4j.comfy.mcp;
@@ -20,35 +8,28 @@
import org.junit.jupiter.api.Test;
-/** Unit tests for {@link ComfyMcpConfig}. @since 1.0.0 */
class ComfyMcpConfigTest {
@Test
- void shouldExposeSensibleDefaults() {
+ void shouldExposeBoundedDefaults() {
ComfyMcpConfig config = new ComfyMcpConfig();
assertEquals("comfy-mcp", config.getLocalExecutable());
assertEquals("2024-11-05", config.getProtocolVersion());
- assertEquals("comfy-java-sdk", config.getClientName());
assertEquals(10_000, config.getConnectTimeoutMillis());
assertEquals(900_000, config.getReadTimeoutMillis());
- assertEquals(1_048_576, config.getMaxFrameChars());
- assertEquals(1_048_576, config.getMaxContentChars());
+ assertEquals(2_000, config.getShutdownTimeoutMillis());
+ assertEquals(16 * 1024 * 1024, config.getMaxFrameChars());
+ assertEquals(4 * 1024 * 1024, config.getMaxContentChars());
}
@Test
- void shouldAcceptExtraArgsAndEnvironment() {
+ void shouldRejectInvalidTimeouts() {
ComfyMcpConfig config = new ComfyMcpConfig();
- config.setMcpArgs(new String[] {"--port", "8188"});
- config.setEnvironment(java.util.Collections.singletonMap("COMFY_BIN", "/opt/venv/bin/comfy"));
- config.validate();
- assertEquals(2, config.getMcpArgs().length);
- assertEquals("/opt/venv/bin/comfy", config.getEnvironment().get("COMFY_BIN"));
- }
+ config.setConnectTimeoutMillis(0);
+ assertThrows(IllegalStateException.class, config::validate);
- @Test
- void shouldRejectNullExecutable() {
- ComfyMcpConfig config = new ComfyMcpConfig();
- config.setLocalExecutable(null);
- assertThrows(NullPointerException.class, config::validate);
+ config = new ComfyMcpConfig();
+ config.setReadTimeoutMillis(0);
+ assertThrows(IllegalStateException.class, config::validate);
}
}
diff --git a/src/test/resources/comfy-envelope.sh b/src/test/resources/comfy-envelope.sh
new file mode 100755
index 0000000..fbae5ee
--- /dev/null
+++ b/src/test/resources/comfy-envelope.sh
@@ -0,0 +1,2 @@
+#!/bin/sh
+printf '%s\n' '{"ok":true,"command":"env","version":"1.15.0","where":"local","data":{"running":true},"error":null}'
diff --git a/src/test/resources/comfy-slow.sh b/src/test/resources/comfy-slow.sh
new file mode 100755
index 0000000..9677378
--- /dev/null
+++ b/src/test/resources/comfy-slow.sh
@@ -0,0 +1,2 @@
+#!/bin/sh
+exec sleep 5
diff --git a/src/test/resources/fake-mcp-server.py b/src/test/resources/fake-mcp-server.py
index 2cbffad..88da22e 100755
--- a/src/test/resources/fake-mcp-server.py
+++ b/src/test/resources/fake-mcp-server.py
@@ -1,70 +1,68 @@
#!/usr/bin/env python3
-"""Fake comfy-mcp server for end-to-end tests.
-
-Speaks newline-delimited JSON-RPC on stdio exactly like an MCP stdio server:
-answers `initialize` (serverInfo) and `notifications/initialized` silently,
-`tools/list` with a small catalog, `tools/call` for `server_info` (fast) and
-`run_workflow` (returns text content + isError=False). Unknown requests get
-a method-not-found error.
-"""
+"""Fake comfy-mcp server used by Java transport/typed-wrapper tests."""
import json
+import os
import sys
+import time
+flood = int(os.environ.get("FAKE_MCP_STDERR_BYTES", "0") or "0")
+if flood > 0:
+ sys.stderr.write("x" * flood)
+ sys.stderr.flush()
def send(payload):
sys.stdout.write(json.dumps(payload) + "\n")
sys.stdout.flush()
-
def reply(req_id, result):
send({"jsonrpc": "2.0", "id": req_id, "result": result})
+for line in sys.stdin:
+ line = line.strip()
+ if not line:
+ continue
+ try:
+ frame = json.loads(line)
+ except ValueError:
+ continue
+ method = frame.get("method", "")
+ req_id = frame.get("id")
+ params = frame.get("params") or {}
-def main():
- for line in sys.stdin:
- line = line.strip()
- if not line:
- continue
- try:
- frame = json.loads(line)
- except ValueError:
- continue
- method = frame.get("method", "")
- req_id = frame.get("id")
- params = frame.get("params") or {}
-
- if method == "initialize":
- reply(req_id, {
- "protocolVersion": params.get("protocolVersion", "2024-11-05"),
- "capabilities": {"tools": {}},
- "serverInfo": {"name": "FakeComfyMcp", "version": "0.0.0-test"},
- })
- elif method == "tools/list":
- reply(req_id, {"tools": [
- {"name": "server_info", "description": "verify ComfyUI is up",
- "inputSchema": {"type": "object", "properties": {}}},
- {"name": "run_workflow", "description": "run a workflow file",
- "inputSchema": {"type": "object",
- "properties": {"workflow_path": {"type": "string"},
- "wait": {"type": "boolean"}}}},
- ]})
- elif method == "tools/call":
- name = params.get("name", "")
- if name == "server_info":
- reply(req_id, {"content": [{"type": "text", "text": "comfyui up"}], "isError": False})
- elif name == "run_workflow":
- path = (params.get("arguments") or {}).get("workflow_path", "")
- reply(req_id, {"content": [
- {"type": "text", "text": "queued "},
- {"type": "text", "text": path},
- ], "isError": False})
- else:
- reply(req_id, {"content": [{"type": "text", "text": "unknown tool: " + name}],
- "isError": True})
- elif req_id is not None:
- send({"jsonrpc": "2.0", "id": req_id,
- "error": {"code": -32601, "message": "method not found: " + method}})
-
-
-if __name__ == "__main__":
- main()
+ if method == "initialize":
+ reply(req_id, {
+ "protocolVersion": params.get("protocolVersion", "2024-11-05"),
+ "capabilities": {"tools": {}},
+ "serverInfo": {"name": "FakeComfyMcp", "version": "0.0.0-test"},
+ })
+ elif method == "tools/list":
+ reply(req_id, {"tools": [
+ {"name": "server_info", "description": "verify ComfyUI is up",
+ "inputSchema": {"type": "object", "properties": {}}},
+ {"name": "run_workflow", "description": "run a workflow file",
+ "inputSchema": {"type": "object", "properties": {"workflow_path": {"type": "string"}}}},
+ {"name": "mixed_content", "description": "mixed MCP content",
+ "inputSchema": {"type": "object", "properties": {}}},
+ ]})
+ elif method == "tools/call":
+ name = params.get("name", "")
+ args = params.get("arguments") or {}
+ if name == "server_info":
+ reply(req_id, {"content": [{"type": "text", "text": "comfyui up"}], "isError": False})
+ elif name == "slow":
+ time.sleep(2.0)
+ reply(req_id, {"content": [{"type": "text", "text": "late"}], "isError": False})
+ elif name == "mixed_content":
+ reply(req_id, {"content": [
+ {"type": "text", "text": "hello"},
+ {"type": "image", "mimeType": "image/png", "data": "aGVsbG8="},
+ {"type": "resource_link", "uri": "file:///tmp/out.png"},
+ ], "isError": False})
+ elif name == "nope":
+ reply(req_id, {"content": [{"type": "text", "text": "unknown tool"}], "isError": True})
+ else:
+ reply(req_id, {"content": [{"type": "text", "text": json.dumps(args, sort_keys=True)}],
+ "isError": False})
+ elif req_id is not None:
+ send({"jsonrpc": "2.0", "id": req_id,
+ "error": {"code": -32601, "message": "method not found: " + method}})