Merge feature/mcp-tools-in-llm-nodes into main

# Conflicts:
#	src/main/java/it/cnr/isti/workflow/manager/llms/LLMProviderCatalogService.java
#	src/main/java/it/cnr/isti/workflow/manager/llms/LLMProviderMetadata.java
#	src/test/java/it/cnr/isti/workflow/manager/llms/LLMProviderCatalogServiceTest.java
This commit is contained in:
Lucio Lelii 2026-09-18 21:27:12 +02:00
commit c044d78a5a
40 changed files with 2591 additions and 39 deletions

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

@ -11,13 +11,16 @@ import java.util.List;
*
* @param requiresEndpoint whether choosing this provider means the credential must also carry a
* base URL - see {@link it.cnr.isti.workflow.manager.llms.providers.LLMProvider#requiresEndpoint()}.
* @param supportsTools whether this provider can be given tools to call, which is what decides
* if a node may bind an MCP server to it - see
* {@link it.cnr.isti.workflow.manager.llms.providers.LLMProvider#supportsTools()}.
* @param supportedParameters which sampling knobs this provider actually applies, by
* {@link ModelParameter} name. Every provider is offered the same five, so
* without this the editor lets a value be set where it does nothing and the
* run only says so afterwards, in a warning nobody was waiting for.
*/
public record LLMProviderMetadata(String name, boolean requiresCredential, boolean requiresEndpoint,
List<String> supportedParameters) {
boolean supportsTools, List<String> supportedParameters) {
public LLMProviderMetadata {
supportedParameters = supportedParameters == null ? List.of() : List.copyOf(supportedParameters);

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

@ -5,6 +5,8 @@
package it.cnr.isti.workflow.manager.llms;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
import java.util.EnumSet;
import java.util.List;
@ -19,12 +21,12 @@ import it.cnr.isti.workflow.manager.llms.providers.LLMProvider;
* What the editor is told about a provider before anyone runs anything.
*
* <p>The capabilities exist so a choice that cannot work is not offered: every provider is handed
* the same five sampling parameters, and the ones it ignores were only ever reported afterwards, in
* an execution warning.
* the same five sampling parameters and the same tool list, and the ones it ignores were only ever
* reported afterwards, in an execution warning.
*/
class LLMProviderCatalogServiceTest {
private static LLMProvider provider(String name, Set<ModelParameter> parameters) {
private static LLMProvider provider(String name, boolean tools, Set<ModelParameter> parameters) {
return new LLMProvider() {
@Override
public String getName() {
@ -41,6 +43,11 @@ class LLMProviderCatalogServiceTest {
throw new UnsupportedOperationException("not exercised by this test");
}
@Override
public boolean supportsTools() {
return tools;
}
@Override
public Set<ModelParameter> supportedParameters() {
return parameters;
@ -51,8 +58,8 @@ class LLMProviderCatalogServiceTest {
@Test
void reportsWhichKnobsEachProviderActuallyApplies() {
LLMProviderCatalogService catalog = new LLMProviderCatalogService(Map.of(
"a", provider("Everything", EnumSet.allOf(ModelParameter.class)),
"b", provider("NoSeed", EnumSet.of(ModelParameter.TEMPERATURE, ModelParameter.TOP_K))));
"a", provider("Everything", true, EnumSet.allOf(ModelParameter.class)),
"b", provider("NoSeed", false, EnumSet.of(ModelParameter.TEMPERATURE, ModelParameter.TOP_K))));
Map<String, LLMProviderMetadata> byName = catalog.list().stream()
.collect(java.util.stream.Collectors.toMap(LLMProviderMetadata::name, metadata -> metadata));
@ -60,5 +67,7 @@ class LLMProviderCatalogServiceTest {
assertEquals(List.of("MAX_TOKENS", "SEED", "TEMPERATURE", "TOP_K", "TOP_P"),
byName.get("Everything").supportedParameters(), "sorted, so the payload is stable");
assertEquals(List.of("TEMPERATURE", "TOP_K"), byName.get("NoSeed").supportedParameters());
assertTrue(byName.get("Everything").supportsTools());
assertFalse(byName.get("NoSeed").supportsTools());
}
}

View File

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

View File

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

View File

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

View File

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