From ab94a4b132f4320dedf01803ade648f52f316b53 Mon Sep 17 00:00:00 2001 From: Felix Weinberger Date: Mon, 6 Jul 2026 14:57:20 +0000 Subject: [PATCH] Add SDK-owned resource subscription tracking to McpServer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An app that declares resources.subscribe today gets no help from the high-level server: it must reach through mcpServer.server to register the resources/subscribe and resources/unsubscribe handlers, keep its own Set of subscribed URIs, and call server.server.sendResourceUpdated, because the facade has sendResourceListChanged but no updated sibling. Declaring the resources.subscribe capability is now the whole opt-in: at connect() time, McpServer installs the resources/subscribe and resources/unsubscribe handlers when neither verb has one yet, and records subscribed URIs in the new read-only per-connection resourceSubscriptions set. Deferring the install to connect keeps hand-registered handlers (the previous escape-hatch pattern) winning by already existing, so servers that answer the verbs themselves keep their exact behavior; the install is skipped as a pair. connect() resets the set, so a reused instance cannot leak one connection's subscriptions into the next, and repeat connects find the first connect's handlers and only clear the set. trackResourceSubscriptions({ onSubscribe?, onUnsubscribe? }) is the explicit configuration path: called before connect it declares the capability, installs the handlers, and attaches veto hooks that run before the set changes — a hook throw is returned to the client as the request's error, and a request cancelled while a hook runs leaves the set unchanged (the handlers re-check the abort signal). The method is assert-first: a duplicate call, a manually installed subscribe handler, or a conflicting low-level resources/list handler throws before any capability or handler state changes. A sendResourceUpdated facade completes the notification trio, and Protocol gains a public hasRequestHandler(method) accessor backing the connect-time check. On 2026-07-28 connections the two verbs do not exist (clients filter subscriptions/listen and the serving entries own that bookkeeping), and the era gate rejects them before any handler runs, so the automatic install is harmless on dual-era servers. Apps that declared resources.subscribe without registering handlers previously advertised the capability while answering -32601; they now get working subscribe/unsubscribe — documented in the migration guide. The resources and todos-server examples and the resources guide recipe now rely on the automatic path; the low-level guide points readers at it for this pair. --- .changeset/track-resource-subscriptions.md | 19 ++ docs/advanced/low-level-server.md | 6 +- docs/migration/upgrade-to-v2.md | 13 + docs/servers/resources.md | 26 +- examples/guides/servers/resources.examples.ts | 18 +- examples/resources/server.ts | 19 +- examples/todos-server/todos.ts | 20 +- packages/core-internal/src/shared/protocol.ts | 9 +- packages/server/src/server/mcp.examples.ts | 20 ++ packages/server/src/server/mcp.ts | 151 ++++++++++ .../resourceSubscriptionTracking.test.ts | 273 ++++++++++++++++++ 11 files changed, 514 insertions(+), 60 deletions(-) create mode 100644 .changeset/track-resource-subscriptions.md create mode 100644 packages/server/test/server/resourceSubscriptionTracking.test.ts diff --git a/.changeset/track-resource-subscriptions.md b/.changeset/track-resource-subscriptions.md new file mode 100644 index 0000000000..14fcbddc85 --- /dev/null +++ b/.changeset/track-resource-subscriptions.md @@ -0,0 +1,19 @@ +--- +'@modelcontextprotocol/server': minor +--- + +Add SDK-owned bookkeeping for 2025-era per-resource subscriptions to `McpServer`. +Declaring the `resources.subscribe` capability now activates it automatically at +`connect()`: the SDK installs the `resources/subscribe` and `resources/unsubscribe` +handlers (unless either verb already has a hand-registered handler, which wins) and +records subscribed URIs in the new per-connection `resourceSubscriptions` read-only +set. `trackResourceSubscriptions({ onSubscribe?, onUnsubscribe? })` is the explicit +configuration path — it declares the capability, installs the handlers, and attaches +veto hooks that run before the set changes and can refuse a request by throwing. A +`sendResourceUpdated` facade joins `sendResourceListChanged` so apps no longer reach +through `.server` to deliver `notifications/resources/updated`. + +Behavior change: apps that declared `resources: { subscribe: true }` but never +registered subscribe handlers previously advertised the capability while answering +`-32601 Method not found`; they now get working SDK-owned subscribe/unsubscribe. +See the migration guide's "Server (McpServer / Streamable HTTP behavior)" section. diff --git a/docs/advanced/low-level-server.md b/docs/advanced/low-level-server.md index 106abe05d3..873b36f30c 100644 --- a/docs/advanced/low-level-server.md +++ b/docs/advanced/low-level-server.md @@ -121,7 +121,7 @@ Every serving recipe — [stdio](../serving/stdio.md), [HTTP](../serving/http.md ## Reach the low level from `McpServer` -Every `McpServer` owns its `Server` as `mcp.server`, so drop down per method, never per program. Declare the extra capability in the constructor, keep `registerTool` for the tools, and hand-register the one method `McpServer` has no API for. +Every `McpServer` owns its `Server` as `mcp.server`, so drop down per method, never per program. Declare the extra capability in the constructor, keep `registerTool` for the tools, and hand-register the method you want to answer yourself. ```ts source="../../examples/guides/advanced/low-level-server.examples.ts#lowLevel_escapeHatch" const mcp = new McpServer({ name: 'catalog', version: '1.0.0' }, { capabilities: { resources: { subscribe: true } } }); @@ -144,6 +144,10 @@ mcp.server.setRequestHandler('resources/subscribe', async request => { `registerTool` still answers `tools/list` and `tools/call`; `resources/subscribe` reaches the handler you wrote. On the 2026-07-28 revision resource subscriptions arrive on a `subscriptions/listen` stream the serving entries answer for you — see [Protocol versions](../protocol-versions.md). +::: info +Per-resource subscriptions specifically no longer need this escape hatch: declaring `resources: { subscribe: true }` alone makes the SDK install both subscription verbs at connect time and own the subscribed-URI set, and `trackResourceSubscriptions()` adds veto hooks — see [Resources](../servers/resources.md#serve-per-resource-subscriptions). Hand-registered handlers like the one above still win: the automatic install skips when either verb already has a handler. The mechanics shown here stay the same for any method you take over. +::: + ## Decide which layer to build on Default to `McpServer`. `registerTool`, `registerResource`, and `registerPrompt` cover everything this page rebuilt — schema derivation, argument validation, typed handler arguments — plus the bookkeeping it skipped: `listChanged` notifications, [completions](../servers/completion.md), and the list/read/get dispatch for every registry. diff --git a/docs/migration/upgrade-to-v2.md b/docs/migration/upgrade-to-v2.md index a149e5b407..0d17ae8767 100644 --- a/docs/migration/upgrade-to-v2.md +++ b/docs/migration/upgrade-to-v2.md @@ -1611,6 +1611,19 @@ not found`. Low-level `Server` users remain responsible for registering handlers tests need re-baselining. To advertise without the default, set `listChanged: false` explicitly; capabilities declared on the low-level `Server` are advertised verbatim. +- **Declared `resources.subscribe` now activates SDK-owned subscription tracking.** At + `McpServer.connect()`, a declared `resources: { subscribe: true }` capability installs + the `resources/subscribe` and `resources/unsubscribe` handlers when neither verb has + one yet: both verbs answer `{}` instead of v1's `-32601 Method not found`, and the + subscribed URIs are recorded in the per-connection `mcpServer.resourceSubscriptions` + read-only set (cleared on every connect). Hand-registered handlers win — register + them before `connect()` (as in v1) and the automatic install skips, so existing + escape-hatch servers keep their exact behavior. To veto subscriptions, call + `mcpServer.trackResourceSubscriptions({ onSubscribe })` before the **first** + `connect()`: the hook runs before a URI is recorded and a throw is returned to the + client as the request's error (after a connect has auto-installed the handlers, the + explicit call throws its duplicate-registration error). Tests pinning `-32601` for + `resources/subscribe` on servers that declare the capability need re-baselining. - **`WebStandardStreamableHTTPServerTransport` store-first `eventStore` semantics.** Request-related events emitted after `closeSSE()` — and the final response when no per-request stream is connected — are now persisted to the configured `eventStore` for diff --git a/docs/servers/resources.md b/docs/servers/resources.md index bb6cacd141..e76b936e64 100644 --- a/docs/servers/resources.md +++ b/docs/servers/resources.md @@ -214,11 +214,14 @@ The notification tells connected clients to call `resources/list` again. A chang ## Serve per-resource subscriptions -A 2025-era client opts into `notifications/resources/updated` for one URI with `resources/subscribe`. The SDK routes the verb; the bookkeeping is yours: advertise the capability, track the URIs per connection, and send the notification to subscribers only. +A 2025-era client opts into `notifications/resources/updated` for one URI with `resources/subscribe`. Declaring `resources: { subscribe: true }` is the whole opt-in: at connect time the SDK installs the `resources/subscribe` and `resources/unsubscribe` handlers and records the subscribed URIs in `resourceSubscriptions`. Sending the notification to subscribers only stays your call. ```ts source="../../examples/guides/servers/resources.examples.ts#sendResourceUpdated_subscribers" let deployStatus = 'idle'; +// Declaring `resources.subscribe` is the whole opt-in: at connect time the SDK +// installs the subscribe/unsubscribe handlers and records this connection's +// subscribed URIs in `deploys.resourceSubscriptions`. const deploys = new McpServer({ name: 'deploys', version: '1.0.0' }, { capabilities: { resources: { subscribe: true } } }); deploys.registerResource( @@ -228,26 +231,15 @@ deploys.registerResource( async uri => ({ contents: [{ uri: uri.href, text: deployStatus }] }) ); -// The SDK routes the two verbs; which URIs this connection watches is yours to track. -const subscribedUris = new Set(); -deploys.server.setRequestHandler('resources/subscribe', request => { - subscribedUris.add(request.params.uri); - return {}; -}); -deploys.server.setRequestHandler('resources/unsubscribe', request => { - subscribedUris.delete(request.params.uri); - return {}; -}); - async function setDeployStatus(status: string): Promise { deployStatus = status; - if (subscribedUris.has('deploys://status')) { - await deploys.server.sendResourceUpdated({ uri: 'deploys://status' }); + if (deploys.resourceSubscriptions.has('deploys://status')) { + await deploys.sendResourceUpdated({ uri: 'deploys://status' }); } } ``` -The `Set` belongs to one server instance, and each connection gets its own instance from your factory — a subscription never leaks across connections. Send `resources/updated` only to connections that subscribed; unsolicited per-resource updates are wrong on 2025-era connections. +The set belongs to one server instance, and each connection gets its own instance from your factory — a subscription never leaks across connections. Send `resources/updated` only to connections that subscribed; unsolicited per-resource updates are wrong on 2025-era connections. To refuse a subscription, call `trackResourceSubscriptions({ onSubscribe: uri => { ... } })` before connecting instead of relying on the automatic install: the hook runs before the URI is recorded, and a throw is returned to the client as the request's error (`onUnsubscribe` is symmetric). Handlers you hand-register on `server.server` always win — the automatic install skips when either verb already has one. The pattern needs a connection that outlives the subscribe call: over stdio (and any sessionful wiring) the instance and its `Set` live as long as the connection. Behind `createMcpHandler`'s stateless legacy fallback each POST gets a fresh instance, so `resources/subscribe` succeeds and the `Set` is discarded with it — no update can ever be delivered on that posture. [Support legacy clients](../serving/legacy-clients.md) covers the serving postures. @@ -255,7 +247,7 @@ The pattern needs a connection that outlives the subscribe call: over stdio (and On [2026-07-28](../protocol-versions.md) connections the verb does not exist: clients name resource URIs in their `subscriptions/listen` filter, and the entry filters delivery itself — `serveStdio` routes the instance's own `sendResourceUpdated` call onto matching streams, and `createMcpHandler` delivers what you publish on its notifier ([Notifications](./notifications.md#publish-a-resource-update-through-the-handler)). ::: -A dual-era server therefore still calls `sendResourceUpdated` on 2026-07-28 connections, where the subscribe set is always empty — gate on the connection's era as well as the set. The [`resources` example](https://github.com/modelcontextprotocol/typescript-sdk/tree/main/examples/resources) guards with `reqCtx.era === 'modern' || subscribedUris.has(uri)` in its factory and runs as a self-verifying pair: delivery is asserted over stdio on both eras and over HTTP on the 2026-07-28 listen path; the stateless legacy HTTP leg asserts only that the subscribe calls succeed. +A dual-era server therefore still calls `sendResourceUpdated` on 2026-07-28 connections, where the subscribe set is always empty — gate on the connection's era as well as the set. The [`resources` example](https://github.com/modelcontextprotocol/typescript-sdk/tree/main/examples/resources) guards with `reqCtx.era === 'modern' || server.resourceSubscriptions.has(uri)` in its factory and runs as a self-verifying pair: delivery is asserted over stdio on both eras and over HTTP on the 2026-07-28 listen path; the stateless legacy HTTP leg asserts only that the subscribe calls succeed. ## Recap @@ -265,4 +257,4 @@ A dual-era server therefore still calls `sendResourceUpdated` on 2026-07-28 conn - A template's `list` callback is what makes its instances appear in `resources/list`. - Resolve file-backed paths to their real location and reject anything outside the root before reading. - Registration changes emit `notifications/resources/list_changed` automatically. -- `resources/subscribe` bookkeeping is the server's: advertise `resources: { subscribe: true }`, track URIs per connection, send `resources/updated` to subscribers only. +- Declaring `resources: { subscribe: true }` gives you SDK-owned `resources/subscribe` bookkeeping automatically; the per-connection URI set is `resourceSubscriptions`, veto hooks come via `trackResourceSubscriptions({ onSubscribe })`, and you send `resources/updated` to subscribers only. diff --git a/examples/guides/servers/resources.examples.ts b/examples/guides/servers/resources.examples.ts index ec74419055..b53fdc4eed 100644 --- a/examples/guides/servers/resources.examples.ts +++ b/examples/guides/servers/resources.examples.ts @@ -124,6 +124,9 @@ server.registerResource( //#region sendResourceUpdated_subscribers let deployStatus = 'idle'; +// Declaring `resources.subscribe` is the whole opt-in: at connect time the SDK +// installs the subscribe/unsubscribe handlers and records this connection's +// subscribed URIs in `deploys.resourceSubscriptions`. const deploys = new McpServer({ name: 'deploys', version: '1.0.0' }, { capabilities: { resources: { subscribe: true } } }); deploys.registerResource( @@ -133,21 +136,10 @@ deploys.registerResource( async uri => ({ contents: [{ uri: uri.href, text: deployStatus }] }) ); -// The SDK routes the two verbs; which URIs this connection watches is yours to track. -const subscribedUris = new Set(); -deploys.server.setRequestHandler('resources/subscribe', request => { - subscribedUris.add(request.params.uri); - return {}; -}); -deploys.server.setRequestHandler('resources/unsubscribe', request => { - subscribedUris.delete(request.params.uri); - return {}; -}); - async function setDeployStatus(status: string): Promise { deployStatus = status; - if (subscribedUris.has('deploys://status')) { - await deploys.server.sendResourceUpdated({ uri: 'deploys://status' }); + if (deploys.resourceSubscriptions.has('deploys://status')) { + await deploys.sendResourceUpdated({ uri: 'deploys://status' }); } } //#endregion sendResourceUpdated_subscribers diff --git a/examples/resources/server.ts b/examples/resources/server.ts index 2ac10338d6..bd73af4256 100644 --- a/examples/resources/server.ts +++ b/examples/resources/server.ts @@ -24,6 +24,9 @@ const COUNTER_URI = 'counter://value'; let counter = 0; function buildServer(reqCtx: McpRequestContext, publishUpdated?: (uri: string) => void): McpServer { + // Declaring `resources.subscribe` is the whole opt-in: at connect time the SDK + // installs the resources/subscribe and resources/unsubscribe handlers and + // tracks the URIs THIS connection watches in `server.resourceSubscriptions`. const server = new McpServer( { name: 'resources-example', version: '1.0.0' }, { capabilities: { resources: { subscribe: true, listChanged: true } } } @@ -53,18 +56,6 @@ function buildServer(reqCtx: McpRequestContext, publishUpdated?: (uri: string) = async uri => ({ contents: [{ uri: uri.href, mimeType: 'text/plain', text: String(counter) }] }) ); - // resources/subscribe bookkeeping is the application's: the SDK routes the - // two verbs, and which URIs THIS connection watches lives here. - const subscribedUris = new Set(); - server.server.setRequestHandler('resources/subscribe', request => { - subscribedUris.add(request.params.uri); - return {}; - }); - server.server.setRequestHandler('resources/unsubscribe', request => { - subscribedUris.delete(request.params.uri); - return {}; - }); - server.registerTool('increment', { description: `Bump ${COUNTER_URI} by one` }, async () => { counter += 1; if (publishUpdated) { @@ -72,11 +63,11 @@ function buildServer(reqCtx: McpRequestContext, publishUpdated?: (uri: string) = // so the change is published on the entry's notifier for the listen // streams other requests hold open. publishUpdated(COUNTER_URI); - } else if (reqCtx.era === 'modern' || subscribedUris.has(COUNTER_URI)) { + } else if (reqCtx.era === 'modern' || server.resourceSubscriptions.has(COUNTER_URI)) { // Connection serving (stdio): announce in-band. The entry routes it // onto 2026-07-28 listen streams; on a 2025-era connection it goes // only to subscribers — unsolicited per-resource updates are wrong. - await server.server.sendResourceUpdated({ uri: COUNTER_URI }).catch(() => {}); + await server.sendResourceUpdated({ uri: COUNTER_URI }).catch(() => {}); } return { content: [{ type: 'text', text: String(counter) }] }; }); diff --git a/examples/todos-server/todos.ts b/examples/todos-server/todos.ts index 1877c93eda..aa6ceb8a0d 100644 --- a/examples/todos-server/todos.ts +++ b/examples/todos-server/todos.ts @@ -231,18 +231,10 @@ export function buildServer(reqCtx: McpRequestContext): McpServer { } ); - // Per-resource subscriptions: 2025-era clients call resources/subscribe (tracked here so - // updates only go to subscribers); 2026-07-28 clients use a subscriptions/listen filter and - // the serving entry routes the same notification onto it. - const subscribedUris = new Set(); - server.server.setRequestHandler('resources/subscribe', request => { - subscribedUris.add(request.params.uri); - return {}; - }); - server.server.setRequestHandler('resources/unsubscribe', request => { - subscribedUris.delete(request.params.uri); - return {}; - }); + // Per-resource subscriptions: the declared `resources.subscribe` capability makes the SDK + // track 2025-era resources/subscribe calls in `server.resourceSubscriptions` (so updates only + // go to subscribers); 2026-07-28 clients use a subscriptions/listen filter and the serving + // entry routes the same notification onto it. /** Tell connected clients the board changed: the resource list, and the board resource for watchers. */ const announceBoardChange = async (): Promise => { @@ -251,10 +243,10 @@ export function buildServer(reqCtx: McpRequestContext): McpServer { // Per-request HTTP serving: cross-request delivery goes through the entry's notifier. publishBoardChanged?.(); publishBoardUpdated('todos://board'); - } else if (reqCtx.era === 'modern' || subscribedUris.has('todos://board')) { + } else if (reqCtx.era === 'modern' || server.resourceSubscriptions.has('todos://board')) { // stdio: the serving entry routes this onto open listen subscriptions (2026-07-28); // on 2025-era connections it goes out unsolicited, so only send it to subscribers. - await server.server.sendResourceUpdated({ uri: 'todos://board' }).catch(() => {}); + await server.sendResourceUpdated({ uri: 'todos://board' }).catch(() => {}); } }; diff --git a/packages/core-internal/src/shared/protocol.ts b/packages/core-internal/src/shared/protocol.ts index ffab6e8df8..ac8074da89 100644 --- a/packages/core-internal/src/shared/protocol.ts +++ b/packages/core-internal/src/shared/protocol.ts @@ -1759,11 +1759,18 @@ export abstract class Protocol { this._requestHandlers.delete(method); } + /** + * Whether a request handler is currently registered for the given method. + */ + hasRequestHandler(method: RequestMethod | string): boolean { + return this._requestHandlers.has(method); + } + /** * Asserts that a request handler has not already been set for the given method, in preparation for a new one being automatically installed. */ assertCanSetRequestHandler(method: RequestMethod | string): void { - if (this._requestHandlers.has(method)) { + if (this.hasRequestHandler(method)) { throw new Error(`A request handler for ${method} already exists, which would be overridden`); } } diff --git a/packages/server/src/server/mcp.examples.ts b/packages/server/src/server/mcp.examples.ts index c4abb9ccb7..0cf9848f7f 100644 --- a/packages/server/src/server/mcp.examples.ts +++ b/packages/server/src/server/mcp.examples.ts @@ -122,6 +122,26 @@ async function McpServer_sendLoggingMessage_basic(server: McpServer) { //#endregion McpServer_sendLoggingMessage_basic } +/** + * Example: Tracking per-resource subscriptions (2025-era `resources/subscribe`). + */ +async function McpServer_trackResourceSubscriptions_basic(server: McpServer) { + //#region McpServer_trackResourceSubscriptions_basic + server.trackResourceSubscriptions({ + onSubscribe: uri => { + if (!uri.startsWith('status://')) { + throw new Error(`Subscriptions to ${uri} are not supported`); + } + } + }); + + // ...later, when the resource changes: + if (server.resourceSubscriptions.has('status://deploy')) { + await server.sendResourceUpdated({ uri: 'status://deploy' }); + } + //#endregion McpServer_trackResourceSubscriptions_basic +} + /** * Example: Logging from inside a tool handler via ctx.mcpReq.log(). */ diff --git a/packages/server/src/server/mcp.ts b/packages/server/src/server/mcp.ts index e0d10b5a8e..59370ebcdb 100644 --- a/packages/server/src/server/mcp.ts +++ b/packages/server/src/server/mcp.ts @@ -18,6 +18,7 @@ import type { ReadResourceResult, Resource, ResourceTemplateReference, + ResourceUpdatedNotification, Result, ServerContext, StandardSchemaWithJSON, @@ -146,6 +147,14 @@ export class McpServer { * ``` */ async connect(transport: Transport): Promise { + // A declared `resources.subscribe` capability gets SDK-owned subscription + // bookkeeping automatically; deferring the install to connect time lets a + // hand-registered `resources/subscribe` handler win by already existing. + this._autoTrackResourceSubscriptions(); + // Per-resource subscriptions are connection state: a reused instance must + // not carry the previous connection's subscriptions onto a new transport, + // where they would route unsolicited `notifications/resources/updated`. + this._resourceSubscriptions.clear(); return await this.server.connect(transport); } @@ -507,6 +516,137 @@ export class McpServer { this._resourceHandlersInitialized = true; } + private readonly _resourceSubscriptions = new Set(); + + /** + * The resource URIs the connected client has subscribed to via `resources/subscribe`, + * as recorded by the SDK's subscription tracking — automatic on {@linkcode connect} + * when the `resources.subscribe` capability is declared, or explicit via + * {@linkcode trackResourceSubscriptions}. Empty while no tracking is active, and + * always empty on 2026-07-28 connections (per-resource subscriptions travel on + * `subscriptions/listen` there, handled by the serving entries). + * Subscriptions are connection state: {@linkcode connect} resets the set. + */ + get resourceSubscriptions(): ReadonlySet { + return this._resourceSubscriptions; + } + + /** + * Explicitly enables SDK-owned bookkeeping for per-resource subscriptions on + * 2025-era connections, with optional veto hooks: installs the + * `resources/subscribe` and `resources/unsubscribe` request handlers, records + * subscribed URIs in {@linkcode resourceSubscriptions}, and declares the + * `resources.subscribe` capability (so, like any capability registration, it + * must be called before {@linkcode connect}). + * + * Declaring the `resources.subscribe` capability alone already activates the + * same tracking automatically at {@linkcode connect} time — call this method + * only to attach hooks, or to declare the capability and install the handlers + * in one step. Handlers hand-registered on the underlying {@linkcode server} + * always win: the automatic install skips when either subscription verb + * already has a handler. + * + * `onSubscribe` runs before the URI is recorded: throw to refuse the + * subscription — the error propagates as the handler error, so the client + * receives it as the JSON-RPC error response (throw a {@linkcode ProtocolError} + * to control the error code) and the set is left unchanged. `onUnsubscribe` + * is symmetric: it runs before the URI is removed, and a throw leaves the + * subscription in place. + * + * Calling this more than once throws: the first call installed the + * `resources/subscribe` handler, so the duplicate-registration guard rejects + * the second. The same guard rejects it when a subscribe handler was + * installed manually on the underlying server, or when a prior + * {@linkcode connect} already auto-installed the handlers for a declared + * `resources.subscribe` capability — attach hooks before the first connect. + * + * On 2026-07-28 connections the two verbs do not exist (clients name resource + * URIs in their `subscriptions/listen` filter and the serving entries own that + * bookkeeping), and requests for them are rejected before any handler runs — + * enabling tracking on a dual-era server is harmless there. + * + * @example + * ```ts source="./mcp.examples.ts#McpServer_trackResourceSubscriptions_basic" + * server.trackResourceSubscriptions({ + * onSubscribe: uri => { + * if (!uri.startsWith('status://')) { + * throw new Error(`Subscriptions to ${uri} are not supported`); + * } + * } + * }); + * + * // ...later, when the resource changes: + * if (server.resourceSubscriptions.has('status://deploy')) { + * await server.sendResourceUpdated({ uri: 'status://deploy' }); + * } + * ``` + */ + trackResourceSubscriptions(options?: { + /** Veto hook: runs before the URI is recorded; a throw refuses the subscription. */ + onSubscribe?: (uri: string, ctx: ServerContext) => void | Promise; + /** Symmetric hook for `resources/unsubscribe`; a throw leaves the subscription in place. */ + onUnsubscribe?: (uri: string, ctx: ServerContext) => void | Promise; + }): void { + // Assert-first, mutate-after: a repeat call (or a manually installed + // subscribe handler) throws here before any state changes. + this.server.assertCanSetRequestHandler('resources/subscribe'); + this.server.assertCanSetRequestHandler('resources/unsubscribe'); + + // A server advertising the resources capability must also answer its + // list/read methods, so those handlers are installed first when nothing has + // done so yet. This call asserts before it mutates, which keeps the whole + // method atomic: any throw leaves no capability half-declared. + this.setResourceRequestHandlers(); + + // Declaring `resources.subscribe` is what advertises the two verbs; doing it + // here (merged with any previously declared resources bits) makes the opt-in + // a single call, the same way registerTool declares `tools`. + this.server.registerCapabilities({ resources: { subscribe: true } }); + + this._installResourceSubscriptionHandlers(options); + } + + /** + * The {@linkcode connect}-time automatic arm of the subscription tracking: a + * declared `resources.subscribe` capability gets the same handlers an explicit + * {@linkcode trackResourceSubscriptions} call installs (without hooks). The + * install is skipped when either subscription verb already has a handler — a + * hand-registered handler wins, and a repeat {@linkcode connect} finds the + * first connect's handlers and does nothing (the set is cleared separately). + */ + private _autoTrackResourceSubscriptions(): void { + if (!this.server.getCapabilities().resources?.subscribe) { + return; + } + if (this.server.hasRequestHandler('resources/subscribe') || this.server.hasRequestHandler('resources/unsubscribe')) { + return; + } + this._installResourceSubscriptionHandlers(); + } + + private _installResourceSubscriptionHandlers(options?: { + onSubscribe?: (uri: string, ctx: ServerContext) => void | Promise; + onUnsubscribe?: (uri: string, ctx: ServerContext) => void | Promise; + }): void { + this.server.setRequestHandler('resources/subscribe', async (request, ctx) => { + await options?.onSubscribe?.(request.params.uri, ctx); + // A request cancelled while the hook ran must not mutate the set — + // the protocol suppresses the aborted request's response, so the + // client would never learn a subscription was recorded. + ctx.mcpReq.signal.throwIfAborted(); + this._resourceSubscriptions.add(request.params.uri); + return {}; + }); + + this.server.setRequestHandler('resources/unsubscribe', async (request, ctx) => { + await options?.onUnsubscribe?.(request.params.uri, ctx); + // Symmetric to subscribe: no set mutation for a cancelled request. + ctx.mcpReq.signal.throwIfAborted(); + this._resourceSubscriptions.delete(request.params.uri); + return {}; + }); + } + private _promptHandlersInitialized = false; private setPromptRequestHandlers() { @@ -1123,6 +1263,17 @@ export class McpServer { } } + /** + * Sends a resource updated event for the given resource URI to the client, if connected. + * On 2025-era connections, send this only for URIs the client subscribed to — see + * {@linkcode trackResourceSubscriptions} and {@linkcode resourceSubscriptions}. + */ + async sendResourceUpdated(params: ResourceUpdatedNotification['params']): Promise { + if (this.isConnected()) { + await this.server.sendResourceUpdated(params); + } + } + /** * Sends a tool list changed event to the client, if connected. */ diff --git a/packages/server/test/server/resourceSubscriptionTracking.test.ts b/packages/server/test/server/resourceSubscriptionTracking.test.ts new file mode 100644 index 0000000000..1291c6e62a --- /dev/null +++ b/packages/server/test/server/resourceSubscriptionTracking.test.ts @@ -0,0 +1,273 @@ +import { ProtocolError, ProtocolErrorCode } from '@modelcontextprotocol/core-internal'; +import { describe, expect, it, vi } from 'vitest'; +import { McpServer } from '../../src/index'; +import { errorOf, legacyInitialize, resultOf, wireLegacy } from './legacyShimHarness'; + +const subscribeRequest = (id: number, uri: string) => ({ jsonrpc: '2.0', id, method: 'resources/subscribe', params: { uri } }) as const; +const unsubscribeRequest = (id: number, uri: string) => ({ jsonrpc: '2.0', id, method: 'resources/unsubscribe', params: { uri } }) as const; + +describe('trackResourceSubscriptions', () => { + it('a declared resources.subscribe capability auto-installs tracking at connect', async () => { + // Declaring the capability is the whole opt-in: connect() installs the + // SDK's subscribe/unsubscribe handlers when nothing else claimed them. + const server = new McpServer({ name: 's', version: '1.0.0' }, { capabilities: { resources: { subscribe: true } } }); + const wire = await wireLegacy(server); + await wire.request(legacyInitialize(1)); + + const subscribed = await wire.request(subscribeRequest(2, 'demo://a')); + expect(resultOf(subscribed)).toEqual({}); + expect([...server.resourceSubscriptions]).toEqual(['demo://a']); + + const unsubscribed = await wire.request(unsubscribeRequest(3, 'demo://a')); + expect(resultOf(unsubscribed)).toEqual({}); + expect(server.resourceSubscriptions.size).toBe(0); + + await server.close(); + }); + + it('baseline: no capability and no explicit call leaves resources/subscribe method-not-found', async () => { + const server = new McpServer({ name: 's', version: '1.0.0' }); + const wire = await wireLegacy(server); + await wire.request(legacyInitialize(1)); + + const response = await wire.request(subscribeRequest(2, 'demo://a')); + expect(errorOf(response)).toEqual({ code: -32601, message: 'Method not found' }); + expect(server.resourceSubscriptions.size).toBe(0); + + await server.close(); + }); + + it('a hand-registered resources/subscribe handler wins over the automatic install', async () => { + const server = new McpServer({ name: 's', version: '1.0.0' }, { capabilities: { resources: { subscribe: true } } }); + const manualUris: string[] = []; + server.server.setRequestHandler('resources/subscribe', request => { + manualUris.push(request.params.uri); + return {}; + }); + + const wire = await wireLegacy(server); + await wire.request(legacyInitialize(1)); + + const subscribed = await wire.request(subscribeRequest(2, 'demo://a')); + expect(resultOf(subscribed)).toEqual({}); + expect(manualUris).toEqual(['demo://a']); + // The SDK's tracking stayed out of the way entirely. + expect(server.resourceSubscriptions.size).toBe(0); + // The automatic install skips as a pair: the sibling verb keeps its + // pre-existing behavior instead of getting half of the SDK's pair. + expect(errorOf(await wire.request(unsubscribeRequest(3, 'demo://a'))).code).toBe(-32601); + + await server.close(); + }); + + it('auto-tracking also serves low-level servers that answer list/read themselves', async () => { + // The explicit call refuses this posture (it installs the registry-backed + // list/read handlers); the automatic arm only adds the subscribe pair, so + // declaring the capability is enough for hand-rolled resource servers. + const server = new McpServer({ name: 's', version: '1.0.0' }); + server.server.registerCapabilities({ resources: { subscribe: true } }); + server.server.setRequestHandler('resources/list', () => ({ resources: [] })); + + const wire = await wireLegacy(server); + await wire.request(legacyInitialize(1)); + + const subscribed = await wire.request(subscribeRequest(2, 'demo://a')); + expect(resultOf(subscribed)).toEqual({}); + expect([...server.resourceSubscriptions]).toEqual(['demo://a']); + + await server.close(); + }); + + it('records subscribed URIs and removes them on unsubscribe', async () => { + const server = new McpServer({ name: 's', version: '1.0.0' }); + server.trackResourceSubscriptions(); + const wire = await wireLegacy(server); + await wire.request(legacyInitialize(1)); + + const subscribed = await wire.request(subscribeRequest(2, 'demo://a')); + expect(resultOf(subscribed)).toEqual({}); + expect([...server.resourceSubscriptions]).toEqual(['demo://a']); + + await wire.request(subscribeRequest(3, 'demo://b')); + expect([...server.resourceSubscriptions].sort()).toEqual(['demo://a', 'demo://b']); + + const unsubscribed = await wire.request(unsubscribeRequest(4, 'demo://a')); + expect(resultOf(unsubscribed)).toEqual({}); + expect([...server.resourceSubscriptions]).toEqual(['demo://b']); + + await server.close(); + }); + + it('onSubscribe veto propagates as the handler error and leaves the set unchanged', async () => { + const server = new McpServer({ name: 's', version: '1.0.0' }); + server.trackResourceSubscriptions({ + onSubscribe: uri => { + if (uri.startsWith('secret://')) { + throw new ProtocolError(ProtocolErrorCode.InvalidParams, `Subscriptions to ${uri} are refused`); + } + } + }); + const wire = await wireLegacy(server); + await wire.request(legacyInitialize(1)); + + const refused = await wire.request(subscribeRequest(2, 'secret://a')); + expect(errorOf(refused).code).toBe(ProtocolErrorCode.InvalidParams); + expect(errorOf(refused).message).toContain('Subscriptions to secret://a are refused'); + expect(server.resourceSubscriptions.size).toBe(0); + + // The veto is per-URI: a URI the hook allows is still recorded. + const allowed = await wire.request(subscribeRequest(3, 'demo://a')); + expect(resultOf(allowed)).toEqual({}); + expect([...server.resourceSubscriptions]).toEqual(['demo://a']); + + await server.close(); + }); + + it('onUnsubscribe veto keeps the subscription in place', async () => { + const server = new McpServer({ name: 's', version: '1.0.0' }); + server.trackResourceSubscriptions({ + onUnsubscribe: () => { + throw new Error('unsubscribe refused'); + } + }); + const wire = await wireLegacy(server); + await wire.request(legacyInitialize(1)); + + await wire.request(subscribeRequest(2, 'demo://a')); + expect([...server.resourceSubscriptions]).toEqual(['demo://a']); + + const refused = await wire.request(unsubscribeRequest(3, 'demo://a')); + expect(errorOf(refused).message).toContain('unsubscribe refused'); + expect([...server.resourceSubscriptions]).toEqual(['demo://a']); + + await server.close(); + }); + + it('sendResourceUpdated facade delivers notifications/resources/updated with the uri', async () => { + const server = new McpServer({ name: 's', version: '1.0.0' }); + server.trackResourceSubscriptions(); + const wire = await wireLegacy(server); + await wire.request(legacyInitialize(1)); + + await wire.request(subscribeRequest(2, 'demo://a')); + await server.sendResourceUpdated({ uri: 'demo://a' }); + + await vi.waitFor(() => expect(wire.notifications.some(n => n.method === 'notifications/resources/updated')).toBe(true)); + const notification = wire.notifications.find(n => n.method === 'notifications/resources/updated'); + expect(notification?.params?.uri).toBe('demo://a'); + + await server.close(); + }); + + it('a request cancelled during the onSubscribe hook does not record the subscription', async () => { + let releaseHook!: () => void; + const hookGate = new Promise(resolve => (releaseHook = resolve)); + let hookSignal: AbortSignal | undefined; + + const server = new McpServer({ name: 's', version: '1.0.0' }); + server.trackResourceSubscriptions({ + onSubscribe: async (_uri, ctx) => { + hookSignal = ctx.mcpReq.signal; + await hookGate; + } + }); + const wire = await wireLegacy(server); + await wire.request(legacyInitialize(1)); + + // The response of a cancelled request is suppressed, so do not await it. + void wire.request(subscribeRequest(2, 'demo://a')); + await vi.waitFor(() => expect(hookSignal).toBeDefined()); + + await wire.notifyFromPeer({ jsonrpc: '2.0', method: 'notifications/cancelled', params: { requestId: 2, reason: 'test' } }); + await vi.waitFor(() => expect(hookSignal?.aborted).toBe(true)); + + // Only now does the hook return — the handler resumes on an already- + // aborted request and must leave the set unchanged. + releaseHook(); + await new Promise(resolve => setTimeout(resolve, 10)); + expect(server.resourceSubscriptions.size).toBe(0); + + await server.close(); + }); + + it('calling trackResourceSubscriptions twice throws via the duplicate-handler guard', () => { + const server = new McpServer({ name: 's', version: '1.0.0' }); + server.trackResourceSubscriptions(); + expect(() => server.trackResourceSubscriptions()).toThrow( + 'A request handler for resources/subscribe already exists, which would be overridden' + ); + }); + + it('declares the resources.subscribe capability, merged with existing resources bits', async () => { + // No resources capability declared anywhere: tracking declares it (and + // installs the resource handlers a declared capability requires). + const bare = new McpServer({ name: 's', version: '1.0.0' }); + bare.trackResourceSubscriptions(); + expect(bare.server.getCapabilities().resources?.subscribe).toBe(true); + + // Previously declared bits survive the merge and reach the client on initialize. + const declared = new McpServer({ name: 's', version: '1.0.0' }, { capabilities: { resources: { listChanged: true } } }); + declared.trackResourceSubscriptions(); + expect(declared.server.getCapabilities().resources).toEqual({ listChanged: true, subscribe: true }); + + const wire = await wireLegacy(declared); + const initialized = resultOf(await wire.request(legacyInitialize(1))) as { + capabilities?: { resources?: Record }; + }; + expect(initialized.capabilities?.resources).toEqual({ listChanged: true, subscribe: true }); + + await declared.close(); + }); + + it('throws when called after connecting (capabilities are fixed at connect time)', async () => { + const server = new McpServer({ name: 's', version: '1.0.0' }); + await wireLegacy(server); + + expect(() => server.trackResourceSubscriptions()).toThrow('Cannot register capabilities after connecting to transport'); + + await server.close(); + }); + + it('throws without declaring the capability when low-level resource handlers are already installed', async () => { + // A server answering resources/list at the low level conflicts with the + // registry-backed handlers this method installs; the throw must leave no + // trace — advertising resources.subscribe with no subscribe handler would + // be the worst of both worlds. + const server = new McpServer({ name: 's', version: '1.0.0' }); + server.server.registerCapabilities({ resources: {} }); + server.server.setRequestHandler('resources/list', () => ({ resources: [] })); + + expect(() => server.trackResourceSubscriptions()).toThrow('A request handler for resources/list already exists'); + expect(server.server.getCapabilities().resources?.subscribe).toBeUndefined(); + + const wire = await wireLegacy(server); + await wire.request(legacyInitialize(1)); + expect(errorOf(await wire.request(subscribeRequest(2, 'demo://a'))).code).toBe(-32601); + + await server.close(); + }); + + it('reconnect is idempotent: no duplicate-handler throw, and the set resets each connect', async () => { + // Auto path: the first connect installs the handlers; the second finds + // them already installed and only clears the connection-scoped set. + const server = new McpServer({ name: 's', version: '1.0.0' }, { capabilities: { resources: { subscribe: true } } }); + + const first = await wireLegacy(server); + await first.request(legacyInitialize(1)); + await first.request(subscribeRequest(2, 'demo://a')); + expect([...server.resourceSubscriptions]).toEqual(['demo://a']); + await first.close(); + + // Reconnecting the same instance serves a NEW client; the previous + // client's subscriptions must not survive into the new connection. + const second = await wireLegacy(server); + await second.request(legacyInitialize(1)); + expect(server.resourceSubscriptions.size).toBe(0); + + // The surviving handlers still work on the new connection. + await second.request(subscribeRequest(2, 'demo://b')); + expect([...server.resourceSubscriptions]).toEqual(['demo://b']); + + await server.close(); + }); +});