feat(binding-mcp): mcp · proxy cache option (#1737)

runtime/binding-kafka/src/main/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/KafkaClientConnectionPool.java

@@ -20,6 +20,7 @@
 import static io.aklivity.zilla.runtime.engine.concurrent.Signaler.NO_CANCEL_ID;
 import static java.lang.System.currentTimeMillis;
 
+import java.time.Instant;
 import java.util.List;
 import java.util.function.Consumer;
 import java.util.function.IntConsumer;
@@ -591,6 +592,15 @@ public long signalAt(
             return delegate.signalAt(timeMillis, signalId, handler);
         }
 
+        @Override
+        public long signalAt(
+            Instant time,
+            int signalId,
+            IntConsumer handler)
+        {
+            return delegate.signalAt(time, signalId, handler);
+        }
+
         @Override
         public void signalNow(
             long originId,
@@ -639,6 +649,19 @@ public long signalAt(
             return stream.doStreamSignalAt(traceId, timeMillis, signalId);
         }
 
+        @Override
+        public long signalAt(
+            Instant time,
+            long originId,
+            long routedId,
+            long streamId,
+            long traceId,
+            int signalId,
+            int contextId)
+        {
+            return signalAt(time.toEpochMilli(), originId, routedId, streamId, traceId, signalId, contextId);
+        }
+
         @Override
         public long signalTask(
             Runnable task,

runtime/binding-mcp/src/main/java/io/aklivity/zilla/runtime/binding/mcp/config/McpAuthorizationConfig.java

@@ -21,6 +21,7 @@
 public final class McpAuthorizationConfig
 {
     public final String name;
+    public final String credentials;
 
     public transient String qname;
 
@@ -36,8 +37,10 @@ public static <T> McpAuthorizationConfigBuilder<T> builder(
     }
 
     McpAuthorizationConfig(
-        String name)
+        String name,
+        String credentials)
     {
         this.name = name;
+        this.credentials = credentials;
     }
 }

runtime/binding-mcp/src/main/java/io/aklivity/zilla/runtime/binding/mcp/config/McpAuthorizationConfigBuilder.java

@@ -23,6 +23,7 @@ public final class McpAuthorizationConfigBuilder<T> extends ConfigBuilder<T, Mcp
     private final Function<McpAuthorizationConfig, T> mapper;
 
     private String name;
+    private String credentials;
 
     McpAuthorizationConfigBuilder(
         Function<McpAuthorizationConfig, T> mapper)
@@ -44,9 +45,16 @@ public McpAuthorizationConfigBuilder<T> name(
         return this;
     }
 
+    public McpAuthorizationConfigBuilder<T> credentials(
+        String credentials)
+    {
+        this.credentials = credentials;
+        return this;
+    }
+
     @Override
     public T build()
     {
-        return mapper.apply(new McpAuthorizationConfig(name));
+        return mapper.apply(new McpAuthorizationConfig(name, credentials));
     }
 }

runtime/binding-mcp/src/main/java/io/aklivity/zilla/runtime/binding/mcp/config/McpCacheConfig.java

@@ -0,0 +1,48 @@
+/*
+ * Copyright 2021-2024 Aklivity Inc
+ *
+ * Licensed under the Aklivity Community License (the "License"); you may not use
+ * this file except in compliance with the License.  You may obtain a copy of the
+ * License at
+ *
+ *   https://www.aklivity.io/aklivity-community-license/
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+package io.aklivity.zilla.runtime.binding.mcp.config;
+
+import static java.util.function.Function.identity;
+
+import java.time.Duration;
+import java.util.function.Function;
+
+public final class McpCacheConfig
+{
+    public final String store;
+    public final Duration ttl;
+    public final McpAuthorizationConfig authorization;
+
+    McpCacheConfig(
+        String store,
+        Duration ttl,
+        McpAuthorizationConfig authorization)
+    {
+        this.store = store;
+        this.ttl = ttl;
+        this.authorization = authorization;
+    }
+
+    public static McpCacheConfigBuilder<McpCacheConfig> builder()
+    {
+        return new McpCacheConfigBuilder<>(identity());
+    }
+
+    public static <T> McpCacheConfigBuilder<T> builder(
+        Function<McpCacheConfig, T> mapper)
+    {
+        return new McpCacheConfigBuilder<>(mapper);
+    }
+}

runtime/binding-mcp/src/main/java/io/aklivity/zilla/runtime/binding/mcp/config/McpCacheConfigBuilder.java

@@ -0,0 +1,74 @@
+/*
+ * Copyright 2021-2024 Aklivity Inc
+ *
+ * Licensed under the Aklivity Community License (the "License"); you may not use
+ * this file except in compliance with the License.  You may obtain a copy of the
+ * License at
+ *
+ *   https://www.aklivity.io/aklivity-community-license/
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+package io.aklivity.zilla.runtime.binding.mcp.config;
+
+import java.time.Duration;
+import java.util.function.Function;
+
+import io.aklivity.zilla.runtime.engine.config.ConfigBuilder;
+
+public final class McpCacheConfigBuilder<T> extends ConfigBuilder<T, McpCacheConfigBuilder<T>>
+{
+    private final Function<McpCacheConfig, T> mapper;
+
+    private String store;
+    private Duration ttl;
+    private McpAuthorizationConfig authorization;
+
+    McpCacheConfigBuilder(
+        Function<McpCacheConfig, T> mapper)
+    {
+        this.mapper = mapper;
+    }
+
+    @Override
+    @SuppressWarnings("unchecked")
+    protected Class<McpCacheConfigBuilder<T>> thisType()
+    {
+        return (Class<McpCacheConfigBuilder<T>>) getClass();
+    }
+
+    public McpCacheConfigBuilder<T> store(
+        String store)
+    {
+        this.store = store;
+        return this;
+    }
+
+    public McpCacheConfigBuilder<T> ttl(
+        Duration ttl)
+    {
+        this.ttl = ttl;
+        return this;
+    }
+
+    public McpCacheConfigBuilder<T> authorization(
+        McpAuthorizationConfig authorization)
+    {
+        this.authorization = authorization;
+        return this;
+    }
+
+    public McpAuthorizationConfigBuilder<McpCacheConfigBuilder<T>> authorization()
+    {
+        return McpAuthorizationConfig.builder(this::authorization);
+    }
+
+    @Override
+    public T build()
+    {
+        return mapper.apply(new McpCacheConfig(store, ttl, authorization));
+    }
+}

runtime/binding-mcp/src/main/java/io/aklivity/zilla/runtime/binding/mcp/config/McpOptionsConfig.java

@@ -24,15 +24,18 @@ public final class McpOptionsConfig extends OptionsConfig
     public final List<McpPromptConfig> prompts;
     public final McpElicitationConfig elicitation;
     public final McpAuthorizationConfig authorization;
+    public final McpCacheConfig cache;
 
-    public McpOptionsConfig(
+    McpOptionsConfig(
         List<McpPromptConfig> prompts,
         McpElicitationConfig elicitation,
-        McpAuthorizationConfig authorization)
+        McpAuthorizationConfig authorization,
+        McpCacheConfig cache)
     {
         this.prompts = prompts;
         this.elicitation = elicitation;
         this.authorization = authorization;
+        this.cache = cache;
     }
 
     public static McpOptionsConfigBuilder<McpOptionsConfig> builder()

runtime/binding-mcp/src/main/java/io/aklivity/zilla/runtime/binding/mcp/config/McpOptionsConfigBuilder.java

@@ -28,6 +28,7 @@ public final class McpOptionsConfigBuilder<T> extends ConfigBuilder<T, McpOption
     private List<McpPromptConfig> prompts;
     private McpElicitationConfig elicitation;
     private McpAuthorizationConfig authorization;
+    private McpCacheConfig cache;
 
     public McpOptionsConfigBuilder(
         Function<OptionsConfig, T> mapper)
@@ -71,6 +72,18 @@ public McpAuthorizationConfigBuilder<McpOptionsConfigBuilder<T>> authorization()
         return McpAuthorizationConfig.builder(this::authorization);
     }
 
+    public McpOptionsConfigBuilder<T> cache(
+        McpCacheConfig cache)
+    {
+        this.cache = cache;
+        return this;
+    }
+
+    public McpCacheConfigBuilder<McpOptionsConfigBuilder<T>> cache()
+    {
+        return McpCacheConfig.builder(this::cache);
+    }
+
     @Override
     @SuppressWarnings("unchecked")
     protected Class<McpOptionsConfigBuilder<T>> thisType()
@@ -81,6 +94,6 @@ protected Class<McpOptionsConfigBuilder<T>> thisType()
     @Override
     public T build()
     {
-        return mapper.apply(new McpOptionsConfig(prompts, elicitation, authorization));
+        return mapper.apply(new McpOptionsConfig(prompts, elicitation, authorization, cache));
     }
 }

runtime/binding-mcp/src/main/java/io/aklivity/zilla/runtime/binding/mcp/internal/McpConfiguration.java

@@ -14,15 +14,21 @@
  */
 package io.aklivity.zilla.runtime.binding.mcp.internal;
 
+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_RESOURCES_LIST;
+import static io.aklivity.zilla.runtime.binding.mcp.internal.types.stream.McpBeginExFW.KIND_TOOLS_LIST;
 import static io.aklivity.zilla.runtime.engine.EngineConfiguration.ENGINE_WORKERS;
 
 import java.lang.invoke.MethodHandle;
 import java.lang.invoke.MethodHandles;
 import java.lang.invoke.MethodType;
 import java.security.SecureRandom;
 import java.time.Duration;
+import java.util.HashSet;
 import java.util.HexFormat;
+import java.util.Set;
 import java.util.UUID;
+import java.util.function.IntPredicate;
 import java.util.function.Supplier;
 
 import org.agrona.LangUtil;
@@ -46,6 +52,9 @@ public class McpConfiguration extends Configuration
     public static final PropertyDef<Duration> MCP_SSE_KEEPALIVE_INTERVAL;
     public static final BooleanPropertyDef MCP_ALT_SVC_ENABLED;
     public static final PropertyDef<Duration> MCP_ALT_SVC_MAX_AGE;
+    public static final PropertyDef<IntPredicate> MCP_HYDRATE_FILTER;
+    public static final PropertyDef<Duration> MCP_LEASE_TTL;
+    public static final PropertyDef<Duration> MCP_LEASE_RETRY;
 
     static
     {
@@ -72,6 +81,12 @@ public class McpConfiguration extends Configuration
         MCP_ALT_SVC_ENABLED = config.property("alt.svc.enabled", McpConfiguration::defaultAltSvcEnabled);
         MCP_ALT_SVC_MAX_AGE = config.property(Duration.class, "alt.svc.max.age",
             (c, v) -> Duration.parse(v), "PT24H");
+        MCP_HYDRATE_FILTER = config.property(IntPredicate.class, "hydrate.filter",
+            McpConfiguration::decodeHydrateFilter, McpConfiguration::defaultHydrateFilter);
+        MCP_LEASE_TTL = config.property(Duration.class, "lease.ttl",
+            (c, v) -> Duration.parse(v), "PT30S");
+        MCP_LEASE_RETRY = config.property(Duration.class, "lease.retry",
+            (c, v) -> Duration.parse(v), "PT0.1S");
         MCP_CONFIG = config;
     }
 
@@ -146,6 +161,21 @@ public Duration altSvcMaxAge()
         return MCP_ALT_SVC_MAX_AGE.get(this);
     }
 
+    public IntPredicate hydrateFilter()
+    {
+        return MCP_HYDRATE_FILTER.get(this);
+    }
+
+    public Duration leaseTtl()
+    {
+        return MCP_LEASE_TTL.get(this);
+    }
+
+    public Duration leaseRetry()
+    {
+        return MCP_LEASE_RETRY.get(this);
+    }
+
     @FunctionalInterface
     public interface SessionIdSupplier
     {
@@ -258,6 +288,36 @@ private static ElicitationIdSupplier decodeElicitationIdSupplier(
         return supplier;
     }
 
+    private static IntPredicate decodeHydrateFilter(
+        String value)
+    {
+        final Set<Integer> kinds = new HashSet<>();
+        for (String name : value.split("\\s+"))
+        {
+            switch (name)
+            {
+            case "tools":
+                kinds.add(KIND_TOOLS_LIST);
+                break;
+            case "resources":
+                kinds.add(KIND_RESOURCES_LIST);
+                break;
+            case "prompts":
+                kinds.add(KIND_PROMPTS_LIST);
+                break;
+            default:
+                break;
+            }
+        }
+        return kinds::contains;
+    }
+
+    private static boolean defaultHydrateFilter(
+        int kind)
+    {
+        return true;
+    }
+
     private static String defaultElicitationIdSupplier()
     {
         final byte[] bytes = new byte[4];