diff --git a/runtime/binding-mcp-http/src/main/java/io/aklivity/zilla/runtime/binding/mcp/http/internal/config/McpHttpBindingConfig.java b/runtime/binding-mcp-http/src/main/java/io/aklivity/zilla/runtime/binding/mcp/http/internal/config/McpHttpBindingConfig.java index 8a0f6c78b9..6000cb5a91 100644 --- a/runtime/binding-mcp-http/src/main/java/io/aklivity/zilla/runtime/binding/mcp/http/internal/config/McpHttpBindingConfig.java +++ b/runtime/binding-mcp-http/src/main/java/io/aklivity/zilla/runtime/binding/mcp/http/internal/config/McpHttpBindingConfig.java @@ -14,6 +14,8 @@ */ package io.aklivity.zilla.runtime.binding.mcp.http.internal.config; +import static io.aklivity.zilla.runtime.binding.mcp.http.internal.types.McpCapabilities.SERVER_RESOURCES; +import static io.aklivity.zilla.runtime.binding.mcp.http.internal.types.McpCapabilities.SERVER_TOOLS; import static java.util.stream.Collectors.toList; import static java.util.stream.Collectors.toMap; @@ -234,6 +236,24 @@ public Collection resources() return resourcesByName.values(); } + // real, declared capabilities -- this binding is configured with whichever of tools/resources + // its own routes actually resolve to (McpOpenapiCompositeGenerator only ever emits one or the + // other, or both), never prompts, so this reflects real config rather than echoing back + // whatever a connecting north forwarded for its own client's elicitation support + public int serverCapabilities() + { + int bits = 0; + if (!toolsByName.isEmpty()) + { + bits |= SERVER_TOOLS.value(); + } + if (!resourcesByName.isEmpty()) + { + bits |= SERVER_RESOURCES.value(); + } + return bits; + } + public byte[] toolsListJson() { return toolsListJson; diff --git a/runtime/binding-mcp-http/src/main/java/io/aklivity/zilla/runtime/binding/mcp/http/internal/stream/McpHttpProxyFactory.java b/runtime/binding-mcp-http/src/main/java/io/aklivity/zilla/runtime/binding/mcp/http/internal/stream/McpHttpProxyFactory.java index 1e49bd6161..ea25f6b2bf 100644 --- a/runtime/binding-mcp-http/src/main/java/io/aklivity/zilla/runtime/binding/mcp/http/internal/stream/McpHttpProxyFactory.java +++ b/runtime/binding-mcp-http/src/main/java/io/aklivity/zilla/runtime/binding/mcp/http/internal/stream/McpHttpProxyFactory.java @@ -232,9 +232,8 @@ public MessageConsumer newStream( if (kind == KIND_LIFECYCLE) { - final int capabilities = beginEx.lifecycle().capabilities(); final String sessionId = newSessionId(); - newStream = new McpSession(sessionId, capabilities, + newStream = new McpSession(sessionId, binding.serverCapabilities(), sender, originId, routedId, initialId, authorization, affinity)::onMcpMessage; } else @@ -2279,6 +2278,9 @@ private void doMcpReplyBegin( { if (!McpHttpState.replyOpened(state)) { + // capabilities is this binding's own declared real capabilities (McpHttpBindingConfig. + // serverCapabilities, computed from its configured tools/resources), not an echo of + // whatever the connecting north forwarded for its own client's elicitation support final McpBeginExFW lifecycleEx = mcpBeginExRW.wrap(extBuffer, 0, extBuffer.capacity()) .typeId(mcpTypeId) .lifecycle(l -> l diff --git a/runtime/binding-mcp-http/src/test/java/io/aklivity/zilla/runtime/binding/mcp/http/internal/config/McpHttpBindingConfigTest.java b/runtime/binding-mcp-http/src/test/java/io/aklivity/zilla/runtime/binding/mcp/http/internal/config/McpHttpBindingConfigTest.java index e574b4fb71..42b0b8b80f 100644 --- a/runtime/binding-mcp-http/src/test/java/io/aklivity/zilla/runtime/binding/mcp/http/internal/config/McpHttpBindingConfigTest.java +++ b/runtime/binding-mcp-http/src/test/java/io/aklivity/zilla/runtime/binding/mcp/http/internal/config/McpHttpBindingConfigTest.java @@ -15,6 +15,8 @@ package io.aklivity.zilla.runtime.binding.mcp.http.internal.config; import static io.aklivity.zilla.runtime.binding.mcp.http.internal.config.McpHttpBindingConfig.argPathValid; +import static io.aklivity.zilla.runtime.binding.mcp.http.internal.types.McpCapabilities.SERVER_RESOURCES; +import static io.aklivity.zilla.runtime.binding.mcp.http.internal.types.McpCapabilities.SERVER_TOOLS; import static org.hamcrest.MatcherAssert.assertThat; import static org.hamcrest.Matchers.empty; import static org.hamcrest.Matchers.equalTo; @@ -29,6 +31,9 @@ import org.junit.Test; import io.aklivity.zilla.config.binding.mcp.http.McpHttpConditionConfig; +import io.aklivity.zilla.config.binding.mcp.http.McpHttpOptionsConfig; +import io.aklivity.zilla.config.binding.mcp.http.McpHttpResourceConfig; +import io.aklivity.zilla.config.binding.mcp.http.McpHttpToolConfig; import io.aklivity.zilla.config.binding.mcp.http.McpHttpWithConfig; import io.aklivity.zilla.config.engine.BindingConfig; import io.aklivity.zilla.config.engine.GenericBindingConfig; @@ -78,6 +83,20 @@ private static McpHttpBindingConfig binding( return new McpHttpBindingConfig(config, null, "sys:http_client"); } + private static McpHttpBindingConfig bindingWithOptions( + McpHttpOptionsConfig options) + { + BindingConfig config = GenericBindingConfig.builder() + .namespace("test") + .name("app0") + .type("mcp-http") + .kind(KindConfig.PROXY) + .options(options) + .routes(List.of()) + .build(); + return new McpHttpBindingConfig(config, null, "sys:http_client"); + } + @Test public void shouldRejectToolWhenGlobalGuardOnlyLayerFails() { @@ -311,4 +330,46 @@ public void shouldAcceptWhenSchemaMalformed() { assertTrue(argPathValid("{ not json", "owner")); } + + @Test + public void shouldDeclareToolsCapabilityWhenOnlyToolsConfigured() + { + McpHttpOptionsConfig options = McpHttpOptionsConfig.builder() + .tools(List.of(McpHttpToolConfig.builder().name("create_pr").build())) + .build(); + McpHttpBindingConfig binding = bindingWithOptions(options); + + assertThat(binding.serverCapabilities(), equalTo(SERVER_TOOLS.value())); + } + + @Test + public void shouldDeclareResourcesCapabilityWhenOnlyResourcesConfigured() + { + McpHttpOptionsConfig options = McpHttpOptionsConfig.builder() + .resources(List.of(McpHttpResourceConfig.builder().name("order").uri("/orders/{id}").build())) + .build(); + McpHttpBindingConfig binding = bindingWithOptions(options); + + assertThat(binding.serverCapabilities(), equalTo(SERVER_RESOURCES.value())); + } + + @Test + public void shouldDeclareBothCapabilitiesWhenToolsAndResourcesConfigured() + { + McpHttpOptionsConfig options = McpHttpOptionsConfig.builder() + .tools(List.of(McpHttpToolConfig.builder().name("create_pr").build())) + .resources(List.of(McpHttpResourceConfig.builder().name("order").uri("/orders/{id}").build())) + .build(); + McpHttpBindingConfig binding = bindingWithOptions(options); + + assertThat(binding.serverCapabilities(), equalTo(SERVER_TOOLS.value() | SERVER_RESOURCES.value())); + } + + @Test + public void shouldDeclareNoCapabilitiesWhenNeitherToolsNorResourcesConfigured() + { + McpHttpBindingConfig binding = binding(List.of()); + + assertThat(binding.serverCapabilities(), equalTo(0)); + } } diff --git a/runtime/binding-mcp-kafka/src/main/java/io/aklivity/zilla/runtime/binding/mcp/kafka/internal/stream/McpKafkaProxyFactory.java b/runtime/binding-mcp-kafka/src/main/java/io/aklivity/zilla/runtime/binding/mcp/kafka/internal/stream/McpKafkaProxyFactory.java index 7c7475efda..4a775c4723 100644 --- a/runtime/binding-mcp-kafka/src/main/java/io/aklivity/zilla/runtime/binding/mcp/kafka/internal/stream/McpKafkaProxyFactory.java +++ b/runtime/binding-mcp-kafka/src/main/java/io/aklivity/zilla/runtime/binding/mcp/kafka/internal/stream/McpKafkaProxyFactory.java @@ -468,9 +468,8 @@ public MessageConsumer newStream( { case KIND_LIFECYCLE: { - final int capabilities = mcpBeginEx.lifecycle().capabilities(); final McpLifecycleProxy lifecycle = new McpLifecycleProxy( - sender, originId, routedId, initialId, authorization, affinity, capabilities); + sender, originId, routedId, initialId, authorization, affinity); newStream = lifecycle::onMcpMessage; break; } @@ -2083,7 +2082,6 @@ private final class McpLifecycleProxy private final long replyId; private final long authorization; private final long affinity; - private final int capabilities; private int state; @@ -2093,8 +2091,7 @@ private McpLifecycleProxy( long routedId, long initialId, long authorization, - long affinity, - int capabilities) + long affinity) { this.mcp = mcp; this.originId = originId; @@ -2103,7 +2100,6 @@ private McpLifecycleProxy( this.replyId = supplyReplyId.applyAsLong(initialId); this.authorization = authorization; this.affinity = affinity; - this.capabilities = capabilities; } private void onMcpMessage( @@ -2166,11 +2162,14 @@ private void doLifecycleReply( long traceId) { final String sessionId = supplySessionId.get(); + // this binding type is tools-only by construction (its route condition schema + // rejects prompt:/resource: selectors), so it always declares CAPABILITIES_TOOLS + // here rather than echoing whatever capabilities the connecting north forwarded final McpBeginExFW lifecycleEx = mcpBeginExRW.wrap(extBuffer, 0, extBuffer.capacity()) .typeId(mcpTypeId) .lifecycle(l -> l .sessionId(sessionId) - .capabilities(capabilities != 0 ? capabilities : CAPABILITIES_TOOLS)) + .capabilities(CAPABILITIES_TOOLS)) .build(); doBegin(mcp, originId, routedId, replyId, traceId, authorization, affinity, lifecycleEx); diff --git a/runtime/binding-mcp-kafka/src/test/java/io/aklivity/zilla/runtime/binding/mcp/kafka/internal/stream/McpKafkaProxyFactoryTest.java b/runtime/binding-mcp-kafka/src/test/java/io/aklivity/zilla/runtime/binding/mcp/kafka/internal/stream/McpKafkaProxyFactoryTest.java index a35f99f9a7..349a222478 100644 --- a/runtime/binding-mcp-kafka/src/test/java/io/aklivity/zilla/runtime/binding/mcp/kafka/internal/stream/McpKafkaProxyFactoryTest.java +++ b/runtime/binding-mcp-kafka/src/test/java/io/aklivity/zilla/runtime/binding/mcp/kafka/internal/stream/McpKafkaProxyFactoryTest.java @@ -44,6 +44,7 @@ import io.aklivity.zilla.config.engine.RouteConfig; import io.aklivity.zilla.runtime.binding.mcp.kafka.internal.McpKafkaConfiguration; import io.aklivity.zilla.runtime.binding.mcp.kafka.internal.types.KafkaOffsetFW; +import io.aklivity.zilla.runtime.binding.mcp.kafka.internal.types.McpCapabilities; import io.aklivity.zilla.runtime.binding.mcp.kafka.internal.types.stream.AbortFW; import io.aklivity.zilla.runtime.binding.mcp.kafka.internal.types.stream.BeginFW; import io.aklivity.zilla.runtime.binding.mcp.kafka.internal.types.stream.DataFW; @@ -92,6 +93,7 @@ public class McpKafkaProxyFactoryTest private final EndFW endRO = new EndFW(); private final ResetFW resetRO = new ResetFW(); private final KafkaBeginExFW kafkaBeginExRO = new KafkaBeginExFW(); + private final McpBeginExFW mcpBeginExRO = new McpBeginExFW(); private final McpResetExFW mcpResetExRO = new McpResetExFW(); private final McpEndExFW mcpEndExRO = new McpEndExFW(); @@ -219,6 +221,38 @@ private MessageConsumer beginToolsCall( return stream; } + private MessageConsumer beginLifecycle( + int capabilities) + { + final McpBeginExFW beginEx = mcpBeginExRW.wrap(extScratch, 0, extScratch.capacity()) + .typeId(MCP_TYPE_ID) + .lifecycle(l -> l.capabilities(capabilities)) + .build(); + + final BeginFW begin = beginRW.wrap(scratch, 0, scratch.capacity()) + .originId(ORIGIN_ID) + .routedId(BINDING_ID) + .streamId(INITIAL_ID) + .sequence(0) + .acknowledge(0) + .maximum(0) + .traceId(1L) + .authorization(AUTHORIZATION) + .affinity(AFFINITY) + .extension(beginEx.buffer(), beginEx.offset(), beginEx.sizeof()) + .build(); + + final MessageConsumer stream = factory.newStream( + begin.typeId(), begin.buffer(), begin.offset(), begin.sizeof(), mcp); + + if (stream != null) + { + stream.accept(begin.typeId(), begin.buffer(), begin.offset(), begin.sizeof()); + } + + return stream; + } + private void data( MessageConsumer stream, long streamId, @@ -392,6 +426,15 @@ private KafkaBeginExFW kafkaBeginEx( return kafkaBeginExRO.wrap(begin.extension().buffer(), begin.extension().offset(), begin.extension().limit()); } + private McpBeginExFW mcpBeginEx( + Recorded recorded) + { + final UnsafeBufferEx buffer = new UnsafeBufferEx(recorded.bytes); + final BeginFW begin = beginRO.wrap(buffer, 0, recorded.bytes.length); + + return mcpBeginExRO.wrap(begin.extension().buffer(), begin.extension().offset(), begin.extension().limit()); + } + private McpResetExFW mcpResetEx( Recorded recorded) { @@ -655,4 +698,26 @@ public void shouldRejectToolsCallWithNoMatchingRoute() throws Exception assertEquals(0, countOf(kafkaSent, BeginFW.TYPE_ID)); } + @Test + public void shouldDeclareToolsOnlyCapabilitiesRegardlessOfClientCapabilities() throws Exception + { + factory.attach(newBinding("produce_message")); + + beginLifecycle(McpCapabilities.CLIENT_ELICITATION.value()); + + final McpBeginExFW lifecycleEx = mcpBeginEx(nthOf(mcpSent, BeginFW.TYPE_ID, 1)); + assertEquals(McpCapabilities.SERVER_TOOLS.value(), lifecycleEx.lifecycle().capabilities()); + } + + @Test + public void shouldDeclareToolsOnlyCapabilitiesWhenClientSendsNone() throws Exception + { + factory.attach(newBinding("produce_message")); + + beginLifecycle(0); + + final McpBeginExFW lifecycleEx = mcpBeginEx(nthOf(mcpSent, BeginFW.TYPE_ID, 1)); + assertEquals(McpCapabilities.SERVER_TOOLS.value(), lifecycleEx.lifecycle().capabilities()); + } + } diff --git a/runtime/binding-mcp/src/main/java/io/aklivity/zilla/runtime/binding/mcp/internal/config/McpBindingConfig.java b/runtime/binding-mcp/src/main/java/io/aklivity/zilla/runtime/binding/mcp/internal/config/McpBindingConfig.java index e9f9842a5b..8d11bc2fea 100644 --- a/runtime/binding-mcp/src/main/java/io/aklivity/zilla/runtime/binding/mcp/internal/config/McpBindingConfig.java +++ b/runtime/binding-mcp/src/main/java/io/aklivity/zilla/runtime/binding/mcp/internal/config/McpBindingConfig.java @@ -42,6 +42,7 @@ import io.aklivity.zilla.config.engine.ModelConfig; import io.aklivity.zilla.runtime.binding.mcp.internal.McpConfiguration; import io.aklivity.zilla.runtime.binding.mcp.internal.stream.cache.McpProxyCache; +import io.aklivity.zilla.runtime.binding.mcp.internal.types.McpCapabilities; import io.aklivity.zilla.runtime.binding.mcp.internal.types.String8FW; import io.aklivity.zilla.runtime.binding.mcp.internal.types.stream.HttpBeginExFW; import io.aklivity.zilla.runtime.binding.mcp.internal.types.stream.McpBearerError; @@ -72,6 +73,10 @@ public final class McpBindingConfig private static final Map> EMPTY_ROLES = Map.of(); + private static final int SERVER_CAPABILITIES_MASK = McpCapabilities.SERVER_TOOLS.value() + | McpCapabilities.SERVER_PROMPTS.value() + | McpCapabilities.SERVER_RESOURCES.value(); + public static final String CREDENTIALS_PLACEHOLDER = "{credentials}"; public final long id; @@ -494,6 +499,28 @@ public void recordServerCapabilities( } } + // a route's static when:-derived capability set (McpRouteConfig.serves) treats an + // unrestricted condition as serving every capability, regardless of what the south + // exit it routes to can actually serve. Once that south exit's own KIND_LIFECYCLE + // handshake has recorded its real capabilities, narrow the static set down to what + // south actually declared; before that handshake completes, realCapabilitiesByRoute + // holds no bits yet for this route and the static set is used as-is. + private boolean routeServes( + McpRouteConfig route, + String capability) + { + boolean serves = route.serves(capability); + if (serves) + { + final long realCapabilities = realCapabilitiesByRoute.get(route.id) & SERVER_CAPABILITIES_MASK; + if (realCapabilities != 0L) + { + serves = (realCapabilities & McpRouteConfig.capabilityBit(capability)) != 0L; + } + } + return serves; + } + public McpRouteConfig resolve( McpBeginExFW beginEx, long authorization) @@ -522,7 +549,7 @@ else if (identifier != null) { for (McpRouteConfig route : routes) { - if (route.authorized(authorization) && route.serves(capability)) + if (route.authorized(authorization) && routeServes(route, capability)) { resolved = route; break; @@ -611,7 +638,7 @@ public List resolveAll( for (McpRouteConfig route : routes) { final long authorization = routeCacheAuthorization(traceId, route.id); - if (route.authorized(authorization) && route.serves(capability)) + if (route.authorized(authorization) && routeServes(route, capability)) { result.add(new McpRoutePrefix(route.id, new String8FW(route.prefix(kind)), route)); } @@ -634,7 +661,7 @@ public List resolveAll( { for (McpRouteConfig route : routes) { - if (route.authorized(authorization) && route.serves(capability)) + if (route.authorized(authorization) && routeServes(route, capability)) { result.add(route); } diff --git a/runtime/binding-mcp/src/main/java/io/aklivity/zilla/runtime/binding/mcp/internal/config/McpRouteConfig.java b/runtime/binding-mcp/src/main/java/io/aklivity/zilla/runtime/binding/mcp/internal/config/McpRouteConfig.java index 34421e474e..2aa3b64699 100644 --- a/runtime/binding-mcp/src/main/java/io/aklivity/zilla/runtime/binding/mcp/internal/config/McpRouteConfig.java +++ b/runtime/binding-mcp/src/main/java/io/aklivity/zilla/runtime/binding/mcp/internal/config/McpRouteConfig.java @@ -279,6 +279,18 @@ static String capabilityOf( }; } + static int capabilityBit( + String capability) + { + return switch (capability) + { + case CAPABILITY_TOOLS -> SERVER_TOOLS.value(); + case CAPABILITY_PROMPTS -> SERVER_PROMPTS.value(); + case CAPABILITY_RESOURCES -> SERVER_RESOURCES.value(); + default -> 0; + }; + } + static String identifierOf( McpBeginExFW beginEx) { diff --git a/runtime/binding-mcp/src/test/java/io/aklivity/zilla/runtime/binding/mcp/internal/config/McpBindingConfigTest.java b/runtime/binding-mcp/src/test/java/io/aklivity/zilla/runtime/binding/mcp/internal/config/McpBindingConfigTest.java index e0bb9f8389..af289f48fd 100644 --- a/runtime/binding-mcp/src/test/java/io/aklivity/zilla/runtime/binding/mcp/internal/config/McpBindingConfigTest.java +++ b/runtime/binding-mcp/src/test/java/io/aklivity/zilla/runtime/binding/mcp/internal/config/McpBindingConfigTest.java @@ -19,6 +19,9 @@ import static io.aklivity.zilla.runtime.binding.mcp.internal.config.McpBindingConfig.naturalAuthority; import static io.aklivity.zilla.runtime.binding.mcp.internal.config.McpBindingConfig.pathOf; import static io.aklivity.zilla.runtime.binding.mcp.internal.config.McpBindingConfig.rolesForTool; +import static io.aklivity.zilla.runtime.binding.mcp.internal.types.McpCapabilities.SERVER_TOOLS; +import static io.aklivity.zilla.runtime.binding.mcp.internal.types.stream.McpBeginExFW.KIND_PROMPTS_LIST; +import static io.aklivity.zilla.runtime.binding.mcp.internal.types.stream.McpBeginExFW.KIND_TOOLS_LIST; import static org.hamcrest.MatcherAssert.assertThat; import static org.hamcrest.Matchers.contains; import static org.hamcrest.Matchers.empty; @@ -100,6 +103,21 @@ private static RouteConfig rawRoute( return config; } + private static RouteConfig unrestrictedRoute( + long id, + String toolkit) + { + RouteConfig config = RouteConfig.builder() + .exit("test") + .when(McpConditionConfig.builder() + .toolkit(toolkit) + .build()) + .build(); + config.id = id; + config.authorized = (authorization, credentials) -> true; + return config; + } + private static McpBindingConfig binding( GuardHandler guard, String sharedCredentials, @@ -267,4 +285,34 @@ public void shouldUseRouteOverrideAuthorizationWhenConfigured() verify(guard, times(1)).reauthorize(anyLong(), anyLong(), anyLong(), eq("{override}")); verify(guard, times(0)).reauthorize(anyLong(), anyLong(), anyLong(), eq("{shared}")); } + + @Test + public void shouldResolveUnrestrictedRouteForAnyCapabilityBeforeRealCapabilitiesRecorded() + { + GuardHandler guard = mock(GuardHandler.class); + RouteConfig route = unrestrictedRoute(300L, "alpha"); + McpBindingConfig binding = binding(guard, null, List.of(route)); + + List prompts = binding.resolveAll(0L, KIND_PROMPTS_LIST); + + assertThat(prompts.size(), equalTo(1)); + assertThat(prompts.get(0).resolvedId(), equalTo(300L)); + } + + @Test + public void shouldExcludeUnrestrictedRouteForCapabilityNotInRecordedRealCapabilities() + { + GuardHandler guard = mock(GuardHandler.class); + RouteConfig route = unrestrictedRoute(400L, "alpha"); + McpBindingConfig binding = binding(guard, null, List.of(route)); + + binding.recordServerCapabilities(400L, SERVER_TOOLS.value()); + + List prompts = binding.resolveAll(0L, KIND_PROMPTS_LIST); + List tools = binding.resolveAll(0L, KIND_TOOLS_LIST); + + assertThat(prompts, empty()); + assertThat(tools.size(), equalTo(1)); + assertThat(tools.get(0).resolvedId(), equalTo(400L)); + } }