feat(binding-mcp): add resources/subscribe, resources/unsubscribe pass-through - #2227
Merged
Merged
Conversation
…s-through Adds pass-through support for MCP resources/subscribe, resources/unsubscribe requests and notifications/resources/updated notifications across all three mcp binding kinds (server, client, proxy), per the 2025-11-25 MCP spec. - mcp.idl: new SERVER_RESOURCES_SUBSCRIBE capability bit, McpResourcesSubscribeBeginEx/ McpResourcesUnsubscribeBeginEx union cases, McpResourcesUpdatedFlushEx union case - server kind (McpServerFactory): dispatch resources/subscribe|unsubscribe, relay notifications/resources/updated over SSE, advertise resources.subscribe in the initialize response only when the app/backend actually declares the capability - client kind (McpClientFactory): forward resources/subscribe|unsubscribe to the upstream server, parse the upstream's resources.subscribe capability during initialize, relay upstream notifications/resources/updated back up the pipeline - proxy kind: route subscribe/unsubscribe to the correct upstream toolkit (same URI-prefix mechanism as resources/read), re-prefix the uri on relayed notifications/resources/updated, and capture each toolkit's real negotiated capabilities (previously discarded) so resources.subscribe is aggregated from what upstreams actually support rather than the static route-config bits alone - specs/binding-mcp.spec: new application/network scenario scripts for subscribe, unsubscribe, unsubscribe-of-unknown-uri (network-level, documents the wire behavior), and notifications/resources/updated, wired into NetworkIT/ ApplicationIT and the runtime McpServerIT/McpClientIT/McpProxyIT suites Note: this sandbox has no credentials for maven.packages.aklivity.io, so flyweight-maven-plugin could not be resolved to regenerate the IDL-derived Java sources, and none of this change could be locally compiled or run. Every edit was written by direct, careful mirroring of the existing analogous code path (resources/read, notifications/*/list_changed) rather than verified by a build. A CI run and careful review are needed before merge. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_0157qpppRcorMgMSxTddQV9a
…tion
Now that this sandbox's flyweight-maven-plugin resolution is fixed, the
previous commit's implementation could actually be compiled and tested
against a live engine. This fixes what that first real build turned up:
- McpServerFactory.java: missing import for McpResourcesUpdatedFlushExFW
(compile error)
- McpBindingConfig.java: Long2IntHashMap doesn't exist in the pinned Agrona
version; switched to Long2ObjectHashMap<Integer> with explicit boxing to
avoid an overload-resolution ambiguity between Agrona's long/Long put()
overloads
- McpClientFactory.java: the newStream() dispatch never set
request.contentLength/timeout for KIND_RESOURCES_SUBSCRIBE/UNSUBSCRIBE,
leaving contentLength at its -1 default and causing
doEncodeStreamedRequestEnd() to compute a negative padding length,
crashing with IndexOutOfBoundsException on every subscribe/unsubscribe
request from the client kind
- application/{resources.subscribe,resources.unsubscribe,
lifecycle.notify.resources.updated}/server.rpt: the app-side lifecycle
begin never declared the SERVER_RESOURCES_SUBSCRIBE capability, so the
network-side scripts' asserted "subscribe":true in the initialize
response never matched what the server kind actually emitted
Verified: `./mvnw clean verify -pl runtime/binding-mcp` is a clean
BUILD SUCCESS (279 tests, all coverage checks met). `./mvnw clean verify
-pl specs/binding-mcp.spec` passes all 248 tests but currently fails its
coverage gate — jacoco reports 0% instruction coverage on the new
McpFunctions builder/matcher inner classes for subscribe/unsubscribe/
updated despite the protocol-level tests that exercise them passing,
which looks like a coverage-attribution quirk rather than a real gap;
flagging for a maintainer to confirm rather than asserting a cause.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0157qpppRcorMgMSxTddQV9a
Long2ObjectHashMap<Integer> boxed every capability bitmask read/write. Agrona has a real primitive fit here: Long2LongHashMap, storing the int bitmask widened to long, with get()/put() fully primitive overloads. Hit a real edge case along the way: Long2LongHashMap's missingValue sentinel (0L here, matching "no capabilities recorded" for an absent route) cannot itself be put() — throws IllegalArgumentException. Since "absent" and "merged == 0" already mean the same thing for a bitmask that defaults to 0, recordServerCapabilities() now just skips the put when the merged value would be the sentinel, verified against the full runtime/binding-mcp test suite (279 tests, all coverage checks met). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_0157qpppRcorMgMSxTddQV9a
Exercises resources/subscribe, resources/unsubscribe, and the relayed notifications/resources/updated (#2220) against the everything reference server through mcp(server) -> mcp(proxy) -> mcp(client). Adds a headless demo client, a compose service to run it on demand, a README walkthrough, and a smoke-test assertion in .github/test.sh. Not yet verified against a live docker compose run in this environment; verification is in progress.
…s directly CI failed the jacoco coverage gate (0.86 vs 0.96 required, 6 classes at 0%) on binding-mcp.spec. Root cause: the new resourcesSubscribe/ resourcesUnsubscribe/resourcesUpdated builder and matcher classes were only exercised indirectly via .rpt scripts, which k3po runs in a separate process the jacoco agent never instruments -- unlike sibling classes (e.g. resourcesRead), which are also covered by a direct McpFunctionsTest unit test that runs under Surefire/jacoco. Adds the missing direct unit tests, mirroring the existing resourcesRead / resourcesListChanged pattern.
….json Matches the checked-in lockfile convention used by the sibling tools-list-client and url-elicit demo clients.
…-implementation-7bgf9h # Conflicts: # examples/mcp.proxy/.github/test.sh
…-id code, for resources/updated notifications/resources/updated was being re-prefixed with McpAggregateRoute.prefix() -- a short CRC32C-derived code used only to disambiguate resumable event/correlation ids across aggregated toolkits (e.g. "S" for a toolkit named "bluesky") -- instead of the toolkit's real "<toolkit>+" URI prefix that resources/list and resources/read already use. A caller subscribed to "everything+demo://..." therefore received updates for "Vdemo://..." (or similar), silently mismatching its own subscription. Found via a live end-to-end run against examples/mcp.proxy: capability negotiation and the subscribe/toggle calls all succeeded, but the relayed notification's uri never matched what was subscribed. Confirmed directly against the real "everything" reference server (bypassing Zilla) that it sends a clean, unprefixed uri, isolating the corruption to the proxy's own relay path. McpBindingConfig gains routeResourcesPrefix(routedId), a lookup by McpRouteConfig.id mirroring the existing routeCacheCredentials pattern, returning the same McpConditionMatcher-derived prefix resources/read already relies on. McpProxyLifecycleFactory stores it as a new resourcesPrefix field, separate from the existing prefix field (which correctly keeps using the short aggregate code for elicitation correlation-id prefixing -- that usage was never wrong). Existing single-route notify tests never caught this because computeRouteByPrefix only activates aggregation once routes.size() > 1, and the two prefix schemes are indistinguishable with only one configured route. Adds a regression scenario (lifecycle.notify.resources.updated.toolkit.multi) against proxy.toolkit.multi.yaml asserting the relayed uri carries the real toolkit prefix; confirmed it fails against the prior code (uri prefixed "S" instead of "bluesky+") before the fix and passes after.
…SE events decode McpClientFactory's HttpEventStream (the persistent GET SSE listener used to receive server-pushed notifications, including resources/updated) set replyAck once from the initial BEGIN and never advanced it afterward. Since its receive window is a fixed replyAck+decodeMax ceiling, any sufficiently long-lived listener eventually exceeds it purely from cumulative bytes -- independent of any individual message's content -- and self-resets via cleanupNet(), tearing down the listener and losing whatever notification triggered the overflow. Add flushNetWindow to HttpEventStream, mirroring the sibling HttpStream's existing correct pattern, and call it after each decode so the window keeps pace with consumed bytes instead of starving. Regression test: McpClientIT.shouldNotifyResourcesUpdatedRepeated drives six resources/updated notifications over one GET listener with a small configured buffer slot capacity, reproducing the starvation deterministically. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_0157qpppRcorMgMSxTddQV9a
The synthetic k3po scenario added in 7628b0e to regress the GET-listener window-starvation fix proved unreliable to construct: shrinking the engine's global buffer slot capacity small enough to trip the bug within a test also destabilizes unrelated flow-control layers beneath the mcp binding (observed as hangs unrelated to the fix itself, confirmed by CI). The production fix in McpClientFactory (flushNetWindow on HttpEventStream) stays -- it was verified directly against the real aklivity/server-everything reference server via a live zilla dump packet capture during the original investigation, and the full binding-mcp verify suite (282 tests) passes cleanly with it applied. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_0157qpppRcorMgMSxTddQV9a
…ent SSE decode A string or number lexeme whose bytes fill the input window before completing is delivered by JsonParserEx as repeated fragments (deferredBytes() true) rather than a single complete event. The mcp client's SSE JSON-RPC decoders were treating the first fragment as the whole value, so a long value (e.g. a notified resource uri) split across a network read boundary was truncated, corrupting the notification and triggering a decode error downstream. Add a shared accumulateJsonRpcStringValue helper, backed by a new sseJsonValue StringBuilder (kept independent of the existing sseSmallValue field, which is never cleared after a completed id:/retry: read and would otherwise leak stale bytes into accumulated values), and use it in every JSON-RPC decoder that reads a scalar: version, id, method, and the params token/message/ elicitationId/url/mode/uri fields. Adds a regression IT (shouldNotifyResourcesUpdatedFragmented) with a small buffer slot capacity to force a long resource uri to split across a decode window, plus the paired client/server spec scripts and ApplicationIT peer-consistency check.
…edging the decoder
decodeJsonRpcParamsNext's default case (an unrecognized params key) looped
back to itself by reassigning the same decoder field reference, which left
decodeNet's "previous != decoder" progress check unable to tell that
anything happened. The outer decode loop exited immediately, buffering the
still-unconsumed value in the decode slot. The next onNetData call resumed
decodeJsonRpcParamsNext directly on that orphaned value token, which the
switch only expects to see KEY_NAME/END_OBJECT for, so it raised a parse
error and left the stream unable to make further progress -- silently
dropping every later event on the same SSE connection, including
notifications/resources/updated. This reproduced live against the
`aklivity/server-everything` reference server, whose notifications/message
event carries params keys ("level", "data") this decoder didn't recognize,
sent immediately before any resources/updated notification on the same
GET stream.
Add a dedicated decodeJsonRpcParamsSkipValue decoder, using the same
depth-tracked skip pattern as decodeJsonRpcSkipObject, that consumes exactly
one JSON value of any shape (scalar, object, or array, including a scalar
that itself spans a decode window) before returning to
decodeJsonRpcParamsNext -- as a distinct decoder reference, it lets the
outer loop's progress check register the transition correctly.
Confirmed against the live mcp.proxy example end-to-end (resources/subscribe
+ toggle-subscriber-updates against the real reference server), which had
been failing in CI even after the prior JSON-fragmentation fix.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Description
Implements #2220: pass-through support for MCP
resources/subscribe,resources/unsubscribe, andnotifications/resources/updatedacross allthree
mcpbinding kinds (server,client,proxy), following theexisting pattern used for other mcp-specific streams (e.g.
resources/read).Targets the
2025-11-25MCP spec (the version this binding currentlyimplements). Support for the newer
2026-07-28stateless spec (whichreplaces subscribe/unsubscribe with a unified
subscriptions/listen) andversion-mapping in the proxy's one-to-many toolkit case are explicitly
deferred to future work.
Changes
mcp.idl: newMcpResourcesSubscribeBeginEx/McpResourcesUnsubscribeBeginEx/
McpResourcesUpdatedFlushExextension types, and aSERVER_RESOURCES_SUBSCRIBEcapability bit.
McpFunctions: builder/matcher support for the new extension types, plusdirect
McpFunctionsTestunit-test coverage for each (see "Verification")..rptscripts (network + application, paired client/server) andNetworkIT/ApplicationITmethods, written before the implementation perthis repo's test-first discipline.
McpServerFactory: decodesresources/subscribe/unsubscribe, encodes thesubscribecapability in theinitializeresponse only when thedownstream binding actually advertises it, and relays
notifications/resources/updatedas an SSE event.McpClientFactory: encodesresources/subscribe/unsubscribeas JSON-RPCto the real upstream MCP server, parses the
resources.subscribecapability from its
initializeresponse, and relaysnotifications/resources/updatedback as an application-layer flush.McpProxyFactory/ newMcpProxyResourcesSubscribeFactory/McpProxyResourcesUnsubscribeFactory: routes the two new request kinds tothe correct toolkit route, same as
resources/read.McpProxyLifecycleFactory: re-prefixes the URI on a relayednotifications/resources/updated(<toolkit>+<uri>, same convention asevery other aggregated resource), and now records each south route's
real server capabilities (learned from that route's own upstream
handshake) instead of only reflecting a static, route-config-declared
capability.
McpBindingConfig: aggregates per-route real capabilities via aboxing-free
Long2LongHashMap, merged with each route's staticcapabilities when computing what to advertise for a given caller.
examples/mcp.proxy: newresource-subscribe-clientheadless demo clientexercising the full round-trip against the
everythingreference server(which implements subscribe/unsubscribe/updated specifically to support
test clients), wired into
compose.yaml, the README, and.github/test.sh.Verification
./mvnw clean verify -pl runtime/binding-mcp(279 tests) and./mvnw clean verify -pl specs/binding-mcp.spec(248 tests) both pass,including all jacoco coverage gates. A full top-level build
(
./mvnw clean install -DskipITs -DskipTests) also succeeds.(CI initially caught a real gap here: the new
McpFunctionsbuilder/matcherclasses were only exercised indirectly through
.rptscripts, which k3poruns in a process jacoco doesn't instrument. Fixed by adding direct
McpFunctionsTestunit tests mirroring the existingresourcesRead/resourcesListChangedcoverage pattern.)The
examples/mcp.proxyend-to-end demo is being verified live against alocally built Docker image in this environment; will report back or push a
follow-up fix if that surfaces anything.
Fixes #2220