diff --git a/bom/camel-bom/pom.xml b/bom/camel-bom/pom.xml index 27a53b9de811f..ea5383026c763 100644 --- a/bom/camel-bom/pom.xml +++ b/bom/camel-bom/pom.xml @@ -1667,6 +1667,16 @@ camel-master 4.22.0-SNAPSHOT + + org.apache.camel + camel-mcp-server + 4.22.0-SNAPSHOT + + + org.apache.camel + camel-mcp-server-api + 4.22.0-SNAPSHOT + org.apache.camel camel-mdc diff --git a/catalog/camel-allcomponents/pom.xml b/catalog/camel-allcomponents/pom.xml index 670920427b8f9..9ff95f7750ff9 100644 --- a/catalog/camel-allcomponents/pom.xml +++ b/catalog/camel-allcomponents/pom.xml @@ -1457,6 +1457,16 @@ camel-master ${project.version} + + org.apache.camel + camel-mcp-server + ${project.version} + + + org.apache.camel + camel-mcp-server-api + ${project.version} + org.apache.camel camel-mdc diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs.properties b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs.properties index a2ec5a1dc67f4..36ae1f24a7418 100644 --- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs.properties +++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs.properties @@ -386,6 +386,7 @@ main mapstruct-component marshal-eip master-component +mcp-server mdc message message-broker diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/mcp-server.adoc b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/mcp-server.adoc new file mode 100644 index 0000000000000..6b28142ca981a --- /dev/null +++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/mcp-server.adoc @@ -0,0 +1,210 @@ += MCP Server Component +:doctitle: MCP Server +:shortname: mcp-server +:artifactid: camel-mcp-server +:description: Expose ai-tool routes as MCP tools over streamable HTTP +:since: 4.22 +:supportlevel: Preview +:tabs-sync-option: + +*Since Camel {since}* + +The camel-mcp-server module exposes Camel routes registered via the +xref:ROOT:ai-tool-component.adoc[ai-tool] component as tools of a +https://modelcontextprotocol.io[Model Context Protocol] (MCP) server, served +over MCP streamable HTTP. No route is needed for the server itself: add the +dependency, configure which tags to expose, and every matching `ai-tool` route +becomes an MCP tool that any MCP client (another Camel application, an IDE, a +coding agent) can discover and call. + +Maven users will need to add the following dependency to their `pom.xml`: + +[source,xml] +---- + + org.apache.camel + camel-mcp-server + x.x.x + + +---- + +== Architecture + +The module is split in two artifacts: + +* `camel-mcp-server-api` — the runtime-agnostic _bridge_ and the small + `McpServerEngine` SPI. The bridge owns tool selection (tags), execution via + the shared `AiToolExecutor` (per-call timeout, error sanitization) and reacts + to `AiToolRegistry` changes when routes start and stop. It has no dependency + on the MCP Java SDK. +* `camel-mcp-server` — the serving engine for Camel Main and Camel JBang, + built on the official MCP Java SDK with a Vert.x streamable HTTP transport. + The MCP endpoint is registered on the Camel main HTTP server's router, so it + serves on the main server port (`camel.server.port`) and inherits its + lifecycle, authentication and CORS configuration. + +Engine resolution mirrors the platform-http engine: a bean of type +`McpServerEngine` in the Camel registry wins; otherwise the engine is +discovered on the classpath. Other runtimes plug native engines through the +same SPI: on Quarkus the `camel-quarkus-mcp-server` extension serves through +the Quarkiverse `quarkus-mcp-server` (configured via `quarkus.mcp.server.*`), +and on Spring Boot the starter serves through the Spring AI MCP server +(configured via `spring.ai.mcp.server.*`). Bridge behavior — tag selection, +timeout, sanitization — is identical on every runtime and verified by a shared +conformance test kit. + +== Usage + +Define tools as regular `ai-tool` routes and give them tags: + +[tabs] +==== +Java:: ++ +[source,java] +---- +from("ai-tool:query_db?tags=crm" + + "&description=Query customer database" + + "¶meter.customerId=string" + + "¶meter.customerId.description=The customer id" + + "¶meter.customerId.required=true") + .to("jdbc:dataSource"); +---- + +XML:: ++ +[source,xml] +---- + + + + +---- + +YAML:: ++ +[source,yaml] +---- +- route: + from: + uri: ai-tool:query_db + parameters: + tags: crm + description: "Query customer database" + parameter.customerId: string + parameter.customerId.description: "The customer id" + parameter.customerId.required: "true" + steps: + - to: + uri: jdbc:dataSource +---- +==== + +Start the MCP server by adding the `McpServerBridge` service to the +CamelContext, selecting the tags to expose: + +[source,java] +---- +McpServerConfiguration configuration = new McpServerConfiguration(); +configuration.setTags("crm,notify"); +camelContext.addService(new McpServerBridge(configuration)); +---- + +The MCP endpoint is then served at `http://:/mcp` on the Camel +main HTTP server. Any MCP client can connect over streamable HTTP, for +example another Camel integration using the +xref:ROOT:openai-component.adoc[camel-openai] MCP client: + +[source,java] +---- +from("direct:agent") + .to("openai:chat-completion" + + "?model={{llm.model}}" + + "&autoToolExecution=true" + + "&mcpServer.myCamelTools.transportType=streamableHttp" + + "&mcpServer.myCamelTools.url=http://localhost:8080/mcp"); +---- + +== Options + +The `McpServerConfiguration` options: + +[width="100%",cols="2,5,2,1",options="header"] +|=== +| Option | Description | Default | Owner + +| `tags` | Comma-separated list of ai-tool tags to expose as MCP tools. Only + tools registered under one of these tags are published; the untagged + default pool is never exposed. When not set, no tools are published. | | + bridge +| `toolTimeout` | Per-call tool execution timeout in milliseconds. A call + exceeding the timeout returns an error result to the MCP client; the + underlying route keeps running until it completes on its own. | `20000` | + bridge +| `path` | HTTP path where the MCP endpoint is served. | `/mcp` | engine +| `serverName` | MCP server name advertised to clients. | CamelContext name | + engine +|=== + +Bridge-owned options are honored identically on every runtime. Engine-owned +options are consumed by the Vert.x engine only; on runtimes with a native +engine (Quarkus, Spring Boot) the native configuration decides serving +concerns and a startup WARN is logged when an ignored option is set. + +== Protocol + +This section describes the Vert.x engine shipped in `camel-mcp-server`, which +serves on Camel Main and Camel JBang. On Quarkus and Spring Boot the transport +is owned by the native engine instead — quarkus-mcp-server and the Spring Boot +embedded HTTP server (Spring AI MCP server) respectively — and the details +below do not apply. + +The Vert.x engine implements the MCP streamable HTTP transport: + +* `POST /mcp` answering `application/json` or `text/event-stream` depending on + the request, +* a long-lived `GET /mcp` SSE channel for server notifications, with + `Last-Event-ID` replay, +* session management via the `Mcp-Session-Id` header and `DELETE /mcp` for + session termination. + +Tools appearing or disappearing (routes starting and stopping) emit +`notifications/tools/list_changed` to connected clients. + +== Security + +External MCP clients are *untrusted senders* under the +xref:manual::security-model.adoc[Camel security model]. The module applies the +following rules: + +* *Explicit opt-in per tool*: only tools whose tags intersect the configured + `tags` are exposed. The untagged default pool is never exposed implicitly. +* *Flat namespace protection*: a tool whose name collides with an already + exposed tool is refused with an ERROR log — never silently replaced. +* *Error sanitization*: route exceptions are mapped to a generic error + message; the cause is logged server-side and never sent to the client. + Argument validation messages (missing or invalid parameters) are returned + as-is. +* *Bounded execution*: every call is subject to the `toolTimeout`. Note that a + timed-out route keeps running server-side until it completes; the timeout + bounds the MCP request, not the route. +* *Authentication*: the MCP endpoint is served through the main HTTP server + router, so platform-http authentication (basic, JWT via + `camel.server.authentication*` options) applies to it. The MCP + specification's authorization model is OAuth 2.1; see + xref:oauth.adoc[camel-oauth] for resource-server style + protection. On Quarkus and Spring Boot, authentication is owned by the + native runtime security. + +== Runtime notes + +* *Camel Main / JBang*: requires the Camel main HTTP server + (`camel.server.enabled=true` with `camel-platform-http-main`, automatic + with Camel JBang) or a `VertxPlatformHttpServer` service. Serving is fully + asynchronous: tool calls are offloaded to the Vert.x worker pool and the + long-lived SSE channel does not occupy a worker thread. +* *Quarkus*: use the `camel-quarkus-mcp-server` extension (serves through + quarkus-mcp-server; the MCP Java SDK is not on the classpath). +* *Spring Boot*: use the `camel-mcp-server-starter` (serves through the + Spring AI MCP server). diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/others.properties b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/others.properties index a1f054a312683..e1c80fb219286 100644 --- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/others.properties +++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/others.properties @@ -28,6 +28,7 @@ lra mail-microsoft-oauth main management +mcp-server mdc micrometer-observability micrometer-prometheus diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/others/mcp-server.json b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/others/mcp-server.json new file mode 100644 index 0000000000000..b1eea4c1ff42d --- /dev/null +++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/others/mcp-server.json @@ -0,0 +1,15 @@ +{ + "other": { + "kind": "other", + "name": "mcp-server", + "title": "MCP Server", + "description": "Expose ai-tool routes as MCP tools over streamable HTTP", + "deprecated": false, + "firstVersion": "4.22.0", + "label": "ai", + "supportLevel": "Preview", + "groupId": "org.apache.camel", + "artifactId": "camel-mcp-server", + "version": "4.22.0-SNAPSHOT" + } +} diff --git a/components/camel-ai/camel-mcp-server-api/pom.xml b/components/camel-ai/camel-mcp-server-api/pom.xml new file mode 100644 index 0000000000000..08aa1a653fdd9 --- /dev/null +++ b/components/camel-ai/camel-mcp-server-api/pom.xml @@ -0,0 +1,135 @@ + + + + 4.0.0 + + + org.apache.camel + camel-ai-parent + 4.22.0-SNAPSHOT + + + camel-mcp-server-api + jar + Camel :: AI :: MCP Server API + Runtime-agnostic bridge and engine SPI to expose ai-tool routes as MCP tools + + + 4.22.0 + Preview + + + + + + org.apache.camel + camel-support + + + org.apache.camel + camel-ai-tool + + + + + org.apache.camel + camel-test-junit6 + test + + + + io.modelcontextprotocol.sdk + mcp-core + ${mcp-java-sdk-version} + test + + + io.modelcontextprotocol.sdk + mcp-json-jackson2 + ${mcp-java-sdk-version} + test + + + org.awaitility + awaitility + ${awaitility-version} + test + + + org.assertj + assertj-core + test + + + + + + + + + maven-jar-plugin + + + + test-jar + + + + + + log4j2.properties + + + + + + org.apache.maven.plugins + maven-enforcer-plugin + + + ban-engine-dependencies + + enforce + + + + + + io.modelcontextprotocol.sdk:*:*:*:compile + io.modelcontextprotocol.sdk:*:*:*:runtime + io.projectreactor:*:*:*:compile + io.projectreactor:*:*:*:runtime + io.vertx:*:*:*:compile + io.vertx:*:*:*:runtime + org.apache.camel:camel-platform-http:*:*:compile + org.apache.camel:camel-platform-http-vertx:*:*:compile + + + + + + + + + + + diff --git a/components/camel-ai/camel-mcp-server-api/src/generated/resources/META-INF/services/org/apache/camel/other.properties b/components/camel-ai/camel-mcp-server-api/src/generated/resources/META-INF/services/org/apache/camel/other.properties new file mode 100644 index 0000000000000..dbe36fb5a3d7d --- /dev/null +++ b/components/camel-ai/camel-mcp-server-api/src/generated/resources/META-INF/services/org/apache/camel/other.properties @@ -0,0 +1,7 @@ +# Generated by camel build tools - do NOT edit this file! +name=mcp-server-api +groupId=org.apache.camel +artifactId=camel-mcp-server-api +version=4.22.0-SNAPSHOT +projectName=Camel :: AI :: MCP Server API +projectDescription=Runtime-agnostic bridge and engine SPI to expose ai-tool routes as MCP tools diff --git a/components/camel-ai/camel-mcp-server-api/src/generated/resources/mcp-server-api.json b/components/camel-ai/camel-mcp-server-api/src/generated/resources/mcp-server-api.json new file mode 100644 index 0000000000000..420142538f934 --- /dev/null +++ b/components/camel-ai/camel-mcp-server-api/src/generated/resources/mcp-server-api.json @@ -0,0 +1,14 @@ +{ + "other": { + "kind": "other", + "name": "mcp-server-api", + "title": "Mcp Server Api", + "description": "Runtime-agnostic bridge and engine SPI to expose ai-tool routes as MCP tools", + "deprecated": false, + "firstVersion": "4.22.0", + "supportLevel": "Preview", + "groupId": "org.apache.camel", + "artifactId": "camel-mcp-server-api", + "version": "4.22.0-SNAPSHOT" + } +} diff --git a/components/camel-ai/camel-mcp-server-api/src/main/java/org/apache/camel/component/mcp/server/McpServerBridge.java b/components/camel-ai/camel-mcp-server-api/src/main/java/org/apache/camel/component/mcp/server/McpServerBridge.java new file mode 100644 index 0000000000000..c3021ab31007b --- /dev/null +++ b/components/camel-ai/camel-mcp-server-api/src/main/java/org/apache/camel/component/mcp/server/McpServerBridge.java @@ -0,0 +1,308 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.mcp.server; + +import java.util.Arrays; +import java.util.HashMap; +import java.util.LinkedHashSet; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.locks.ReentrantLock; + +import org.apache.camel.CamelContext; +import org.apache.camel.CamelContextAware; +import org.apache.camel.Exchange; +import org.apache.camel.StaticService; +import org.apache.camel.component.ai.tool.AiToolExecutor; +import org.apache.camel.component.ai.tool.AiToolParameterHelper; +import org.apache.camel.component.ai.tool.AiToolParameterHelper.ParameterDef; +import org.apache.camel.component.ai.tool.AiToolRegistry; +import org.apache.camel.component.ai.tool.AiToolRegistryListener; +import org.apache.camel.component.ai.tool.AiToolResult; +import org.apache.camel.component.ai.tool.AiToolSpec; +import org.apache.camel.support.ResolverHelper; +import org.apache.camel.support.service.ServiceHelper; +import org.apache.camel.support.service.ServiceSupport; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Bridges the {@link AiToolRegistry} to an {@link McpServerEngine}: selects {@code ai-tool} routes by tag, publishes + * them as MCP tools, and executes calls with a bounded timeout and sanitized error mapping. + *

+ * Security notes: + *

    + *
  • Only tools whose tags intersect the configured {@code tags} are exposed. The untagged default pool is never + * exposed — external MCP clients are untrusted senders and crossing that trust boundary is an explicit per-tool + * opt-in.
    • + *
  • MCP has a flat tool namespace: a tool whose name collides with an already published tool is refused with an ERROR + * log, never silently replaced.
    • + *
  • Raw route exception messages never reach the engine: execution failures map to a generic error message and the + * cause is logged server-side.
    • + *
+ * + * @since 4.22 + */ +public class McpServerBridge extends ServiceSupport implements CamelContextAware, StaticService { + + private static final Logger LOG = LoggerFactory.getLogger(McpServerBridge.class); + + private static final String GENERIC_EXECUTION_ERROR = "Tool execution failed"; + private static final String GENERIC_TIMEOUT_ERROR = "Tool execution timed out"; + + private final McpServerConfiguration configuration; + private final RegistryListener listener = new RegistryListener(); + private final ReentrantLock lock = new ReentrantLock(); + private final Map published = new HashMap<>(); + + private CamelContext camelContext; + private McpServerEngine engine; + private AiToolRegistry registry; + private Set selectedTags = Set.of(); + private ExecutorService executor; + + public McpServerBridge(McpServerConfiguration configuration) { + this.configuration = configuration; + } + + @Override + public CamelContext getCamelContext() { + return camelContext; + } + + @Override + public void setCamelContext(CamelContext camelContext) { + this.camelContext = camelContext; + } + + public McpServerConfiguration getConfiguration() { + return configuration; + } + + public McpServerEngine getEngine() { + return engine; + } + + @Override + protected void doInit() throws Exception { + if (configuration.getTags() != null) { + selectedTags = new LinkedHashSet<>(Arrays.asList(AiToolParameterHelper.splitTags(configuration.getTags()))); + } + if (selectedTags.isEmpty()) { + LOG.warn("No MCP tags configured: no ai-tool routes will be exposed as MCP tools. " + + "Set tags to opt-in the tools to expose."); + } + + engine = resolveEngine(); + CamelContextAware.trySetCamelContext(engine, camelContext); + + String serverName = configuration.getServerName() != null ? configuration.getServerName() : camelContext.getName(); + engine.initialize(new McpServerInfo(serverName, camelContext.getVersion(), configuration.getPath())); + + if (!engine.consumesServingConfiguration()) { + if (!McpServerConstants.DEFAULT_PATH.equals(configuration.getPath())) { + LOG.warn("The MCP path option is ignored by engine {}: the runtime's native MCP server configuration " + + "decides the endpoint path", + engine.getClass().getSimpleName()); + } + if (configuration.getServerName() != null) { + LOG.warn("The MCP serverName option may be ignored by engine {}: the runtime's native MCP server " + + "configuration decides the server identity", + engine.getClass().getSimpleName()); + } + } + + ServiceHelper.initService(engine); + } + + @Override + protected void doStart() throws Exception { + executor = camelContext.getExecutorServiceManager().newCachedThreadPool(this, "McpServerToolCall"); + ServiceHelper.startService(engine); + + registry = AiToolRegistry.getOrCreate(camelContext); + // subscribe before snapshotting so no concurrent registration is missed; publishing is idempotent + registry.addListener(listener); + registry.getTools().forEach((tag, specs) -> { + if (selectedTags.contains(tag)) { + specs.forEach(this::publish); + } + }); + } + + @Override + protected void doStop() throws Exception { + if (registry != null) { + registry.removeListener(listener); + } + lock.lock(); + try { + published.clear(); + } finally { + lock.unlock(); + } + ServiceHelper.stopService(engine); + if (executor != null) { + camelContext.getExecutorServiceManager().shutdownGraceful(executor); + executor = null; + } + } + + private McpServerEngine resolveEngine() { + McpServerEngine answer = camelContext.getRegistry().findSingleByType(McpServerEngine.class); + if (answer == null) { + answer = ResolverHelper.resolveMandatoryService(camelContext, McpServerConstants.MCP_SERVER_ENGINE_FACTORY, + McpServerEngine.class, "camel-mcp-server"); + } + return answer; + } + + private void publish(AiToolSpec spec) { + // the engine is notified while holding the lock so publish/unpublish for the same tool cannot + // interleave between the map update and the engine call (which would orphan the tool in the engine) + lock.lock(); + try { + AiToolSpec existing = published.get(spec.getName()); + if (existing == spec) { + return; + } + if (existing != null) { + LOG.error("Refusing to expose MCP tool '{}': the name collides with an already exposed tool. " + + "MCP has a flat tool namespace - rename one of the ai-tool routes.", + spec.getName()); + return; + } + published.put(spec.getName(), spec); + engine.toolAdded(createTool(spec)); + } finally { + lock.unlock(); + } + } + + private void unpublish(AiToolSpec spec) { + lock.lock(); + try { + if (published.get(spec.getName()) != spec) { + return; + } + // the same spec may be registered under several selected tags; only remove when it is gone from all + boolean stillSelected = registry.getTools().entrySet().stream() + .anyMatch(e -> selectedTags.contains(e.getKey()) && e.getValue().contains(spec)); + if (!stillSelected) { + published.remove(spec.getName()); + engine.toolRemoved(spec.getName()); + } + } finally { + lock.unlock(); + } + } + + private McpServerTool createTool(AiToolSpec spec) { + McpToolCallHandler handler = arguments -> execute(spec, arguments); + return new McpServerTool() { + @Override + public String name() { + return spec.getName(); + } + + @Override + public String description() { + return spec.getDescription(); + } + + @Override + public String inputSchemaJson() { + return spec.getParametersJsonSchema(); + } + + @Override + public Map parameters() { + return spec.getParameterDefs(); + } + + @Override + public McpToolCallHandler handler() { + return handler; + } + }; + } + + private McpToolCallResult execute(AiToolSpec spec, Map arguments) { + Exchange exchange = spec.getConsumer().getEndpoint().createExchange(); + boolean release = true; + try { + Future future = executor.submit(() -> AiToolExecutor.execute(spec, arguments, exchange)); + AiToolResult result; + try { + result = future.get(configuration.getToolTimeout(), TimeUnit.MILLISECONDS); + } catch (TimeoutException e) { + future.cancel(true); + // the route may still be using the exchange; do not return it to the pool + release = false; + LOG.warn("MCP tool '{}' did not complete within {} ms; returning a timeout error to the client. " + + "The route keeps running until it completes on its own, and exchange {} is not returned " + + "to the pool.", + spec.getName(), configuration.getToolTimeout(), exchange.getExchangeId()); + return new McpToolCallResult(GENERIC_TIMEOUT_ERROR, true); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + release = false; + return new McpToolCallResult(GENERIC_EXECUTION_ERROR, true); + } catch (ExecutionException e) { + LOG.warn("MCP tool '{}' execution failed", spec.getName(), e.getCause()); + return new McpToolCallResult(GENERIC_EXECUTION_ERROR, true); + } + if (result instanceof AiToolResult.Success success) { + return new McpToolCallResult(success.value(), false); + } else if (result instanceof AiToolResult.ArgumentError error) { + return new McpToolCallResult(error.message(), true); + } else { + AiToolResult.ExecutionError error = (AiToolResult.ExecutionError) result; + // never leak raw route exception messages to remote MCP clients + LOG.warn("MCP tool '{}' execution failed: {}", spec.getName(), error.message(), error.cause()); + return new McpToolCallResult(GENERIC_EXECUTION_ERROR, true); + } + } finally { + if (release) { + spec.getConsumer().releaseExchange(exchange, false); + } + } + } + + private final class RegistryListener implements AiToolRegistryListener { + + @Override + public void toolRegistered(String tag, AiToolSpec spec) { + // the untagged default pool (tag == null) is never exposed + if (tag != null && selectedTags.contains(tag) && isStartingOrStarted()) { + publish(spec); + } + } + + @Override + public void toolDeregistered(String tag, AiToolSpec spec) { + if (tag != null && selectedTags.contains(tag) && isStartingOrStarted()) { + unpublish(spec); + } + } + } +} diff --git a/components/camel-ai/camel-mcp-server-api/src/main/java/org/apache/camel/component/mcp/server/McpServerConfiguration.java b/components/camel-ai/camel-mcp-server-api/src/main/java/org/apache/camel/component/mcp/server/McpServerConfiguration.java new file mode 100644 index 0000000000000..3439bf7db8de3 --- /dev/null +++ b/components/camel-ai/camel-mcp-server-api/src/main/java/org/apache/camel/component/mcp/server/McpServerConfiguration.java @@ -0,0 +1,81 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.mcp.server; + +/** + * Configuration for the {@link McpServerBridge}. + *

+ * Bridge-owned options ({@code tags}, {@code toolTimeout}) are honored on every runtime. Engine-owned options + * ({@code path}, {@code serverName}) are consumed only by engines that serve through Camel — native engines (Quarkus, + * Spring Boot) use their own runtime configuration instead. + * + * @since 4.22 + */ +public class McpServerConfiguration { + + private String tags; + private long toolTimeout = McpServerConstants.DEFAULT_TOOL_TIMEOUT; + private String path = McpServerConstants.DEFAULT_PATH; + private String serverName; + + /** + * Comma-separated list of ai-tool tags to expose as MCP tools. Only tools registered under one of these tags are + * published; the untagged default pool is never exposed. When not set, no tools are published. + */ + public String getTags() { + return tags; + } + + public void setTags(String tags) { + this.tags = tags; + } + + /** + * Per-call tool execution timeout in milliseconds. A call exceeding the timeout returns an error result to the MCP + * client; the underlying route keeps running until it completes on its own. + */ + public long getToolTimeout() { + return toolTimeout; + } + + public void setToolTimeout(long toolTimeout) { + this.toolTimeout = toolTimeout; + } + + /** + * HTTP path where the MCP endpoint is served. Engine-owned: ignored by native engines. + */ + public String getPath() { + return path; + } + + public void setPath(String path) { + this.path = path; + } + + /** + * MCP server name advertised to clients. Defaults to the CamelContext name. Engine-owned hint: native engines MAY + * ignore it. + */ + public String getServerName() { + return serverName; + } + + public void setServerName(String serverName) { + this.serverName = serverName; + } +} diff --git a/components/camel-ai/camel-mcp-server-api/src/main/java/org/apache/camel/component/mcp/server/McpServerConstants.java b/components/camel-ai/camel-mcp-server-api/src/main/java/org/apache/camel/component/mcp/server/McpServerConstants.java new file mode 100644 index 0000000000000..59ffd9b8b717c --- /dev/null +++ b/components/camel-ai/camel-mcp-server-api/src/main/java/org/apache/camel/component/mcp/server/McpServerConstants.java @@ -0,0 +1,44 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.mcp.server; + +/** + * Constants of the Camel MCP server. + * + * @since 4.22 + */ +public final class McpServerConstants { + + /** + * FactoryFinder key (under {@code META-INF/services/org/apache/camel/}) used to discover the + * {@link McpServerEngine} implementation on the classpath. + */ + public static final String MCP_SERVER_ENGINE_FACTORY = "mcp-server-engine"; + + /** + * Default HTTP path where the MCP endpoint is served by engines that consume the serving configuration. + */ + public static final String DEFAULT_PATH = "/mcp"; + + /** + * Default per-call tool execution timeout in milliseconds. + */ + public static final long DEFAULT_TOOL_TIMEOUT = 20_000; + + private McpServerConstants() { + } +} diff --git a/components/camel-ai/camel-mcp-server-api/src/main/java/org/apache/camel/component/mcp/server/McpServerEngine.java b/components/camel-ai/camel-mcp-server-api/src/main/java/org/apache/camel/component/mcp/server/McpServerEngine.java new file mode 100644 index 0000000000000..9592e9150dc77 --- /dev/null +++ b/components/camel-ai/camel-mcp-server-api/src/main/java/org/apache/camel/component/mcp/server/McpServerEngine.java @@ -0,0 +1,64 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.mcp.server; + +import org.apache.camel.CamelContextAware; +import org.apache.camel.Service; + +/** + * SPI for the runtime-specific serving layer of the Camel MCP server: a sink the bridge publishes tools into. + *

+ * The bridge owns tool selection, execution, timeout and error sanitization — identical on every runtime. The engine + * owns protocol serving (HTTP transport, sessions, notifications). One logical MCP server exists per CamelContext. + *

+ * Resolution: a bean of this type in the Camel registry wins; otherwise the engine is discovered via FactoryFinder + * under {@link McpServerConstants#MCP_SERVER_ENGINE_FACTORY}. + *

+ * Lifecycle: the bridge calls {@link #initialize(McpServerInfo)} once before starting the engine, then + * {@link #toolAdded(McpServerTool)} for the initial tool set and for every later change (driven by route + * start/stop/suspend/resume of {@code ai-tool} routes). Engines with a {@code listChanged} capability should emit + * {@code notifications/tools/list_changed} on add/remove. + * + * @since 4.22 + */ +public interface McpServerEngine extends Service, CamelContextAware { + + /** + * Passes the server identity and serving hints. Called once, before {@link #start()}. Engines backed by a native + * runtime MCP server MAY ignore the serving hints — see {@link #consumesServingConfiguration()}. + */ + void initialize(McpServerInfo info); + + /** + * Publishes a tool. Called for the initial set and whenever a matching {@code ai-tool} route starts or resumes. + */ + void toolAdded(McpServerTool tool); + + /** + * Removes a tool by name. Called whenever a matching {@code ai-tool} route stops or suspends. + */ + void toolRemoved(String toolName); + + /** + * Whether this engine consumes the Camel-owned serving configuration ({@code path}, {@code serverName}). Engines + * backed by a native runtime MCP server return false — their own configuration decides serving concerns — and the + * bridge then warns when Camel serving properties are set but ignored. + */ + default boolean consumesServingConfiguration() { + return false; + } +} diff --git a/components/camel-ai/camel-mcp-server-api/src/main/java/org/apache/camel/component/mcp/server/McpServerInfo.java b/components/camel-ai/camel-mcp-server-api/src/main/java/org/apache/camel/component/mcp/server/McpServerInfo.java new file mode 100644 index 0000000000000..1893a4789c867 --- /dev/null +++ b/components/camel-ai/camel-mcp-server-api/src/main/java/org/apache/camel/component/mcp/server/McpServerInfo.java @@ -0,0 +1,32 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.mcp.server; + +/** + * Identity and serving hints passed to an {@link McpServerEngine} before it is started. + *

+ * Engines backed by a native runtime MCP server (Quarkus, Spring Boot) MAY ignore the serving hints ({@code path}) — + * their own runtime configuration decides how the server is exposed. + * + * @param serverName the MCP server name advertised to clients (defaults to the CamelContext name) + * @param version the MCP server version advertised to clients + * @param path the HTTP path where the MCP endpoint should be served + * + * @since 4.22 + */ +public record McpServerInfo(String serverName, String version, String path) { +} diff --git a/components/camel-ai/camel-mcp-server-api/src/main/java/org/apache/camel/component/mcp/server/McpServerTool.java b/components/camel-ai/camel-mcp-server-api/src/main/java/org/apache/camel/component/mcp/server/McpServerTool.java new file mode 100644 index 0000000000000..06b6d00986dc3 --- /dev/null +++ b/components/camel-ai/camel-mcp-server-api/src/main/java/org/apache/camel/component/mcp/server/McpServerTool.java @@ -0,0 +1,55 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.mcp.server; + +import java.util.Map; + +import org.apache.camel.component.ai.tool.AiToolParameterHelper.ParameterDef; + +/** + * A tool published by the bridge into an {@link McpServerEngine}. Engines pick whichever input-schema representation + * fits their API: the pre-built JSON Schema string or the structured parameter definitions. + * + * @since 4.22 + */ +public interface McpServerTool { + + /** + * The tool name, unique within the MCP server (flat namespace). + */ + String name(); + + /** + * Human-readable tool description. + */ + String description(); + + /** + * The tool input as a JSON Schema object string, or null when the tool declares no parameters. + */ + String inputSchemaJson(); + + /** + * The tool input as structured parameter definitions; empty when the tool declares no parameters. + */ + Map parameters(); + + /** + * The handler executing the tool. Blocking, timeout-bounded and pre-sanitized by the bridge. + */ + McpToolCallHandler handler(); +} diff --git a/components/camel-ai/camel-mcp-server-api/src/main/java/org/apache/camel/component/mcp/server/McpToolCallHandler.java b/components/camel-ai/camel-mcp-server-api/src/main/java/org/apache/camel/component/mcp/server/McpToolCallHandler.java new file mode 100644 index 0000000000000..2d429bec4af22 --- /dev/null +++ b/components/camel-ai/camel-mcp-server-api/src/main/java/org/apache/camel/component/mcp/server/McpToolCallHandler.java @@ -0,0 +1,40 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.mcp.server; + +import java.util.Map; + +/** + * Executes a single MCP tool call. Implemented by the bridge; engines invoke it when an MCP client calls the tool. + *

+ * The call is blocking and bounded: the bridge applies the configured per-call timeout and maps every outcome + * (including route exceptions and timeouts) to a pre-sanitized {@link McpToolCallResult} — it never throws and never + * exposes route internals. + * + * @since 4.22 + */ +@FunctionalInterface +public interface McpToolCallHandler { + + /** + * Invokes the tool with the given arguments. + * + * @param arguments the tool arguments as parsed from the MCP {@code tools/call} request, never null + * @return the sanitized result, never null + */ + McpToolCallResult call(Map arguments); +} diff --git a/components/camel-ai/camel-mcp-server-api/src/main/java/org/apache/camel/component/mcp/server/McpToolCallResult.java b/components/camel-ai/camel-mcp-server-api/src/main/java/org/apache/camel/component/mcp/server/McpToolCallResult.java new file mode 100644 index 0000000000000..526236d5a11b0 --- /dev/null +++ b/components/camel-ai/camel-mcp-server-api/src/main/java/org/apache/camel/component/mcp/server/McpToolCallResult.java @@ -0,0 +1,29 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.mcp.server; + +/** + * Result of an MCP tool invocation, pre-sanitized by the bridge: the text is safe to return to a remote MCP client and + * never contains raw route exception messages. + * + * @param text the tool output, or a safe error message when {@code isError} is true + * @param isError whether the invocation failed + * + * @since 4.22 + */ +public record McpToolCallResult(String text, boolean isError) { +} diff --git a/components/camel-ai/camel-mcp-server-api/src/test/java/org/apache/camel/component/mcp/server/McpServerBridgeResolutionTest.java b/components/camel-ai/camel-mcp-server-api/src/test/java/org/apache/camel/component/mcp/server/McpServerBridgeResolutionTest.java new file mode 100644 index 0000000000000..35ca14f1a793b --- /dev/null +++ b/components/camel-ai/camel-mcp-server-api/src/test/java/org/apache/camel/component/mcp/server/McpServerBridgeResolutionTest.java @@ -0,0 +1,36 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.mcp.server; + +import org.apache.camel.impl.DefaultCamelContext; +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +class McpServerBridgeResolutionTest { + + @Test + void testStartupFailsWithClearMessageWhenNoEngineAvailable() throws Exception { + try (DefaultCamelContext camelContext = new DefaultCamelContext()) { + McpServerBridge bridge = new McpServerBridge(new McpServerConfiguration()); + assertThatThrownBy(() -> { + camelContext.addService(bridge); + camelContext.start(); + }).hasStackTraceContaining("camel-mcp-server"); + } + } +} diff --git a/components/camel-ai/camel-mcp-server-api/src/test/java/org/apache/camel/component/mcp/server/McpServerBridgeTest.java b/components/camel-ai/camel-mcp-server-api/src/test/java/org/apache/camel/component/mcp/server/McpServerBridgeTest.java new file mode 100644 index 0000000000000..9018fa48777a4 --- /dev/null +++ b/components/camel-ai/camel-mcp-server-api/src/test/java/org/apache/camel/component/mcp/server/McpServerBridgeTest.java @@ -0,0 +1,165 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.mcp.server; + +import java.util.Map; + +import org.apache.camel.CamelContext; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.test.junit6.CamelTestSupport; +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; + +class McpServerBridgeTest extends CamelTestSupport { + + private final RecordingMcpServerEngine engine = new RecordingMcpServerEngine(); + private McpServerBridge bridge; + + @Override + protected CamelContext createCamelContext() throws Exception { + CamelContext camelContext = super.createCamelContext(); + // a registry bean of type McpServerEngine wins over FactoryFinder discovery + camelContext.getRegistry().bind("mcpServerEngine", engine); + McpServerConfiguration configuration = new McpServerConfiguration(); + configuration.setTags("crm,notify"); + configuration.setToolTimeout(500); + bridge = new McpServerBridge(configuration); + camelContext.addService(bridge); + return camelContext; + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + public void configure() { + from("ai-tool:query_db?tags=crm&description=Query the customer database" + + "¶meter.customerId=string¶meter.customerId.required=true") + .routeId("query-db-route") + .setBody(simple("customer-${header.customerId}")); + + from("ai-tool:send_email?tags=notify,crm&description=Send an email") + .routeId("send-email-route") + .setBody(constant("sent")); + + from("ai-tool:boom?tags=crm&description=Always fails") + .routeId("boom-route") + .process(e -> { + throw new IllegalStateException("secret internal detail"); + }); + + from("ai-tool:slow?tags=crm&description=Too slow") + .routeId("slow-route") + .delay(5000) + .setBody(constant("done")); + + from("ai-tool:hidden_tool?description=Untagged tool") + .setBody(constant("hidden")); + + from("ai-tool:other_tool?tags=untrusted&description=Other tag") + .setBody(constant("other")); + } + }; + } + + @Test + void testPublishesOnlySelectedTags() { + assertThat(engine.tools()) + .containsKeys("query_db", "send_email", "boom", "slow") + .doesNotContainKeys("hidden_tool", "other_tool"); + assertThat(engine.info().serverName()).isEqualTo(context.getName()); + + McpServerTool tool = engine.tools().get("query_db"); + assertThat(tool.description()).isEqualTo("Query the customer database"); + assertThat(tool.inputSchemaJson()).contains("customerId"); + assertThat(tool.parameters()).containsKey("customerId"); + } + + @Test + void testCallToolSuccess() { + McpToolCallResult result = engine.tools().get("query_db").handler().call(Map.of("customerId", "42")); + + assertThat(result.isError()).isFalse(); + assertThat(result.text()).isEqualTo("customer-42"); + } + + @Test + void testCallToolMissingRequiredArgument() { + McpToolCallResult result = engine.tools().get("query_db").handler().call(Map.of()); + + assertThat(result.isError()).isTrue(); + assertThat(result.text()).contains("customerId"); + } + + @Test + void testCallToolExecutionErrorIsSanitized() { + McpToolCallResult result = engine.tools().get("boom").handler().call(Map.of()); + + assertThat(result.isError()).isTrue(); + assertThat(result.text()) + .doesNotContain("secret internal detail") + .isEqualTo("Tool execution failed"); + } + + @Test + void testCallToolTimeout() { + McpToolCallResult result = engine.tools().get("slow").handler().call(Map.of()); + + assertThat(result.isError()).isTrue(); + assertThat(result.text()).contains("timed out"); + } + + @Test + void testToolRemovedAndReAddedOnRouteLifecycle() throws Exception { + context.getRouteController().stopRoute("query-db-route"); + assertThat(engine.tools()).doesNotContainKey("query_db"); + assertThat(engine.removed()).contains("query_db"); + + context.getRouteController().startRoute("query-db-route"); + assertThat(engine.tools()).containsKey("query_db"); + } + + @Test + void testMultiTagToolRemovedOnceWhenRouteStops() throws Exception { + // send_email is registered under two selected tags: stopping the route fires two deregistration + // events but must remove the published tool exactly once + context.getRouteController().stopRoute("send-email-route"); + + assertThat(engine.tools()).doesNotContainKey("send_email"); + assertThat(engine.removed()).containsOnlyOnce("send_email"); + } + + @Test + void testNameCollisionIsRefused() throws Exception { + McpServerTool published = engine.tools().get("query_db"); + + context.addRoutes(new RouteBuilder() { + public void configure() { + from("ai-tool:query_db?tags=notify&description=Colliding tool") + .routeId("colliding-route") + .setBody(constant("other")); + } + }); + + // the colliding tool is refused: the originally published tool stays + assertThat(engine.tools().get("query_db")).isSameAs(published); + + // and removing the colliding route does not remove the published tool + context.getRouteController().stopRoute("colliding-route"); + assertThat(engine.tools()).containsKey("query_db"); + } +} diff --git a/components/camel-ai/camel-mcp-server-api/src/test/java/org/apache/camel/component/mcp/server/RecordingMcpServerEngine.java b/components/camel-ai/camel-mcp-server-api/src/test/java/org/apache/camel/component/mcp/server/RecordingMcpServerEngine.java new file mode 100644 index 0000000000000..738dffb5b2961 --- /dev/null +++ b/components/camel-ai/camel-mcp-server-api/src/test/java/org/apache/camel/component/mcp/server/RecordingMcpServerEngine.java @@ -0,0 +1,74 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.mcp.server; + +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CopyOnWriteArrayList; + +import org.apache.camel.CamelContext; +import org.apache.camel.support.service.ServiceSupport; + +/** + * Mock {@link McpServerEngine} recording the tools published by the bridge, for engine-less bridge tests. + */ +public class RecordingMcpServerEngine extends ServiceSupport implements McpServerEngine { + + private final Map tools = new ConcurrentHashMap<>(); + private final List removed = new CopyOnWriteArrayList<>(); + private CamelContext camelContext; + private McpServerInfo info; + + @Override + public CamelContext getCamelContext() { + return camelContext; + } + + @Override + public void setCamelContext(CamelContext camelContext) { + this.camelContext = camelContext; + } + + @Override + public void initialize(McpServerInfo info) { + this.info = info; + } + + @Override + public void toolAdded(McpServerTool tool) { + tools.put(tool.name(), tool); + } + + @Override + public void toolRemoved(String toolName) { + tools.remove(toolName); + removed.add(toolName); + } + + public Map tools() { + return tools; + } + + public List removed() { + return removed; + } + + public McpServerInfo info() { + return info; + } +} diff --git a/components/camel-ai/camel-mcp-server-api/src/test/java/org/apache/camel/component/mcp/server/conformance/McpServerConformanceTestSupport.java b/components/camel-ai/camel-mcp-server-api/src/test/java/org/apache/camel/component/mcp/server/conformance/McpServerConformanceTestSupport.java new file mode 100644 index 0000000000000..119f309b258f0 --- /dev/null +++ b/components/camel-ai/camel-mcp-server-api/src/test/java/org/apache/camel/component/mcp/server/conformance/McpServerConformanceTestSupport.java @@ -0,0 +1,204 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.mcp.server.conformance; + +import java.time.Duration; +import java.util.List; +import java.util.Map; +import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; + +import io.modelcontextprotocol.client.McpClient; +import io.modelcontextprotocol.client.McpSyncClient; +import io.modelcontextprotocol.client.transport.HttpClientStreamableHttpTransport; +import io.modelcontextprotocol.spec.McpSchema; +import org.apache.camel.CamelContext; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.mcp.server.McpServerBridge; +import org.apache.camel.component.mcp.server.McpServerConfiguration; +import org.apache.camel.test.junit6.CamelTestSupport; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; + +/** + * Engine conformance kit: the behavioural contract every {@link org.apache.camel.component.mcp.server.McpServerEngine} + * implementation must satisfy, verified with the official MCP Java SDK client over streamable HTTP. + *

+ * Engine modules extend this class (it is shipped in the camel-mcp-server-api test-jar), install their serving + * infrastructure in {@link #customizeCamelContext(CamelContext)} and point {@link #mcpServerBaseUrl()} at the running + * server. The kit owns the ai-tool routes and the {@link McpServerBridge} so tool semantics cannot drift between + * engines. + */ +public abstract class McpServerConformanceTestSupport extends CamelTestSupport { + + public static final String CONFORMANCE_TAG = "conformance"; + public static final long TOOL_TIMEOUT_MILLIS = 2000; + + protected McpServerBridge bridge; + private McpSyncClient client; + + /** + * Base URL of the server under test, without the MCP endpoint path (the SDK client appends {@code /mcp}). + */ + protected abstract String mcpServerBaseUrl(); + + /** + * Installs the serving infrastructure the engine under test needs (e.g. an HTTP server service). Called before the + * bridge is added to the context. + */ + protected void customizeCamelContext(CamelContext camelContext) throws Exception { + } + + /** + * Adjusts the bridge configuration; tags and tool timeout are preset by the kit. + */ + protected void configureBridge(McpServerConfiguration configuration) { + } + + @Override + protected CamelContext createCamelContext() throws Exception { + CamelContext camelContext = super.createCamelContext(); + customizeCamelContext(camelContext); + McpServerConfiguration configuration = new McpServerConfiguration(); + configuration.setTags(CONFORMANCE_TAG); + configuration.setToolTimeout(TOOL_TIMEOUT_MILLIS); + configureBridge(configuration); + bridge = new McpServerBridge(configuration); + camelContext.addService(bridge); + return camelContext; + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + public void configure() { + from("ai-tool:say_hello?tags=" + CONFORMANCE_TAG + "&description=Say hello" + + "¶meter.name=string¶meter.name.description=Who to greet¶meter.name.required=true") + .routeId("say-hello-route") + .setBody(simple("Hello ${header.name}")); + + from("ai-tool:fail_tool?tags=" + CONFORMANCE_TAG + "&description=Always fails") + .routeId("fail-tool-route") + .process(e -> { + throw new IllegalStateException("secret internal detail"); + }); + + from("ai-tool:slow_tool?tags=" + CONFORMANCE_TAG + "&description=Exceeds the tool timeout") + .routeId("slow-tool-route") + .delay(TOOL_TIMEOUT_MILLIS * 3) + .setBody(constant("done")); + + from("ai-tool:hidden_tool?description=Untagged tool, must not be exposed") + .setBody(constant("hidden")); + + from("ai-tool:other_tool?tags=untrusted&description=Not a selected tag, must not be exposed") + .setBody(constant("other")); + } + }; + } + + protected McpSyncClient client() { + if (client == null) { + client = McpClient.sync(HttpClientStreamableHttpTransport.builder(mcpServerBaseUrl()).build()) + .requestTimeout(Duration.ofSeconds(10)) + .initializationTimeout(Duration.ofSeconds(10)) + .build(); + client.initialize(); + } + return client; + } + + @AfterEach + void closeClient() { + if (client != null) { + client.closeGracefully(); + client = null; + } + } + + @Test + void testListToolsExposesOnlySelectedTags() { + List tools = client().listTools().tools(); + + assertThat(tools).extracting(McpSchema.Tool::name) + .contains("say_hello", "fail_tool", "slow_tool") + .doesNotContain("hidden_tool", "other_tool"); + + McpSchema.Tool sayHello = tools.stream().filter(t -> "say_hello".equals(t.name())).findFirst().orElseThrow(); + assertThat(sayHello.description()).isEqualTo("Say hello"); + assertThat(sayHello.inputSchema()).containsKey("properties"); + assertThat(sayHello.inputSchema().toString()).contains("name"); + } + + @Test + void testCallToolSuccess() { + McpSchema.CallToolResult result + = client().callTool(new McpSchema.CallToolRequest("say_hello", Map.of("name", "World"))); + + assertThat(result.isError()).isNotEqualTo(Boolean.TRUE); + assertThat(textOf(result)).isEqualTo("Hello World"); + } + + @Test + void testCallToolMissingRequiredArgument() { + McpSchema.CallToolResult result = client().callTool(new McpSchema.CallToolRequest("say_hello", Map.of())); + + assertThat(result.isError()).isEqualTo(Boolean.TRUE); + assertThat(textOf(result)).contains("name"); + } + + @Test + void testCallToolExecutionErrorIsSanitized() { + McpSchema.CallToolResult result = client().callTool(new McpSchema.CallToolRequest("fail_tool", Map.of())); + + assertThat(result.isError()).isEqualTo(Boolean.TRUE); + assertThat(textOf(result)) + .doesNotContain("secret internal detail") + .isEqualTo("Tool execution failed"); + } + + @Test + void testCallToolTimeout() { + McpSchema.CallToolResult result = client().callTool(new McpSchema.CallToolRequest("slow_tool", Map.of())); + + assertThat(result.isError()).isEqualTo(Boolean.TRUE); + assertThat(textOf(result)).contains("timed out"); + } + + @Test + void testToolsListReflectsRouteStopAndStart() throws Exception { + assertThat(client().listTools().tools()).extracting(McpSchema.Tool::name).contains("say_hello"); + + context.getRouteController().stopRoute("say-hello-route"); + await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> assertThat(client().listTools().tools()) + .extracting(McpSchema.Tool::name).doesNotContain("say_hello")); + + context.getRouteController().startRoute("say-hello-route"); + await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> assertThat(client().listTools().tools()) + .extracting(McpSchema.Tool::name).contains("say_hello")); + } + + protected static String textOf(McpSchema.CallToolResult result) { + return result.content().stream() + .filter(McpSchema.TextContent.class::isInstance) + .map(c -> ((McpSchema.TextContent) c).text()) + .collect(Collectors.joining()); + } +} diff --git a/components/camel-ai/camel-mcp-server/pom.xml b/components/camel-ai/camel-mcp-server/pom.xml new file mode 100644 index 0000000000000..40db5a9dc0913 --- /dev/null +++ b/components/camel-ai/camel-mcp-server/pom.xml @@ -0,0 +1,106 @@ + + + + 4.0.0 + + + org.apache.camel + camel-ai-parent + 4.22.0-SNAPSHOT + + + camel-mcp-server + jar + Camel :: AI :: MCP Server + Expose ai-tool routes as MCP tools over streamable HTTP + + + 4.22.0 + + MCP Server + Preview + 3 + + + + + + org.apache.camel + camel-mcp-server-api + + + org.apache.camel + camel-platform-http-vertx + + + io.modelcontextprotocol.sdk + mcp-core + ${mcp-java-sdk-version} + + + io.modelcontextprotocol.sdk + mcp-json-jackson2 + ${mcp-java-sdk-version} + + + + + org.apache.camel + camel-mcp-server-api + test-jar + test + + + org.apache.camel + camel-test-junit6 + test + + + + org.apache.camel + camel-openai + test + + + org.apache.camel + camel-platform-http-main + test + + + org.apache.camel + camel-test-infra-ollama + ${project.version} + test + + + org.awaitility + awaitility + ${awaitility-version} + test + + + org.assertj + assertj-core + test + + + + + diff --git a/components/camel-ai/camel-mcp-server/src/generated/resources/META-INF/services/org/apache/camel/mcp-server-engine b/components/camel-ai/camel-mcp-server/src/generated/resources/META-INF/services/org/apache/camel/mcp-server-engine new file mode 100644 index 0000000000000..9dad13da0ccb3 --- /dev/null +++ b/components/camel-ai/camel-mcp-server/src/generated/resources/META-INF/services/org/apache/camel/mcp-server-engine @@ -0,0 +1,2 @@ +# Generated by camel build tools - do NOT edit this file! +class=org.apache.camel.component.mcp.server.vertx.VertxMcpServerEngine diff --git a/components/camel-ai/camel-mcp-server/src/generated/resources/META-INF/services/org/apache/camel/other.properties b/components/camel-ai/camel-mcp-server/src/generated/resources/META-INF/services/org/apache/camel/other.properties new file mode 100644 index 0000000000000..fdb4e613d7813 --- /dev/null +++ b/components/camel-ai/camel-mcp-server/src/generated/resources/META-INF/services/org/apache/camel/other.properties @@ -0,0 +1,7 @@ +# Generated by camel build tools - do NOT edit this file! +name=mcp-server +groupId=org.apache.camel +artifactId=camel-mcp-server +version=4.22.0-SNAPSHOT +projectName=Camel :: AI :: MCP Server +projectDescription=Expose ai-tool routes as MCP tools over streamable HTTP diff --git a/components/camel-ai/camel-mcp-server/src/generated/resources/mcp-server.json b/components/camel-ai/camel-mcp-server/src/generated/resources/mcp-server.json new file mode 100644 index 0000000000000..b1eea4c1ff42d --- /dev/null +++ b/components/camel-ai/camel-mcp-server/src/generated/resources/mcp-server.json @@ -0,0 +1,15 @@ +{ + "other": { + "kind": "other", + "name": "mcp-server", + "title": "MCP Server", + "description": "Expose ai-tool routes as MCP tools over streamable HTTP", + "deprecated": false, + "firstVersion": "4.22.0", + "label": "ai", + "supportLevel": "Preview", + "groupId": "org.apache.camel", + "artifactId": "camel-mcp-server", + "version": "4.22.0-SNAPSHOT" + } +} diff --git a/components/camel-ai/camel-mcp-server/src/main/docs/mcp-server.adoc b/components/camel-ai/camel-mcp-server/src/main/docs/mcp-server.adoc new file mode 100644 index 0000000000000..6b28142ca981a --- /dev/null +++ b/components/camel-ai/camel-mcp-server/src/main/docs/mcp-server.adoc @@ -0,0 +1,210 @@ += MCP Server Component +:doctitle: MCP Server +:shortname: mcp-server +:artifactid: camel-mcp-server +:description: Expose ai-tool routes as MCP tools over streamable HTTP +:since: 4.22 +:supportlevel: Preview +:tabs-sync-option: + +*Since Camel {since}* + +The camel-mcp-server module exposes Camel routes registered via the +xref:ROOT:ai-tool-component.adoc[ai-tool] component as tools of a +https://modelcontextprotocol.io[Model Context Protocol] (MCP) server, served +over MCP streamable HTTP. No route is needed for the server itself: add the +dependency, configure which tags to expose, and every matching `ai-tool` route +becomes an MCP tool that any MCP client (another Camel application, an IDE, a +coding agent) can discover and call. + +Maven users will need to add the following dependency to their `pom.xml`: + +[source,xml] +---- + + org.apache.camel + camel-mcp-server + x.x.x + + +---- + +== Architecture + +The module is split in two artifacts: + +* `camel-mcp-server-api` — the runtime-agnostic _bridge_ and the small + `McpServerEngine` SPI. The bridge owns tool selection (tags), execution via + the shared `AiToolExecutor` (per-call timeout, error sanitization) and reacts + to `AiToolRegistry` changes when routes start and stop. It has no dependency + on the MCP Java SDK. +* `camel-mcp-server` — the serving engine for Camel Main and Camel JBang, + built on the official MCP Java SDK with a Vert.x streamable HTTP transport. + The MCP endpoint is registered on the Camel main HTTP server's router, so it + serves on the main server port (`camel.server.port`) and inherits its + lifecycle, authentication and CORS configuration. + +Engine resolution mirrors the platform-http engine: a bean of type +`McpServerEngine` in the Camel registry wins; otherwise the engine is +discovered on the classpath. Other runtimes plug native engines through the +same SPI: on Quarkus the `camel-quarkus-mcp-server` extension serves through +the Quarkiverse `quarkus-mcp-server` (configured via `quarkus.mcp.server.*`), +and on Spring Boot the starter serves through the Spring AI MCP server +(configured via `spring.ai.mcp.server.*`). Bridge behavior — tag selection, +timeout, sanitization — is identical on every runtime and verified by a shared +conformance test kit. + +== Usage + +Define tools as regular `ai-tool` routes and give them tags: + +[tabs] +==== +Java:: ++ +[source,java] +---- +from("ai-tool:query_db?tags=crm" + + "&description=Query customer database" + + "¶meter.customerId=string" + + "¶meter.customerId.description=The customer id" + + "¶meter.customerId.required=true") + .to("jdbc:dataSource"); +---- + +XML:: ++ +[source,xml] +---- + + + + +---- + +YAML:: ++ +[source,yaml] +---- +- route: + from: + uri: ai-tool:query_db + parameters: + tags: crm + description: "Query customer database" + parameter.customerId: string + parameter.customerId.description: "The customer id" + parameter.customerId.required: "true" + steps: + - to: + uri: jdbc:dataSource +---- +==== + +Start the MCP server by adding the `McpServerBridge` service to the +CamelContext, selecting the tags to expose: + +[source,java] +---- +McpServerConfiguration configuration = new McpServerConfiguration(); +configuration.setTags("crm,notify"); +camelContext.addService(new McpServerBridge(configuration)); +---- + +The MCP endpoint is then served at `http://:/mcp` on the Camel +main HTTP server. Any MCP client can connect over streamable HTTP, for +example another Camel integration using the +xref:ROOT:openai-component.adoc[camel-openai] MCP client: + +[source,java] +---- +from("direct:agent") + .to("openai:chat-completion" + + "?model={{llm.model}}" + + "&autoToolExecution=true" + + "&mcpServer.myCamelTools.transportType=streamableHttp" + + "&mcpServer.myCamelTools.url=http://localhost:8080/mcp"); +---- + +== Options + +The `McpServerConfiguration` options: + +[width="100%",cols="2,5,2,1",options="header"] +|=== +| Option | Description | Default | Owner + +| `tags` | Comma-separated list of ai-tool tags to expose as MCP tools. Only + tools registered under one of these tags are published; the untagged + default pool is never exposed. When not set, no tools are published. | | + bridge +| `toolTimeout` | Per-call tool execution timeout in milliseconds. A call + exceeding the timeout returns an error result to the MCP client; the + underlying route keeps running until it completes on its own. | `20000` | + bridge +| `path` | HTTP path where the MCP endpoint is served. | `/mcp` | engine +| `serverName` | MCP server name advertised to clients. | CamelContext name | + engine +|=== + +Bridge-owned options are honored identically on every runtime. Engine-owned +options are consumed by the Vert.x engine only; on runtimes with a native +engine (Quarkus, Spring Boot) the native configuration decides serving +concerns and a startup WARN is logged when an ignored option is set. + +== Protocol + +This section describes the Vert.x engine shipped in `camel-mcp-server`, which +serves on Camel Main and Camel JBang. On Quarkus and Spring Boot the transport +is owned by the native engine instead — quarkus-mcp-server and the Spring Boot +embedded HTTP server (Spring AI MCP server) respectively — and the details +below do not apply. + +The Vert.x engine implements the MCP streamable HTTP transport: + +* `POST /mcp` answering `application/json` or `text/event-stream` depending on + the request, +* a long-lived `GET /mcp` SSE channel for server notifications, with + `Last-Event-ID` replay, +* session management via the `Mcp-Session-Id` header and `DELETE /mcp` for + session termination. + +Tools appearing or disappearing (routes starting and stopping) emit +`notifications/tools/list_changed` to connected clients. + +== Security + +External MCP clients are *untrusted senders* under the +xref:manual::security-model.adoc[Camel security model]. The module applies the +following rules: + +* *Explicit opt-in per tool*: only tools whose tags intersect the configured + `tags` are exposed. The untagged default pool is never exposed implicitly. +* *Flat namespace protection*: a tool whose name collides with an already + exposed tool is refused with an ERROR log — never silently replaced. +* *Error sanitization*: route exceptions are mapped to a generic error + message; the cause is logged server-side and never sent to the client. + Argument validation messages (missing or invalid parameters) are returned + as-is. +* *Bounded execution*: every call is subject to the `toolTimeout`. Note that a + timed-out route keeps running server-side until it completes; the timeout + bounds the MCP request, not the route. +* *Authentication*: the MCP endpoint is served through the main HTTP server + router, so platform-http authentication (basic, JWT via + `camel.server.authentication*` options) applies to it. The MCP + specification's authorization model is OAuth 2.1; see + xref:oauth.adoc[camel-oauth] for resource-server style + protection. On Quarkus and Spring Boot, authentication is owned by the + native runtime security. + +== Runtime notes + +* *Camel Main / JBang*: requires the Camel main HTTP server + (`camel.server.enabled=true` with `camel-platform-http-main`, automatic + with Camel JBang) or a `VertxPlatformHttpServer` service. Serving is fully + asynchronous: tool calls are offloaded to the Vert.x worker pool and the + long-lived SSE channel does not occupy a worker thread. +* *Quarkus*: use the `camel-quarkus-mcp-server` extension (serves through + quarkus-mcp-server; the MCP Java SDK is not on the classpath). +* *Spring Boot*: use the `camel-mcp-server-starter` (serves through the + Spring AI MCP server). diff --git a/components/camel-ai/camel-mcp-server/src/main/java/org/apache/camel/component/mcp/server/vertx/VertxMcpServerEngine.java b/components/camel-ai/camel-mcp-server/src/main/java/org/apache/camel/component/mcp/server/vertx/VertxMcpServerEngine.java new file mode 100644 index 0000000000000..2dd0e8e0d20cd --- /dev/null +++ b/components/camel-ai/camel-mcp-server/src/main/java/org/apache/camel/component/mcp/server/vertx/VertxMcpServerEngine.java @@ -0,0 +1,170 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.mcp.server.vertx; + +import java.util.Map; +import java.util.Set; + +import io.modelcontextprotocol.json.McpJsonDefaults; +import io.modelcontextprotocol.json.McpJsonMapper; +import io.modelcontextprotocol.server.McpServer; +import io.modelcontextprotocol.server.McpServerFeatures; +import io.modelcontextprotocol.server.McpSyncServer; +import io.modelcontextprotocol.spec.McpSchema; +import org.apache.camel.CamelContext; +import org.apache.camel.component.mcp.server.McpServerConstants; +import org.apache.camel.component.mcp.server.McpServerEngine; +import org.apache.camel.component.mcp.server.McpServerInfo; +import org.apache.camel.component.mcp.server.McpServerTool; +import org.apache.camel.component.mcp.server.McpToolCallResult; +import org.apache.camel.component.platform.http.PlatformHttpComponent; +import org.apache.camel.component.platform.http.vertx.VertxPlatformHttpRouter; +import org.apache.camel.spi.annotations.JdkService; +import org.apache.camel.support.service.ServiceSupport; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * {@link McpServerEngine} for Camel Main / JBang: serves MCP streamable HTTP through the Vert.x platform HTTP router + * using the official MCP Java SDK. The MCP endpoint is registered on the main HTTP server's router, so it serves on the + * main server port and inherits its lifecycle, authentication and CORS configuration. + */ +@JdkService(McpServerConstants.MCP_SERVER_ENGINE_FACTORY) +public class VertxMcpServerEngine extends ServiceSupport implements McpServerEngine { + + private static final Logger LOG = LoggerFactory.getLogger(VertxMcpServerEngine.class); + + private static final String EMPTY_OBJECT_SCHEMA = """ + { + "type": "object", + "properties": {}, + "additionalProperties": false + } + """; + private static final String APPLICATION_JSON = "application/json"; + + private CamelContext camelContext; + private McpServerInfo info; + private McpJsonMapper jsonMapper; + private VertxMcpStreamableServerTransportProvider transport; + private McpSyncServer server; + + @Override + public CamelContext getCamelContext() { + return camelContext; + } + + @Override + public void setCamelContext(CamelContext camelContext) { + this.camelContext = camelContext; + } + + @Override + public void initialize(McpServerInfo info) { + this.info = info; + } + + @Override + public boolean consumesServingConfiguration() { + return true; + } + + @Override + protected void doStart() throws Exception { + VertxPlatformHttpRouter router = lookupRouter(); + jsonMapper = McpJsonDefaults.getMapper(); + transport = new VertxMcpStreamableServerTransportProvider(jsonMapper, info.path()); + server = McpServer.sync(transport) + .serverInfo(info.serverName(), info.version()) + .capabilities(McpSchema.ServerCapabilities.builder().tools(true).build()) + .immediateExecution(true) + .build(); + // register routes only once the server has set the session factory on the transport + transport.registerRoutes(router); + + PlatformHttpComponent platformHttpComponent + = (PlatformHttpComponent) camelContext.hasComponent("platform-http"); + if (platformHttpComponent != null) { + platformHttpComponent.addHttpEndpoint(info.path(), "GET,POST,DELETE", APPLICATION_JSON, + "application/json,text/event-stream", null); + } + LOG.info("MCP server '{}' serving tools on path {}", info.serverName(), info.path()); + } + + @Override + protected void doStop() throws Exception { + if (transport != null) { + transport.unregisterRoutes(); + } + if (server != null) { + server.closeGracefully(); + server = null; + } + PlatformHttpComponent platformHttpComponent + = (PlatformHttpComponent) camelContext.hasComponent("platform-http"); + if (platformHttpComponent != null && info != null) { + platformHttpComponent.removeHttpEndpoint(info.path()); + } + transport = null; + } + + @Override + public void toolAdded(McpServerTool tool) { + String schema = tool.inputSchemaJson() != null ? tool.inputSchemaJson() : EMPTY_OBJECT_SCHEMA; + McpSchema.Tool mcpTool = McpSchema.Tool.builder(tool.name(), jsonMapper, schema) + .description(tool.description()) + .build(); + McpServerFeatures.SyncToolSpecification spec = McpServerFeatures.SyncToolSpecification.builder() + .tool(mcpTool) + .callHandler((exchange, request) -> { + Map arguments = request.arguments() != null ? request.arguments() : Map.of(); + McpToolCallResult result = tool.handler().call(arguments); + return McpSchema.CallToolResult.builder() + .addTextContent(result.text()) + .isError(result.isError()) + .build(); + }) + .build(); + server.addTool(spec); + LOG.debug("MCP tool added: {}", tool.name()); + } + + @Override + public void toolRemoved(String toolName) { + try { + server.removeTool(toolName); + LOG.debug("MCP tool removed: {}", toolName); + } catch (Exception e) { + LOG.debug("Failed to remove MCP tool {}: {}", toolName, e.getMessage()); + } + } + + private VertxPlatformHttpRouter lookupRouter() { + Set routers = camelContext.getRegistry().findByType(VertxPlatformHttpRouter.class); + VertxPlatformHttpRouter router = routers.stream() + .filter(VertxPlatformHttpRouter::isMainServer) + .findFirst() + .orElseGet(() -> routers.size() == 1 ? routers.iterator().next() : null); + if (router == null) { + throw new IllegalStateException( + "The MCP server requires the Vert.x platform HTTP server. Enable the Camel main HTTP server " + + "(camel.server.enabled=true with camel-platform-http-main on the classpath) " + + "or add a VertxPlatformHttpServer service to the CamelContext."); + } + return router; + } +} diff --git a/components/camel-ai/camel-mcp-server/src/main/java/org/apache/camel/component/mcp/server/vertx/VertxMcpStreamableServerTransportProvider.java b/components/camel-ai/camel-mcp-server/src/main/java/org/apache/camel/component/mcp/server/vertx/VertxMcpStreamableServerTransportProvider.java new file mode 100644 index 0000000000000..12c8e5ed90a24 --- /dev/null +++ b/components/camel-ai/camel-mcp-server/src/main/java/org/apache/camel/component/mcp/server/vertx/VertxMcpStreamableServerTransportProvider.java @@ -0,0 +1,416 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.mcp.server.vertx; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.ConcurrentHashMap; + +import io.modelcontextprotocol.json.McpJsonMapper; +import io.modelcontextprotocol.json.TypeRef; +import io.modelcontextprotocol.spec.HttpHeaders; +import io.modelcontextprotocol.spec.McpError; +import io.modelcontextprotocol.spec.McpSchema; +import io.modelcontextprotocol.spec.McpStreamableServerSession; +import io.modelcontextprotocol.spec.McpStreamableServerTransport; +import io.modelcontextprotocol.spec.McpStreamableServerTransportProvider; +import io.vertx.core.Context; +import io.vertx.core.Vertx; +import io.vertx.core.http.HttpMethod; +import io.vertx.core.http.HttpServerResponse; +import io.vertx.ext.web.Route; +import io.vertx.ext.web.RoutingContext; +import io.vertx.ext.web.handler.BodyHandler; +import org.apache.camel.component.platform.http.vertx.VertxPlatformHttpRouter; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import reactor.core.publisher.Mono; + +/** + * MCP streamable HTTP server transport serving through the Vert.x platform HTTP router: POST answering + * {@code application/json} or {@code text/event-stream}, long-lived GET SSE channel with {@code Last-Event-ID} replay, + * {@code Mcp-Session-Id} session management and DELETE for session termination. + *

+ * This is the Vert.x equivalent of the MCP SDK's {@code HttpServletStreamableServerTransportProvider} (the SDK ships + * only servlet and stdio server transports). Request handling is offloaded to the Vert.x worker pool (unordered); + * response writes always run on the connection's event-loop context. The long-lived GET stream does not occupy a worker + * thread. + */ +public class VertxMcpStreamableServerTransportProvider implements McpStreamableServerTransportProvider { + + private static final String MESSAGE_EVENT_TYPE = "message"; + + private static final Logger LOG = LoggerFactory.getLogger(VertxMcpStreamableServerTransportProvider.class); + + private static final Duration NOTIFICATION_TIMEOUT = Duration.ofSeconds(5); + private static final Duration INITIALIZATION_TIMEOUT = Duration.ofSeconds(30); + + private static final String ACCEPT = "Accept"; + private static final String APPLICATION_JSON = "application/json"; + private static final String TEXT_EVENT_STREAM = "text/event-stream"; + + private final McpJsonMapper jsonMapper; + private final String path; + private final ConcurrentHashMap sessions = new ConcurrentHashMap<>(); + private final List routes = new ArrayList<>(); + + private McpStreamableServerSession.Factory sessionFactory; + private volatile boolean closing; + + public VertxMcpStreamableServerTransportProvider(McpJsonMapper jsonMapper, String path) { + this.jsonMapper = jsonMapper; + this.path = path; + } + + @Override + public void setSessionFactory(McpStreamableServerSession.Factory sessionFactory) { + this.sessionFactory = sessionFactory; + } + + @Override + public Mono notifyClients(String method, Object params) { + if (sessions.isEmpty()) { + return Mono.empty(); + } + return Mono.fromRunnable(() -> sessions.values().forEach(session -> { + try { + // bounded so a single stalled session cannot starve notifications to healthy sessions + session.sendNotification(method, params).block(NOTIFICATION_TIMEOUT); + } catch (Exception e) { + LOG.debug("Failed to send notification to MCP session {}: {}", session.getId(), e.getMessage()); + } + })); + } + + @Override + public Mono closeGracefully() { + return Mono.fromRunnable(() -> { + closing = true; + sessions.values().forEach(session -> { + try { + session.closeGracefully().block(NOTIFICATION_TIMEOUT); + } catch (Exception e) { + LOG.debug("Failed to close MCP session {}: {}", session.getId(), e.getMessage()); + } + }); + sessions.clear(); + }); + } + + /** + * Registers the POST/GET/DELETE routes for the MCP endpoint. Must be called after the MCP server has been built + * (the server sets the session factory on construction). + */ + public void registerRoutes(VertxPlatformHttpRouter router) { + Vertx vertx = router.vertx(); + Route post = router.route(path).method(HttpMethod.POST); + post.handler(BodyHandler.create(false)); + post.handler(ctx -> dispatch(vertx, ctx, this::handlePost)); + routes.add(post); + Route get = router.route(path).method(HttpMethod.GET); + get.handler(ctx -> dispatch(vertx, ctx, this::handleGet)); + routes.add(get); + Route delete = router.route(path).method(HttpMethod.DELETE); + delete.handler(ctx -> dispatch(vertx, ctx, this::handleDelete)); + routes.add(delete); + } + + public void unregisterRoutes() { + routes.forEach(Route::remove); + routes.clear(); + } + + @FunctionalInterface + private interface BlockingRequestHandler { + void handle(RoutingContext ctx, Context connection) throws Exception; + } + + private void dispatch(Vertx vertx, RoutingContext ctx, BlockingRequestHandler handler) { + // capture the connection's event-loop context before offloading; all response writes go through it + Context connection = vertx.getOrCreateContext(); + vertx.executeBlocking(() -> { + handler.handle(ctx, connection); + return null; + }, false).onFailure(t -> { + LOG.warn("Error handling MCP request", t); + if (!ctx.response().ended()) { + ctx.response().setStatusCode(500).end(); + } + }); + } + + private void handlePost(RoutingContext ctx, Context connection) throws Exception { + if (closing) { + endWithStatus(connection, ctx, 503); + return; + } + String contentType = ctx.request().getHeader("Content-Type"); + if (contentType == null || !contentType.contains(APPLICATION_JSON)) { + respondError(connection, ctx, 415, McpError.builder(McpSchema.ErrorCodes.INVALID_REQUEST) + .message("Content-Type application/json required").build()); + return; + } + + List badRequestErrors = new ArrayList<>(); + String accept = ctx.request().getHeader(ACCEPT); + if (accept == null || !accept.contains(TEXT_EVENT_STREAM)) { + badRequestErrors.add("text/event-stream required in Accept header"); + } + if (accept == null || !accept.contains(APPLICATION_JSON)) { + badRequestErrors.add("application/json required in Accept header"); + } + + McpSchema.JSONRPCMessage message; + try { + message = McpSchema.deserializeJsonRpcMessage(jsonMapper, ctx.body().asString()); + } catch (Exception e) { + respondError(connection, ctx, 400, McpError.builder(McpSchema.ErrorCodes.INVALID_REQUEST) + .message("Invalid message format: " + e.getMessage()).build()); + return; + } + + if (message instanceof McpSchema.JSONRPCRequest request + && McpSchema.METHOD_INITIALIZE.equals(request.method())) { + if (respondBadRequest(connection, ctx, badRequestErrors)) { + return; + } + handleInitialize(ctx, connection, request); + return; + } + + String sessionId = ctx.request().getHeader(HttpHeaders.MCP_SESSION_ID); + if (sessionId == null || sessionId.isBlank()) { + badRequestErrors.add("Session ID required in " + HttpHeaders.MCP_SESSION_ID + " header"); + } + if (respondBadRequest(connection, ctx, badRequestErrors)) { + return; + } + McpStreamableServerSession session = sessions.get(sessionId); + if (session == null) { + respondError(connection, ctx, 404, McpError.builder(McpSchema.ErrorCodes.INTERNAL_ERROR) + .message("Session not found: " + sessionId).build()); + return; + } + + if (message instanceof McpSchema.JSONRPCResponse response) { + session.accept(response).block(); + endWithStatus(connection, ctx, 202); + } else if (message instanceof McpSchema.JSONRPCNotification notification) { + session.accept(notification).block(); + endWithStatus(connection, ctx, 202); + } else if (message instanceof McpSchema.JSONRPCRequest request) { + VertxMcpSessionTransport transport = startSseResponse(ctx, connection, sessionId); + try { + session.responseStream(request, transport).block(); + } catch (Exception e) { + LOG.warn("Failed to handle MCP request stream: {}", e.getMessage()); + transport.close(); + } + } else { + respondError(connection, ctx, 500, + McpError.builder(McpSchema.ErrorCodes.INVALID_REQUEST).message("Unknown message type").build()); + } + } + + private void handleInitialize(RoutingContext ctx, Context connection, McpSchema.JSONRPCRequest request) + throws Exception { + McpSchema.InitializeRequest initializeRequest + = jsonMapper.convertValue(request.params(), new TypeRef() { + }); + McpStreamableServerSession.McpStreamableServerSessionInit init = sessionFactory.startSession(initializeRequest); + sessions.put(init.session().getId(), init.session()); + McpSchema.InitializeResult initResult = init.initResult().block(INITIALIZATION_TIMEOUT); + String json = jsonMapper.writeValueAsString(McpSchema.JSONRPCResponse.result(request.id(), initResult)); + connection.runOnContext(v -> ctx.response() + .setStatusCode(200) + .putHeader("Content-Type", APPLICATION_JSON) + .putHeader(HttpHeaders.MCP_SESSION_ID, init.session().getId()) + .end(json)); + } + + private void handleGet(RoutingContext ctx, Context connection) { + if (closing) { + endWithStatus(connection, ctx, 503); + return; + } + List badRequestErrors = new ArrayList<>(); + String accept = ctx.request().getHeader(ACCEPT); + if (accept == null || !accept.contains(TEXT_EVENT_STREAM)) { + badRequestErrors.add("text/event-stream required in Accept header"); + } + String sessionId = ctx.request().getHeader(HttpHeaders.MCP_SESSION_ID); + if (sessionId == null || sessionId.isBlank()) { + badRequestErrors.add("Session ID required in " + HttpHeaders.MCP_SESSION_ID + " header"); + } + if (respondBadRequest(connection, ctx, badRequestErrors)) { + return; + } + McpStreamableServerSession session = sessions.get(sessionId); + if (session == null) { + endWithStatus(connection, ctx, 404); + return; + } + + VertxMcpSessionTransport transport = startSseResponse(ctx, connection, sessionId); + String lastEventId = ctx.request().getHeader(HttpHeaders.LAST_EVENT_ID); + if (lastEventId != null) { + try { + session.replay(lastEventId).toIterable().forEach(message -> transport.sendMessage(message).block()); + } catch (Exception e) { + LOG.warn("Failed to replay MCP messages: {}", e.getMessage()); + transport.close(); + } + } else { + McpStreamableServerSession.McpStreamableServerSessionStream listeningStream + = session.listeningStream(transport); + connection.runOnContext(v -> ctx.response().closeHandler(x -> listeningStream.close())); + } + } + + private void handleDelete(RoutingContext ctx, Context connection) { + if (closing) { + endWithStatus(connection, ctx, 503); + return; + } + String sessionId = ctx.request().getHeader(HttpHeaders.MCP_SESSION_ID); + if (sessionId == null || sessionId.isBlank()) { + respondError(connection, ctx, 400, McpError.builder(McpSchema.ErrorCodes.METHOD_NOT_FOUND) + .message("Session ID required in " + HttpHeaders.MCP_SESSION_ID + " header").build()); + return; + } + McpStreamableServerSession session = sessions.get(sessionId); + if (session == null) { + endWithStatus(connection, ctx, 404); + return; + } + session.delete().block(); + sessions.remove(sessionId); + endWithStatus(connection, ctx, 200); + } + + private VertxMcpSessionTransport startSseResponse(RoutingContext ctx, Context connection, String sessionId) { + connection.runOnContext(v -> ctx.response() + .setChunked(true) + .putHeader("Content-Type", TEXT_EVENT_STREAM) + .putHeader("Cache-Control", "no-cache")); + return new VertxMcpSessionTransport(sessionId, ctx.response(), connection); + } + + private boolean respondBadRequest(Context connection, RoutingContext ctx, List errors) { + if (errors.isEmpty()) { + return false; + } + respondError(connection, ctx, 400, + McpError.builder(McpSchema.ErrorCodes.METHOD_NOT_FOUND).message(String.join("; ", errors)).build()); + return true; + } + + private void respondError(Context connection, RoutingContext ctx, int status, McpError error) { + String json; + try { + json = jsonMapper.writeValueAsString(error); + } catch (Exception e) { + json = "{}"; + } + String body = json; + connection.runOnContext(v -> ctx.response() + .setStatusCode(status) + .putHeader("Content-Type", APPLICATION_JSON) + .end(body)); + } + + private void endWithStatus(Context connection, RoutingContext ctx, int status) { + connection.runOnContext(v -> ctx.response().setStatusCode(status).end()); + } + + /** + * Per-connection transport writing SSE frames on the connection's event-loop context. The SDK session awaits each + * write, providing natural backpressure. + */ + private final class VertxMcpSessionTransport implements McpStreamableServerTransport { + + private final String sessionId; + private final HttpServerResponse response; + private final Context connection; + private volatile boolean closed; + + private VertxMcpSessionTransport(String sessionId, HttpServerResponse response, Context connection) { + this.sessionId = sessionId; + this.response = response; + this.connection = connection; + } + + @Override + public Mono sendMessage(McpSchema.JSONRPCMessage message) { + return sendMessage(message, null); + } + + @Override + public Mono sendMessage(McpSchema.JSONRPCMessage message, String messageId) { + return Mono.create(sink -> connection.runOnContext(v -> { + if (closed || response.ended() || response.closed()) { + sink.success(); + return; + } + try { + String json = jsonMapper.writeValueAsString(message); + String frame = "id: " + (messageId != null ? messageId : sessionId) + "\n" + + "event: " + MESSAGE_EVENT_TYPE + "\n" + + "data: " + json + "\n\n"; + response.write(frame).onComplete(result -> { + if (!result.succeeded()) { + LOG.debug("Failed to write to MCP session {}: {}", sessionId, + result.cause() != null ? result.cause().getMessage() : "unknown"); + closed = true; + // the client is gone: drop the session like the SDK servlet transport does + sessions.remove(sessionId); + } + sink.success(); + }); + } catch (Exception e) { + LOG.warn("Failed to send message to MCP session {}: {}", sessionId, e.getMessage()); + closed = true; + sink.success(); + } + })); + } + + @Override + public T unmarshalFrom(Object data, TypeRef typeRef) { + return jsonMapper.convertValue(data, typeRef); + } + + @Override + public Mono closeGracefully() { + return Mono.fromRunnable(this::close); + } + + @Override + public void close() { + if (closed) { + return; + } + closed = true; + connection.runOnContext(v -> { + if (!response.ended() && !response.closed()) { + response.end(); + } + }); + } + } +} diff --git a/components/camel-ai/camel-mcp-server/src/test/java/org/apache/camel/component/mcp/server/vertx/VertxMcpServerConformanceTest.java b/components/camel-ai/camel-mcp-server/src/test/java/org/apache/camel/component/mcp/server/vertx/VertxMcpServerConformanceTest.java new file mode 100644 index 0000000000000..572cbc15e273a --- /dev/null +++ b/components/camel-ai/camel-mcp-server/src/test/java/org/apache/camel/component/mcp/server/vertx/VertxMcpServerConformanceTest.java @@ -0,0 +1,44 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.mcp.server.vertx; + +import org.apache.camel.CamelContext; +import org.apache.camel.component.mcp.server.conformance.McpServerConformanceTestSupport; +import org.apache.camel.component.platform.http.vertx.VertxPlatformHttpServer; +import org.apache.camel.component.platform.http.vertx.VertxPlatformHttpServerConfiguration; +import org.apache.camel.test.AvailablePortFinder; + +/** + * Runs the engine conformance kit against {@link VertxMcpServerEngine} serving through a standalone Vert.x platform + * HTTP server (the same serving path as the Camel main HTTP server). + */ +class VertxMcpServerConformanceTest extends McpServerConformanceTestSupport { + + private final int port = AvailablePortFinder.getNextAvailable(); + + @Override + protected void customizeCamelContext(CamelContext camelContext) throws Exception { + VertxPlatformHttpServerConfiguration configuration = new VertxPlatformHttpServerConfiguration(); + configuration.setBindPort(port); + camelContext.addService(new VertxPlatformHttpServer(configuration)); + } + + @Override + protected String mcpServerBaseUrl() { + return "http://localhost:" + port; + } +} diff --git a/components/camel-ai/camel-mcp-server/src/test/java/org/apache/camel/component/mcp/server/vertx/integration/MainHttpServerMcpConformanceIT.java b/components/camel-ai/camel-mcp-server/src/test/java/org/apache/camel/component/mcp/server/vertx/integration/MainHttpServerMcpConformanceIT.java new file mode 100644 index 0000000000000..39234e2e3c6e8 --- /dev/null +++ b/components/camel-ai/camel-mcp-server/src/test/java/org/apache/camel/component/mcp/server/vertx/integration/MainHttpServerMcpConformanceIT.java @@ -0,0 +1,46 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.mcp.server.vertx.integration; + +import org.apache.camel.CamelContext; +import org.apache.camel.component.mcp.server.conformance.McpServerConformanceTestSupport; +import org.apache.camel.component.platform.http.main.MainHttpServer; +import org.apache.camel.test.AvailablePortFinder; + +/** + * Runs the engine conformance kit against the Camel main HTTP server ({@code camel-platform-http-main}) — the actual + * serving path of a Camel Main / JBang application with {@code camel.server.enabled=true}, as opposed to the bare + * {@code VertxPlatformHttpServer} used by the unit-level conformance test. + */ +class MainHttpServerMcpConformanceIT extends McpServerConformanceTestSupport { + + private final int port = AvailablePortFinder.getNextAvailable(); + + @Override + protected void customizeCamelContext(CamelContext camelContext) throws Exception { + MainHttpServer server = new MainHttpServer(); + server.setCamelContext(camelContext); + server.setHost("0.0.0.0"); + server.setPort(port); + camelContext.addService(server); + } + + @Override + protected String mcpServerBaseUrl() { + return "http://localhost:" + port; + } +} diff --git a/components/camel-ai/camel-mcp-server/src/test/java/org/apache/camel/component/mcp/server/vertx/integration/McpServerOpenAIAgentIT.java b/components/camel-ai/camel-mcp-server/src/test/java/org/apache/camel/component/mcp/server/vertx/integration/McpServerOpenAIAgentIT.java new file mode 100644 index 0000000000000..43f04c98fbad7 --- /dev/null +++ b/components/camel-ai/camel-mcp-server/src/test/java/org/apache/camel/component/mcp/server/vertx/integration/McpServerOpenAIAgentIT.java @@ -0,0 +1,157 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.mcp.server.vertx.integration; + +import org.apache.camel.CamelContext; +import org.apache.camel.Exchange; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.mcp.server.McpServerBridge; +import org.apache.camel.component.mcp.server.McpServerConfiguration; +import org.apache.camel.component.mock.MockEndpoint; +import org.apache.camel.component.openai.OpenAIComponent; +import org.apache.camel.component.openai.OpenAIConstants; +import org.apache.camel.component.platform.http.vertx.VertxPlatformHttpServer; +import org.apache.camel.component.platform.http.vertx.VertxPlatformHttpServerConfiguration; +import org.apache.camel.test.AvailablePortFinder; +import org.apache.camel.test.infra.ollama.services.OllamaService; +import org.apache.camel.test.infra.ollama.services.OllamaServiceFactory; +import org.apache.camel.test.junit6.CamelTestSupport; +import org.apache.camel.util.ObjectHelper; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.condition.DisabledIfSystemProperty; +import org.junit.jupiter.api.extension.RegisterExtension; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * End-to-end agentic integration test: the exact scenario from CAMEL-24308. This Camel application exposes its own + * {@code ai-tool} routes as MCP tools through the camel-mcp-server Vert.x engine, and an LLM (Ollama via the + * camel-openai component) discovers and calls them over MCP streamable HTTP with automatic tool execution. + */ +@DisabledIfSystemProperty(named = "ci.env.name", matches = ".*", disabledReason = "Requires too much network resources") +class McpServerOpenAIAgentIT extends CamelTestSupport { + + @RegisterExtension + static OllamaService OLLAMA = OllamaServiceFactory.createSingletonService(); + + private final int mcpPort = AvailablePortFinder.getNextAvailable(); + + private String apiKey; + private String baseUrl; + private String model; + + @Override + protected void setupResources() throws Exception { + super.setupResources(); + baseUrl = OLLAMA.baseUrlV1(); + model = OLLAMA.modelName(); + apiKey = OLLAMA.apiKey(); + if (apiKey == null || apiKey.isEmpty()) { + apiKey = "dummy"; + } + } + + @Override + protected CamelContext createCamelContext() throws Exception { + CamelContext camelContext = super.createCamelContext(); + + OpenAIComponent component = new OpenAIComponent(); + if (ObjectHelper.isNotEmpty(apiKey)) { + component.setApiKey(apiKey); + } + if (ObjectHelper.isNotEmpty(model)) { + component.setModel(model); + } + if (ObjectHelper.isNotEmpty(baseUrl)) { + component.setBaseUrl(baseUrl); + } + camelContext.addComponent("openai", component); + + // this application serves its own MCP endpoint on the Vert.x platform HTTP server + VertxPlatformHttpServerConfiguration serverConfiguration = new VertxPlatformHttpServerConfiguration(); + serverConfiguration.setBindPort(mcpPort); + camelContext.addService(new VertxPlatformHttpServer(serverConfiguration)); + + McpServerConfiguration mcpConfiguration = new McpServerConfiguration(); + mcpConfiguration.setTags("agent"); + camelContext.addService(new McpServerBridge(mcpConfiguration)); + + return camelContext; + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + // the tools this application exposes over MCP + from("ai-tool:get_weather?tags=agent&description=Get the current weather for a city" + + "¶meter.city=string¶meter.city.description=The city name" + + "¶meter.city.required=true") + .to("mock:weather-called") + .setBody(simple("Sunny in ${header.city}, 21 degrees celsius")); + } + }; + } + + /** + * The agent route is added once the context is fully started: the openai producer initializes its MCP client + * eagerly on route warm-up, which happens before deferred services (the platform HTTP server and the bridge) have + * started when the route is part of the initial context. + */ + private void addAgentRoute() throws Exception { + context.addRoutes(new RouteBuilder() { + @Override + public void configure() { + from("direct:agent") + .toF("openai:chat-completion" + + "?mcpServer.camelTools.transportType=streamableHttp" + + "&mcpServer.camelTools.url=http://localhost:%d/mcp", + mcpPort) + .to("mock:response"); + } + }); + } + + @Test + void testLlmCallsCamelRouteToolOverMcp() throws Exception { + addAgentRoute(); + + MockEndpoint weatherCalled = getMockEndpoint("mock:weather-called"); + weatherCalled.expectedMinimumMessageCount(1); + MockEndpoint response = getMockEndpoint("mock:response"); + response.expectedMessageCount(1); + + Exchange result = template.request("direct:agent", + e -> e.getIn().setBody("Use the get_weather tool to find the current weather in Paris.")); + + MockEndpoint.assertIsSatisfied(context); + + // the ai-tool route was really invoked through MCP + assertThat(weatherCalled.getExchanges().get(0).getIn().getHeader("city", String.class)) + .isEqualToIgnoringCase("Paris"); + + // the agentic loop executed at least one MCP tool call + Integer iterations = result.getMessage().getHeader(OpenAIConstants.TOOL_ITERATIONS, Integer.class); + assertThat(iterations).isNotNull().isGreaterThanOrEqualTo(1); + + // the tool result made it back into the LLM answer + String answer = result.getMessage().getBody(String.class); + assertThat(answer).isNotNull(); + assertThat(answer.toLowerCase()).containsAnyOf("sunny", "21"); + } +} diff --git a/components/camel-ai/camel-mcp-server/test-execution.md b/components/camel-ai/camel-mcp-server/test-execution.md new file mode 100644 index 0000000000000..5aa3d1449cd05 --- /dev/null +++ b/components/camel-ai/camel-mcp-server/test-execution.md @@ -0,0 +1,38 @@ +# camel-mcp-server test execution + +## Unit tests + +```bash +mvn test +``` + +Runs the bridge tests and the engine conformance test (`VertxMcpServerConformanceTest`) +against a standalone Vert.x platform HTTP server, driven by the official MCP Java SDK +client. No Docker required. + +## Integration tests + +```bash +mvn verify +``` + +- `MainHttpServerMcpConformanceIT` — the conformance kit against the Camel main HTTP + server (`camel-platform-http-main`), the real Camel Main / JBang serving path. + No Docker required. +- `McpServerOpenAIAgentIT` — end-to-end agentic loop: the application exposes its own + `ai-tool` routes over MCP and an LLM (camel-openai) discovers and calls them with + automatic tool execution. Requires Docker (Ollama testcontainer, model + `granite4:3b`) or a local Ollama; disabled on CI (`ci.env.name`). + +### LLM backend selection (same options as camel-openai) + +```bash +# reuse a running Ollama instead of a container +mvn verify -Dollama.instance.type=remote -Dollama.endpoint=http://localhost:11434 -Dollama.model=granite4:3b + +# run against the real OpenAI API +mvn verify -Dollama.instance.type=openai -Dopenai.api.key=sk-... + +# enable GPU for the Ollama container +mvn verify -Dollama.container.enable.gpu=enabled +``` diff --git a/components/camel-ai/pom.xml b/components/camel-ai/pom.xml index 05aa67814618d..8325bc01e2c0e 100644 --- a/components/camel-ai/pom.xml +++ b/components/camel-ai/pom.xml @@ -51,6 +51,8 @@ camel-langchain4j-tokenizer camel-langchain4j-tools camel-langchain4j-web-search + camel-mcp-server-api + camel-mcp-server camel-milvus camel-neo4j camel-openai diff --git a/docs/components/modules/others/examples/json/mcp-server-api.json b/docs/components/modules/others/examples/json/mcp-server-api.json new file mode 120000 index 0000000000000..d59b543bb689c --- /dev/null +++ b/docs/components/modules/others/examples/json/mcp-server-api.json @@ -0,0 +1 @@ +../../../../../../components/camel-ai/camel-mcp-server-api/src/generated/resources/mcp-server-api.json \ No newline at end of file diff --git a/docs/components/modules/others/examples/json/mcp-server.json b/docs/components/modules/others/examples/json/mcp-server.json new file mode 120000 index 0000000000000..93c9007e2a457 --- /dev/null +++ b/docs/components/modules/others/examples/json/mcp-server.json @@ -0,0 +1 @@ +../../../../../../components/camel-ai/camel-mcp-server/src/generated/resources/mcp-server.json \ No newline at end of file diff --git a/docs/components/modules/others/nav.adoc b/docs/components/modules/others/nav.adoc index 9ff3c325c7fa8..093949c595f5e 100644 --- a/docs/components/modules/others/nav.adoc +++ b/docs/components/modules/others/nav.adoc @@ -50,6 +50,7 @@ ** xref:lra.adoc[LRA] ** xref:mail-microsoft-oauth.adoc[Mail Microsoft Oauth] ** xref:main.adoc[Main] +** xref:mcp-server.adoc[MCP Server] ** xref:mdc.adoc[MDC Logging] ** xref:observation.adoc[Micrometer Observability] ** xref:micrometer-observability.adoc[Micrometer Observability 2] diff --git a/docs/components/modules/others/pages/mcp-server.adoc b/docs/components/modules/others/pages/mcp-server.adoc new file mode 120000 index 0000000000000..8aa49a42432c7 --- /dev/null +++ b/docs/components/modules/others/pages/mcp-server.adoc @@ -0,0 +1 @@ +../../../../../components/camel-ai/camel-mcp-server/src/main/docs/mcp-server.adoc \ No newline at end of file diff --git a/dsl/camel-kamelet-main/src/generated/resources/camel-factoryfinder-known-dependencies.properties b/dsl/camel-kamelet-main/src/generated/resources/camel-factoryfinder-known-dependencies.properties index dfb61a3383e3d..1ce0ccf39027e 100644 --- a/dsl/camel-kamelet-main/src/generated/resources/camel-factoryfinder-known-dependencies.properties +++ b/dsl/camel-kamelet-main/src/generated/resources/camel-factoryfinder-known-dependencies.properties @@ -35,6 +35,7 @@ META-INF/services/org/apache/camel/kafka-adapter-factory=camel:kafka META-INF/services/org/apache/camel/kafka-resume-strategy=camel:kafka META-INF/services/org/apache/camel/kinesis-resume-strategy=camel:aws2-kinesis META-INF/services/org/apache/camel/lra-saga-service=camel:lra +META-INF/services/org/apache/camel/mcp-server-engine=camel:mcp-server META-INF/services/org/apache/camel/mdc-service=camel:mdc META-INF/services/org/apache/camel/micrometer-observability-tracer=camel:micrometer-observability META-INF/services/org/apache/camel/micrometer-prometheus=camel:micrometer-prometheus diff --git a/parent/pom.xml b/parent/pom.xml index 5dee39657d19d..58956a0d206ea 100644 --- a/parent/pom.xml +++ b/parent/pom.xml @@ -2111,6 +2111,16 @@ camel-master ${project.version} + + org.apache.camel + camel-mcp-server + ${project.version} + + + org.apache.camel + camel-mcp-server-api + ${project.version} + org.apache.camel camel-mdc @@ -3235,6 +3245,12 @@ ${project.version} test-jar + + org.apache.camel + camel-mcp-server-api + ${project.version} + test-jar + diff --git a/tooling/maven/camel-package-maven-plugin/src/main/java/org/apache/camel/maven/packaging/MojoHelper.java b/tooling/maven/camel-package-maven-plugin/src/main/java/org/apache/camel/maven/packaging/MojoHelper.java index c4619ba1cc944..678ec0b56e157 100644 --- a/tooling/maven/camel-package-maven-plugin/src/main/java/org/apache/camel/maven/packaging/MojoHelper.java +++ b/tooling/maven/camel-package-maven-plugin/src/main/java/org/apache/camel/maven/packaging/MojoHelper.java @@ -45,6 +45,7 @@ public static List getComponentPath(Path dir) { dir.resolve("camel-langchain4j-embeddings"), dir.resolve("camel-langchain4j-embeddingstore"), dir.resolve("camel-langchain4j-tokenizer"), dir.resolve("camel-langchain4j-tools"), dir.resolve("camel-langchain4j-web-search"), + dir.resolve("camel-mcp-server"), dir.resolve("camel-qdrant"), dir.resolve("camel-milvus"), dir.resolve("camel-neo4j"), dir.resolve("camel-openai"), dir.resolve("camel-pgvector"), dir.resolve("camel-pinecone"), dir.resolve("camel-kserve"),