Repository navigation
add a client helper for subscription streams - #679
Yusufihsangorgel wants to merge 48 commits into
Conversation
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request implements MCP subscriptions and per-request cancellation support. It introduces the Subscription class, adds the listen method to ServerConnection to handle subscriptions/listen streams, and integrates request cancellation tracking within MCPBase and streamable HTTP channels. Additionally, the SubscriptionsListenRequest factory now requires MetaWithRequestEnvelope. Feedback on the changes suggests enhancing type safety in the _forward method of Subscription by utilizing type promotion instead of explicit casting and null-assertion.
|
#678 puts a |
jakemac53
left a comment
jakemac53
left a comment
There was a problem hiding this comment.
Still trying to wrap my head around this one, left a few comments though
Read the client's reply to the server ping through a Completer instead of a plain variable, and await it, since an async function's future completes synchronously on the VM but only after an await on dart2js and one microtask later on dart2wasm. Answer the fake server's ping in a scheduled microtask instead of inside the write: on a sync channel json_rpc_2 writes a request before registering it, so a reply delivered inside that write is dropped and the pending ping times out. A real transport never replies inside the write.
_forwardSubscriptionNotification now reads the subscription ID off the raw _meta map and delivers only to that subscription, instead of letting every open subscription filter the same params by its own ID. Subscription._forward keeps only its acknowledged-and-open check. Add a test that a notification whose subscription ID matches no open subscription reaches none of them.
A synchronous listener could otherwise send progress for the request before it was marked cancelled, and that progress went out on the wire.
There was a problem hiding this comment.
Can we add an example for how we expect users to deal with subscriptions across various versions now?
Lets say I want to subscribe to the tools list, across versions. Can we abstract this away better? Maybe the existing toolsListChanged (and others) should go away, and we should force everything to go through the this new listen method, which would only send an actual subscriptions/listen request on the versions that require it? And otherwise it just provides a view over the normal notifications stream which filters out the unwanted things.
|
A follow up possible design, could be to move the various streams we expose today to the |
|
Before 2026-07-28, listen now sends nothing. It acks locally from the server capabilities and filters the list-changed notifications (24c0df1). multi_version_client.dart has the tools list case. I kept the connection-level streams here, and can do typed ones on Subscription as the follow-up you suggested. |
| /// requested list-changed types [serverCapabilities] supports. Notifications | ||
| /// are a local view of the connection's list-changed streams, and closing | ||
| /// the subscription sends nothing. Its [Subscription.id] is `null`, and | ||
| /// [meta] is not sent. `resourceSubscriptions` is never acknowledged on |
There was a problem hiding this comment.
Can't this also support resource subscriptions on older versions?
| /// requested list-changed types [serverCapabilities] supports. Notifications | ||
| /// are a local view of the connection's list-changed streams, and closing | ||
| /// the subscription sends nothing. Its [Subscription.id] is `null`, and | ||
| /// [meta] is not sent. `resourceSubscriptions` is never acknowledged on |
There was a problem hiding this comment.
Can't this also support resource subscriptions on older versions?
| }) { | ||
| final version = protocolVersion; | ||
| if (version != null && | ||
| !version.methodIsValid(SubscriptionsListenRequest.methodName)) { |
There was a problem hiding this comment.
I think we should specifically check for legacy version ranges here. Otherwise if subscriptions/listen gets removed in a later version, we could start incorrectly going down this code path, when we should be throwing an error instead.
We should still check if it is a valid method, but return an UnsupportedError or something instead, after checking for the legacy code path.
| final version = protocolVersion; | ||
| if (version != null && | ||
| !version.methodIsValid(SubscriptionsListenRequest.methodName)) { | ||
| return Subscription._local( |
There was a problem hiding this comment.
Lets make this be a factory which takes the subscriptions filter and serverCapabilities object so that there is less stuff hard coded here at the call site.
I don't think we need to recreate the subscription filter either - the ACK should be based on the server capabilities and originally provided filter. The current implementation risks losing additional data that may be attached to the filter (if for instance more is added later).
2 participants
Adds
ServerConnection.listen, the client half of the subscription streams the server has served since #636. A handle comes back right away, reports the filter the server agreed to, and routes each notification to the subscription that opened it. Closing one subscription ends only that stream. Over stdio it sendsnotifications/cancelledfor that request, over Streamable HTTP it drops the SSE response. A cancellation arriving for one subscription ends that one and leaves the connection alone.Three helper names here are this package's own and not the schema's. Reading the outgoing request ID off the channel sink leaves
package:json_rpc_2untouched.Tracked in #668.