diff --git a/src/main/java/it/cnr/isti/workflow/manager/assistant/BlockDraftNormalizer.java b/src/main/java/it/cnr/isti/workflow/manager/assistant/BlockDraftNormalizer.java index 3111981..5b23424 100644 --- a/src/main/java/it/cnr/isti/workflow/manager/assistant/BlockDraftNormalizer.java +++ b/src/main/java/it/cnr/isti/workflow/manager/assistant/BlockDraftNormalizer.java @@ -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())) { diff --git a/src/main/java/it/cnr/isti/workflow/manager/blocks/configurations/LLMBlockConfiguration.java b/src/main/java/it/cnr/isti/workflow/manager/blocks/configurations/LLMBlockConfiguration.java index f04808c..ab12e08 100644 --- a/src/main/java/it/cnr/isti/workflow/manager/blocks/configurations/LLMBlockConfiguration.java +++ b/src/main/java/it/cnr/isti/workflow/manager/blocks/configurations/LLMBlockConfiguration.java @@ -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 { @JsonProperty(required = false) List 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 mcpServers = List.of(); + @Builder public LLMBlockConfiguration(@NonNull String name, @JsonProperty(value = "llmDescriptor", required = false) LLMDescriptor llmDescriptor, String prompt, - List skills) { + List skills, + List 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 { 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 { } @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 { .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 names = mcpServers.stream() + .map(MCPToolServerBinding::serverName) + .filter(serverName -> serverName != null && !serverName.isBlank()) + .toList(); + return names.stream().distinct().count() == names.size(); + } + } diff --git a/src/main/java/it/cnr/isti/workflow/manager/configurations/retrievers/MCPServersFieldRetriever.java b/src/main/java/it/cnr/isti/workflow/manager/configurations/retrievers/MCPServersFieldRetriever.java index d204884..7a5a683 100644 --- a/src/main/java/it/cnr/isti/workflow/manager/configurations/retrievers/MCPServersFieldRetriever.java +++ b/src/main/java/it/cnr/isti/workflow/manager/configurations/retrievers/MCPServersFieldRetriever.java @@ -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); }; diff --git a/src/main/java/it/cnr/isti/workflow/manager/executions/ExecutionEventType.java b/src/main/java/it/cnr/isti/workflow/manager/executions/ExecutionEventType.java index 898706b..b99a111 100644 --- a/src/main/java/it/cnr/isti/workflow/manager/executions/ExecutionEventType.java +++ b/src/main/java/it/cnr/isti/workflow/manager/executions/ExecutionEventType.java @@ -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 } diff --git a/src/main/java/it/cnr/isti/workflow/manager/executions/executors/blocks/LLMExecutor.java b/src/main/java/it/cnr/isti/workflow/manager/executions/executors/blocks/LLMExecutor.java index 10fbd0a..515c477 100644 --- a/src/main/java/it/cnr/isti/workflow/manager/executions/executors/blocks/LLMExecutor.java +++ b/src/main/java/it/cnr/isti/workflow/manager/executions/executors/blocks/LLMExecutor.java @@ -44,6 +44,9 @@ public class LLMExecutor implements BlockExecutor { @Autowired private LLMCredentialResolver credentialResolver; + @Autowired + private LLMToolLoop llmToolLoop; + @Override public Map execute(Block block, List inputs, Map authorizations, Map executionVariables, Map executionVariableDescriptors, @@ -82,6 +85,15 @@ public class LLMExecutor implements BlockExecutor { 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()); diff --git a/src/main/java/it/cnr/isti/workflow/manager/executions/executors/blocks/LLMToolLoop.java b/src/main/java/it/cnr/isti/workflow/manager/executions/executors/blocks/LLMToolLoop.java new file mode 100644 index 0000000..05e883c --- /dev/null +++ b/src/main/java/it/cnr/isti/workflow/manager/executions/executors/blocks/LLMToolLoop.java @@ -0,0 +1,321 @@ +// SPDX-FileCopyrightText: 2025-2026 Lucio Lelii - 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. + * + *

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 bindings, Map 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 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 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. + * + *

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. + * + *

{@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. + * + *

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. + * + *

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 messages, int protectFromIndex, + Set 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 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 tools) { + int total = 0; + for (ToolDefinition tool : tools) { + total += tool.name().length() + tool.description().length() + tool.inputSchema().toString().length(); + } + return total; + } + + private static Map eventDetailsFor(LLMDescriptor descriptor, int iteration, ToolChatResult turn) { + Map 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 details) { + if (eventLogger != null) { + eventLogger.info(type, message, details); + } + } +} diff --git a/src/main/java/it/cnr/isti/workflow/manager/flows/validation/FlowDataValidator.java b/src/main/java/it/cnr/isti/workflow/manager/flows/validation/FlowDataValidator.java index 6900edf..8466795 100644 --- a/src/main/java/it/cnr/isti/workflow/manager/flows/validation/FlowDataValidator.java +++ b/src/main/java/it/cnr/isti/workflow/manager/flows/validation/FlowDataValidator.java @@ -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 llmProviders; + @Autowired + MCPServersProvider mcpServersProvider; + @Override public boolean isValid(FlowData flowData, ConstraintValidatorContext context) { if (flowData == null) { @@ -198,6 +203,7 @@ public class FlowDataValidator implements ConstraintValidator block) { + if (!(block.getSpecificConfiguration() instanceof LLMBlockConfiguration configuration)) { + return; + } + List 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 llmDescriptorsOf(Block block) { Object configuration = block.getSpecificConfiguration(); if (configuration instanceof LLMBlockConfiguration llmConfiguration) { diff --git a/src/main/java/it/cnr/isti/workflow/manager/flows/validation/ValidationErrorCode.java b/src/main/java/it/cnr/isti/workflow/manager/flows/validation/ValidationErrorCode.java index 1e92f5d..e1ab6fb 100644 --- a/src/main/java/it/cnr/isti/workflow/manager/flows/validation/ValidationErrorCode.java +++ b/src/main/java/it/cnr/isti/workflow/manager/flows/validation/ValidationErrorCode.java @@ -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 } diff --git a/src/main/java/it/cnr/isti/workflow/manager/llms/ChatMessage.java b/src/main/java/it/cnr/isti/workflow/manager/llms/ChatMessage.java index 65dad4c..f7648ff 100644 --- a/src/main/java/it/cnr/isti/workflow/manager/llms/ChatMessage.java +++ b/src/main/java/it/cnr/isti/workflow/manager/llms/ChatMessage.java @@ -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. + * + *

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 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 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(); } } diff --git a/src/main/java/it/cnr/isti/workflow/manager/llms/LLMProviderCatalogService.java b/src/main/java/it/cnr/isti/workflow/manager/llms/LLMProviderCatalogService.java index 3f704fa..c8dc7e7 100644 --- a/src/main/java/it/cnr/isti/workflow/manager/llms/LLMProviderCatalogService.java +++ b/src/main/java/it/cnr/isti/workflow/manager/llms/LLMProviderCatalogService.java @@ -24,7 +24,7 @@ public class LLMProviderCatalogService { public List list() { return providers.values().stream() .map(provider -> new LLMProviderMetadata(provider.getName(), provider.requiresAuthorization(), - provider.requiresEndpoint(), + provider.requiresEndpoint(), provider.supportsTools(), provider.supportedParameters().stream().map(Enum::name).sorted().toList())) .filter(provider -> StringUtils.hasText(provider.name())) .distinct() diff --git a/src/main/java/it/cnr/isti/workflow/manager/llms/LLMProviderMetadata.java b/src/main/java/it/cnr/isti/workflow/manager/llms/LLMProviderMetadata.java index 3c7c66c..eeab7f4 100644 --- a/src/main/java/it/cnr/isti/workflow/manager/llms/LLMProviderMetadata.java +++ b/src/main/java/it/cnr/isti/workflow/manager/llms/LLMProviderMetadata.java @@ -11,13 +11,16 @@ import java.util.List; * * @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()}. * @param supportedParameters which sampling knobs this provider actually applies, by * {@link ModelParameter} name. Every provider is offered the same five, so * without this the editor lets a value be set where it does nothing and the * run only says so afterwards, in a warning nobody was waiting for. */ public record LLMProviderMetadata(String name, boolean requiresCredential, boolean requiresEndpoint, - List supportedParameters) { + boolean supportsTools, List supportedParameters) { public LLMProviderMetadata { supportedParameters = supportedParameters == null ? List.of() : List.copyOf(supportedParameters); diff --git a/src/main/java/it/cnr/isti/workflow/manager/llms/ToolCall.java b/src/main/java/it/cnr/isti/workflow/manager/llms/ToolCall.java new file mode 100644 index 0000000..72b54db --- /dev/null +++ b/src/main/java/it/cnr/isti/workflow/manager/llms/ToolCall.java @@ -0,0 +1,36 @@ +// SPDX-FileCopyrightText: 2025-2026 Lucio Lelii - 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. + * + *

{@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. + * + *

{@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); + } +} diff --git a/src/main/java/it/cnr/isti/workflow/manager/llms/ToolChatResult.java b/src/main/java/it/cnr/isti/workflow/manager/llms/ToolChatResult.java new file mode 100644 index 0000000..13e9196 --- /dev/null +++ b/src/main/java/it/cnr/isti/workflow/manager/llms/ToolChatResult.java @@ -0,0 +1,52 @@ +// SPDX-FileCopyrightText: 2025-2026 Lucio Lelii - 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. + * + *

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 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 toolCalls) { + return new ToolChatResult(content, toolCalls, null); + } + + public static ToolChatResult toolCalls(String content, List toolCalls, String finishReason) { + return new ToolChatResult(content, toolCalls, finishReason); + } + + public boolean hasToolCalls() { + return !toolCalls.isEmpty(); + } +} diff --git a/src/main/java/it/cnr/isti/workflow/manager/llms/ToolDefinition.java b/src/main/java/it/cnr/isti/workflow/manager/llms/ToolDefinition.java new file mode 100644 index 0000000..4577997 --- /dev/null +++ b/src/main/java/it/cnr/isti/workflow/manager/llms/ToolDefinition.java @@ -0,0 +1,38 @@ +// SPDX-FileCopyrightText: 2025-2026 Lucio Lelii - 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. + * + *

{@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; + } +} diff --git a/src/main/java/it/cnr/isti/workflow/manager/llms/providers/LLMProvider.java b/src/main/java/it/cnr/isti/workflow/manager/llms/providers/LLMProvider.java index c7f314d..1191917 100644 --- a/src/main/java/it/cnr/isti/workflow/manager/llms/providers/LLMProvider.java +++ b/src/main/java/it/cnr/isti/workflow/manager/llms/providers/LLMProvider.java @@ -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. + * + *

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. + * + *

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 messages, + List 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 diff --git a/src/main/java/it/cnr/isti/workflow/manager/llms/providers/google/GeminiLLMProvider.java b/src/main/java/it/cnr/isti/workflow/manager/llms/providers/google/GeminiLLMProvider.java index e277dbc..d3a71c8 100644 --- a/src/main/java/it/cnr/isti/workflow/manager/llms/providers/google/GeminiLLMProvider.java +++ b/src/main/java/it/cnr/isti/workflow/manager/llms/providers/google/GeminiLLMProvider.java @@ -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()); }; } diff --git a/src/main/java/it/cnr/isti/workflow/manager/llms/providers/ollama/OllamaProtocolProvider.java b/src/main/java/it/cnr/isti/workflow/manager/llms/providers/ollama/OllamaProtocolProvider.java index e525e37..6004764 100644 --- a/src/main/java/it/cnr/isti/workflow/manager/llms/providers/ollama/OllamaProtocolProvider.java +++ b/src/main/java/it/cnr/isti/workflow/manager/llms/providers/ollama/OllamaProtocolProvider.java @@ -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 messages, List tools, + ProviderCredential credential, ModelParameters parameters) { + return parseChatResult(chatCallRaw(model, messages, tools, credential, parameters)); + } + private String chatCall(String model, List messages, ProviderCredential credential, ModelParameters parameters) { + return parseChatResponse(chatCallRaw(model, messages, List.of(), credential, parameters)); + } + + private String chatCallRaw(String model, List messages, List tools, + ProviderCredential credential, ModelParameters parameters) { Objects.requireNonNull(model, "model cannot be null"); Objects.requireNonNull(messages, "messages cannot be null"); - Map body = buildChatBody(model, messages, parameters); + Map body = buildChatBody(model, messages, parameters, tools); Mono 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 buildChatBody(String model, List messages, ModelParameters parameters) { + return buildChatBody(model, messages, parameters, List.of()); + } + + /** + * The same body, plus the tools the model may call. + * + *

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 buildChatBody(String model, List messages, ModelParameters parameters, + List tools) { Map 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 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 toOllamaMessage(ChatMessage message) { + Map 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. + * + *

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 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 toToolCalls(ChatResponseMessage message) { + if (message == null || message.getToolCalls() == null) { + return List.of(); + } + List 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); } } diff --git a/src/main/java/it/cnr/isti/workflow/manager/llms/providers/ollama/response/ChatResponse.java b/src/main/java/it/cnr/isti/workflow/manager/llms/providers/ollama/response/ChatResponse.java index 24cb300..11a47c1 100644 --- a/src/main/java/it/cnr/isti/workflow/manager/llms/providers/ollama/response/ChatResponse.java +++ b/src/main/java/it/cnr/isti/workflow/manager/llms/providers/ollama/response/ChatResponse.java @@ -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; } diff --git a/src/main/java/it/cnr/isti/workflow/manager/llms/providers/ollama/response/ChatResponseMessage.java b/src/main/java/it/cnr/isti/workflow/manager/llms/providers/ollama/response/ChatResponseMessage.java index db301ec..1c1eeb2 100644 --- a/src/main/java/it/cnr/isti/workflow/manager/llms/providers/ollama/response/ChatResponseMessage.java +++ b/src/main/java/it/cnr/isti/workflow/manager/llms/providers/ollama/response/ChatResponseMessage.java @@ -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 toolCalls; } diff --git a/src/main/java/it/cnr/isti/workflow/manager/llms/providers/ollama/response/ToolCallFunction.java b/src/main/java/it/cnr/isti/workflow/manager/llms/providers/ollama/response/ToolCallFunction.java new file mode 100644 index 0000000..8841bb0 --- /dev/null +++ b/src/main/java/it/cnr/isti/workflow/manager/llms/providers/ollama/response/ToolCallFunction.java @@ -0,0 +1,27 @@ +// SPDX-FileCopyrightText: 2025-2026 Lucio Lelii - 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. + * + *

{@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; +} diff --git a/src/main/java/it/cnr/isti/workflow/manager/llms/providers/ollama/response/ToolCallResponse.java b/src/main/java/it/cnr/isti/workflow/manager/llms/providers/ollama/response/ToolCallResponse.java new file mode 100644 index 0000000..91ddb33 --- /dev/null +++ b/src/main/java/it/cnr/isti/workflow/manager/llms/providers/ollama/response/ToolCallResponse.java @@ -0,0 +1,18 @@ +// SPDX-FileCopyrightText: 2025-2026 Lucio Lelii - 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; +} diff --git a/src/main/java/it/cnr/isti/workflow/manager/llms/providers/openai/OpenAIProtocolProvider.java b/src/main/java/it/cnr/isti/workflow/manager/llms/providers/openai/OpenAIProtocolProvider.java index b971ea8..f2b3970 100644 --- a/src/main/java/it/cnr/isti/workflow/manager/llms/providers/openai/OpenAIProtocolProvider.java +++ b/src/main/java/it/cnr/isti/workflow/manager/llms/providers/openai/OpenAIProtocolProvider.java @@ -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()); }; } diff --git a/src/main/java/it/cnr/isti/workflow/manager/mcp/MCPAgentService.java b/src/main/java/it/cnr/isti/workflow/manager/mcp/MCPAgentService.java index a0e735e..005bc1f 100644 --- a/src/main/java/it/cnr/isti/workflow/manager/mcp/MCPAgentService.java +++ b/src/main/java/it/cnr/isti/workflow/manager/mcp/MCPAgentService.java @@ -748,19 +748,12 @@ public class MCPAgentService { private Map resolveRuntimeTemplates(Map values, Map templateValues, Map executionVariables) { - LinkedHashMap 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 templateValues, Map 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 resolveTemplatedObject(Map values, Map templateValues, diff --git a/src/main/java/it/cnr/isti/workflow/manager/mcp/MCPRuntimeTemplates.java b/src/main/java/it/cnr/isti/workflow/manager/mcp/MCPRuntimeTemplates.java new file mode 100644 index 0000000..ba69c2f --- /dev/null +++ b/src/main/java/it/cnr/isti/workflow/manager/mcp/MCPRuntimeTemplates.java @@ -0,0 +1,44 @@ +// SPDX-FileCopyrightText: 2025-2026 Lucio Lelii - 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. + * + *

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 resolveAll(Map values, Map templateValues, + Map executionVariables) { + LinkedHashMap resolved = new LinkedHashMap<>(); + values.forEach((key, value) -> resolved.put(key, resolve(value, templateValues, executionVariables))); + return resolved; + } + + public static String resolve(String template, Map templateValues, + Map executionVariables) { + if (!StringUtils.hasText(template)) { + return template; + } + return ExecutionTemplateResolver.resolve(template, + templateValues == null ? Map.of() : templateValues, + executionVariables == null ? Map.of() : executionVariables); + } +} diff --git a/src/main/java/it/cnr/isti/workflow/manager/mcp/MCPToolServerBinding.java b/src/main/java/it/cnr/isti/workflow/manager/mcp/MCPToolServerBinding.java new file mode 100644 index 0000000..8b1ada4 --- /dev/null +++ b/src/main/java/it/cnr/isti/workflow/manager/mcp/MCPToolServerBinding.java @@ -0,0 +1,49 @@ +// SPDX-FileCopyrightText: 2025-2026 Lucio Lelii - 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. + * + *

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. + * + *

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; +} diff --git a/src/main/java/it/cnr/isti/workflow/manager/mcp/client/MCPToolCallResult.java b/src/main/java/it/cnr/isti/workflow/manager/mcp/client/MCPToolCallResult.java new file mode 100644 index 0000000..f45d105 --- /dev/null +++ b/src/main/java/it/cnr/isti/workflow/manager/mcp/client/MCPToolCallResult.java @@ -0,0 +1,15 @@ +// SPDX-FileCopyrightText: 2025-2026 Lucio Lelii - 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. + * + *

{@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) { +} diff --git a/src/main/java/it/cnr/isti/workflow/manager/mcp/client/MCPToolResults.java b/src/main/java/it/cnr/isti/workflow/manager/mcp/client/MCPToolResults.java new file mode 100644 index 0000000..5ffdbd8 --- /dev/null +++ b/src/main/java/it/cnr/isti/workflow/manager/mcp/client/MCPToolResults.java @@ -0,0 +1,66 @@ +// SPDX-FileCopyrightText: 2025-2026 Lucio Lelii - 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 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]"; + } +} diff --git a/src/main/java/it/cnr/isti/workflow/manager/mcp/client/MCPToolSession.java b/src/main/java/it/cnr/isti/workflow/manager/mcp/client/MCPToolSession.java new file mode 100644 index 0000000..7924ebc --- /dev/null +++ b/src/main/java/it/cnr/isti/workflow/manager/mcp/client/MCPToolSession.java @@ -0,0 +1,276 @@ +// SPDX-FileCopyrightText: 2025-2026 Lucio Lelii - 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. + * + *

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 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 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 listTools() { + List 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>() {}) + .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) + "..."; + } +} diff --git a/src/main/java/it/cnr/isti/workflow/manager/mcp/client/MCPToolSessionFactory.java b/src/main/java/it/cnr/isti/workflow/manager/mcp/client/MCPToolSessionFactory.java new file mode 100644 index 0000000..e847522 --- /dev/null +++ b/src/main/java/it/cnr/isti/workflow/manager/mcp/client/MCPToolSessionFactory.java @@ -0,0 +1,95 @@ +// SPDX-FileCopyrightText: 2025-2026 Lucio Lelii - 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. + * + *

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> 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 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 templateValues = configuration == null || configuration.isNull() + ? Map.of() + : objectMapper.convertValue(configuration, STRING_OBJECT_MAP); + String resolvedUrl = MCPRuntimeTemplates.resolve(definition.url(), templateValues, executionVariables); + Map 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; + } +} diff --git a/src/main/java/it/cnr/isti/workflow/manager/mcp/client/MCPToolbox.java b/src/main/java/it/cnr/isti/workflow/manager/mcp/client/MCPToolbox.java new file mode 100644 index 0000000..fee0906 --- /dev/null +++ b/src/main/java/it/cnr/isti/workflow/manager/mcp/client/MCPToolbox.java @@ -0,0 +1,106 @@ +// SPDX-FileCopyrightText: 2025-2026 Lucio Lelii - 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. + * + *

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 sessions; + private final Map sessionsByTool; + private final List tools; + + private MCPToolbox(List sessions, Map sessionsByTool, + List tools) { + this.sessions = sessions; + this.sessionsByTool = sessionsByTool; + this.tools = tools; + } + + public static MCPToolbox open(List bindings, Map executionVariables, + MCPToolSessionFactory factory) { + List opened = new ArrayList<>(); + Map byTool = new LinkedHashMap<>(); + List 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 tools() { + return tools; + } + + public List 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 sessions) { + for (MCPToolSession session : sessions) { + try { + session.close(); + } catch (RuntimeException e) { + log.debug("Closing MCP session on {} failed: {}", session.serverId(), e.toString()); + } + } + } +} diff --git a/src/main/resources/application.properties b/src/main/resources/application.properties index 44633ae..39fb53b 100644 --- a/src/main/resources/application.properties +++ b/src/main/resources/application.properties @@ -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} diff --git a/src/main/resources/skills-catalog.json b/src/main/resources/skills-catalog.json index ec42231..50f621f 100644 --- a/src/main/resources/skills-catalog.json +++ b/src/main/resources/skills-catalog.json @@ -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" } ] } diff --git a/src/main/resources/skills/mcp-context-economy/SKILL.md b/src/main/resources/skills/mcp-context-economy/SKILL.md new file mode 100644 index 0000000..974d688 --- /dev/null +++ b/src/main/resources/skills/mcp-context-economy/SKILL.md @@ -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 diff --git a/src/test/java/it/cnr/isti/workflow/manager/controllers/RetrieverControllerTest.java b/src/test/java/it/cnr/isti/workflow/manager/controllers/RetrieverControllerTest.java index d87ed16..482361f 100644 --- a/src/test/java/it/cnr/isti/workflow/manager/controllers/RetrieverControllerTest.java +++ b/src/test/java/it/cnr/isti/workflow/manager/controllers/RetrieverControllerTest.java @@ -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 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 diff --git a/src/test/java/it/cnr/isti/workflow/manager/executions/executors/blocks/LLMToolLoopTest.java b/src/test/java/it/cnr/isti/workflow/manager/executions/executors/blocks/LLMToolLoopTest.java new file mode 100644 index 0000000..4431f30 --- /dev/null +++ b/src/test/java/it/cnr/isti/workflow/manager/executions/executors/blocks/LLMToolLoopTest.java @@ -0,0 +1,489 @@ +// SPDX-FileCopyrightText: 2025-2026 Lucio Lelii - 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 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 turns = new ArrayDeque<>(); + private final List> 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 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 messages, List 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 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 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 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 loggedTypes = new ArrayList<>(); + List> 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 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); + } + } + } +} diff --git a/src/test/java/it/cnr/isti/workflow/manager/llms/LLMProviderCatalogServiceTest.java b/src/test/java/it/cnr/isti/workflow/manager/llms/LLMProviderCatalogServiceTest.java index fb010fb..5fb9e8b 100644 --- a/src/test/java/it/cnr/isti/workflow/manager/llms/LLMProviderCatalogServiceTest.java +++ b/src/test/java/it/cnr/isti/workflow/manager/llms/LLMProviderCatalogServiceTest.java @@ -5,6 +5,8 @@ package it.cnr.isti.workflow.manager.llms; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; import java.util.EnumSet; import java.util.List; @@ -19,12 +21,12 @@ import it.cnr.isti.workflow.manager.llms.providers.LLMProvider; * What the editor is told about a provider before anyone runs anything. * *

The capabilities exist so a choice that cannot work is not offered: every provider is handed - * the same five sampling parameters, and the ones it ignores were only ever reported afterwards, in - * an execution warning. + * the same five sampling parameters and the same tool list, and the ones it ignores were only ever + * reported afterwards, in an execution warning. */ class LLMProviderCatalogServiceTest { - private static LLMProvider provider(String name, Set parameters) { + private static LLMProvider provider(String name, boolean tools, Set parameters) { return new LLMProvider() { @Override public String getName() { @@ -41,6 +43,11 @@ class LLMProviderCatalogServiceTest { throw new UnsupportedOperationException("not exercised by this test"); } + @Override + public boolean supportsTools() { + return tools; + } + @Override public Set supportedParameters() { return parameters; @@ -51,8 +58,8 @@ class LLMProviderCatalogServiceTest { @Test void reportsWhichKnobsEachProviderActuallyApplies() { LLMProviderCatalogService catalog = new LLMProviderCatalogService(Map.of( - "a", provider("Everything", EnumSet.allOf(ModelParameter.class)), - "b", provider("NoSeed", EnumSet.of(ModelParameter.TEMPERATURE, ModelParameter.TOP_K)))); + "a", provider("Everything", true, EnumSet.allOf(ModelParameter.class)), + "b", provider("NoSeed", false, EnumSet.of(ModelParameter.TEMPERATURE, ModelParameter.TOP_K)))); Map byName = catalog.list().stream() .collect(java.util.stream.Collectors.toMap(LLMProviderMetadata::name, metadata -> metadata)); @@ -60,5 +67,7 @@ class LLMProviderCatalogServiceTest { assertEquals(List.of("MAX_TOKENS", "SEED", "TEMPERATURE", "TOP_K", "TOP_P"), byName.get("Everything").supportedParameters(), "sorted, so the payload is stable"); assertEquals(List.of("TEMPERATURE", "TOP_K"), byName.get("NoSeed").supportedParameters()); + assertTrue(byName.get("Everything").supportsTools()); + assertFalse(byName.get("NoSeed").supportsTools()); } } diff --git a/src/test/java/it/cnr/isti/workflow/manager/llms/providers/ToolCallingDefaultsTest.java b/src/test/java/it/cnr/isti/workflow/manager/llms/providers/ToolCallingDefaultsTest.java new file mode 100644 index 0000000..4da26e0 --- /dev/null +++ b/src/test/java/it/cnr/isti/workflow/manager/llms/providers/ToolCallingDefaultsTest.java @@ -0,0 +1,87 @@ +// SPDX-FileCopyrightText: 2025-2026 Lucio Lelii - 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 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 messages, ProviderCredential credential, + ModelParameters parameters) { + return "chatted:" + messages.size(); + } + } + + private final LLMProvider textOnly = new TextOnlyProvider(); + private final List 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 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()); + } +} diff --git a/src/test/java/it/cnr/isti/workflow/manager/llms/providers/ollama/InternalOllamaLLMProviderBodyTest.java b/src/test/java/it/cnr/isti/workflow/manager/llms/providers/ollama/InternalOllamaLLMProviderBodyTest.java index afee65e..9b53b29 100644 --- a/src/test/java/it/cnr/isti/workflow/manager/llms/providers/ollama/InternalOllamaLLMProviderBodyTest.java +++ b/src/test/java/it/cnr/isti/workflow/manager/llms/providers/ollama/InternalOllamaLLMProviderBodyTest.java @@ -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 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 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 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 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")); + } + } diff --git a/src/test/java/it/cnr/isti/workflow/manager/llms/providers/ollama/InternalOllamaLLMProviderHttpTest.java b/src/test/java/it/cnr/isti/workflow/manager/llms/providers/ollama/InternalOllamaLLMProviderHttpTest.java index 4ec6a48..e05f1c0 100644 --- a/src/test/java/it/cnr/isti/workflow/manager/llms/providers/ollama/InternalOllamaLLMProviderHttpTest.java +++ b/src/test/java/it/cnr/isti/workflow/manager/llms/providers/ollama/InternalOllamaLLMProviderHttpTest.java @@ -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"); diff --git a/src/test/java/it/cnr/isti/workflow/manager/mcp/client/MCPToolSessionTest.java b/src/test/java/it/cnr/isti/workflow/manager/mcp/client/MCPToolSessionTest.java new file mode 100644 index 0000000..1c3bfb5 --- /dev/null +++ b/src/test/java/it/cnr/isti/workflow/manager/mcp/client/MCPToolSessionTest.java @@ -0,0 +1,278 @@ +// SPDX-FileCopyrightText: 2025-2026 Lucio Lelii - 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 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 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 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); + } + } +}