fix(binding-mcp): honor south exits' real capabilities in route resolution

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<McpHttpResourceConfig> 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;

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

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));
+    }
 }

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);

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());
+    }
+
 }

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<String, List<String>> 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<McpRoutePrefix> 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<McpRouteConfig> resolveAll(
         {
             for (McpRouteConfig route : routes)
             {
-                if (route.authorized(authorization) && route.serves(capability))
+                if (route.authorized(authorization) && routeServes(route, capability))
                 {
                     result.add(route);
                 }

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)
     {