diff --git a/components/camel-ai/camel-ai-tool/src/main/java/org/apache/camel/component/ai/tool/AiToolRegistry.java b/components/camel-ai/camel-ai-tool/src/main/java/org/apache/camel/component/ai/tool/AiToolRegistry.java index c47562fd820c6..8c3d71372e935 100644 --- a/components/camel-ai/camel-ai-tool/src/main/java/org/apache/camel/component/ai/tool/AiToolRegistry.java +++ b/components/camel-ai/camel-ai-tool/src/main/java/org/apache/camel/component/ai/tool/AiToolRegistry.java @@ -19,11 +19,15 @@ import java.util.HashMap; import java.util.LinkedHashMap; import java.util.LinkedHashSet; +import java.util.List; import java.util.Map; import java.util.Set; +import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.locks.ReentrantLock; import org.apache.camel.CamelContext; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** * CamelContext-scoped registry mapping tags to {@link AiToolSpec} instances. AI components (LangChain4j, Spring AI) @@ -32,6 +36,9 @@ * Each {@link CamelContext} gets its own registry instance, registered as a context plugin. Use * {@link #getOrCreate(CamelContext)} to obtain the instance for a given context. *

+ * Adapters that need to react to tools appearing or disappearing (e.g. to push MCP {@code tools/list_changed} + * notifications) can register an {@link AiToolRegistryListener} instead of polling. + *

* Replaces the duplicated {@code CamelToolExecutorCache} singletons from {@code camel-langchain4j-tools} and * {@code camel-spring-ai-tools}. * @@ -39,11 +46,14 @@ */ public final class AiToolRegistry { + private static final Logger LOG = LoggerFactory.getLogger(AiToolRegistry.class); + private static final ReentrantLock FACTORY_LOCK = new ReentrantLock(); private final ReentrantLock lock = new ReentrantLock(); private final Map> tools; private final Set defaultTools; + private final List listeners = new CopyOnWriteArrayList<>(); AiToolRegistry() { tools = new HashMap<>(); @@ -71,6 +81,7 @@ public static AiToolRegistry getOrCreate(CamelContext context) { } public void put(String tag, AiToolSpec spec) { + boolean added; lock.lock(); try { Set set = tools.computeIfAbsent(tag, k -> new LinkedHashSet<>()); @@ -81,18 +92,22 @@ public void put(String tag, AiToolSpec spec) { + "': tool names must be unique per tag"); } } - set.add(spec); + added = set.add(spec); } finally { lock.unlock(); } + if (added) { + notifyRegistered(tag, spec); + } } public void remove(String tag, AiToolSpec spec) { + boolean removed = false; lock.lock(); try { Set set = tools.get(tag); if (set != null) { - set.remove(spec); + removed = set.remove(spec); if (set.isEmpty()) { tools.remove(tag); } @@ -100,9 +115,13 @@ public void remove(String tag, AiToolSpec spec) { } finally { lock.unlock(); } + if (removed) { + notifyDeregistered(tag, spec); + } } public void putDefault(AiToolSpec spec) { + boolean added; lock.lock(); try { for (AiToolSpec existing : defaultTools) { @@ -112,19 +131,60 @@ public void putDefault(AiToolSpec spec) { + "' in the default pool: tool names must be unique"); } } - defaultTools.add(spec); + added = defaultTools.add(spec); } finally { lock.unlock(); } + if (added) { + notifyRegistered(null, spec); + } } public void removeDefault(AiToolSpec spec) { + boolean removed; lock.lock(); try { - defaultTools.remove(spec); + removed = defaultTools.remove(spec); } finally { lock.unlock(); } + if (removed) { + notifyDeregistered(null, spec); + } + } + + /** + * Adds a listener notified on tool registration changes. See {@link AiToolRegistryListener} for the callback + * contract and the subscribe-then-snapshot idiom to observe current state without missing events. + */ + public void addListener(AiToolRegistryListener listener) { + listeners.add(listener); + } + + public void removeListener(AiToolRegistryListener listener) { + listeners.remove(listener); + } + + private void notifyRegistered(String tag, AiToolSpec spec) { + for (AiToolRegistryListener listener : listeners) { + try { + listener.toolRegistered(tag, spec); + } catch (Exception e) { + LOG.warn("AiToolRegistryListener {} failed on toolRegistered for tool '{}': {}", + listener, spec.getName(), e.getMessage(), e); + } + } + } + + private void notifyDeregistered(String tag, AiToolSpec spec) { + for (AiToolRegistryListener listener : listeners) { + try { + listener.toolDeregistered(tag, spec); + } catch (Exception e) { + LOG.warn("AiToolRegistryListener {} failed on toolDeregistered for tool '{}': {}", + listener, spec.getName(), e.getMessage(), e); + } + } } /** diff --git a/components/camel-ai/camel-ai-tool/src/main/java/org/apache/camel/component/ai/tool/AiToolRegistryListener.java b/components/camel-ai/camel-ai-tool/src/main/java/org/apache/camel/component/ai/tool/AiToolRegistryListener.java new file mode 100644 index 0000000000000..378842f53e270 --- /dev/null +++ b/components/camel-ai/camel-ai-tool/src/main/java/org/apache/camel/component/ai/tool/AiToolRegistryListener.java @@ -0,0 +1,52 @@ +/* + * 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.ai.tool; + +/** + * Listener notified when tool specifications are registered to or deregistered from an {@link AiToolRegistry}. + *

+ * Registration events are driven by the {@code ai-tool} consumer lifecycle: a tool is registered when its route starts + * or resumes, and deregistered when its route stops or suspends. A tool endpoint declaring multiple tags fires one + * event per tag. + *

+ * Callbacks are invoked outside the registry lock, on the thread performing the (de)registration — typically a route + * lifecycle thread. Implementations must be thread-safe and non-blocking; a callback that throws is logged and does not + * affect the registration itself or other listeners. + *

+ * To observe the current registry state without missing concurrent changes, add the listener first and then read a + * snapshot (e.g. {@link AiToolRegistry#getTools()}), tolerating events that duplicate snapshot content. + * + * @since 4.22 + */ +public interface AiToolRegistryListener { + + /** + * Called after a tool specification has been registered. + * + * @param tag the tag the tool was registered under, or {@code null} for the default (untagged) pool + * @param spec the registered tool specification + */ + void toolRegistered(String tag, AiToolSpec spec); + + /** + * Called after a tool specification has been deregistered. + * + * @param tag the tag the tool was deregistered from, or {@code null} for the default (untagged) pool + * @param spec the deregistered tool specification + */ + void toolDeregistered(String tag, AiToolSpec spec); +} diff --git a/components/camel-ai/camel-ai-tool/src/test/java/org/apache/camel/component/ai/tool/AiToolRegistryListenerLifecycleTest.java b/components/camel-ai/camel-ai-tool/src/test/java/org/apache/camel/component/ai/tool/AiToolRegistryListenerLifecycleTest.java new file mode 100644 index 0000000000000..bba34ceea07de --- /dev/null +++ b/components/camel-ai/camel-ai-tool/src/test/java/org/apache/camel/component/ai/tool/AiToolRegistryListenerLifecycleTest.java @@ -0,0 +1,144 @@ +/* + * 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.ai.tool; + +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; + +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; +import static org.assertj.core.api.Assertions.tuple; + +/** + * Verifies that {@link AiToolRegistryListener} callbacks are driven by the {@code ai-tool} consumer lifecycle: route + * start/resume registers, route stop/suspend deregisters. + */ +class AiToolRegistryListenerLifecycleTest extends CamelTestSupport { + + private final RecordingListener listener = new RecordingListener(); + + @Override + protected CamelContext createCamelContext() throws Exception { + CamelContext camelContext = super.createCamelContext(); + AiToolRegistry.getOrCreate(camelContext).addListener(listener); + return camelContext; + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + public void configure() { + from("ai-tool:getWeather?tags=weather&description=Get the weather") + .routeId("weather-route") + .setBody(constant("sunny")); + } + }; + } + + @Test + void testEventsOnRouteStartAndStop() throws Exception { + assertThat(listener.events) + .as("Route start should fire toolRegistered") + .extracting(Event::type, Event::tag, Event::toolName) + .containsExactly(tuple("registered", "weather", "getWeather")); + + context.getRouteController().stopRoute("weather-route"); + + assertThat(listener.events) + .extracting(Event::type, Event::tag, Event::toolName) + .containsExactly( + tuple("registered", "weather", "getWeather"), + tuple("deregistered", "weather", "getWeather")); + } + + @Test + void testEventsOnSuspendAndResume() throws Exception { + context.getRouteController().suspendRoute("weather-route"); + context.getRouteController().resumeRoute("weather-route"); + + assertThat(listener.events) + .extracting(Event::type, Event::tag, Event::toolName) + .containsExactly( + tuple("registered", "weather", "getWeather"), + tuple("deregistered", "weather", "getWeather"), + tuple("registered", "weather", "getWeather")); + } + + @Test + void testMultiTagEndpointFiresOneEventPerTag() throws Exception { + context.addRoutes(new RouteBuilder() { + public void configure() { + from("ai-tool:sendEmail?tags=notify,crm&description=Send an email") + .routeId("email-route") + .setBody(constant("sent")); + } + }); + + assertThat(listener.events) + .filteredOn(e -> "sendEmail".equals(e.toolName())) + .extracting(Event::type, Event::tag) + .containsExactlyInAnyOrder( + tuple("registered", "notify"), + tuple("registered", "crm")); + } + + @Test + void testUntaggedEndpointFiresDefaultPoolEvent() throws Exception { + context.addRoutes(new RouteBuilder() { + public void configure() { + from("ai-tool:lookupOrder?description=Look up an order") + .routeId("order-route") + .setBody(constant("order")); + } + }); + + assertThat(listener.events) + .filteredOn(e -> "lookupOrder".equals(e.toolName())) + .extracting(Event::type, Event::tag) + .containsExactly(tuple("registered", null)); + + context.getRouteController().stopRoute("order-route"); + + assertThat(listener.events) + .filteredOn(e -> "lookupOrder".equals(e.toolName())) + .extracting(Event::type, Event::tag) + .containsExactly( + tuple("registered", null), + tuple("deregistered", null)); + } + + private record Event(String type, String tag, String toolName) { + } + + private static final class RecordingListener implements AiToolRegistryListener { + private final List events = new CopyOnWriteArrayList<>(); + + @Override + public void toolRegistered(String tag, AiToolSpec spec) { + events.add(new Event("registered", tag, spec.getName())); + } + + @Override + public void toolDeregistered(String tag, AiToolSpec spec) { + events.add(new Event("deregistered", tag, spec.getName())); + } + } +} diff --git a/components/camel-ai/camel-ai-tool/src/test/java/org/apache/camel/component/ai/tool/AiToolRegistryListenerTest.java b/components/camel-ai/camel-ai-tool/src/test/java/org/apache/camel/component/ai/tool/AiToolRegistryListenerTest.java new file mode 100644 index 0000000000000..3d1c5e009acf0 --- /dev/null +++ b/components/camel-ai/camel-ai-tool/src/test/java/org/apache/camel/component/ai/tool/AiToolRegistryListenerTest.java @@ -0,0 +1,158 @@ +/* + * 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.ai.tool; + +import java.util.ArrayList; +import java.util.List; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +class AiToolRegistryListenerTest { + + private AiToolRegistry registry; + private RecordingListener listener; + + @BeforeEach + void setUp() { + registry = new AiToolRegistry(); + listener = new RecordingListener(); + registry.addListener(listener); + } + + @Test + void testRegisteredEventOnPut() { + AiToolSpec spec = spec("getWeather"); + registry.put("weather", spec); + + assertThat(listener.events).containsExactly(new Event("registered", "weather", spec)); + } + + @Test + void testNoDuplicateEventOnRepeatedPutOfSameSpec() { + AiToolSpec spec = spec("getWeather"); + registry.put("weather", spec); + registry.put("weather", spec); + + assertThat(listener.events) + .as("Re-adding the same spec instance should not fire a second event") + .hasSize(1); + } + + @Test + void testDeregisteredEventOnRemove() { + AiToolSpec spec = spec("getWeather"); + registry.put("weather", spec); + registry.remove("weather", spec); + + assertThat(listener.events).containsExactly( + new Event("registered", "weather", spec), + new Event("deregistered", "weather", spec)); + } + + @Test + void testNoEventOnRemovingAbsentSpec() { + registry.remove("weather", spec("getWeather")); + registry.removeDefault(spec("getWeather")); + + assertThat(listener.events) + .as("Removing a spec that was never registered should not fire events") + .isEmpty(); + } + + @Test + void testDefaultPoolEventsUseNullTag() { + AiToolSpec spec = spec("getWeather"); + registry.putDefault(spec); + registry.removeDefault(spec); + + assertThat(listener.events).containsExactly( + new Event("registered", null, spec), + new Event("deregistered", null, spec)); + } + + @Test + void testNoEventWhenPutThrowsOnDuplicateName() { + registry.put("weather", spec("getWeather")); + listener.events.clear(); + + assertThatThrownBy(() -> registry.put("weather", spec("getWeather"))) + .isInstanceOf(IllegalArgumentException.class); + assertThat(listener.events) + .as("A rejected registration should not fire an event") + .isEmpty(); + } + + @Test + void testListenerExceptionDoesNotBreakRegistrationOrOtherListeners() { + registry.addListener(new AiToolRegistryListener() { + @Override + public void toolRegistered(String tag, AiToolSpec spec) { + throw new IllegalStateException("boom"); + } + + @Override + public void toolDeregistered(String tag, AiToolSpec spec) { + throw new IllegalStateException("boom"); + } + }); + RecordingListener second = new RecordingListener(); + registry.addListener(second); + + AiToolSpec spec = spec("getWeather"); + registry.put("weather", spec); + + assertThat(registry.getToolsByTag("weather")) + .as("Registration should succeed despite a failing listener") + .contains(spec); + assertThat(second.events) + .as("Listeners after the failing one should still be notified") + .containsExactly(new Event("registered", "weather", spec)); + } + + @Test + void testRemovedListenerReceivesNoFurtherEvents() { + registry.removeListener(listener); + registry.put("weather", spec("getWeather")); + + assertThat(listener.events).isEmpty(); + } + + private static AiToolSpec spec(String name) { + return new AiToolSpec(name, name + " description", null, null, null); + } + + private record Event(String type, String tag, AiToolSpec spec) { + } + + private static final class RecordingListener implements AiToolRegistryListener { + private final List events = new ArrayList<>(); + + @Override + public void toolRegistered(String tag, AiToolSpec spec) { + events.add(new Event("registered", tag, spec)); + } + + @Override + public void toolDeregistered(String tag, AiToolSpec spec) { + events.add(new Event("deregistered", tag, spec)); + } + } +}