From 2217ea188a7b8072b8542132e4eaffb2cc9b4839 Mon Sep 17 00:00:00 2001 From: Lucio Lelii Date: Fri, 18 Sep 2026 21:22:42 +0200 Subject: [PATCH] Let an LLM node call tools directly, bypassing the MCP bridge The bridge coerces a tool argument from string to object whenever the string happens to be valid JSON, even though the model emits it correctly and the tool's own schema declares it as a string. The coercion happens inside the bridge, between the model's response and the downstream MCP call: verified by calling the model directly (arguments.content stays a string, byte for byte) and by comparing a JSON-valid value (rejected) against a malformed or plain text one (passed through untouched). The failing call never reaches the MCP server, and the agent's retries under a different encoding are what turn an unfinished operation into one that returns status: completed with the model's last preamble as if it were the answer. There is no fix available on our side for the bridge itself, so this removes it from the path instead. An LLMBlock can now bind MCP servers straight from the catalog and run its own tool-calling loop in this service: - LLMProvider gains chatWithTools/supportsTools; only OllamaProtocolProvider implements it for now. Tool arguments stay JsonNode end to end - never a string, never re-parsed - which is the one change that actually closes the bridge's bug rather than working around it. - A native streamable-http MCP client (mcp/client/) talks to a server without the bridge: initialize, tools/list, tools/call, session header handling, both response shapes the spec allows. - MCPToolServerBinding is a narrower binding than MCPAgent's, restricted to catalog servers reachable over streamable-http - the ones this service can call directly, not the stdio ones the bridge still hosts a process for. - LLMToolLoop runs the model/tool/model cycle with real budgets: a wall-clock deadline and iteration cap that fail the block explicitly rather than return a partial answer, and a character-based context budget that replaces older tool results with a placeholder once the conversation - plus the tool schemas sent on every call, which do not appear in the conversation but are not free either - grows past it. The iteration just completed is never pruned, and a result under ~500 characters is left alone: shrinking it would cost about as much as it saves. - Ollama's done_reason now travels back as ToolChatResult.finishReason, so a turn that answers nothing can say whether the model chose silence or num_predict cut it off mid-thought - two different problems with two different fixes, previously indistinguishable from the error alone. - A new skill, mcp-context-economy, carries the operating rules a real run against a 24-task plan exposed the hard way: write_file to create a file, apply_patch only to edit one that exists, and never read a file straight back after writing it or re-pull an already-inline document into the conversation - each halves the context a node needs for the same work. MCPAgent and the bridge are untouched: this is a second path, not a replacement, for the one transport (streamable-http) this service can reach without it. Co-Authored-By: Claude Opus 5 (1M context) --- .../assistant/BlockDraftNormalizer.java | 42 ++ .../configurations/LLMBlockConfiguration.java | 33 +- .../retrievers/MCPServersFieldRetriever.java | 8 + .../executions/ExecutionEventType.java | 2 + .../executors/blocks/LLMExecutor.java | 12 + .../executors/blocks/LLMToolLoop.java | 321 ++++++++++++ .../flows/validation/FlowDataValidator.java | 55 ++ .../flows/validation/ValidationErrorCode.java | 5 +- .../workflow/manager/llms/ChatMessage.java | 51 +- .../llms/LLMProviderCatalogService.java | 2 +- .../manager/llms/LLMProviderMetadata.java | 6 +- .../isti/workflow/manager/llms/ToolCall.java | 36 ++ .../workflow/manager/llms/ToolChatResult.java | 52 ++ .../workflow/manager/llms/ToolDefinition.java | 38 ++ .../manager/llms/providers/LLMProvider.java | 31 ++ .../providers/google/GeminiLLMProvider.java | 5 + .../ollama/OllamaProtocolProvider.java | 136 ++++- .../ollama/response/ChatResponse.java | 6 + .../ollama/response/ChatResponseMessage.java | 7 + .../ollama/response/ToolCallFunction.java | 27 + .../ollama/response/ToolCallResponse.java | 18 + .../openai/OpenAIProtocolProvider.java | 7 + .../workflow/manager/mcp/MCPAgentService.java | 11 +- .../manager/mcp/MCPRuntimeTemplates.java | 44 ++ .../manager/mcp/MCPToolServerBinding.java | 49 ++ .../manager/mcp/client/MCPToolCallResult.java | 15 + .../manager/mcp/client/MCPToolResults.java | 66 +++ .../manager/mcp/client/MCPToolSession.java | 276 ++++++++++ .../mcp/client/MCPToolSessionFactory.java | 95 ++++ .../manager/mcp/client/MCPToolbox.java | 106 ++++ src/main/resources/application.properties | 25 + src/main/resources/skills-catalog.json | 6 + .../skills/mcp-context-economy/SKILL.md | 8 + .../controllers/RetrieverControllerTest.java | 12 + .../executors/blocks/LLMToolLoopTest.java | 489 ++++++++++++++++++ .../llms/LLMProviderCatalogServiceTest.java | 62 +++ .../providers/ToolCallingDefaultsTest.java | 87 ++++ .../InternalOllamaLLMProviderBodyTest.java | 60 +++ .../InternalOllamaLLMProviderHttpTest.java | 85 +++ .../mcp/client/MCPToolSessionTest.java | 278 ++++++++++ 40 files changed, 2640 insertions(+), 34 deletions(-) create mode 100644 src/main/java/it/cnr/isti/workflow/manager/executions/executors/blocks/LLMToolLoop.java create mode 100644 src/main/java/it/cnr/isti/workflow/manager/llms/ToolCall.java create mode 100644 src/main/java/it/cnr/isti/workflow/manager/llms/ToolChatResult.java create mode 100644 src/main/java/it/cnr/isti/workflow/manager/llms/ToolDefinition.java create mode 100644 src/main/java/it/cnr/isti/workflow/manager/llms/providers/ollama/response/ToolCallFunction.java create mode 100644 src/main/java/it/cnr/isti/workflow/manager/llms/providers/ollama/response/ToolCallResponse.java create mode 100644 src/main/java/it/cnr/isti/workflow/manager/mcp/MCPRuntimeTemplates.java create mode 100644 src/main/java/it/cnr/isti/workflow/manager/mcp/MCPToolServerBinding.java create mode 100644 src/main/java/it/cnr/isti/workflow/manager/mcp/client/MCPToolCallResult.java create mode 100644 src/main/java/it/cnr/isti/workflow/manager/mcp/client/MCPToolResults.java create mode 100644 src/main/java/it/cnr/isti/workflow/manager/mcp/client/MCPToolSession.java create mode 100644 src/main/java/it/cnr/isti/workflow/manager/mcp/client/MCPToolSessionFactory.java create mode 100644 src/main/java/it/cnr/isti/workflow/manager/mcp/client/MCPToolbox.java create mode 100644 src/main/resources/skills/mcp-context-economy/SKILL.md create mode 100644 src/test/java/it/cnr/isti/workflow/manager/executions/executors/blocks/LLMToolLoopTest.java create mode 100644 src/test/java/it/cnr/isti/workflow/manager/llms/LLMProviderCatalogServiceTest.java create mode 100644 src/test/java/it/cnr/isti/workflow/manager/llms/providers/ToolCallingDefaultsTest.java create mode 100644 src/test/java/it/cnr/isti/workflow/manager/mcp/client/MCPToolSessionTest.java 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 4ba8699..c0992fd 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())) .filter(provider -> StringUtils.hasText(provider.name())) .distinct() .sorted(java.util.Comparator.comparing(LLMProviderMetadata::name, String.CASE_INSENSITIVE_ORDER)) 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 c9db54f..3062032 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 @@ -9,6 +9,10 @@ package it.cnr.isti.workflow.manager.llms; * * @param requiresEndpoint whether choosing this provider means the credential must also carry a * base URL - see {@link it.cnr.isti.workflow.manager.llms.providers.LLMProvider#requiresEndpoint()}. + * @param supportsTools whether this provider can be given tools to call, which is what decides + * if a node may bind an MCP server to it - see + * {@link it.cnr.isti.workflow.manager.llms.providers.LLMProvider#supportsTools()}. */ -public record LLMProviderMetadata(String name, boolean requiresCredential, boolean requiresEndpoint) { +public record LLMProviderMetadata(String name, boolean requiresCredential, boolean requiresEndpoint, + boolean supportsTools) { } 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 new file mode 100644 index 0000000..a0b38a7 --- /dev/null +++ b/src/test/java/it/cnr/isti/workflow/manager/llms/LLMProviderCatalogServiceTest.java @@ -0,0 +1,62 @@ +// 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 static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.List; +import java.util.Map; + +import org.junit.jupiter.api.Test; + +import it.cnr.isti.workflow.manager.llms.providers.LLMProvider; + +/** + * What the editor is told about a provider before anyone runs anything. + * + *

The capability exists so a choice that cannot work is not offered: a node cannot bind an MCP + * server to a provider that never supports tools, and the editor needs to know that before the flow + * is even saved, not from an execution failure afterwards. + */ +class LLMProviderCatalogServiceTest { + + private static LLMProvider provider(String name, boolean tools) { + return new LLMProvider() { + @Override + public String getName() { + return name; + } + + @Override + public List getRegisteredModels() { + return List.of(); + } + + @Override + public String generate(String model, String prompt) { + throw new UnsupportedOperationException("not exercised by this test"); + } + + @Override + public boolean supportsTools() { + return tools; + } + }; + } + + @Test + void reportsWhichProvidersCanCallTools() { + LLMProviderCatalogService catalog = new LLMProviderCatalogService(Map.of( + "a", provider("WithTools", true), + "b", provider("NoTools", false))); + + Map byName = catalog.list().stream() + .collect(java.util.stream.Collectors.toMap(LLMProviderMetadata::name, metadata -> metadata)); + + assertTrue(byName.get("WithTools").supportsTools()); + assertFalse(byName.get("NoTools").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); + } + } +}