lelii
/
humainflow-service
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Let an LLM node call tools directly, bypassing the MCP bridge 

The bridge coerces a tool argument from string to object whenever the string
happens to be valid JSON, even though the model emits it correctly and the
tool's own schema declares it as a string. The coercion happens inside the
bridge, between the model's response and the downstream MCP call: verified by
calling the model directly (arguments.content stays a string, byte for byte)
and by comparing a JSON-valid value (rejected) against a malformed or plain
text one (passed through untouched). The failing call never reaches the MCP
server, and the agent's retries under a different encoding are what turn an
unfinished operation into one that returns status: completed with the model's
last preamble as if it were the answer.

There is no fix available on our side for the bridge itself, so this removes
it from the path instead. An LLMBlock can now bind MCP servers straight from
the catalog and run its own tool-calling loop in this service:

- LLMProvider gains chatWithTools/supportsTools; only OllamaProtocolProvider
  implements it for now. Tool arguments stay JsonNode end to end - never a
  string, never re-parsed - which is the one change that actually closes the
  bridge's bug rather than working around it.
- A native streamable-http MCP client (mcp/client/) talks to a server without
  the bridge: initialize, tools/list, tools/call, session header handling,
  both response shapes the spec allows.
- MCPToolServerBinding is a narrower binding than MCPAgent's, restricted to
  catalog servers reachable over streamable-http - the ones this service can
  call directly, not the stdio ones the bridge still hosts a process for.
- LLMToolLoop runs the model/tool/model cycle with real budgets: a wall-clock
  deadline and iteration cap that fail the block explicitly rather than
  return a partial answer, and a character-based context budget that replaces
  older tool results with a placeholder once the conversation - plus the tool
  schemas sent on every call, which do not appear in the conversation but are
  not free either - grows past it. The iteration just completed is never
  pruned, and a result under ~500 characters is left alone: shrinking it would
  cost about as much as it saves.
- Ollama's done_reason now travels back as ToolChatResult.finishReason, so a
  turn that answers nothing can say whether the model chose silence or
  num_predict cut it off mid-thought - two different problems with two
  different fixes, previously indistinguishable from the error alone.
- A new skill, mcp-context-economy, carries the operating rules a real run
  against a 24-task plan exposed the hard way: write_file to create a file,
  apply_patch only to edit one that exists, and never read a file straight
  back after writing it or re-pull an already-inline document into the
  conversation - each halves the context a node needs for the same work.

MCPAgent and the bridge are untouched: this is a second path, not a
replacement, for the one transport (streamable-http) this service can reach
without it.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Lucio Lelii 2026-09-18 21:22:42 +02:00
parent f4dbb866c0
commit 2217ea188a
40 changed files with 2640 additions and 34 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

42
src/main/java/it/cnr/isti/workflow/manager/assistant/BlockDraftNormalizer.java
View File

@ -23,6 +23,7 @@ import it.cnr.isti.workflow.manager.assistant.FlowAssistantService.AssistantConf
import it.cnr.isti.workflow.manager.blocks.Block;
import it.cnr.isti.workflow.manager.blocks.configurations.BlockConfiguration;
import it.cnr.isti.workflow.manager.blocks.factories.BlockFactory;
import it.cnr.isti.workflow.manager.blocks.types.LLMBlockType;
import it.cnr.isti.workflow.manager.mcp.MCPServersProvider;
final class BlockDraftNormalizer {
@ -64,6 +65,7 @@ final class BlockDraftNormalizer {
normalizeHumanDecisionOptions(normalizedConfig, descriptor);
normalizeHttpServerCallAuthorization(normalizedConfig, blockPlan);
normalizeMcpAgentServers(normalizedConfig, blockPlan, mcpServersProvider);
normalizeLlmToolServers(normalizedConfig, blockPlan, mcpServersProvider);
normalizeMcpAgentSharedMemory(normalizedConfig, blockPlan, model, requireSharedMemorySemantics);
removeMcpServerBindingsForSharedSessionConsumer(normalizedConfig, blockPlan);
ensureMcpAgentModelConfigured(normalizedConfig, blockPlan, model);
@ -174,6 +176,46 @@ final class BlockDraftNormalizer {
}
}
/**
* The same protection for an LLM node's tool bindings, which are narrower: catalog only, and only
* the servers this service can call directly. A stdio server here would pass block validation of
* its own shape and then be refused when the flow is saved, so it is dropped while drafting.
*/
private static void normalizeLlmToolServers(ObjectNode config, AssistantBlockPlan blockPlan,
MCPServersProvider mcpServersProvider) {
if (!LLMBlockType.TYPE.equals(blockPlan.blockType())) {
return;
}
JsonNode serversNode = config.get("mcpServers");
if (serversNode == null || !serversNode.isArray()) {
return;
}
ArrayNode kept = ObjectMapperHolder.mapper.createArrayNode();
for (JsonNode server : serversNode) {
String serverName = AssistantTextSupport.textOrEmpty(server.path("serverName"));
if (isStreamableHttpMcpServer(serverName, mcpServersProvider)) {
kept.add(server);
} else {
log.warn("Dropping MCP server '{}' chosen for LLM block {}: unknown or not streamable-http",
serverName, blockPlan.blockId());
}
}
if (kept.isEmpty()) {
config.remove("mcpServers");
} else {
config.set("mcpServers", kept);
}
}
private static boolean isStreamableHttpMcpServer(String serverName, MCPServersProvider mcpServersProvider) {
if (serverName == null || serverName.isBlank()) {
return false;
}
return mcpServersProvider.getServers().stream()
.anyMatch(server -> serverName.equalsIgnoreCase(server.id())
&& "streamable-http".equalsIgnoreCase(server.transport()));
}
private static void normalizeMcpAgentSharedMemory(ObjectNode config, AssistantBlockPlan blockPlan, String model,
boolean requireSharedMemorySemantics) {
if (!requireSharedMemorySemantics || !"MCPAgent".equals(blockPlan.blockType())) {

33
src/main/java/it/cnr/isti/workflow/manager/blocks/configurations/LLMBlockConfiguration.java
View File

@ -15,6 +15,7 @@ import it.cnr.isti.workflow.manager.configurations.annotations.LongText;
import it.cnr.isti.workflow.manager.configurations.annotations.Structural;
import it.cnr.isti.workflow.manager.configurations.annotations.UiUniqueItemsBy;
import it.cnr.isti.workflow.manager.llms.LLMDescriptor;
import it.cnr.isti.workflow.manager.mcp.MCPToolServerBinding;
import it.cnr.isti.workflow.manager.skills.SkillBinding;
import it.cnr.isti.workflow.manager.skills.SkillsCatalogService;
import jakarta.validation.Valid;
@ -51,15 +52,28 @@ public class LLMBlockConfiguration extends BlockConfiguration<LLMBlockType> {
@JsonProperty(required = false)
List<SkillBinding> skills = List.of();
/**
* MCP servers whose tools this node may call. Empty leaves the node exactly as it was: one call
* to the model and its answer. With any bound, the node runs an agent loop instead.
*/
@Structural
@Valid
@Size(max = MCPToolServerBinding.MAX_PER_BLOCK)
@UiUniqueItemsBy("serverName")
@JsonProperty(required = false)
List<MCPToolServerBinding> mcpServers = List.of();
@Builder
public LLMBlockConfiguration(@NonNull String name,
@JsonProperty(value = "llmDescriptor", required = false) LLMDescriptor llmDescriptor,
String prompt,
List<SkillBinding> skills) {
List<SkillBinding> skills,
List<MCPToolServerBinding> mcpServers) {
super(name);
this.llmDescriptor = llmDescriptor;
this.prompt = prompt;
this.skills = skills == null ? List.of() : List.copyOf(skills);
this.mcpServers = mcpServers == null ? List.of() : List.copyOf(mcpServers);
}
@Override
@ -71,6 +85,7 @@ public class LLMBlockConfiguration extends BlockConfiguration<LLMBlockType> {
LLMBlockConfiguration configuration = new LLMBlockConfiguration();
configuration.name = LLMBlockType.TYPE;
configuration.skills = List.of();
configuration.mcpServers = List.of();
return configuration;
}
@ -87,6 +102,7 @@ public class LLMBlockConfiguration extends BlockConfiguration<LLMBlockType> {
}
@AssertTrue(message = "skills must have unique skillId values")
@JsonIgnore
boolean areSkillIdsUnique() {
if (skills == null || skills.isEmpty()) {
return true;
@ -101,4 +117,19 @@ public class LLMBlockConfiguration extends BlockConfiguration<LLMBlockType> {
.count();
}
@AssertTrue(message = "mcpServers must have unique serverName values")
@JsonIgnore
boolean areMcpServerNamesUnique() {
if (mcpServers == null || mcpServers.isEmpty()) {
return true;
}
// Binding the same server twice would open two sessions offering the same tool names, which
// the toolbox then refuses as ambiguous - better to say so while the flow is being edited.
List<String> names = mcpServers.stream()
.map(MCPToolServerBinding::serverName)
.filter(serverName -> serverName != null && !serverName.isBlank())
.toList();
return names.stream().distinct().count() == names.size();
}
}

8
src/main/java/it/cnr/isti/workflow/manager/configurations/retrievers/MCPServersFieldRetriever.java
View File

@ -34,6 +34,14 @@ public class MCPServersFieldRetriever implements DynamicFieldRetriever {
.map(MCPServersProvider.MCPServerDefinition::id)
.sorted()
.toList();
// Its own parameter rather than a filter on "servers": an unknown query parameter would
// silently return the unfiltered list, while an unknown path is a 404 that says so. It
// also leaves "servers" byte for byte what MCPAgent already retrieves.
case "streamableHttpServers" -> mcpServersProvider.getServers().stream()
.filter(server -> "streamable-http".equalsIgnoreCase(server.transport()))
.map(MCPServersProvider.MCPServerDefinition::id)
.sorted()
.toList();
default -> throw new ResponseStatusException(HttpStatus.NOT_FOUND,
"Unknown MCPServers retriever parameter: " + parameter);
};

2
src/main/java/it/cnr/isti/workflow/manager/executions/ExecutionEventType.java
View File

@ -33,6 +33,8 @@ public enum ExecutionEventType {
MCP_SESSION_OPENED,
MCP_SESSION_REUSED,
MCP_SESSION_CLOSED,
MCP_TOOL_CALL,
LLM_CONTEXT_PRUNED,
BIAS_EXPERIMENT_APPLIED,
BIAS_SIDE_EFFECT_MOCKED
}

12
src/main/java/it/cnr/isti/workflow/manager/executions/executors/blocks/LLMExecutor.java
View File

@ -44,6 +44,9 @@ public class LLMExecutor implements BlockExecutor<LLMBlockType> {
@Autowired
private LLMCredentialResolver credentialResolver;
@Autowired
private LLMToolLoop llmToolLoop;
@Override
public Map<String, Object> execute(Block<LLMBlockType> block, List<Input> inputs, Map<String, Object> authorizations,
Map<String, Object> executionVariables, Map<String, ExecutionVariableDescriptor> executionVariableDescriptors,
@ -82,6 +85,15 @@ public class LLMExecutor implements BlockExecutor<LLMBlockType> {
ModelParameterReporting.reportUnsupported(llmProvider, llmDescriptor.parameters(), llmDescriptor.model(),
eventLogger);
// With no MCP server bound this stays one call and one answer, exactly as it always was; the
// loop and its events only exist for a node that was given tools.
if (config.getMcpServers() != null && !config.getMcpServers().isEmpty()) {
String response = llmToolLoop.run(llmProvider, llmDescriptor, credential, prompt,
config.getMcpServers(), executionVariables, eventLogger);
return Map.of(LLMBlockFactory.OUTPUT_NAME, response);
}
String response = llmProvider.generate(llmDescriptor.model(), prompt, credential,
llmDescriptor.parameters());

321
src/main/java/it/cnr/isti/workflow/manager/executions/executors/blocks/LLMToolLoop.java Normal file
View File

@ -0,0 +1,321 @@
// SPDX-FileCopyrightText: 2025-2026 Lucio Lelii <lucio.lelii@isti.cnr.it> - ISTI-CNR
// SPDX-License-Identifier: AGPL-3.0-or-later
// Attribution term under AGPL-3.0 section 7(b): see LICENSE-ADDENDUM.
package it.cnr.isti.workflow.manager.executions.executors.blocks;
import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import it.cnr.isti.workflow.manager.executions.ExecutionEventLogger;
import it.cnr.isti.workflow.manager.executions.ExecutionEventType;
import it.cnr.isti.workflow.manager.llms.ChatMessage;
import it.cnr.isti.workflow.manager.llms.LLMDescriptor;
import it.cnr.isti.workflow.manager.llms.ProviderCredential;
import it.cnr.isti.workflow.manager.llms.ToolCall;
import it.cnr.isti.workflow.manager.llms.ToolChatResult;
import it.cnr.isti.workflow.manager.llms.ToolDefinition;
import it.cnr.isti.workflow.manager.llms.providers.LLMProvider;
import it.cnr.isti.workflow.manager.mcp.MCPToolServerBinding;
import it.cnr.isti.workflow.manager.mcp.client.MCPToolCallResult;
import it.cnr.isti.workflow.manager.mcp.client.MCPToolSessionFactory;
import it.cnr.isti.workflow.manager.mcp.client.MCPToolbox;
/**
* Runs an LLM node that has MCP servers bound: model, tools, model again, until it answers.
*
* <p>The loop lives here rather than in a remote agent service on purpose. Every budget is enforced
* where it can be seen, every tool call becomes an execution event, and running out of budget fails
* the block instead of returning whatever the model last said - which, when the last thing it said
* was "now I will write the file", reads exactly like an answer and is not one.
*/
@Component
public class LLMToolLoop {
private static final Logger logger = LoggerFactory.getLogger(LLMToolLoop.class);
private final MCPToolSessionFactory sessionFactory;
private final int maxIterations;
private final Duration maxDuration;
private final int contextBudgetChars;
public LLMToolLoop(MCPToolSessionFactory sessionFactory,
@Value("${app.llm.tools.max-iterations:10}") int maxIterations,
@Value("${app.llm.tools.max-duration-seconds:900}") long maxDurationSeconds,
@Value("${app.llm.tools.context-budget-chars:60000}") int contextBudgetChars) {
this.sessionFactory = sessionFactory;
this.maxIterations = maxIterations;
this.maxDuration = Duration.ofSeconds(maxDurationSeconds);
this.contextBudgetChars = contextBudgetChars;
}
public String run(LLMProvider provider, LLMDescriptor descriptor, ProviderCredential credential, String prompt,
List<MCPToolServerBinding> bindings, Map<String, Object> executionVariables,
ExecutionEventLogger eventLogger) {
if (!provider.supportsTools()) {
// Checked here and not only when the flow is saved: the provider can arrive from an input
// port, and then no save-time check ever saw it.
throw new IllegalArgumentException("LLM provider " + provider.getName()
+ " cannot call tools, so this node cannot use the MCP servers bound to it");
}
Instant deadline = Instant.now().plus(maxDuration);
try (MCPToolbox toolbox = MCPToolbox.open(bindings, executionVariables, sessionFactory)) {
logEvent(eventLogger, ExecutionEventType.MCP_SESSION_OPENED,
"Opened MCP sessions for " + String.join(", ", toolbox.serverIds()),
Map.of("servers", toolbox.serverIds(), "tools", toolbox.tools().size()));
List<ChatMessage> messages = new ArrayList<>();
// One user message, because skills and bias decoration are already part of the prompt:
// the conversation the model sees starts identical to the single-call path.
messages.add(ChatMessage.user(prompt));
// Indices already replaced by a placeholder, so a second pass never re-shrinks the same
// message (and never mistakes a short placeholder for a result still worth pruning).
Set<Integer> prunedIndices = new HashSet<>();
// The tool definitions are sent on every single call and never change for the life of this
// loop, but they are not free: a real catalog (14 tools on the one this was built against)
// costs several thousand characters every time, and the first version of this budget
// ignored that entirely - it measured only the conversation and missed roughly a fifth of
// what the provider actually received, which was enough on its own to reach the gateway's
// real ceiling one call after the budget believed there was still room.
int toolsOverheadChars = estimateToolsChars(toolbox.tools());
for (int iteration = 1; iteration <= maxIterations; iteration++) {
stopIfCancelled(iteration);
stopIfOutOfTime(deadline, iteration);
ToolChatResult turn = provider.chatWithTools(descriptor.model(), messages, toolbox.tools(),
credential, descriptor.parameters());
logEvent(eventLogger, ExecutionEventType.LLM_REQUEST,
"Called LLM model " + descriptor.model(),
eventDetailsFor(descriptor, iteration, turn));
if (!turn.hasToolCalls()) {
if (turn.content().isBlank()) {
throw new IllegalStateException(describeEmptyTurn(descriptor.model(), turn.finishReason()));
}
return turn.content();
}
// Marks where this iteration's own messages begin, so pruning below never touches the
// tool results the model just produced - only ones from earlier iterations, which by
// now the model has already acted on.
int iterationStart = messages.size();
messages.add(ChatMessage.assistantToolCalls(turn.content(), turn.toolCalls()));
for (ToolCall toolCall : turn.toolCalls()) {
messages.add(runTool(toolbox, toolCall, iteration, eventLogger));
}
pruneOlderToolResultsIfOverBudget(messages, iterationStart, prunedIndices, toolsOverheadChars,
iteration, eventLogger);
}
} finally {
logEvent(eventLogger, ExecutionEventType.MCP_SESSION_CLOSED, "Closed MCP sessions", Map.of());
}
throw new IllegalStateException("LLM node used all " + maxIterations
+ " tool iterations without producing an answer"
+ " (raise app.llm.tools.max-iterations if the task genuinely needs more)");
}
private ChatMessage runTool(MCPToolbox toolbox, ToolCall toolCall, int iteration,
ExecutionEventLogger eventLogger) {
// A turn can ask for several tools, and a cancellation arriving between two of them should
// stop here rather than run the rest of the batch first.
stopIfCancelled(iteration);
long startedAt = System.currentTimeMillis();
MCPToolCallResult result;
try {
result = toolbox.call(toolCall);
} catch (RuntimeException e) {
// An interrupt landing inside the HTTP call surfaces as a reactive wrapper around
// InterruptedException, which says nothing about why the block stopped.
if (Thread.currentThread().isInterrupted()) {
throw new IllegalStateException(
"LLM tool loop was cancelled while calling tool " + toolCall.name(), e);
}
throw e;
}
long elapsed = System.currentTimeMillis() - startedAt;
// Names and sizes only. The arguments can carry whatever the server's configuration carries,
// including its api key, and an execution event is readable by anyone who can see the run.
logEvent(eventLogger, ExecutionEventType.MCP_TOOL_CALL,
"Called MCP tool " + toolCall.name(),
Map.of("tool", toolCall.name(),
"server", String.valueOf(result.serverId()),
"iteration", iteration,
"durationMs", elapsed,
"resultChars", result.text().length(),
"error", result.error()));
logger.debug("MCP tool {} on {} finished in {}ms (error={})", toolCall.name(), result.serverId(), elapsed,
result.error());
return ChatMessage.toolResult(toolCall.id(), toolCall.name(), result.text());
}
/**
* Below this, a placeholder costs about as much as the result it replaces - pruning it would move
* the count without moving the needle, and would only make the reported "reclaimed" figure noise.
*/
private static final int MIN_PRUNABLE_CHARS = 500;
/**
* Replaces the content of older tool results with a placeholder once the conversation grows past
* {@link #contextBudgetChars}, oldest first, stopping as soon as the count is back under budget or
* there is nothing left worth pruning.
*
* <p>Only tool results are touched, never the user prompt (index 0, always protected) and never
* anything from the iteration just completed (below {@code protectFromIndex}): the model has not
* acted on this iteration's own results yet, so those are exactly the ones it still needs in full.
* An older result the model already used to decide its next step is a much safer thing to shrink -
* and the file it came from is still on the node's workspace for the rest of this execution, so a
* model that genuinely needs it again can simply call the tool a second time.
*
* <p>{@code toolsOverheadChars} counts what the tool definitions themselves cost on every call -
* fixed for the loop's whole run, and easy to forget precisely because it never shows up in the
* conversation being trimmed. Missing it here once meant the budget looked satisfied at a total
* that, with the tool schemas added back in, was already at the provider's real ceiling.
*
* <p>Character count, not tokens: this codebase has no tokenizer for whatever model is configured,
* and a conservative character budget is a workable proxy for the one thing this exists to avoid -
* the provider silently truncating or context-shifting mid-generation, which reads as a model that
* did the work and then produced no answer at all.
*
* <p>Pruning can still run out of room to give: everything eligible may already be a placeholder,
* or may all be too small to bother with, while the protected current iteration alone is what pushed
* the total over budget. That is logged as a warning rather than silently proceeding, because the
* next call is then a real risk, not a resolved one.
*/
private void pruneOlderToolResultsIfOverBudget(List<ChatMessage> messages, int protectFromIndex,
Set<Integer> prunedIndices, int toolsOverheadChars, int iteration, ExecutionEventLogger eventLogger) {
if (contextBudgetChars <= 0) {
return;
}
int total = totalContentChars(messages) + toolsOverheadChars;
if (total <= contextBudgetChars) {
return;
}
int prunedCount = 0;
long reclaimedChars = 0;
for (int i = 1; i < protectFromIndex && total > contextBudgetChars; i++) {
if (prunedIndices.contains(i)) {
continue;
}
ChatMessage message = messages.get(i);
if (message.role() != ChatMessage.Role.TOOL || message.content().length() < MIN_PRUNABLE_CHARS) {
continue;
}
int originalChars = message.content().length();
String placeholder = "[tool result pruned to save context: " + originalChars
+ " characters from " + message.toolName() + ". Call the tool again if you need to see it.]";
messages.set(i, ChatMessage.toolResult(message.toolCallId(), message.toolName(), placeholder));
prunedIndices.add(i);
int reclaimed = Math.max(0, originalChars - placeholder.length());
total -= reclaimed;
reclaimedChars += reclaimed;
prunedCount++;
}
if (prunedCount > 0) {
logEvent(eventLogger, ExecutionEventType.LLM_CONTEXT_PRUNED,
"Pruned " + prunedCount + " older tool result(s) to stay under the context budget",
Map.of("iteration", iteration, "prunedCount", prunedCount,
"reclaimedChars", reclaimedChars, "totalCharsAfter", total,
"budgetChars", contextBudgetChars));
}
if (total > contextBudgetChars && eventLogger != null) {
eventLogger.warning(ExecutionEventType.LLM_CONTEXT_PRUNED,
"Still " + (total - contextBudgetChars) + " character(s) over budget after pruning everything"
+ " available; the tool results just produced this iteration are protected and cannot"
+ " be shrunk, so the next call may be rejected by the provider",
Map.of("iteration", iteration, "totalCharsAfter", total, "budgetChars", contextBudgetChars));
}
}
private static int totalContentChars(List<ChatMessage> messages) {
int total = 0;
for (ChatMessage message : messages) {
total += message.content().length();
}
return total;
}
/**
* A rough, provider-agnostic stand-in for how big the {@code tools} field of the request will be.
* Not exact - it does not replicate any one provider's JSON wrapper - but it is the dominant cost
* (name, description, and the schema itself dwarf a few bytes of {"type":"function",...} framing),
* and unlike that framing it scales with the catalog actually bound to this node.
*/
private static int estimateToolsChars(List<ToolDefinition> tools) {
int total = 0;
for (ToolDefinition tool : tools) {
total += tool.name().length() + tool.description().length() + tool.inputSchema().toString().length();
}
return total;
}
private static Map<String, Object> eventDetailsFor(LLMDescriptor descriptor, int iteration, ToolChatResult turn) {
Map<String, Object> details = new LinkedHashMap<>();
details.put("provider", descriptor.provider());
details.put("model", descriptor.model());
details.put("iteration", iteration);
details.put("toolCalls", turn.toolCalls().size());
if (turn.finishReason() != null) {
details.put("finishReason", turn.finishReason());
}
return details;
}
/**
* What actually happened, not just that nothing did. A blank turn with no tool call always looks
* the same to the loop, but "the model chose to say nothing" and "num_predict cut it off mid
* thought" call for different fixes - the second happened while reasoning was still ahead of the
* answer, so raising the prompt's instructions again would not help; the model never got that far.
*/
private static String describeEmptyTurn(String model, String finishReason) {
if ("length".equals(finishReason)) {
return "LLM model " + model + " ran out of output tokens before answering or calling a tool"
+ " (finish reason: length). max_tokens is likely too close to the model's context"
+ " window (num_ctx): leave enough headroom for the prompt and the tool-calling"
+ " history that grows with every iteration, not only for the answer itself.";
}
String reasonSuffix = finishReason == null || finishReason.isBlank() ? "" : " (finish reason: " + finishReason + ")";
return "LLM model " + model + " ended the conversation without an answer and without calling a tool"
+ reasonSuffix;
}
private static void stopIfCancelled(int iteration) {
// isInterrupted rather than interrupted(): the latter clears the flag, and the executor
// shutting the pool down is the only cancellation signal a block executor ever gets.
if (Thread.currentThread().isInterrupted()) {
throw new IllegalStateException("LLM tool loop was cancelled before iteration " + iteration);
}
}
private static void stopIfOutOfTime(Instant deadline, int iteration) {
if (Instant.now().isAfter(deadline)) {
throw new IllegalStateException("LLM node ran out of time before iteration " + iteration
+ " (raise app.llm.tools.max-duration-seconds if the task genuinely needs longer)");
}
}
private static void logEvent(ExecutionEventLogger eventLogger, ExecutionEventType type, String message,
Map<String, Object> details) {
if (eventLogger != null) {
eventLogger.info(type, message, details);
}
}
}

55
src/main/java/it/cnr/isti/workflow/manager/flows/validation/FlowDataValidator.java
View File

@ -51,6 +51,8 @@ import it.cnr.isti.workflow.manager.executions.bias.runtime.BiasBehaviorAdapterR
import it.cnr.isti.workflow.manager.ios.IODescriptor;
import it.cnr.isti.workflow.manager.llms.LLMDescriptor;
import it.cnr.isti.workflow.manager.llms.providers.LLMProvider;
import it.cnr.isti.workflow.manager.mcp.MCPServersProvider;
import it.cnr.isti.workflow.manager.mcp.MCPToolServerBinding;
import jakarta.validation.ConstraintValidator;
import jakarta.validation.ConstraintValidatorContext;
@ -72,6 +74,9 @@ public class FlowDataValidator implements ConstraintValidator<ValidFlowStructure
@Autowired
Map<String, LLMProvider> llmProviders;
@Autowired
MCPServersProvider mcpServersProvider;
@Override
public boolean isValid(FlowData flowData, ConstraintValidatorContext context) {
if (flowData == null) {
@ -198,6 +203,7 @@ public class FlowDataValidator implements ConstraintValidator<ValidFlowStructure
throw validationError(error(ValidationErrorCode.BLOCK_OUTPUTS_MISMATCH, "block", block.getId(), "outputs", "Outputs do not match its configuration"));
}
validateLlmDescriptorProviders(block);
validateLlmToolBindings(block);
validateBiasAnnotations(block);
}
@ -224,6 +230,55 @@ public class FlowDataValidator implements ConstraintValidator<ValidFlowStructure
}
}
/**
* An LLM node that binds MCP servers can only run if its provider can call tools and every bound
* server is one this service can reach directly. Both are knowable while editing, and neither
* fails in a way that reads as a configuration problem at execution time: a provider without tool
* support refuses mid-run, and a stdio server fails when the session is opened.
*/
private void validateLlmToolBindings(Block<?> block) {
if (!(block.getSpecificConfiguration() instanceof LLMBlockConfiguration configuration)) {
return;
}
List<MCPToolServerBinding> bindings = configuration.getMcpServers();
if (bindings == null || bindings.isEmpty()) {
return;
}
LLMDescriptor descriptor = configuration.getLlmDescriptor();
String providerName = descriptor == null ? null : descriptor.provider();
// Same escape as validateLlmDescriptorProviders: a templated provider is only known at
// execution time, and the loop checks supportsTools() again before it opens anything.
if (providerName != null && !providerName.isBlank() && !providerName.contains("${{")) {
LLMProvider provider = resolveProvider(providerName);
if (provider != null && !provider.supportsTools()) {
throw validationError(error(ValidationErrorCode.LLM_PROVIDER_DOES_NOT_SUPPORT_TOOLS, "block",
block.getId(), "llmDescriptor.provider",
"LLM provider " + providerName + " cannot call tools, so this node cannot bind MCP servers"));
}
}
for (MCPToolServerBinding binding : bindings) {
String serverName = binding.serverName();
if (serverName == null || serverName.isBlank()) {
continue;
}
MCPServersProvider.MCPServerDefinition definition;
try {
definition = mcpServersProvider.getServer(serverName);
} catch (IllegalArgumentException e) {
throw validationError(error(ValidationErrorCode.MCP_SERVER_NOT_FOUND, "block", block.getId(),
"mcpServers", "No MCP server is registered under the name " + serverName));
}
if (!"streamable-http".equalsIgnoreCase(definition.transport())) {
throw validationError(error(ValidationErrorCode.MCP_SERVER_TRANSPORT_NOT_SUPPORTED, "block",
block.getId(), "mcpServers", "MCP server " + serverName + " uses transport "
+ definition.transport() + ", which an LLM node cannot call directly;"
+ " use an MCPAgent node instead"));
}
}
}
private List<LLMDescriptor> llmDescriptorsOf(Block<?> block) {
Object configuration = block.getSpecificConfiguration();
if (configuration instanceof LLMBlockConfiguration llmConfiguration) {

5
src/main/java/it/cnr/isti/workflow/manager/flows/validation/ValidationErrorCode.java
View File

@ -100,5 +100,8 @@ public enum ValidationErrorCode {
PROJECT_CONTEXT_NAME_INVALID,
PROJECT_CONTEXT_NAME_RESERVED,
PROJECT_CONTEXT_NAME_DUPLICATE,
LLM_PROVIDER_NOT_FOUND
LLM_PROVIDER_NOT_FOUND,
LLM_PROVIDER_DOES_NOT_SUPPORT_TOOLS,
MCP_SERVER_NOT_FOUND,
MCP_SERVER_TRANSPORT_NOT_SUPPORTED
}

51
src/main/java/it/cnr/isti/workflow/manager/llms/ChatMessage.java
View File

@ -4,18 +4,65 @@
package it.cnr.isti.workflow.manager.llms;
import java.util.List;
import java.util.Objects;
public record ChatMessage(Role role, String content) {
/**
* One message in a conversation sent to a provider.
*
* <p>The last three components carry a tool-calling turn: {@code toolCalls} on an
* {@link Role#ASSISTANT} message is what the model asked to run, and {@code toolCallId} /
* {@code toolName} on a {@link Role#TOOL} message say which of those calls the message is the
* result of. They are empty for every plain text message, which is why the two-argument
* constructor below is kept: the many call sites that only ever build text messages read exactly
* as they did before tools existed.
*/
public record ChatMessage(Role role, String content, List<ToolCall> toolCalls, String toolCallId,
String toolName) {
public ChatMessage {
role = role == null ? Role.USER : role;
content = Objects.requireNonNullElse(content, "");
toolCalls = toolCalls == null ? List.of() : List.copyOf(toolCalls);
}
public ChatMessage(Role role, String content) {
this(role, content, List.of(), null, null);
}
public enum Role {
SYSTEM,
USER,
ASSISTANT
ASSISTANT,
TOOL
}
public static ChatMessage system(String content) {
return new ChatMessage(Role.SYSTEM, content);
}
public static ChatMessage user(String content) {
return new ChatMessage(Role.USER, content);
}
public static ChatMessage assistant(String content) {
return new ChatMessage(Role.ASSISTANT, content);
}
/**
* The model's turn when it asked for tools. {@code content} is whatever it said alongside the
* calls, often a preamble and often empty; it is kept because dropping it would leave a gap in
* the conversation the model sees on the next turn.
*/
public static ChatMessage assistantToolCalls(String content, List<ToolCall> toolCalls) {
return new ChatMessage(Role.ASSISTANT, content, toolCalls, null, null);
}
public static ChatMessage toolResult(String toolCallId, String toolName, String content) {
return new ChatMessage(Role.TOOL, content, List.of(), toolCallId, toolName);
}
public boolean hasToolCalls() {
return !toolCalls.isEmpty();
}
}

2
src/main/java/it/cnr/isti/workflow/manager/llms/LLMProviderCatalogService.java
View File

@ -24,7 +24,7 @@ public class LLMProviderCatalogService {
public List<LLMProviderMetadata> list() {
return providers.values().stream()
.map(provider -> new LLMProviderMetadata(provider.getName(), provider.requiresAuthorization(),
provider.requiresEndpoint()))
provider.requiresEndpoint(), provider.supportsTools()))
.filter(provider -> StringUtils.hasText(provider.name()))
.distinct()
.sorted(java.util.Comparator.comparing(LLMProviderMetadata::name, String.CASE_INSENSITIVE_ORDER))

6
src/main/java/it/cnr/isti/workflow/manager/llms/LLMProviderMetadata.java
View File

@ -9,6 +9,10 @@ package it.cnr.isti.workflow.manager.llms;
*
* @param requiresEndpoint whether choosing this provider means the credential must also carry a
* base URL - see {@link it.cnr.isti.workflow.manager.llms.providers.LLMProvider#requiresEndpoint()}.
* @param supportsTools whether this provider can be given tools to call, which is what decides
* if a node may bind an MCP server to it - see
* {@link it.cnr.isti.workflow.manager.llms.providers.LLMProvider#supportsTools()}.
*/
public record LLMProviderMetadata(String name, boolean requiresCredential, boolean requiresEndpoint) {
public record LLMProviderMetadata(String name, boolean requiresCredential, boolean requiresEndpoint,
boolean supportsTools) {
}

36
src/main/java/it/cnr/isti/workflow/manager/llms/ToolCall.java Normal file
View File

@ -0,0 +1,36 @@
// SPDX-FileCopyrightText: 2025-2026 Lucio Lelii <lucio.lelii@isti.cnr.it> - ISTI-CNR
// SPDX-License-Identifier: AGPL-3.0-or-later
// Attribution term under AGPL-3.0 section 7(b): see LICENSE-ADDENDUM.
package it.cnr.isti.workflow.manager.llms;
import java.util.Objects;
import tools.jackson.databind.JsonNode;
import tools.jackson.databind.node.JsonNodeFactory;
/**
* One tool invocation a model asked for.
*
* <p>{@code arguments} stays a {@link JsonNode} from the provider's response all the way to the MCP
* server, never a string and never re-parsed. See {@link ToolDefinition} for why that matters.
*
* <p>{@code id} is the provider's correlation id when it sends one. Ollama does not, so
* {@link #of(int, String, JsonNode)} synthesizes one from the call's position in the batch: it only
* ever has to pair a result with its call inside a single turn of our own loop, so any value unique
* within that turn will do.
*/
public record ToolCall(String id, String name, JsonNode arguments) {
public ToolCall {
Objects.requireNonNull(id, "id cannot be null");
Objects.requireNonNull(name, "name cannot be null");
arguments = arguments == null || arguments.isNull()
? JsonNodeFactory.instance.objectNode()
: arguments;
}
public static ToolCall of(int index, String name, JsonNode arguments) {
return new ToolCall("call_" + index, name, arguments);
}
}

52
src/main/java/it/cnr/isti/workflow/manager/llms/ToolChatResult.java Normal file
View File

@ -0,0 +1,52 @@
// SPDX-FileCopyrightText: 2025-2026 Lucio Lelii <lucio.lelii@isti.cnr.it> - ISTI-CNR
// SPDX-License-Identifier: AGPL-3.0-or-later
// Attribution term under AGPL-3.0 section 7(b): see LICENSE-ADDENDUM.
package it.cnr.isti.workflow.manager.llms;
import java.util.List;
import java.util.Objects;
/**
* One turn of a model that may call tools: either an answer, or a batch of invocations to run
* before asking again.
*
* <p>The two are not exclusive - a model may narrate what it is about to do and call a tool in the
* same turn - so {@code content} can be non-blank alongside tool calls. Only {@link #hasToolCalls()}
* decides whether the agent loop keeps going, never the presence of text: reading a preamble as if
* it were the final answer is exactly how the MCP bridge turns an unfinished run into a plausible
* looking result.
*
* @param finishReason why the provider stopped generating, in its own words when it says one at all
* (Ollama's {@code done_reason}: {@code "stop"}, {@code "length"}, ...). Null for
* a provider that reports nothing, or for the no-tools default in
* {@link it.cnr.isti.workflow.manager.llms.providers.LLMProvider#chatWithTools}.
* Exists only so a turn with neither an answer nor a tool call can say why
* instead of just that: {@code "length"} means the model ran out of output
* tokens - typically while still reasoning, before writing anything to
* {@code content} - which reads identically to a model that simply went silent
* unless something carries the reason along.
*/
public record ToolChatResult(String content, List<ToolCall> toolCalls, String finishReason) {
public ToolChatResult {
content = Objects.requireNonNullElse(content, "");
toolCalls = toolCalls == null ? List.of() : List.copyOf(toolCalls);
}
public static ToolChatResult text(String content) {
return new ToolChatResult(content, List.of(), null);
}
public static ToolChatResult toolCalls(String content, List<ToolCall> toolCalls) {
return new ToolChatResult(content, toolCalls, null);
}
public static ToolChatResult toolCalls(String content, List<ToolCall> toolCalls, String finishReason) {
return new ToolChatResult(content, toolCalls, finishReason);
}
public boolean hasToolCalls() {
return !toolCalls.isEmpty();
}
}

38
src/main/java/it/cnr/isti/workflow/manager/llms/ToolDefinition.java Normal file
View File

@ -0,0 +1,38 @@
// SPDX-FileCopyrightText: 2025-2026 Lucio Lelii <lucio.lelii@isti.cnr.it> - ISTI-CNR
// SPDX-License-Identifier: AGPL-3.0-or-later
// Attribution term under AGPL-3.0 section 7(b): see LICENSE-ADDENDUM.
package it.cnr.isti.workflow.manager.llms;
import java.util.Objects;
import tools.jackson.databind.JsonNode;
import tools.jackson.databind.node.JsonNodeFactory;
import tools.jackson.databind.node.ObjectNode;
/**
* One tool a model may call, as the MCP server declares it.
*
* <p>{@code inputSchema} is the server's own JSON Schema, carried through untouched: it is handed
* to the provider as-is and the arguments that come back are handed to the server as-is. Nothing
* in this codebase rewrites, normalizes or re-types it. That is deliberate - the MCP bridge this
* type exists to replace parses a string argument into an object whenever the string happens to be
* valid JSON, which makes the server reject its own schema's {@code "type": "string"} fields. The
* cheapest way not to have that bug is to have no conversion step at all.
*/
public record ToolDefinition(String name, String description, JsonNode inputSchema) {
public ToolDefinition {
Objects.requireNonNull(name, "name cannot be null");
description = Objects.requireNonNullElse(description, "");
// A tool that takes no arguments may declare no schema; providers still want an object.
inputSchema = inputSchema == null || inputSchema.isNull() ? emptyObjectSchema() : inputSchema;
}
private static JsonNode emptyObjectSchema() {
ObjectNode schema = JsonNodeFactory.instance.objectNode();
schema.put("type", "object");
schema.set("properties", JsonNodeFactory.instance.objectNode());
return schema;
}
}

31
src/main/java/it/cnr/isti/workflow/manager/llms/providers/LLMProvider.java
View File

@ -13,6 +13,8 @@ import it.cnr.isti.workflow.manager.llms.ChatMessage;
import it.cnr.isti.workflow.manager.llms.ModelParameter;
import it.cnr.isti.workflow.manager.llms.ModelParameters;
import it.cnr.isti.workflow.manager.llms.ProviderCredential;
import it.cnr.isti.workflow.manager.llms.ToolChatResult;
import it.cnr.isti.workflow.manager.llms.ToolDefinition;
public interface LLMProvider {
@ -87,6 +89,35 @@ public interface LLMProvider {
return chat(model, messages, credential == null ? null : credential.value(), parameters);
}
/**
* Whether this provider can be given tools to call. False unless a provider says otherwise, so
* a provider that knows nothing about tools keeps behaving exactly as it does today.
*/
default boolean supportsTools() {
return false;
}
/**
* One turn of a conversation the model may answer with tool calls instead of text.
*
* <p>Only declared at this level, not in the four overload shapes the text calls have: this is
* the level the executors actually call, and a tool-calling turn has no meaning for the
* prompt-only entry points.
*
* <p>With no tools this is {@link #chat} with its answer wrapped, so a caller can drive both
* cases through one method and a provider that does not override it stays correct. With tools it
* fails loudly rather than quietly dropping them: silently discarding part of a request and
* returning a plausible answer anyway is the exact failure this whole path exists to get away
* from, and a caller that reaches here with tools has already ignored {@link #supportsTools()}.
*/
default ToolChatResult chatWithTools(String model, List<ChatMessage> messages,
List<ToolDefinition> tools, ProviderCredential credential, ModelParameters parameters) {
if (tools != null && !tools.isEmpty()) {
throw new UnsupportedOperationException(getName() + " does not support tool calling");
}
return ToolChatResult.text(chat(model, messages, credential, parameters));
}
/**
* Which sampling parameters this provider actually applies. Everything, by default: a provider
* that maps none of them still behaves as it always has, and only one that knowingly leaves a

5
src/main/java/it/cnr/isti/workflow/manager/llms/providers/google/GeminiLLMProvider.java
View File

@ -204,6 +204,11 @@ public class GeminiLLMProvider extends AbstractHttpLLMProvider {
return switch (role) {
case SYSTEM, USER -> "user";
case ASSISTANT -> "model";
// supportsTools() is false here, so a TOOL message is our bug rather than bad input.
// Gemini also spells a tool result as a functionResponse part rather than a role, so
// there is nothing sensible to map it to in this builder.
case TOOL -> throw new IllegalArgumentException(
"Tool messages are not supported by " + getName());
};
}

136
src/main/java/it/cnr/isti/workflow/manager/llms/providers/ollama/OllamaProtocolProvider.java
View File

@ -4,6 +4,7 @@
package it.cnr.isti.workflow.manager.llms.providers.ollama;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
@ -18,8 +19,13 @@ import tools.jackson.databind.ObjectMapper;
import it.cnr.isti.workflow.manager.llms.ChatMessage;
import it.cnr.isti.workflow.manager.llms.ModelParameters;
import it.cnr.isti.workflow.manager.llms.ProviderCredential;
import it.cnr.isti.workflow.manager.llms.ToolCall;
import it.cnr.isti.workflow.manager.llms.ToolChatResult;
import it.cnr.isti.workflow.manager.llms.ToolDefinition;
import it.cnr.isti.workflow.manager.llms.providers.AbstractHttpLLMProvider;
import it.cnr.isti.workflow.manager.llms.providers.ollama.response.ChatResponse;
import it.cnr.isti.workflow.manager.llms.providers.ollama.response.ChatResponseMessage;
import it.cnr.isti.workflow.manager.llms.providers.ollama.response.ToolCallResponse;
import it.cnr.isti.workflow.manager.llms.providers.ollama.response.GenerateResponse;
import it.cnr.isti.workflow.manager.llms.providers.ollama.response.ModelInfo;
import it.cnr.isti.workflow.manager.llms.providers.ollama.response.ModelResponse;
@ -133,18 +139,34 @@ public abstract class OllamaProtocolProvider extends AbstractHttpLLMProvider {
return parseGenerateResponse(blockingCall(request, "generate", model));
}
@Override
public boolean supportsTools() {
return true;
}
@Override
public ToolChatResult chatWithTools(String model, List<ChatMessage> messages, List<ToolDefinition> tools,
ProviderCredential credential, ModelParameters parameters) {
return parseChatResult(chatCallRaw(model, messages, tools, credential, parameters));
}
private String chatCall(String model, List<ChatMessage> messages, ProviderCredential credential,
ModelParameters parameters) {
return parseChatResponse(chatCallRaw(model, messages, List.of(), credential, parameters));
}
private String chatCallRaw(String model, List<ChatMessage> messages, List<ToolDefinition> tools,
ProviderCredential credential, ModelParameters parameters) {
Objects.requireNonNull(model, "model cannot be null");
Objects.requireNonNull(messages, "messages cannot be null");
Map<String, Object> body = buildChatBody(model, messages, parameters);
Map<String, Object> body = buildChatBody(model, messages, parameters, tools);
Mono<String> request = withErrorHandling(clientFor(baseUrl(credential)).post()
.uri("/chat")
.header("Authorization", "Bearer " + resolveApiKey(credential))
.contentType(MediaType.APPLICATION_JSON)
.bodyValue(body)
.retrieve());
return parseChatResponse(blockingCall(request, "chat", model));
return blockingCall(request, "chat", model);
}
/**
@ -218,17 +240,32 @@ public abstract class OllamaProtocolProvider extends AbstractHttpLLMProvider {
/** Mutable, unlike the Map.of it replaces: with nothing set the body is the one it always sent. */
Map<String, Object> buildChatBody(String model, List<ChatMessage> messages, ModelParameters parameters) {
return buildChatBody(model, messages, parameters, List.of());
}
/**
* The same body, plus the tools the model may call.
*
* <p>With an empty tool list this is byte for byte the body the three-argument overload has
* always sent, which is what keeps the text path - every existing caller - unchanged.
*/
Map<String, Object> buildChatBody(String model, List<ChatMessage> messages, ModelParameters parameters,
List<ToolDefinition> tools) {
Map<String, Object> bodyMap = new LinkedHashMap<>();
bodyMap.put("model", model);
bodyMap.put("messages", messages.stream()
.map(message -> Map.of(
"role", message.role().name().toLowerCase(),
"content", message.content()))
.toList());
bodyMap.put("messages", messages.stream().map(OllamaProtocolProvider::toOllamaMessage).toList());
bodyMap.put("stream", false);
// See buildGenerateBody: keeps reasoning out of message.content without asking the model to
// reason less.
bodyMap.put("think", true);
if (tools != null && !tools.isEmpty()) {
bodyMap.put("tools", tools.stream()
.map(tool -> Map.of("type", "function", "function", Map.of(
"name", tool.name(),
"description", tool.description(),
"parameters", tool.inputSchema())))
.toList());
}
Map<String, Object> options = optionsFor(parameters);
if (!options.isEmpty()) {
bodyMap.put("options", options);
@ -236,6 +273,32 @@ public abstract class OllamaProtocolProvider extends AbstractHttpLLMProvider {
return bodyMap;
}
private static Map<String, Object> toOllamaMessage(ChatMessage message) {
Map<String, Object> mapped = new LinkedHashMap<>();
mapped.put("role", message.role().name().toLowerCase());
mapped.put("content", message.content());
if (message.hasToolCalls()) {
mapped.put("tool_calls", message.toolCalls().stream()
.map(call -> Map.of("function", Map.of(
"name", call.name(),
"arguments", call.arguments())))
.toList());
}
if (message.role() == ChatMessage.Role.TOOL) {
// Both keys, because which one Ollama reads depends on its version and the extra one is
// ignored: older builds match the result to the call by tool_name, newer ones by
// tool_call_id. Getting this wrong does not fail loudly - the model simply sees a result
// it cannot attribute - so it is cheaper to send both than to pin a version.
if (message.toolName() != null) {
mapped.put("tool_name", message.toolName());
}
if (message.toolCallId() != null) {
mapped.put("tool_call_id", message.toolCallId());
}
}
return mapped;
}
private String parseGenerateResponse(String responseBody) {
if (responseBody == null || responseBody.isBlank()) {
throw new RuntimeException("Empty response body from Ollama generate endpoint");
@ -257,23 +320,58 @@ public abstract class OllamaProtocolProvider extends AbstractHttpLLMProvider {
}
private String parseChatResponse(String responseBody) {
ToolChatResult result = parseChatResult(responseBody);
if (result.hasToolCalls() && result.content().isBlank()) {
// Only reachable if the model called tools nobody offered it, so there is no result to
// return and no loop waiting to run them.
throw new RuntimeException("Ollama chat response has tool calls but no text");
}
return result.content();
}
/**
* A chat turn that may be an answer, tool calls, or both.
*
* <p>A turn carrying only tool calls has no {@code message.content} at all, which is why this
* cannot simply reuse the text path: that one treats a missing content as a broken response, and
* for a model that just asked to call something it is the normal shape.
*/
private ToolChatResult parseChatResult(String responseBody) {
if (responseBody == null || responseBody.isBlank()) {
throw new RuntimeException("Empty response body from Ollama chat endpoint");
}
String trimmedResponse = responseBody.trim();
if (trimmedResponse.startsWith("{")) {
try {
ChatResponse response = objectMapper.readValue(trimmedResponse, ChatResponse.class);
if (response.getMessage() == null || response.getMessage().getContent() == null) {
throw new RuntimeException("Missing 'message.content' field in Ollama chat response");
}
return response.getMessage().getContent();
} catch (Exception e) {
throw new RuntimeException("Unable to parse Ollama chat response", e);
}
if (!trimmedResponse.startsWith("{")) {
log.debug("Ollama chat endpoint returned text/plain response");
return ToolChatResult.text(trimmedResponse);
}
log.debug("Ollama chat endpoint returned text/plain response");
return trimmedResponse;
try {
ChatResponse response = objectMapper.readValue(trimmedResponse, ChatResponse.class);
ChatResponseMessage message = response.getMessage();
List<ToolCall> toolCalls = toToolCalls(message);
if (message == null || (message.getContent() == null && toolCalls.isEmpty())) {
throw new RuntimeException("Missing 'message.content' field in Ollama chat response");
}
return ToolChatResult.toolCalls(message.getContent(), toolCalls, response.getDoneReason());
} catch (Exception e) {
throw new RuntimeException("Unable to parse Ollama chat response", e);
}
}
private static List<ToolCall> toToolCalls(ChatResponseMessage message) {
if (message == null || message.getToolCalls() == null) {
return List.of();
}
List<ToolCall> calls = new ArrayList<>();
int index = 0;
for (ToolCallResponse raw : message.getToolCalls()) {
if (raw == null || raw.getFunction() == null || raw.getFunction().getName() == null) {
continue;
}
// Arguments travel on as the node Ollama sent, never re-read or re-typed.
calls.add(ToolCall.of(index++, raw.getFunction().getName(), raw.getFunction().getArguments()));
}
return List.copyOf(calls);
}
}

6
src/main/java/it/cnr/isti/workflow/manager/llms/providers/ollama/response/ChatResponse.java
View File

@ -5,6 +5,7 @@
package it.cnr.isti.workflow.manager.llms.providers.ollama.response;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.annotation.JsonProperty;
import lombok.Data;
import lombok.NoArgsConstructor;
@ -14,4 +15,9 @@ import lombok.NoArgsConstructor;
@NoArgsConstructor
public class ChatResponse {
private ChatResponseMessage message;
// "stop" for a natural end, "length" when num_predict cut generation short - often while the
// model was still reasoning, before it wrote anything to message.content. Sits on the envelope,
// not the message, because it describes how the turn ended, not what the assistant said.
@JsonProperty("done_reason")
private String doneReason;
}

7
src/main/java/it/cnr/isti/workflow/manager/llms/providers/ollama/response/ChatResponseMessage.java
View File

@ -4,7 +4,10 @@
package it.cnr.isti.workflow.manager.llms.providers.ollama.response;
import java.util.List;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.annotation.JsonProperty;
import lombok.Data;
import lombok.NoArgsConstructor;
@ -15,4 +18,8 @@ import lombok.NoArgsConstructor;
public class ChatResponseMessage {
private String role;
private String content;
// Spelled out because this mapper has no naming strategy: without it the field silently stays
// null and a turn that only called tools reads as a turn that answered nothing.
@JsonProperty("tool_calls")
private List<ToolCallResponse> toolCalls;
}

27
src/main/java/it/cnr/isti/workflow/manager/llms/providers/ollama/response/ToolCallFunction.java Normal file
View File

@ -0,0 +1,27 @@
// SPDX-FileCopyrightText: 2025-2026 Lucio Lelii <lucio.lelii@isti.cnr.it> - ISTI-CNR
// SPDX-License-Identifier: AGPL-3.0-or-later
// Attribution term under AGPL-3.0 section 7(b): see LICENSE-ADDENDUM.
package it.cnr.isti.workflow.manager.llms.providers.ollama.response;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import lombok.Data;
import lombok.NoArgsConstructor;
import tools.jackson.databind.JsonNode;
/**
* The tool and arguments of one call Ollama reports.
*
* <p>{@code arguments} is a {@link JsonNode} because Ollama sends it as a JSON object already
* parsed, unlike the OpenAI protocol which sends it as a string that has to be parsed once. Binding
* it to a node and passing that node on means this codebase never decides what the values inside
* are: a string stays a string even when it happens to hold valid JSON.
*/
@JsonIgnoreProperties(ignoreUnknown = true)
@Data
@NoArgsConstructor
public class ToolCallFunction {
private String name;
private JsonNode arguments;
}

18
src/main/java/it/cnr/isti/workflow/manager/llms/providers/ollama/response/ToolCallResponse.java Normal file
View File

@ -0,0 +1,18 @@
// SPDX-FileCopyrightText: 2025-2026 Lucio Lelii <lucio.lelii@isti.cnr.it> - ISTI-CNR
// SPDX-License-Identifier: AGPL-3.0-or-later
// Attribution term under AGPL-3.0 section 7(b): see LICENSE-ADDENDUM.
package it.cnr.isti.workflow.manager.llms.providers.ollama.response;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import lombok.Data;
import lombok.NoArgsConstructor;
/** One entry of {@code message.tool_calls}; Ollama nests the call under a {@code function} key. */
@JsonIgnoreProperties(ignoreUnknown = true)
@Data
@NoArgsConstructor
public class ToolCallResponse {
private ToolCallFunction function;
}

7
src/main/java/it/cnr/isti/workflow/manager/llms/providers/openai/OpenAIProtocolProvider.java
View File

@ -187,6 +187,13 @@ public abstract class OpenAIProtocolProvider extends AbstractHttpLLMProvider {
case SYSTEM -> "system";
case USER -> "user";
case ASSISTANT -> "assistant";
// This provider reports supportsTools() false, so it is never handed a conversation
// containing tool turns: a TOOL message arriving here is our bug, not bad input. It
// would also be unusable on its own - OpenAI rejects a tool message that is not
// preceded by the assistant message whose tool_calls it answers, and this builder
// does not carry those - so failing is the only honest option.
case TOOL -> throw new IllegalArgumentException(
"Tool messages are not supported by " + getName());
};
}

11
src/main/java/it/cnr/isti/workflow/manager/mcp/MCPAgentService.java
View File

@ -748,19 +748,12 @@ public class MCPAgentService {
private Map<String, String> resolveRuntimeTemplates(Map<String, String> values, Map<String, Object> templateValues,
Map<String, Object> executionVariables) {
LinkedHashMap<String, String> resolved = new LinkedHashMap<>();
values.forEach((key, value) -> resolved.put(key, resolveRuntimeTemplate(value, templateValues, executionVariables)));
return resolved;
return MCPRuntimeTemplates.resolveAll(values, templateValues, executionVariables);
}
private String resolveRuntimeTemplate(String template, Map<String, Object> templateValues,
Map<String, Object> executionVariables) {
if (!StringUtils.hasText(template)) {
return template;
}
return it.cnr.isti.workflow.manager.executions.ExecutionTemplateResolver.resolve(template,
templateValues == null ? Map.of() : templateValues,
executionVariables == null ? Map.of() : executionVariables);
return MCPRuntimeTemplates.resolve(template, templateValues, executionVariables);
}
private Map<String, Object> resolveTemplatedObject(Map<String, Object> values, Map<String, Object> templateValues,

44
src/main/java/it/cnr/isti/workflow/manager/mcp/MCPRuntimeTemplates.java Normal file
View File

@ -0,0 +1,44 @@
// SPDX-FileCopyrightText: 2025-2026 Lucio Lelii <lucio.lelii@isti.cnr.it> - ISTI-CNR
// SPDX-License-Identifier: AGPL-3.0-or-later
// Attribution term under AGPL-3.0 section 7(b): see LICENSE-ADDENDUM.
package it.cnr.isti.workflow.manager.mcp;
import java.util.LinkedHashMap;
import java.util.Map;
import org.springframework.util.StringUtils;
import it.cnr.isti.workflow.manager.executions.ExecutionTemplateResolver;
/**
* Resolves the {@code ${{...}}} placeholders a catalog MCP server declares in its url and headers -
* {@code ${{host}}} and {@code ${{key}}} from the block's own configuration, {@code ${{context.*}}}
* from the execution.
*
* <p>Shared so the two ways of reaching an MCP server - through the bridge, and the native client -
* resolve a given server definition to exactly the same url and headers. A second copy of this would
* drift, and the drift would only show up as one path reaching a different workspace than the other.
*/
public final class MCPRuntimeTemplates {
private MCPRuntimeTemplates() {
}
public static Map<String, String> resolveAll(Map<String, String> values, Map<String, Object> templateValues,
Map<String, Object> executionVariables) {
LinkedHashMap<String, String> resolved = new LinkedHashMap<>();
values.forEach((key, value) -> resolved.put(key, resolve(value, templateValues, executionVariables)));
return resolved;
}
public static String resolve(String template, Map<String, Object> templateValues,
Map<String, Object> executionVariables) {
if (!StringUtils.hasText(template)) {
return template;
}
return ExecutionTemplateResolver.resolve(template,
templateValues == null ? Map.of() : templateValues,
executionVariables == null ? Map.of() : executionVariables);
}
}

49
src/main/java/it/cnr/isti/workflow/manager/mcp/MCPToolServerBinding.java Normal file
View File

@ -0,0 +1,49 @@
// SPDX-FileCopyrightText: 2025-2026 Lucio Lelii <lucio.lelii@isti.cnr.it> - ISTI-CNR
// SPDX-License-Identifier: AGPL-3.0-or-later
// Attribution term under AGPL-3.0 section 7(b): see LICENSE-ADDENDUM.
package it.cnr.isti.workflow.manager.mcp;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.annotation.JsonProperty;
import tools.jackson.databind.JsonNode;
import it.cnr.isti.workflow.manager.configurations.annotations.DynamicSchema;
import it.cnr.isti.workflow.manager.configurations.annotations.FieldRetriever;
import it.cnr.isti.workflow.manager.configurations.annotations.UiDescription;
import it.cnr.isti.workflow.manager.configurations.annotations.UiLabel;
import jakarta.validation.constraints.NotBlank;
import jakarta.validation.constraints.NotNull;
import lombok.Builder;
/**
* An MCP server an LLM node may call tools on, chosen from the catalog.
*
* <p>Narrower than the binding an {@code MCPAgent} node uses, and separate from it on purpose. The
* retriever url lives on the component, so it ends up inside the shared schema definition rather
* than on the property that references it: one Java type would mean one server list for every block
* that binds servers, and restricting this node to the servers it can actually reach would restrict
* {@code MCPAgent} to them too.
*
* <p>Catalog only, so no {@code sourceType}: a hand-written server would have the flow author typing
* a transport and a url that this service - not the bridge - then connects to.
*/
@Builder
@JsonIgnoreProperties(ignoreUnknown = true)
public record MCPToolServerBinding(
@NotBlank
@JsonProperty(required = false)
@FieldRetriever(name = "MCPServers", url = "/retriever/MCPServers/streamableHttpServers")
@UiLabel("server")
@UiDescription("Catalog MCP server whose tools this node may call. Only streamable-http servers"
+ " can be called directly; use an MCPAgent node for the others.")
String serverName,
@NotNull
@JsonProperty(required = false)
@DynamicSchema(url = "/retriever/MCPServers/definitions/schema", dependsOn = { "serverName" })
JsonNode configuration) {
/** Kept low on purpose: every bound server is a session opened for the life of the block. */
public static final int MAX_PER_BLOCK = 4;
}

15
src/main/java/it/cnr/isti/workflow/manager/mcp/client/MCPToolCallResult.java Normal file
View File

@ -0,0 +1,15 @@
// SPDX-FileCopyrightText: 2025-2026 Lucio Lelii <lucio.lelii@isti.cnr.it> - ISTI-CNR
// SPDX-License-Identifier: AGPL-3.0-or-later
// Attribution term under AGPL-3.0 section 7(b): see LICENSE-ADDENDUM.
package it.cnr.isti.workflow.manager.mcp.client;
/**
* What one {@code tools/call} produced, already flattened to the text that goes back to the model.
*
* <p>{@code error} is not a failure of the call itself: MCP models a tool that refused or blew up as
* a normal result carrying {@code isError}, precisely so the model can read the message and try
* something else. It is carried here only so the execution event can say what happened.
*/
public record MCPToolCallResult(String serverId, String text, boolean error) {
}

66
src/main/java/it/cnr/isti/workflow/manager/mcp/client/MCPToolResults.java Normal file
View File

@ -0,0 +1,66 @@
// SPDX-FileCopyrightText: 2025-2026 Lucio Lelii <lucio.lelii@isti.cnr.it> - ISTI-CNR
// SPDX-License-Identifier: AGPL-3.0-or-later
// Attribution term under AGPL-3.0 section 7(b): see LICENSE-ADDENDUM.
package it.cnr.isti.workflow.manager.mcp.client;
import java.util.ArrayList;
import java.util.List;
import tools.jackson.databind.JsonNode;
/** Turns the {@code content} array of a {@code tools/call} result into the text the model reads. */
final class MCPToolResults {
private MCPToolResults() {
}
static String toText(JsonNode result, int maxChars) {
if (result == null || result.isNull()) {
return "";
}
List<String> parts = new ArrayList<>();
JsonNode content = result.get("content");
if (content != null && content.isArray()) {
for (JsonNode part : content) {
String text = textOf(part);
if (!text.isEmpty()) {
parts.add(text);
}
}
}
if (parts.isEmpty()) {
// A tool that answers only with structured data still has to say something to the model,
// which reads text and nothing else.
JsonNode structured = result.get("structuredContent");
if (structured != null && !structured.isNull()) {
parts.add(structured.toString());
}
}
return truncate(String.join(System.lineSeparator(), parts), maxChars);
}
private static String textOf(JsonNode part) {
if (part == null || !part.isObject()) {
return "";
}
JsonNode text = part.get("text");
if (text != null && text.isString()) {
return text.asString();
}
// Images, audio and embedded resources have no textual form worth inventing, but dropping
// them silently would let the model believe the tool returned nothing at all.
JsonNode type = part.get("type");
return "[" + (type != null && type.isString() ? type.asString() : "unknown") + " content omitted]";
}
private static String truncate(String text, int maxChars) {
if (maxChars <= 0 || text.length() <= maxChars) {
return text;
}
// Marked, never silent: a model that is handed a cut-off file and is not told so will reason
// confidently about the half it can see.
int dropped = text.length() - maxChars;
return text.substring(0, maxChars) + System.lineSeparator() + "... [truncated " + dropped + " characters]";
}
}

276
src/main/java/it/cnr/isti/workflow/manager/mcp/client/MCPToolSession.java Normal file
View File

@ -0,0 +1,276 @@
// SPDX-FileCopyrightText: 2025-2026 Lucio Lelii <lucio.lelii@isti.cnr.it> - ISTI-CNR
// SPDX-License-Identifier: AGPL-3.0-or-later
// Attribution term under AGPL-3.0 section 7(b): see LICENSE-ADDENDUM.
package it.cnr.isti.workflow.manager.mcp.client;
import java.net.URI;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.http.codec.ServerSentEvent;
import org.springframework.web.reactive.function.client.WebClient;
import it.cnr.isti.workflow.manager.llms.ToolDefinition;
import reactor.core.publisher.Mono;
import tools.jackson.databind.JsonNode;
import tools.jackson.databind.ObjectMapper;
import tools.jackson.databind.node.JsonNodeFactory;
import tools.jackson.databind.node.ObjectNode;
/**
* A conversation with one MCP server over the {@code streamable-http} transport, for the length of a
* single block execution.
*
* <p>Deliberately small. It declares no client capabilities at all, so the server has no reason to
* open the server-to-client channel and this never has to hold the GET stream: every exchange is one
* POST and one response. Sampling, roots and elicitation would each need that channel, and a
* workflow node has no user to ask anyway.
*/
public class MCPToolSession implements AutoCloseable {
private static final Logger log = LoggerFactory.getLogger(MCPToolSession.class);
/** A contract with the server rather than an operational knob, so it lives in code. */
static final String PROTOCOL_VERSION = "2025-06-18";
private static final String SESSION_HEADER = "Mcp-Session-Id";
private static final String PROTOCOL_HEADER = "MCP-Protocol-Version";
/** Enough for any real catalog; a server that keeps handing out cursors is broken, not verbose. */
private static final int MAX_TOOL_PAGES = 20;
private final String serverId;
private final WebClient client;
private final URI url;
private final Map<String, String> headers;
private final Duration timeout;
private final int maxResultChars;
private final ObjectMapper objectMapper;
private final AtomicLong nextRequestId = new AtomicLong(1);
private String sessionId;
MCPToolSession(String serverId, WebClient client, URI url, Map<String, String> headers, Duration timeout,
int maxResultChars, ObjectMapper objectMapper) {
this.serverId = serverId;
this.client = client;
this.url = url;
this.headers = headers;
this.timeout = timeout;
this.maxResultChars = maxResultChars;
this.objectMapper = objectMapper;
}
public String serverId() {
return serverId;
}
void initialize() {
ObjectNode params = JsonNodeFactory.instance.objectNode();
params.put("protocolVersion", PROTOCOL_VERSION);
params.set("capabilities", JsonNodeFactory.instance.objectNode());
ObjectNode clientInfo = params.putObject("clientInfo");
clientInfo.put("name", "humainflow-workflow-manager");
clientInfo.put("version", "1");
request("initialize", params);
notifyServer();
}
public List<ToolDefinition> listTools() {
List<ToolDefinition> tools = new ArrayList<>();
String cursor = null;
for (int page = 0; page < MAX_TOOL_PAGES; page++) {
ObjectNode params = JsonNodeFactory.instance.objectNode();
if (cursor != null) {
params.put("cursor", cursor);
}
JsonNode result = request("tools/list", params);
JsonNode listed = result.get("tools");
if (listed != null && listed.isArray()) {
for (JsonNode tool : listed) {
JsonNode name = tool.get("name");
if (name == null || !name.isString()) {
continue;
}
JsonNode description = tool.get("description");
tools.add(new ToolDefinition(name.asString(),
description != null && description.isString() ? description.asString() : "",
tool.get("inputSchema")));
}
}
JsonNode next = result.get("nextCursor");
if (next == null || !next.isString()) {
break;
}
cursor = next.asString();
}
if (tools.isEmpty()) {
// Carrying on would give the model no way to act and every reason to invent having acted.
throw new IllegalStateException("MCP server " + serverId + " exposes no tools");
}
return List.copyOf(tools);
}
/**
* Runs one tool. A tool that fails is not an exception: MCP reports it as a result carrying
* {@code isError} so the model can read what went wrong and choose differently, and a JSON-RPC
* error here means the same thing to the model even though the protocol spells it differently.
*/
public MCPToolCallResult callTool(String name, JsonNode arguments) {
ObjectNode params = JsonNodeFactory.instance.objectNode();
params.put("name", name);
// Passed through exactly as the model produced it. Nothing here inspects or re-types the
// values: an argument the tool's schema declares as a string stays a string even when its
// text happens to be valid JSON.
params.set("arguments", arguments == null || arguments.isNull()
? JsonNodeFactory.instance.objectNode()
: arguments);
JsonNode envelope = exchange("tools/call", params);
JsonNode error = envelope.get("error");
if (error != null && !error.isNull()) {
return new MCPToolCallResult(serverId, describeRpcError(error), true);
}
JsonNode result = envelope.get("result");
boolean isError = result != null && result.path("isError").asBoolean(false);
return new MCPToolCallResult(serverId, MCPToolResults.toText(result, maxResultChars), isError);
}
@Override
public void close() {
if (sessionId == null) {
return;
}
try {
client.delete()
.uri(url)
.headers(this::applyHeaders)
.retrieve()
.toBodilessEntity()
.block(timeout);
} catch (Exception e) {
// Never fails the block: the session expires on its own, and a server that does not
// implement DELETE answers 405, which is not a problem worth surfacing.
log.debug("Closing MCP session on {} failed, letting it expire instead: {}", serverId, e.toString());
}
}
/** One request/response exchange, failing on a JSON-RPC error. */
private JsonNode request(String method, JsonNode params) {
JsonNode envelope = exchange(method, params);
JsonNode error = envelope.get("error");
if (error != null && !error.isNull()) {
throw new IllegalStateException(
"MCP server " + serverId + " failed " + method + ": " + describeRpcError(error));
}
JsonNode result = envelope.get("result");
return result == null ? JsonNodeFactory.instance.objectNode() : result;
}
private JsonNode exchange(String method, JsonNode params) {
ObjectNode body = JsonNodeFactory.instance.objectNode();
body.put("jsonrpc", "2.0");
body.put("id", nextRequestId.getAndIncrement());
body.put("method", method);
body.set("params", params);
String response = post(body, method);
if (response == null || response.isBlank()) {
throw new IllegalStateException("MCP server " + serverId + " returned an empty body for " + method);
}
try {
return objectMapper.readTree(response);
} catch (Exception e) {
throw new IllegalStateException(
"MCP server " + serverId + " returned an unparsable response for " + method, e);
}
}
/** The handshake's second half: a notification, so no id and no response to read. */
private void notifyServer() {
ObjectNode body = JsonNodeFactory.instance.objectNode();
body.put("jsonrpc", "2.0");
body.put("method", "notifications/initialized");
body.set("params", JsonNodeFactory.instance.objectNode());
post(body, "notifications/initialized");
}
private String post(JsonNode body, String method) {
return client.post()
.uri(url)
.headers(this::applyHeaders)
.contentType(MediaType.APPLICATION_JSON)
.bodyValue(body)
.exchangeToMono(response -> {
captureSessionId(response.headers().asHttpHeaders());
if (response.statusCode().isError()) {
int status = response.statusCode().value();
return response.bodyToMono(String.class).defaultIfEmpty("")
.flatMap(errorBody -> Mono.error(new IllegalStateException(
describeHttpFailure(method, status, errorBody))));
}
// The spec lets a server answer a single request either way, so the content type
// decides how to read it. On the SSE branch only the first event matters and the
// stream is cancelled right after, otherwise it would stay open until the timeout.
MediaType contentType = response.headers().contentType().orElse(null);
if (contentType != null && contentType.isCompatibleWith(MediaType.TEXT_EVENT_STREAM)) {
return response.bodyToFlux(new ParameterizedTypeReference<ServerSentEvent<String>>() {})
.next()
.map(event -> event.data() == null ? "" : event.data())
.defaultIfEmpty("");
}
return response.bodyToMono(String.class).defaultIfEmpty("");
})
.block(timeout);
}
private void applyHeaders(HttpHeaders target) {
headers.forEach(target::set);
// Both are required on every request: without the Accept a server is entitled to answer 406,
// and without the protocol header it may assume an older revision than the one negotiated.
target.set(HttpHeaders.ACCEPT, "application/json, text/event-stream");
target.set(PROTOCOL_HEADER, PROTOCOL_VERSION);
if (sessionId != null) {
target.set(SESSION_HEADER, sessionId);
}
}
private void captureSessionId(HttpHeaders responseHeaders) {
String returned = responseHeaders.getFirst(SESSION_HEADER);
if (returned != null && !returned.isBlank()) {
sessionId = returned;
}
}
private String describeHttpFailure(String method, int status, String body) {
if (status == 404 && sessionId != null) {
// Re-initializing here would silently restart the server-side conversation midway through
// an agent loop, so the block fails instead and says why.
return "MCP session on " + serverId + " is gone (404 on " + method + "); it expired or was closed";
}
return "MCP server " + serverId + " returned HTTP " + status + " for " + method + ": " + abbreviate(body);
}
private static String describeRpcError(JsonNode error) {
JsonNode message = error.get("message");
JsonNode code = error.get("code");
String text = message != null && message.isString() ? message.asString() : error.toString();
return code != null && code.isNumber() ? text + " (code " + code.asInt() + ")" : text;
}
private static String abbreviate(String body) {
if (body == null) {
return "";
}
String normalized = body.replaceAll("\\s+", " ").trim();
return normalized.length() <= 500 ? normalized : normalized.substring(0, 500) + "...";
}
}

95
src/main/java/it/cnr/isti/workflow/manager/mcp/client/MCPToolSessionFactory.java Normal file
View File

@ -0,0 +1,95 @@
// SPDX-FileCopyrightText: 2025-2026 Lucio Lelii <lucio.lelii@isti.cnr.it> - ISTI-CNR
// SPDX-License-Identifier: AGPL-3.0-or-later
// Attribution term under AGPL-3.0 section 7(b): see LICENSE-ADDENDUM.
package it.cnr.isti.workflow.manager.mcp.client;
import java.net.URI;
import java.time.Duration;
import java.util.Map;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.web.reactive.function.client.WebClient;
import it.cnr.isti.workflow.manager.mcp.MCPRuntimeTemplates;
import it.cnr.isti.workflow.manager.mcp.MCPServersProvider;
import tools.jackson.core.type.TypeReference;
import tools.jackson.databind.JsonNode;
import tools.jackson.databind.ObjectMapper;
/**
* Opens a {@link MCPToolSession} against a catalog server, resolving its url and headers exactly the
* way the bridge path does.
*
* <p>Only {@code streamable-http} servers are reachable this way. The {@code stdio} ones in the
* catalog need something to start a process and keep it alive, which is what the bridge is for and
* what a node running one HTTP conversation per execution is not.
*/
@Component
public class MCPToolSessionFactory {
/** Deliberately not the URL: only that we speak it. See the plan's note on outbound calls. */
private static final String SUPPORTED_TRANSPORT = "streamable-http";
private static final TypeReference<Map<String, Object>> STRING_OBJECT_MAP = new TypeReference<>() {
};
private final WebClient client;
private final MCPServersProvider serversProvider;
private final ObjectMapper objectMapper = new ObjectMapper();
private final Duration requestTimeout;
private final int maxToolResultChars;
public MCPToolSessionFactory(@Qualifier("mcpWebClientBuilder") WebClient.Builder webClientBuilder,
MCPServersProvider serversProvider,
@Value("${app.mcp.client.request-timeout-seconds:120}") long requestTimeoutSeconds,
@Value("${app.mcp.client.max-tool-result-chars:32000}") int maxToolResultChars) {
this.client = webClientBuilder.build();
this.serversProvider = serversProvider;
this.requestTimeout = Duration.ofSeconds(requestTimeoutSeconds);
this.maxToolResultChars = maxToolResultChars;
}
public MCPToolSession open(String serverName, JsonNode configuration, Map<String, Object> executionVariables) {
MCPServersProvider.MCPServerDefinition definition = serversProvider.getServer(serverName);
if (!SUPPORTED_TRANSPORT.equalsIgnoreCase(definition.transport())) {
throw new IllegalArgumentException("MCP server " + serverName + " uses transport "
+ definition.transport() + ", which an LLM node cannot call directly; use an MCPAgent node");
}
if (definition.url() == null || definition.url().isBlank()) {
throw new IllegalArgumentException("MCP server " + serverName + " declares no url");
}
Map<String, Object> templateValues = configuration == null || configuration.isNull()
? Map.of()
: objectMapper.convertValue(configuration, STRING_OBJECT_MAP);
String resolvedUrl = MCPRuntimeTemplates.resolve(definition.url(), templateValues, executionVariables);
Map<String, String> resolvedHeaders = definition.headers() == null
? Map.of()
: MCPRuntimeTemplates.resolveAll(definition.headers(), templateValues, executionVariables);
MCPToolSession session = new MCPToolSession(serverName, client, toHttpUri(serverName, resolvedUrl),
resolvedHeaders, requestTimeout, maxToolResultChars, objectMapper);
session.initialize();
return session;
}
private static URI toHttpUri(String serverName, String url) {
URI uri;
try {
uri = URI.create(url);
} catch (IllegalArgumentException e) {
throw new IllegalArgumentException("MCP server " + serverName + " resolved to an invalid url: " + url, e);
}
String scheme = uri.getScheme();
if (scheme == null || !(scheme.equalsIgnoreCase("http") || scheme.equalsIgnoreCase("https"))) {
throw new IllegalArgumentException(
"MCP server " + serverName + " must be reached over http or https, got: " + url);
}
if (uri.getHost() == null || uri.getHost().isBlank()) {
throw new IllegalArgumentException("MCP server " + serverName + " resolved to a url with no host: " + url);
}
return uri;
}
}

106
src/main/java/it/cnr/isti/workflow/manager/mcp/client/MCPToolbox.java Normal file
View File

@ -0,0 +1,106 @@
// SPDX-FileCopyrightText: 2025-2026 Lucio Lelii <lucio.lelii@isti.cnr.it> - ISTI-CNR
// SPDX-License-Identifier: AGPL-3.0-or-later
// Attribution term under AGPL-3.0 section 7(b): see LICENSE-ADDENDUM.
package it.cnr.isti.workflow.manager.mcp.client;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import it.cnr.isti.workflow.manager.llms.ToolCall;
import it.cnr.isti.workflow.manager.llms.ToolDefinition;
import it.cnr.isti.workflow.manager.mcp.MCPToolServerBinding;
/**
* Every tool a node can reach, across the servers it binds, behind one lookup by tool name.
*
* <p>The agent loop talks only to this: it knows tools, not servers, which is also all the model
* knows.
*/
public class MCPToolbox implements AutoCloseable {
private static final Logger log = LoggerFactory.getLogger(MCPToolbox.class);
private final List<MCPToolSession> sessions;
private final Map<String, MCPToolSession> sessionsByTool;
private final List<ToolDefinition> tools;
private MCPToolbox(List<MCPToolSession> sessions, Map<String, MCPToolSession> sessionsByTool,
List<ToolDefinition> tools) {
this.sessions = sessions;
this.sessionsByTool = sessionsByTool;
this.tools = tools;
}
public static MCPToolbox open(List<MCPToolServerBinding> bindings, Map<String, Object> executionVariables,
MCPToolSessionFactory factory) {
List<MCPToolSession> opened = new ArrayList<>();
Map<String, MCPToolSession> byTool = new LinkedHashMap<>();
List<ToolDefinition> allTools = new ArrayList<>();
try {
for (MCPToolServerBinding binding : bindings) {
MCPToolSession session = factory.open(binding.serverName(), binding.configuration(),
executionVariables);
opened.add(session);
for (ToolDefinition tool : session.listTools()) {
MCPToolSession previous = byTool.putIfAbsent(tool.name(), session);
if (previous != null) {
// Renaming one of them would change what the model sees for reasons it has no
// way to know, so the flow has to pick.
throw new IllegalStateException("Tool " + tool.name() + " is offered by both "
+ previous.serverId() + " and " + session.serverId()
+ "; bind only one of them to this node");
}
allTools.add(tool);
}
}
} catch (RuntimeException e) {
closeAll(opened);
throw e;
}
return new MCPToolbox(List.copyOf(opened), byTool, List.copyOf(allTools));
}
public List<ToolDefinition> tools() {
return tools;
}
public List<String> serverIds() {
return sessions.stream().map(MCPToolSession::serverId).toList();
}
/**
* Runs one call the model asked for. A name no server offers comes back as an error result
* rather than an exception: the model made it up, and telling it so is what lets it recover.
*/
public MCPToolCallResult call(ToolCall toolCall) {
MCPToolSession session = sessionsByTool.get(toolCall.name());
if (session == null) {
return new MCPToolCallResult(null,
"No such tool: " + toolCall.name() + ". Available tools: "
+ String.join(", ", sessionsByTool.keySet()),
true);
}
return session.callTool(toolCall.name(), toolCall.arguments());
}
@Override
public void close() {
closeAll(sessions);
}
private static void closeAll(List<MCPToolSession> sessions) {
for (MCPToolSession session : sessions) {
try {
session.close();
} catch (RuntimeException e) {
log.debug("Closing MCP session on {} failed: {}", session.serverId(), e.toString());
}
}
}
}

25
src/main/resources/application.properties
View File

@ -57,6 +57,31 @@ app.mcp.session.cleanup-interval-ms=${MCP_SESSION_CLEANUP_INTERVAL_MS:60000}
app.mcp.bridge.max-steps=${MCP_BRIDGE_MAX_STEPS:100}
app.mcp.servers.file=${MCP_SERVERS_FILE:}
app.mcp.agent.responses-log-file=${MCP_AGENT_RESPONSES_LOG_FILE:logs/mcp-agent-responses.log}
# An LLM node that binds MCP servers runs its tool loop in this service, so the budget is enforced
# here. A single model call is already capped by AbstractHttpLLMProvider.timeout(), which makes the
# wall-clock deadline the limit that actually bites: there is no per-step timeout, and cancelling an
# execution only interrupts the thread, so this is what keeps a runaway node from holding a slot in
# the per-execution pool. Running out of either fails the block rather than returning what the model
# happened to say last.
app.llm.tools.max-iterations=${LLM_TOOLS_MAX_ITERATIONS:10}
app.llm.tools.max-duration-seconds=${LLM_TOOLS_MAX_DURATION_SECONDS:900}
# max-tool-result-chars caps a single result; it does not cap what a node with several tool-heavy
# iterations adds up to. Once the running conversation - plus the tool schemas sent on every call,
# which do not appear in the conversation but are not free either - passes this many characters, the
# loop starts replacing OLDER tool results with a placeholder (newest iteration always kept in full)
# rather than letting the provider find out the hard way that it ran out of context. On the gateway
# this was tuned against, going over does not fail cleanly: it drops the earliest message and answers
# "no user query found in messages", or the model context-shifts mid-generation and answers nothing at
# all. Characters, not tokens: nothing here has a tokenizer for whatever model is configured, so this
# is a conservative proxy, sized with real headroom under a typical 32k-token context window for the
# model's own output and reasoning. 0 disables pruning.
app.llm.tools.context-budget-chars=${LLM_TOOLS_CONTEXT_BUDGET_CHARS:60000}
app.mcp.client.request-timeout-seconds=${MCP_CLIENT_REQUEST_TIMEOUT_SECONDS:120}
# Kept well under the context budget above: a single call reading one large file at this cap already
# uses a fifth of the whole conversation's budget, and the protected "current iteration" tool results
# are the ones pruning can never shrink - a handful of large reads in the same turn is what emptied
# the budget in one step the first time this was tried against a real, 24-task plan.
app.mcp.client.max-tool-result-chars=${MCP_CLIENT_MAX_TOOL_RESULT_CHARS:12000}
app.executions.cache.max-size=${APP_EXECUTIONS_CACHE_MAX_SIZE:1000}
app.executions.cache.final-ttl-ms=${APP_EXECUTIONS_CACHE_FINAL_TTL_MS:1800000}
app.executions.cache.cleanup-interval-ms=${APP_EXECUTIONS_CACHE_CLEANUP_INTERVAL_MS:60000}

6
src/main/resources/skills-catalog.json
View File

@ -23,6 +23,12 @@
"name": "Task Generation",
"description": "Breaks requested work into actionable development tasks with dependency awareness.",
"promptPath": "classpath:skills/task-generation/SKILL.md"
},
{
"id": "mcp-context-economy",
"name": "MCP Context Economy",
"description": "Keeps a multi-turn MCP tool-calling loop within its context budget: the right tool for creating vs. editing a file, and never re-reading what was just written or is already in the prompt.",
"promptPath": "classpath:skills/mcp-context-economy/SKILL.md"
}
]
}

8
src/main/resources/skills/mcp-context-economy/SKILL.md Normal file
View File

@ -0,0 +1,8 @@
When working through MCP file tools across a multi-step task, treat context space as a shared, exhaustible budget: what you keep in the conversation now stays for every later turn too, not just this one.
Apply these rules:
- to create a file, use the write tool (e.g. write_file); reserve a patch tool (e.g. apply_patch) for editing a file that already exists. Applying a patch to create a file is unreliable, and recovering from that mistake is exactly the kind of extra reasoning that can run out your own output budget before you produce an answer
- after writing a file, verify it with a lightweight check only - existence and a plausible size (e.g. file_info) - never with a full read right after the write. You already have the file's exact content from the write call you just made; reading it back adds a second copy of the same bytes for no new information
- if a document is already given to you inline in your instructions (a plan, a spec, a large piece of text) and you write it to a file, do not also pull it back into the conversation with a read tool - refer to the file path from then on instead of re-reading it
- the rule above is only about re-reading what you already have in front of you: when you need to inspect something you did not just write yourself, or that may already exist from earlier work, reading it is the right call
- keep a read as narrow as the task allows - a targeted search or a specific section - rather than reading an entire large file when only part of it answers the question

12
src/test/java/it/cnr/isti/workflow/manager/controllers/RetrieverControllerTest.java
View File

@ -41,6 +41,18 @@ public class RetrieverControllerTest {
assertTrue(values.contains("postgres"));
}
@Test
public void listsOnlyTheServersAnLlmNodeCanCallDirectly() {
// An LLM node opens the connection itself, so it can only offer the http-transport servers.
// The stdio ones need a process started for them, which is what an MCPAgent node's bridge
// does - offering them here would mean a flow that only fails when it is saved.
List<String> values = retrieverController.getValues("MCPServers", "streamableHttpServers", Map.of());
assertTrue(values.contains("coding-agent-mcp"));
assertFalse(values.contains("postgres"));
assertFalse(values.contains("url-downloader"));
}
@Test
public void llmModelsAreOpenForAProviderThatCannotListThem() {
// Gemini's catalogue is not listable without a credential and its names move constantly, so

489
src/test/java/it/cnr/isti/workflow/manager/executions/executors/blocks/LLMToolLoopTest.java Normal file
View File

@ -0,0 +1,489 @@
// SPDX-FileCopyrightText: 2025-2026 Lucio Lelii <lucio.lelii@isti.cnr.it> - ISTI-CNR
// SPDX-License-Identifier: AGPL-3.0-or-later
// Attribution term under AGPL-3.0 section 7(b): see LICENSE-ADDENDUM.
package it.cnr.isti.workflow.manager.executions.executors.blocks;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Deque;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import org.springframework.core.io.DefaultResourceLoader;
import org.springframework.web.reactive.function.client.WebClient;
import com.sun.net.httpserver.HttpExchange;
import com.sun.net.httpserver.HttpServer;
import it.cnr.isti.workflow.manager.executions.ExecutionEventLogger;
import it.cnr.isti.workflow.manager.executions.ExecutionEventType;
import it.cnr.isti.workflow.manager.llms.ChatMessage;
import it.cnr.isti.workflow.manager.llms.LLMDescriptor;
import it.cnr.isti.workflow.manager.llms.ModelParameters;
import it.cnr.isti.workflow.manager.llms.ProviderCredential;
import it.cnr.isti.workflow.manager.llms.ToolCall;
import it.cnr.isti.workflow.manager.llms.ToolChatResult;
import it.cnr.isti.workflow.manager.llms.ToolDefinition;
import it.cnr.isti.workflow.manager.llms.providers.LLMProvider;
import it.cnr.isti.workflow.manager.mcp.MCPServersProvider;
import it.cnr.isti.workflow.manager.mcp.MCPToolServerBinding;
import it.cnr.isti.workflow.manager.mcp.client.MCPToolSessionFactory;
import tools.jackson.databind.JsonNode;
import tools.jackson.databind.ObjectMapper;
import tools.jackson.databind.node.JsonNodeFactory;
import tools.jackson.databind.node.ObjectNode;
/**
* The agent loop an LLM node runs when it has MCP servers bound: what reaches the tool, what reaches
* the model on the next turn, and what happens when it does not finish.
*/
class LLMToolLoopTest {
@TempDir
Path tempDir;
private final ObjectMapper mapper = new ObjectMapper();
private final List<String> toolCallArguments = new ArrayList<>();
private final AtomicInteger deleteCalls = new AtomicInteger();
/** Answers with whatever the test queued, and records the conversation it was given. */
private static class ScriptedProvider implements LLMProvider {
private final Deque<ToolChatResult> turns = new ArrayDeque<>();
private final List<List<ChatMessage>> seenConversations = new ArrayList<>();
private final boolean supportsTools;
private Runnable beforeAnswering = () -> { };
ScriptedProvider(boolean supportsTools) {
this.supportsTools = supportsTools;
}
@Override
public String getName() {
return "Scripted";
}
@Override
public List<String> getRegisteredModels() {
return List.of();
}
@Override
public String generate(String model, String prompt) {
throw new UnsupportedOperationException("the loop must never fall back to the text path");
}
@Override
public boolean supportsTools() {
return supportsTools;
}
@Override
public ToolChatResult chatWithTools(String model, List<ChatMessage> messages, List<ToolDefinition> tools,
ProviderCredential credential, ModelParameters parameters) {
seenConversations.add(List.copyOf(messages));
beforeAnswering.run();
return turns.isEmpty() ? ToolChatResult.text("ran out of script") : turns.removeFirst();
}
}
/** Large enough that no existing behavioural test brushes against it by accident. */
private static final int NO_PRUNING = 10_000_000;
private LLMToolLoop loopFor(HttpServer server, int maxIterations, long maxDurationSeconds) throws Exception {
return new LLMToolLoop(factoryFor(server), maxIterations, maxDurationSeconds, NO_PRUNING);
}
private MCPToolSessionFactory factoryFor(HttpServer server) throws Exception {
String url = "http://127.0.0.1:" + server.getAddress().getPort() + "/mcp";
Path catalog = tempDir.resolve("mcp-servers.json");
Files.writeString(catalog, """
{ "servers": [ { "id": "coding", "name": "Coding", "description": "d",
"transport": "streamable-http", "url": "%s" } ] }
""".formatted(url));
MCPServersProvider provider = new MCPServersProvider(new ObjectMapper(), new DefaultResourceLoader(),
catalog.toString());
return new MCPToolSessionFactory(WebClient.builder(), provider, 5, 32000);
}
private HttpServer startServer() throws IOException {
HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
server.createContext("/mcp", exchange -> {
if ("DELETE".equals(exchange.getRequestMethod())) {
deleteCalls.incrementAndGet();
writeJson(exchange, 200, "");
return;
}
String body = new String(exchange.getRequestBody().readAllBytes(), StandardCharsets.UTF_8);
JsonNode request = mapper.readTree(body);
switch (request.path("method").asString()) {
case "initialize" -> {
exchange.getResponseHeaders().set("Mcp-Session-Id", "sess-1");
writeJson(exchange, 200, rpcResult("{}"));
}
case "notifications/initialized" -> writeJson(exchange, 202, "");
case "tools/list" -> writeJson(exchange, 200, rpcResult("""
{"tools":[{"name":"write_file","description":"writes",
"inputSchema":{"type":"object","properties":{"content":{"type":"string"}}}}]}
"""));
case "tools/call" -> {
toolCallArguments.add(request.path("params").path("arguments").toString());
writeJson(exchange, 200,
rpcResult("{\"content\":[{\"type\":\"text\",\"text\":\"written\"}]}"));
}
default -> writeJson(exchange, 500, "unexpected");
}
});
server.start();
return server;
}
private List<MCPToolServerBinding> bindings() {
return List.of(MCPToolServerBinding.builder()
.serverName("coding")
.configuration(JsonNodeFactory.instance.objectNode())
.build());
}
private static LLMDescriptor descriptor() {
return new LLMDescriptor("Scripted", "test-model", null);
}
private static ObjectNode argumentsWithJsonLookingContent() {
ObjectNode arguments = JsonNodeFactory.instance.objectNode();
arguments.put("path", "state.json");
arguments.put("content", "{\"schema_version\":1}");
return arguments;
}
@Test
void callsTheToolThenReturnsTheAnswerOfTheNextTurn() throws Exception {
HttpServer server = startServer();
ScriptedProvider provider = new ScriptedProvider(true);
provider.turns.add(ToolChatResult.toolCalls("writing it",
List.of(ToolCall.of(0, "write_file", argumentsWithJsonLookingContent()))));
provider.turns.add(ToolChatResult.text("DONE"));
try {
String answer = loopFor(server, 10, 60)
.run(provider, descriptor(), null, "do it", bindings(), Map.of(), null);
assertEquals("DONE", answer);
// Second turn must show the model its own call and the result, or it has no idea the
// tool ran at all.
List<ChatMessage> secondTurn = provider.seenConversations.get(1);
assertEquals(3, secondTurn.size());
assertEquals(ChatMessage.Role.ASSISTANT, secondTurn.get(1).role());
assertEquals(ChatMessage.Role.TOOL, secondTurn.get(2).role());
assertEquals("written", secondTurn.get(2).content());
} finally {
server.stop(0);
}
}
@Test
void anArgumentThatLooksLikeJsonReachesTheToolAsAString() throws Exception {
// The regression the whole native path exists for: the bridge turned this very value into an
// object, which made the server reject its own "type": "string" schema.
HttpServer server = startServer();
ScriptedProvider provider = new ScriptedProvider(true);
provider.turns.add(ToolChatResult.toolCalls("",
List.of(ToolCall.of(0, "write_file", argumentsWithJsonLookingContent()))));
provider.turns.add(ToolChatResult.text("DONE"));
try {
loopFor(server, 10, 60)
.run(provider, descriptor(), null, "do it", bindings(), Map.of(), null);
JsonNode sent = mapper.readTree(toolCallArguments.getFirst());
assertTrue(sent.get("content").isString(),
"content must arrive as a string, was " + sent.get("content").getNodeType());
assertEquals("{\"schema_version\":1}", sent.get("content").asString());
} finally {
server.stop(0);
}
}
@Test
void failsWhenItRunsOutOfIterationsInsteadOfReturningWhatTheModelLastSaid() throws Exception {
HttpServer server = startServer();
ScriptedProvider provider = new ScriptedProvider(true);
for (int i = 0; i < 5; i++) {
provider.turns.add(ToolChatResult.toolCalls("now I will write the file",
List.of(ToolCall.of(0, "write_file", argumentsWithJsonLookingContent()))));
}
try {
IllegalStateException failure = assertThrows(IllegalStateException.class,
() -> loopFor(server, 2, 60)
.run(provider, descriptor(), null, "do it", bindings(), Map.of(), null));
assertTrue(failure.getMessage().contains("max-iterations"),
"the message should name the budget that ran out, got: " + failure.getMessage());
assertEquals(1, deleteCalls.get(), "the session must still be closed");
} finally {
server.stop(0);
}
}
@Test
void namesTheBudgetWhenTheModelRanOutOfOutputTokensMidThought() throws Exception {
// "length" means num_predict cut generation short, typically while the model was still
// reasoning - it never got as far as writing an answer or a tool call. The generic "ended the
// conversation" message reads identically whether the model chose silence or was cut off, and
// only one of those two is fixed by raising max_tokens.
HttpServer server = startServer();
ScriptedProvider provider = new ScriptedProvider(true);
provider.turns.add(new ToolChatResult("", List.of(), "length"));
try {
IllegalStateException failure = assertThrows(IllegalStateException.class,
() -> loopFor(server, 10, 60)
.run(provider, descriptor(), null, "do it", bindings(), Map.of(), null));
assertTrue(failure.getMessage().contains("max_tokens"),
"should point at the parameter to raise, got: " + failure.getMessage());
assertTrue(failure.getMessage().contains("length"), failure.getMessage());
} finally {
server.stop(0);
}
}
@Test
void saysNothingExtraWhenTheProviderGivesNoReason() throws Exception {
// A provider that never reports why it stopped must not be blamed for running out of tokens -
// that guess would be worse than no guess at all.
HttpServer server = startServer();
ScriptedProvider provider = new ScriptedProvider(true);
provider.turns.add(ToolChatResult.text(""));
try {
IllegalStateException failure = assertThrows(IllegalStateException.class,
() -> loopFor(server, 10, 60)
.run(provider, descriptor(), null, "do it", bindings(), Map.of(), null));
assertFalse(failure.getMessage().contains("max_tokens"), failure.getMessage());
assertFalse(failure.getMessage().contains("finish reason"), failure.getMessage());
} finally {
server.stop(0);
}
}
@Test
void stopsWhenTheExecutionIsCancelled() throws Exception {
HttpServer server = startServer();
ScriptedProvider provider = new ScriptedProvider(true);
provider.turns.add(ToolChatResult.toolCalls("one moment",
List.of(ToolCall.of(0, "write_file", argumentsWithJsonLookingContent()))));
provider.turns.add(ToolChatResult.text("never reached"));
// Cancelling an execution interrupts the pool thread, which is the only signal a block gets.
provider.beforeAnswering = () -> Thread.currentThread().interrupt();
try {
IllegalStateException failure = assertThrows(IllegalStateException.class,
() -> loopFor(server, 10, 60)
.run(provider, descriptor(), null, "do it", bindings(), Map.of(), null));
assertTrue(failure.getMessage().contains("cancelled"), failure.getMessage());
assertEquals(1, provider.seenConversations.size(),
"the loop must stop at the interrupted iteration, not run the queued next turn");
// The session is deliberately left to expire rather than closed: closing means a blocking
// HTTP call, and making one on a thread that was just interrupted is how a cancellation
// ends up hanging on the very cleanup meant to tidy up after it.
assertEquals(0, deleteCalls.get());
} finally {
Thread.interrupted();
server.stop(0);
}
}
@Test
void refusesAProviderWithoutToolSupportBeforeOpeningAnything() throws Exception {
HttpServer server = startServer();
try {
IllegalArgumentException failure = assertThrows(IllegalArgumentException.class,
() -> loopFor(server, 10, 60).run(new ScriptedProvider(false),
descriptor(), null, "do it", bindings(), Map.of(), null));
assertTrue(failure.getMessage().contains("cannot call tools"), failure.getMessage());
assertEquals(0, deleteCalls.get(), "no session should have been opened at all");
} finally {
server.stop(0);
}
}
@Test
void tellsTheModelWhenItAsksForAToolThatDoesNotExist() throws Exception {
HttpServer server = startServer();
ScriptedProvider provider = new ScriptedProvider(true);
provider.turns.add(ToolChatResult.toolCalls("",
List.of(ToolCall.of(0, "delete_everything", JsonNodeFactory.instance.objectNode()))));
provider.turns.add(ToolChatResult.text("understood"));
try {
// A hallucinated tool is recoverable: the model is told and gets another turn.
String answer = loopFor(server, 10, 60)
.run(provider, descriptor(), null, "do it", bindings(), Map.of(), null);
assertEquals("understood", answer);
ChatMessage toolResult = provider.seenConversations.get(1).get(2);
assertTrue(toolResult.content().contains("No such tool: delete_everything"), toolResult.content());
assertTrue(toolResult.content().contains("write_file"), "it should list what does exist");
} finally {
server.stop(0);
}
}
/**
* A server whose tool always answers with a result of the given size, so the conversation's
* growth is exactly known instead of estimated.
*/
private HttpServer startServerWithFixedResultSize(int resultChars) throws IOException {
String bigResult = "R".repeat(resultChars);
HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
server.createContext("/mcp", exchange -> {
if ("DELETE".equals(exchange.getRequestMethod())) {
deleteCalls.incrementAndGet();
writeJson(exchange, 200, "");
return;
}
String body = new String(exchange.getRequestBody().readAllBytes(), StandardCharsets.UTF_8);
JsonNode request = mapper.readTree(body);
switch (request.path("method").asString()) {
case "initialize" -> writeJson(exchange, 200, rpcResult("{}"));
case "notifications/initialized" -> writeJson(exchange, 202, "");
case "tools/list" -> writeJson(exchange, 200, rpcResult("""
{"tools":[{"name":"big_tool","description":"d",
"inputSchema":{"type":"object","properties":{}}}]}
"""));
case "tools/call" -> writeJson(exchange, 200,
rpcResult("{\"content\":[{\"type\":\"text\",\"text\":\"" + bigResult + "\"}]}"));
default -> writeJson(exchange, 500, "unexpected");
}
});
server.start();
return server;
}
private static ToolChatResult bigToolCallTurn() {
return ToolChatResult.toolCalls("", List.of(ToolCall.of(0, "big_tool", JsonNodeFactory.instance.objectNode())));
}
@Test
void replacesOnlyOlderToolResultsOnceOverTheContextBudget() throws Exception {
// Two rounds of an 800-character tool result: alone each is comfortably under a 1000-character
// budget, but the sum after the second is not - which is exactly the gap max-tool-result-chars
// cannot close, since it caps one result at a time, not what a multi-iteration node adds up to.
HttpServer server = startServerWithFixedResultSize(800);
ScriptedProvider provider = new ScriptedProvider(true);
provider.turns.add(bigToolCallTurn());
provider.turns.add(bigToolCallTurn());
provider.turns.add(ToolChatResult.text("DONE"));
try {
String answer = new LLMToolLoop(factoryFor(server), 10, 60, 1000)
.run(provider, descriptor(), null, "do it", bindings(), Map.of(), null);
assertEquals("DONE", answer);
// What the model saw on its third call: user, assistant#1, tool#1 (now a placeholder),
// assistant#2, tool#2 (the iteration just completed - kept in full).
List<ChatMessage> thirdTurn = provider.seenConversations.get(2);
ChatMessage firstToolResult = thirdTurn.get(2);
ChatMessage secondToolResult = thirdTurn.get(4);
assertTrue(firstToolResult.content().contains("pruned"), firstToolResult.content());
assertFalse(firstToolResult.content().contains("R"), "the original text must be gone, not just labelled");
assertEquals("R".repeat(800), secondToolResult.content(),
"the most recent iteration's own result must never be touched");
} finally {
server.stop(0);
}
}
@Test
void neverPrunesTheIterationThatJustRan() throws Exception {
// A single result already over budget on its own: there is nothing older to prune yet, so it
// must reach the model untouched rather than being shortened the moment it is produced.
HttpServer server = startServerWithFixedResultSize(2000);
ScriptedProvider provider = new ScriptedProvider(true);
provider.turns.add(bigToolCallTurn());
provider.turns.add(ToolChatResult.text("DONE"));
try {
new LLMToolLoop(factoryFor(server), 10, 60, 1000)
.run(provider, descriptor(), null, "do it", bindings(), Map.of(), null);
ChatMessage toolResult = provider.seenConversations.get(1).get(2);
assertEquals("R".repeat(2000), toolResult.content());
} finally {
server.stop(0);
}
}
@Test
void zeroDisablesPruningEvenWellOverBudget() throws Exception {
HttpServer server = startServerWithFixedResultSize(800);
ScriptedProvider provider = new ScriptedProvider(true);
provider.turns.add(bigToolCallTurn());
provider.turns.add(bigToolCallTurn());
provider.turns.add(ToolChatResult.text("DONE"));
try {
new LLMToolLoop(factoryFor(server), 10, 60, 0)
.run(provider, descriptor(), null, "do it", bindings(), Map.of(), null);
ChatMessage firstToolResult = provider.seenConversations.get(2).get(2);
assertEquals("R".repeat(800), firstToolResult.content());
} finally {
server.stop(0);
}
}
@Test
void reportsWhatItPrunedAsAnEvent() throws Exception {
HttpServer server = startServerWithFixedResultSize(800);
ScriptedProvider provider = new ScriptedProvider(true);
provider.turns.add(bigToolCallTurn());
provider.turns.add(bigToolCallTurn());
provider.turns.add(ToolChatResult.text("DONE"));
List<ExecutionEventType> loggedTypes = new ArrayList<>();
List<Map<String, Object>> loggedDetails = new ArrayList<>();
ExecutionEventLogger recordingLogger = (level, type, message, details) -> {
loggedTypes.add(type);
loggedDetails.add(details);
};
try {
new LLMToolLoop(factoryFor(server), 10, 60, 1000)
.run(provider, descriptor(), null, "do it", bindings(), Map.of(), recordingLogger);
int prunedIndex = loggedTypes.indexOf(ExecutionEventType.LLM_CONTEXT_PRUNED);
assertTrue(prunedIndex >= 0, "expected an LLM_CONTEXT_PRUNED event, got: " + loggedTypes);
Map<String, Object> details = loggedDetails.get(prunedIndex);
assertEquals(1, details.get("prunedCount"));
// 800 minus whatever the placeholder itself costs - loose on purpose, so the assertion
// does not have to be rewritten every time the placeholder's wording changes.
long reclaimed = (long) details.get("reclaimedChars");
assertTrue(reclaimed > 600 && reclaimed < 800, "got: " + reclaimed);
} finally {
server.stop(0);
}
}
private static String rpcResult(String result) {
return "{\"jsonrpc\":\"2.0\",\"id\":1,\"result\":" + result + "}";
}
private static void writeJson(HttpExchange exchange, int status, String body) throws IOException {
byte[] bytes = body.getBytes(StandardCharsets.UTF_8);
exchange.getResponseHeaders().set("Content-Type", "application/json");
exchange.sendResponseHeaders(status, bytes.length == 0 ? -1 : bytes.length);
if (bytes.length > 0) {
try (var output = exchange.getResponseBody()) {
output.write(bytes);
}
}
}
}

62
src/test/java/it/cnr/isti/workflow/manager/llms/LLMProviderCatalogServiceTest.java Normal file
View File

@ -0,0 +1,62 @@
// SPDX-FileCopyrightText: 2025-2026 Lucio Lelii <lucio.lelii@isti.cnr.it> - ISTI-CNR
// SPDX-License-Identifier: AGPL-3.0-or-later
// Attribution term under AGPL-3.0 section 7(b): see LICENSE-ADDENDUM.
package it.cnr.isti.workflow.manager.llms;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
import java.util.List;
import java.util.Map;
import org.junit.jupiter.api.Test;
import it.cnr.isti.workflow.manager.llms.providers.LLMProvider;
/**
* What the editor is told about a provider before anyone runs anything.
*
* <p>The capability exists so a choice that cannot work is not offered: a node cannot bind an MCP
* server to a provider that never supports tools, and the editor needs to know that before the flow
* is even saved, not from an execution failure afterwards.
*/
class LLMProviderCatalogServiceTest {
private static LLMProvider provider(String name, boolean tools) {
return new LLMProvider() {
@Override
public String getName() {
return name;
}
@Override
public List<String> getRegisteredModels() {
return List.of();
}
@Override
public String generate(String model, String prompt) {
throw new UnsupportedOperationException("not exercised by this test");
}
@Override
public boolean supportsTools() {
return tools;
}
};
}
@Test
void reportsWhichProvidersCanCallTools() {
LLMProviderCatalogService catalog = new LLMProviderCatalogService(Map.of(
"a", provider("WithTools", true),
"b", provider("NoTools", false)));
Map<String, LLMProviderMetadata> byName = catalog.list().stream()
.collect(java.util.stream.Collectors.toMap(LLMProviderMetadata::name, metadata -> metadata));
assertTrue(byName.get("WithTools").supportsTools());
assertFalse(byName.get("NoTools").supportsTools());
}
}

87
src/test/java/it/cnr/isti/workflow/manager/llms/providers/ToolCallingDefaultsTest.java Normal file
View File

@ -0,0 +1,87 @@
// SPDX-FileCopyrightText: 2025-2026 Lucio Lelii <lucio.lelii@isti.cnr.it> - ISTI-CNR
// SPDX-License-Identifier: AGPL-3.0-or-later
// Attribution term under AGPL-3.0 section 7(b): see LICENSE-ADDENDUM.
package it.cnr.isti.workflow.manager.llms.providers;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import java.util.List;
import org.junit.jupiter.api.Test;
import it.cnr.isti.workflow.manager.llms.ChatMessage;
import it.cnr.isti.workflow.manager.llms.ModelParameters;
import it.cnr.isti.workflow.manager.llms.ProviderCredential;
import it.cnr.isti.workflow.manager.llms.ToolChatResult;
import it.cnr.isti.workflow.manager.llms.ToolDefinition;
/**
* What {@link LLMProvider#chatWithTools} does for a provider that never heard of tools: with none
* asked for it must be indistinguishable from {@code chat}, and with some asked for it must refuse
* rather than answer as if it had used them.
*/
class ToolCallingDefaultsTest {
/** Knows only the text calls, like every provider that predates tool support. */
private static class TextOnlyProvider implements LLMProvider {
@Override
public String getName() {
return "TextOnly";
}
@Override
public List<String> getRegisteredModels() {
return List.of();
}
@Override
public String generate(String model, String prompt) {
throw new UnsupportedOperationException("not exercised by this test");
}
@Override
public String chat(String model, List<ChatMessage> messages, ProviderCredential credential,
ModelParameters parameters) {
return "chatted:" + messages.size();
}
}
private final LLMProvider textOnly = new TextOnlyProvider();
private final List<ChatMessage> messages = List.of(ChatMessage.user("hi"));
@Test
void aProviderDeclaresNoToolSupportUnlessItSaysOtherwise() {
assertFalse(textOnly.supportsTools());
}
@Test
void withoutToolsItIsPlainChatWithTheAnswerWrapped() {
ToolChatResult result = textOnly.chatWithTools("m", messages, List.of(), null, null);
assertEquals("chatted:1", result.content());
assertFalse(result.hasToolCalls());
}
@Test
void aNullToolListIsTheSameAsNoTools() {
assertEquals("chatted:1", textOnly.chatWithTools("m", messages, null, null, null).content());
}
@Test
void refusesRatherThanQuietlyDroppingToolsItCannotUse() {
// Answering anyway would be indistinguishable, to the caller, from a model that considered
// the tools and chose not to call any - which is how a run ends up looking finished when it
// never started.
List<ToolDefinition> tools = List.of(new ToolDefinition("write_file", "writes", null));
UnsupportedOperationException failure = assertThrows(UnsupportedOperationException.class,
() -> textOnly.chatWithTools("m", messages, tools, null, null));
assertTrue(failure.getMessage().contains("TextOnly"),
"the message should name the provider that cannot do it, got: " + failure.getMessage());
}
}

60
src/test/java/it/cnr/isti/workflow/manager/llms/providers/ollama/InternalOllamaLLMProviderBodyTest.java
View File

@ -15,6 +15,10 @@ import org.springframework.web.reactive.function.client.WebClient;
import it.cnr.isti.workflow.manager.llms.ChatMessage;
import it.cnr.isti.workflow.manager.llms.ModelParameters;
import it.cnr.isti.workflow.manager.llms.ToolCall;
import it.cnr.isti.workflow.manager.llms.ToolDefinition;
import tools.jackson.databind.node.JsonNodeFactory;
import tools.jackson.databind.node.ObjectNode;
/**
* What actually goes over the wire to Ollama.
@ -104,4 +108,60 @@ class InternalOllamaLLMProviderBodyTest {
assertEquals(Map.of("temperature", 0.0), body.get("options"));
}
@Test
void anEmptyToolListLeavesTheChatBodyExactlyAsItWas() {
// The overload with tools is what every chat call now goes through, so with none declared it
// has to produce the body the three-argument one always did - otherwise adding tool support
// would change every existing flow that never asked for a tool.
List<ChatMessage> messages = List.of(new ChatMessage(ChatMessage.Role.USER, "hi"));
assertEquals(ollama.buildChatBody("m", messages, null),
ollama.buildChatBody("m", messages, null, List.of()));
}
@Test
void theChatBodyDeclaresTheToolsWithTheServerSchemaUntouched() {
ObjectNode schema = JsonNodeFactory.instance.objectNode();
schema.put("type", "object");
Map<String, Object> body = ollama.buildChatBody("m",
List.of(new ChatMessage(ChatMessage.Role.USER, "hi")), null,
List.of(new ToolDefinition("write_file", "writes a file", schema)));
assertEquals(List.of(Map.of("type", "function", "function", Map.of(
"name", "write_file", "description", "writes a file", "parameters", schema))),
body.get("tools"));
}
@Test
void anAssistantTurnCarriesBackTheCallsItAskedFor() {
ObjectNode arguments = JsonNodeFactory.instance.objectNode();
arguments.put("path", "state.json");
Map<String, Object> body = ollama.buildChatBody("m",
List.of(ChatMessage.assistantToolCalls("on it", List.of(ToolCall.of(0, "write_file", arguments)))),
null, List.of());
assertEquals(List.of(Map.of(
"role", "assistant",
"content", "on it",
"tool_calls", List.of(Map.of("function", Map.of("name", "write_file", "arguments", arguments))))),
body.get("messages"));
}
@Test
void aToolResultIsSentWithBothCorrelationKeys() {
// Which one Ollama reads depends on its version, and picking wrong fails silently: the model
// just sees a result it cannot attribute to its call.
Map<String, Object> body = ollama.buildChatBody("m",
List.of(ChatMessage.toolResult("call_0", "write_file", "ok")), null, List.of());
assertEquals(List.of(Map.of(
"role", "tool",
"content", "ok",
"tool_name", "write_file",
"tool_call_id", "call_0")),
body.get("messages"));
}
}

85
src/test/java/it/cnr/isti/workflow/manager/llms/providers/ollama/InternalOllamaLLMProviderHttpTest.java
View File

@ -19,7 +19,11 @@ import org.springframework.web.reactive.function.client.WebClient;
import com.sun.net.httpserver.HttpExchange;
import com.sun.net.httpserver.HttpServer;
import it.cnr.isti.workflow.manager.llms.ChatMessage;
import it.cnr.isti.workflow.manager.llms.ToolChatResult;
import it.cnr.isti.workflow.manager.llms.ToolDefinition;
import it.cnr.isti.workflow.manager.llms.providers.LLMProviderHttpException;
import tools.jackson.databind.JsonNode;
/**
* The real HTTP round trip through {@link OllamaProtocolProvider}, which nothing exercised before
@ -111,6 +115,87 @@ class InternalOllamaLLMProviderHttpTest {
}
}
@Test
void aTurnThatOnlyCallsToolsIsNotReadAsABrokenResponse() throws IOException {
// Before tool support this response threw: a turn made of tool calls has no message.content
// at all, and the text path treats that as a malformed answer.
HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
server.createContext("/chat", exchange -> writeJson(exchange, 200,
"{\"message\":{\"role\":\"assistant\",\"tool_calls\":"
+ "[{\"function\":{\"name\":\"write_file\",\"arguments\":{\"path\":\"a.txt\"}}}]}}"));
server.start();
try {
ToolChatResult result = providerFor(server).chatWithTools("m", List.of(ChatMessage.user("hi")),
List.of(new ToolDefinition("write_file", "writes", null)), null, null);
assertTrue(result.hasToolCalls());
assertEquals("", result.content());
assertEquals("write_file", result.toolCalls().getFirst().name());
} finally {
server.stop(0);
}
}
@Test
void anArgumentThatLooksLikeJsonStaysAString() throws IOException {
// The whole reason this path exists. The MCP bridge parses a string argument into an object
// whenever it happens to be valid JSON, which makes the MCP server reject its own schema's
// "type": "string" field and the tool call never runs. Nothing here may do that.
HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
server.createContext("/chat", exchange -> writeJson(exchange, 200,
"{\"message\":{\"role\":\"assistant\",\"tool_calls\":[{\"function\":{\"name\":\"write_file\","
+ "\"arguments\":{\"content\":\"{\\\"schema_version\\\":1}\"}}}]}}"));
server.start();
try {
ToolChatResult result = providerFor(server).chatWithTools("m", List.of(ChatMessage.user("hi")),
List.of(new ToolDefinition("write_file", "writes", null)), null, null);
JsonNode content = result.toolCalls().getFirst().arguments().get("content");
assertTrue(content.isString(), "content must stay a string node, was " + content.getNodeType());
assertEquals("{\"schema_version\":1}", content.asString());
} finally {
server.stop(0);
}
}
@Test
void aTurnWithNeitherTextNorToolCallsStillFails() throws IOException {
HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
server.createContext("/chat", exchange ->
writeJson(exchange, 200, "{\"message\":{\"role\":\"assistant\"}}"));
server.start();
try {
RuntimeException failure = assertThrows(RuntimeException.class,
() -> providerFor(server).chatWithTools("m", List.of(ChatMessage.user("hi")), List.of(),
null, null));
assertTrue(failure.getMessage().contains("Unable to parse Ollama chat response"),
"got: " + failure.getMessage());
} finally {
server.stop(0);
}
}
@Test
void capturesWhyGenerationStoppedWhenTheModelSaysSo() throws IOException {
// done_reason sits on the envelope, not the message: it says how the turn ended, and "length"
// in particular means num_predict cut generation short - often before the model reached
// content at all, which otherwise reads identically to a model that simply answered nothing.
HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
server.createContext("/chat", exchange -> writeJson(exchange, 200,
"{\"message\":{\"role\":\"assistant\",\"tool_calls\":"
+ "[{\"function\":{\"name\":\"write_file\",\"arguments\":{}}}]},"
+ "\"done_reason\":\"length\"}"));
server.start();
try {
ToolChatResult result = providerFor(server).chatWithTools("m", List.of(ChatMessage.user("hi")),
List.of(new ToolDefinition("write_file", "writes", null)), null, null);
assertEquals("length", result.finishReason());
} finally {
server.stop(0);
}
}
private static void writeJson(HttpExchange exchange, int status, String body) throws IOException {
byte[] bytes = body.getBytes(StandardCharsets.UTF_8);
exchange.getResponseHeaders().set("Content-Type", "application/json");

278
src/test/java/it/cnr/isti/workflow/manager/mcp/client/MCPToolSessionTest.java Normal file
View File

@ -0,0 +1,278 @@
// SPDX-FileCopyrightText: 2025-2026 Lucio Lelii <lucio.lelii@isti.cnr.it> - ISTI-CNR
// SPDX-License-Identifier: AGPL-3.0-or-later
// Attribution term under AGPL-3.0 section 7(b): see LICENSE-ADDENDUM.
package it.cnr.isti.workflow.manager.mcp.client;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import org.springframework.core.io.DefaultResourceLoader;
import org.springframework.web.reactive.function.client.WebClient;
import com.sun.net.httpserver.HttpExchange;
import com.sun.net.httpserver.HttpServer;
import it.cnr.isti.workflow.manager.llms.ToolDefinition;
import it.cnr.isti.workflow.manager.mcp.MCPServersProvider;
import tools.jackson.databind.ObjectMapper;
import tools.jackson.databind.node.JsonNodeFactory;
import tools.jackson.databind.node.ObjectNode;
/**
* The native streamable-http conversation with an MCP server: the handshake, the session header a
* server may hand out, and the two shapes a single response is allowed to arrive in.
*/
class MCPToolSessionTest {
@TempDir
Path tempDir;
private final ObjectMapper mapper = new ObjectMapper();
/** Every request the fake server saw, so the handshake can be asserted in order. */
private final List<String> methods = new ArrayList<>();
private MCPToolSessionFactory factoryFor(HttpServer server) throws Exception {
String url = "http://127.0.0.1:" + server.getAddress().getPort() + "/mcp";
Path catalog = tempDir.resolve("mcp-servers.json");
Files.writeString(catalog, """
{ "servers": [ {
"id": "coding", "name": "Coding", "description": "d",
"transport": "streamable-http",
"url": "%s",
"headers": { "Authorization": "Bearer ${{key}}" }
} ] }
""".formatted(url));
MCPServersProvider provider = new MCPServersProvider(new ObjectMapper(), new DefaultResourceLoader(),
catalog.toString());
return new MCPToolSessionFactory(WebClient.builder(), provider, 5, 32000);
}
private ObjectNode configuration() {
ObjectNode configuration = JsonNodeFactory.instance.objectNode();
configuration.put("key", "secret-key");
return configuration;
}
@Test
void handshakesThenListsTools() throws Exception {
HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
server.createContext("/mcp", exchange -> {
String method = recordMethod(exchange);
assertEquals("Bearer secret-key", exchange.getRequestHeaders().getFirst("Authorization"),
"the ${{key}} placeholder must be resolved from the block configuration");
assertTrue(exchange.getRequestHeaders().getFirst("Accept").contains("text/event-stream"),
"the spec requires both media types on every request");
switch (method) {
case "initialize" -> {
exchange.getResponseHeaders().set("Mcp-Session-Id", "sess-1");
writeJson(exchange, 200, rpcResult("{\"protocolVersion\":\"2025-06-18\"}"));
}
case "notifications/initialized" -> writeJson(exchange, 202, "");
case "tools/list" -> writeJson(exchange, 200, rpcResult("""
{"tools":[{"name":"write_file","description":"writes",
"inputSchema":{"type":"object","properties":{"content":{"type":"string"}}}}]}
"""));
default -> writeJson(exchange, 500, "unexpected " + method);
}
});
server.start();
try (MCPToolSession session = factoryFor(server).open("coding", configuration(), Map.of())) {
List<ToolDefinition> tools = session.listTools();
assertEquals(List.of("initialize", "notifications/initialized"), methods.subList(0, 2));
assertEquals(1, tools.size());
assertEquals("write_file", tools.getFirst().name());
assertEquals("string", tools.getFirst().inputSchema().path("properties").path("content")
.path("type").asString());
} finally {
server.stop(0);
}
}
@Test
void sendsBackTheSessionIdTheServerHandedOut() throws Exception {
List<String> seenSessionIds = new ArrayList<>();
HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
server.createContext("/mcp", exchange -> {
String method = recordMethod(exchange);
seenSessionIds.add(String.valueOf(exchange.getRequestHeaders().getFirst("Mcp-Session-Id")));
if ("initialize".equals(method)) {
exchange.getResponseHeaders().set("Mcp-Session-Id", "sess-42");
writeJson(exchange, 200, rpcResult("{}"));
} else {
writeJson(exchange, 200, rpcResult("{\"tools\":[{\"name\":\"t\"}]}"));
}
});
server.start();
try (MCPToolSession session = factoryFor(server).open("coding", configuration(), Map.of())) {
session.listTools();
assertEquals("null", seenSessionIds.getFirst(), "nothing to send before the server names it");
assertTrue(seenSessionIds.subList(1, seenSessionIds.size()).stream().allMatch("sess-42"::equals),
"every later request must carry it, got " + seenSessionIds);
} finally {
server.stop(0);
}
}
@Test
void readsAResponseDeliveredAsAnEventStream() throws Exception {
HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
server.createContext("/mcp", exchange -> {
String method = recordMethod(exchange);
if ("initialize".equals(method)) {
writeJson(exchange, 200, rpcResult("{}"));
} else if ("notifications/initialized".equals(method)) {
writeJson(exchange, 202, "");
} else {
// The same single response, delivered the other way the spec allows.
writeSse(exchange, rpcResult("{\"tools\":[{\"name\":\"read_file\",\"description\":\"reads\"}]}"));
}
});
server.start();
try (MCPToolSession session = factoryFor(server).open("coding", configuration(), Map.of())) {
assertEquals("read_file", session.listTools().getFirst().name());
} finally {
server.stop(0);
}
}
@Test
void aToolThatFailsComesBackAsAResultTheModelCanRead() throws Exception {
HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
server.createContext("/mcp", exchange -> {
String method = recordMethod(exchange);
switch (method) {
case "initialize" -> writeJson(exchange, 200, rpcResult("{}"));
case "notifications/initialized" -> writeJson(exchange, 202, "");
default -> writeJson(exchange, 200, rpcResult("""
{"isError":true,"content":[{"type":"text","text":"ENOENT: no such file"}]}
"""));
}
});
server.start();
try (MCPToolSession session = factoryFor(server).open("coding", configuration(), Map.of())) {
// Not an exception: MCP reports a refused or broken tool as a normal result so the model
// can read the message and choose differently.
MCPToolCallResult result = session.callTool("read_file", JsonNodeFactory.instance.objectNode());
assertTrue(result.error());
assertEquals("ENOENT: no such file", result.text());
} finally {
server.stop(0);
}
}
@Test
void aServerThatOffersNoToolsFailsInsteadOfRunningWithout() throws Exception {
HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
server.createContext("/mcp", exchange -> {
String method = recordMethod(exchange);
switch (method) {
case "initialize" -> writeJson(exchange, 200, rpcResult("{}"));
case "notifications/initialized" -> writeJson(exchange, 202, "");
default -> writeJson(exchange, 200, rpcResult("{\"tools\":[]}"));
}
});
server.start();
try (MCPToolSession session = factoryFor(server).open("coding", configuration(), Map.of())) {
IllegalStateException failure = assertThrows(IllegalStateException.class, session::listTools);
assertTrue(failure.getMessage().contains("no tools"), failure.getMessage());
} finally {
server.stop(0);
}
}
@Test
void refusesAServerItCannotSpeakTo() throws Exception {
Path catalog = tempDir.resolve("mcp-servers.json");
Files.writeString(catalog, """
{ "servers": [ { "id": "pg", "name": "Postgres", "description": "d",
"transport": "stdio", "command": "docker" } ] }
""");
MCPServersProvider provider = new MCPServersProvider(new ObjectMapper(), new DefaultResourceLoader(),
catalog.toString());
MCPToolSessionFactory factory = new MCPToolSessionFactory(WebClient.builder(), provider, 5, 32000);
IllegalArgumentException failure = assertThrows(IllegalArgumentException.class,
() -> factory.open("pg", configuration(), Map.of()));
assertTrue(failure.getMessage().contains("MCPAgent"),
"the message should point at the node that can run stdio servers, got: " + failure.getMessage());
}
@Test
void closingToleratesAServerThatDoesNotImplementDelete() throws Exception {
HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
server.createContext("/mcp", exchange -> {
if ("DELETE".equals(exchange.getRequestMethod())) {
writeJson(exchange, 405, "method not allowed");
return;
}
String method = recordMethod(exchange);
if ("initialize".equals(method)) {
exchange.getResponseHeaders().set("Mcp-Session-Id", "sess-1");
writeJson(exchange, 200, rpcResult("{}"));
} else {
writeJson(exchange, 202, "");
}
});
server.start();
try {
MCPToolSession session = factoryFor(server).open("coding", configuration(), Map.of());
session.close();
assertFalse(methods.isEmpty(), "the handshake should still have happened");
} finally {
server.stop(0);
}
}
private String recordMethod(HttpExchange exchange) throws IOException {
String body = new String(exchange.getRequestBody().readAllBytes(), StandardCharsets.UTF_8);
String method = mapper.readTree(body).path("method").asString();
methods.add(method);
return method;
}
private static String rpcResult(String result) {
return "{\"jsonrpc\":\"2.0\",\"id\":1,\"result\":" + result + "}";
}
private static void writeJson(HttpExchange exchange, int status, String body) throws IOException {
byte[] bytes = body.getBytes(StandardCharsets.UTF_8);
exchange.getResponseHeaders().set("Content-Type", "application/json");
exchange.sendResponseHeaders(status, bytes.length == 0 ? -1 : bytes.length);
if (bytes.length > 0) {
try (var output = exchange.getResponseBody()) {
output.write(bytes);
}
}
}
private static void writeSse(HttpExchange exchange, String data) throws IOException {
byte[] bytes = ("event: message\ndata: " + data + "\n\n").getBytes(StandardCharsets.UTF_8);
exchange.getResponseHeaders().set("Content-Type", "text/event-stream");
exchange.sendResponseHeaders(200, bytes.length);
try (var output = exchange.getResponseBody()) {
output.write(bytes);
}
}
}