diff --git a/.claude/skills/test-servers/SKILL.md b/.claude/skills/test-servers/SKILL.md index fbde45e502..5fa8b2d00f 100644 --- a/.claude/skills/test-servers/SKILL.md +++ b/.claude/skills/test-servers/SKILL.md @@ -269,6 +269,7 @@ usually looks like a missing capability rather than an error. | OAuth token revocation on clear | `oauth-revocation-http.json` (legacy) | | A token endpoint the SDK refuses | `oauth-insecure-token-endpoint-http.json` (legacy) | | Cancelling a call mid-flight | `cancellation-modern-http.json` (modern) | +| A resource update reaching every subscribed session (`--load-subscribe`) | `subscriptions-broadcast-http.json` (legacy) | ## Adding a config or preset diff --git a/clients/cli/README.md b/clients/cli/README.md index 9d400e669f..205aa42898 100644 --- a/clients/cli/README.md +++ b/clients/cli/README.md @@ -658,6 +658,14 @@ mcp-inspector --cli https://example.com/mcp --method tools/call --load \ fails. More than one subscriber needs an HTTP or SSE server, for the reason `--load-connections` does, and the extra sessions open with stored credentials only. +- To try it without writing a server, run the + `subscriptions-broadcast-http.json` test server + ([docs](../../docs/test-servers.md#resource-updates-fanned-out-to-every-session)), + whose `update_resource` notifies every subscribed session: + `--tool-name update_resource --tool-arg uri=test://resource_1 + --tool-arg 'text=v{{seq}}' --load-subscribe test://resource_1`. The stock + composable server notifies only the calling session, so a run against it + reports the other subscribers as missed. - Not combined with `--load-connections`, `--load-churn`, `--load-steps` / `--load-ramp`, `--load-warmup` or `--load-open-loop` (which would time each trigger from its scheduled start). The CLI's own connection is subscribed too, diff --git a/clients/cli/__tests__/subscriptions-broadcast.test.ts b/clients/cli/__tests__/subscriptions-broadcast.test.ts new file mode 100644 index 0000000000..55af0095dc --- /dev/null +++ b/clients/cli/__tests__/subscriptions-broadcast.test.ts @@ -0,0 +1,197 @@ +import { describe, it, expect, afterEach } from "vitest"; +import path from "node:path"; +import { fileURLToPath } from "node:url"; +import { + createResourceUpdateBroadcast, + createTestServerHttp, + loadConfig, + resolveConfig, + setResourceText, + type ResourceUpdateBroadcast, + type ResourceUpdateMember, + type TestServerHttp, +} from "@modelcontextprotocol/inspector-test-server"; +import { runCli } from "./helpers/cli-runner.js"; +import { expectCliSuccess, expectValidJson } from "./helpers/assertions.js"; + +/** + * The `subscriptions-broadcast-http` showcase config (#2964): a resource + * update that reaches every subscribed session, so `--load-subscribe` can be + * tried by hand without writing a server. Without the broadcast, each session + * of the composable server is its own server, so `update_resource` notifies + * only the session that called it and a fan-out run reports the rest missed. + */ + +const CONFIGS = path.resolve( + path.dirname(fileURLToPath(import.meta.url)), + "../../../test-servers/configs", +); +const URI = "test://resource_1"; + +let server: TestServerHttp | null = null; + +afterEach(async () => { + if (server) { + try { + await server.stop(); + } catch { + // ignore + } + server = null; + } +}); + +/** Start a showcase config in-process, on a port the harness picks. */ +async function startConfig(name: string): Promise<{ + url: string; + hub: ResourceUpdateBroadcast | undefined; +}> { + const resolved = resolveConfig(loadConfig(path.join(CONFIGS, name))); + const started = createTestServerHttp({ ...resolved, port: undefined }); + await started.start(); + server = started; + return { url: started.url, hub: resolved.resourceUpdateBroadcast }; +} + +/** A `--load-subscribe` fan-out run triggered by `update_resource`. */ +function fanoutRun(url: string): string[] { + return [ + url, + "--cli", + "--transport", + "http", + "--method", + "tools/call", + "--tool-name", + "update_resource", + "--tool-arg", + `uri=${URI}`, + "--tool-arg", + "text=v{{seq}}", + "--load", + "--load-requests", + "4", + "--load-subscribe", + URI, + "--load-subscribers", + "3", + "--load-subscribe-wait", + "1", + ]; +} + +describe("subscriptions-broadcast-http.json (CLI)", () => { + it("delivers every update to every subscribed session", async () => { + const { url, hub } = await startConfig("subscriptions-broadcast-http.json"); + const result = await runCli(fanoutRun(url)); + expectCliSuccess(result); + expect(expectValidJson(result)).toMatchObject({ + ok: 4, + fanout: { + uri: URI, + mode: "triggered", + subscribed: 3, + updates: 4, + delivered: 12, + missed: 0, + }, + }); + + // A session opened after the run serves the last update, not the preset. + const read = await runCli([ + url, + "--cli", + "--transport", + "http", + "--method", + "resources/read", + "--uri", + URI, + ]); + expectCliSuccess(read); + expect(expectValidJson(read)).toMatchObject({ + contents: [{ uri: URI, text: "v4" }], + }); + + // Every session left the hub when its connection closed. + expect(hub?.size).toBe(0); + }); + + it("does not hold a server whose sessionless POST was rejected", async () => { + const { url, hub } = await startConfig("subscriptions-broadcast-http.json"); + // Not an `initialize`, and no Mcp-Session-Id: the transport refuses it, and + // the server built for it is never registered as a session nor closed. + const response = await fetch(url, { + method: "POST", + headers: { + "content-type": "application/json", + accept: "application/json, text/event-stream", + }, + body: JSON.stringify({ jsonrpc: "2.0", id: 1, method: "tools/list" }), + }); + expect(response.ok).toBe(false); + expect(hub?.size).toBe(0); + }); + + it("is what the stock legacy config lacks: only the caller is notified", async () => { + const { url, hub } = await startConfig("subscriptions-legacy-http.json"); + expect(hub).toBeUndefined(); + const result = await runCli(fanoutRun(url)); + expectCliSuccess(result); + expect(expectValidJson(result)).toMatchObject({ + fanout: { updates: 4, delivered: 4, missed: 8, sessionsMissed: 2 }, + }); + }); +}); + +describe("createResourceUpdateBroadcast", () => { + function member(subscribed: boolean, failing = false) { + const applied: Array<[string, string]> = []; + const m: ResourceUpdateMember = { + apply: (uri, text) => { + applied.push([uri, text]); + }, + notify: async () => { + if (failing) throw new Error("stream gone"); + return subscribed; + }, + }; + return { m, applied }; + } + + it("applies to every member and counts the subscribed ones notified", async () => { + const hub = createResourceUpdateBroadcast(); + const a = member(true); + const b = member(false); + const broken = member(true, true); + hub.join(a.m); + hub.join(b.m); + hub.join(broken.m); + expect(hub.size).toBe(3); + + // A failing send is skipped rather than failing the publish. + await expect(hub.publish(URI, "v1")).resolves.toBe(1); + expect(a.applied).toEqual([[URI, "v1"]]); + expect(b.applied).toEqual([[URI, "v1"]]); + }); + + it("replays the latest update to a late joiner, and stops after leave", async () => { + const hub = createResourceUpdateBroadcast(); + await hub.publish(URI, "v1"); + await hub.publish(URI, "v2"); + const late = member(true); + const leave = hub.join(late.m); + expect(late.applied).toEqual([[URI, "v2"]]); + + leave(); + expect(hub.size).toBe(0); + await expect(hub.publish(URI, "v3")).resolves.toBe(0); + expect(late.applied).toEqual([[URI, "v2"]]); + }); +}); + +describe("setResourceText", () => { + it("returns false for a URI that is not registered", () => { + expect(setResourceText(new Map(), URI, "v1")).toBe(false); + }); +}); diff --git a/docs/test-servers.md b/docs/test-servers.md index d3c7bea80b..94db24b755 100644 --- a/docs/test-servers.md +++ b/docs/test-servers.md @@ -60,6 +60,7 @@ as a missing capability rather than an error. | `oauth-stalled-{token,discovery}-http.json` **(legacy era)** | An OAuth endpoint that accepts the request and never answers, so the OAuth-path request timeouts fire against a real idle socket | [#2382](https://github.com/modelcontextprotocol/inspector/issues/2382) | | `logging-{legacy,modern}-http.json` **(era per file)** | Logging, both eras | [#1629](https://github.com/modelcontextprotocol/inspector/issues/1629) | | `subscriptions-{legacy,modern}-http.json` **(era per file)** | Resource subscriptions, both eras | [#1630](https://github.com/modelcontextprotocol/inspector/issues/1630) | +| `subscriptions-broadcast-http.json` **(legacy era)** | A resource update that reaches **every** subscribed session — the server for a `--load-subscribe` fan-out run | [#2964](https://github.com/modelcontextprotocol/inspector/issues/2964) | | `subscriptions-never-acknowledged-http.json` **(modern era)** | A `subscriptions/listen` answered with a bare result | [#2097](https://github.com/modelcontextprotocol/inspector/issues/2097) | | `subscriptions-listen-refused-http.json` **(modern era)** | A `subscriptions/listen` refused with a bare HTTP 403 | [#2722](https://github.com/modelcontextprotocol/inspector/issues/2722) | | `tasks-{legacy,modern}-http.json` **(era per file)** | Tasks, both eras | [#1631](https://github.com/modelcontextprotocol/inspector/issues/1631) | @@ -717,6 +718,28 @@ The modern config deliberately **omits** `update_resource`. The SDK's modern leg So the live update-notification round-trip is demonstrated on the legacy (stateful-session) server, and the modern server is for the subscribe/listen/badge behavior. The Inspector's _receive_ path is era-transparent, so a real stateful modern server that routes `resources/updated` onto the listen stream drives the subscribed tile the same way. +## Resource updates fanned out to every session + +`subscriptions-broadcast-http.json` serves the same three `numbered_resources` and `update_resource` tool as `subscriptions-legacy-http.json`, plus `broadcastResourceUpdates: true`. It is the server to point `--load-subscribe` ([CLI README](../clients/cli/README.md#notification-fan-out---load-subscribe)) at by hand. Connect with the legacy era. + +Over Streamable HTTP each session of the composable server is its own server instance, so on the stock legacy config `update_resource` changes and notifies **only the session that called it**. A fan-out run against it reports every other subscriber as missed — correctly, since nothing was sent to them: + +```sh +node test-servers/build/server-composable.js --config test-servers/configs/subscriptions-legacy-http.json +mcp-inspector --cli http://127.0.0.1:3221/mcp --transport http --method tools/call \ + --tool-name update_resource --tool-arg uri=test://resource_1 --tool-arg 'text=v{{seq}}' \ + --load --load-requests 10 --load-subscribe test://resource_1 --load-subscribers 5 --format text +# Delivered: 10 of 50 notifications (10 updates × 5), 40 missed across 4 sessions, 0 unexpected +``` + +`broadcastResourceUpdates` gives the server one hub shared by every session (`test-servers/src/resource-update-broadcast.ts`). Each session joins it once initialized and leaves when its connection closes, and `update_resource` publishes through it: every session's copy of the resource takes the new text, and every session subscribed to it gets `notifications/resources/updated`. The same run against port `3225` delivers them all: + +``` +# Delivered: 50 of 50 notifications (10 updates × 5), 0 missed across 0 sessions, 0 unexpected +``` + +The latest text per URI is kept, so a session opened after an update — a `resources/read` from a fresh CLI call, or a subscriber re-reading on notification — serves the updated content rather than the preset's. The tool's result says how many subscribed sessions it notified. In-process, pass `resourceUpdateBroadcast: createResourceUpdateBroadcast()` on the `ServerConfig` instead. + ## A listen that is never acknowledged `subscriptions-never-acknowledged-http.json` serves the same three `numbered_resources` on the modern leg. It acknowledges the first `subscriptions/listen` **that subscribes to a resource** normally, and answers every resource-subscription listen after it with a bare JSON-RPC `result` instead of a `notifications/subscriptions/acknowledged`. A listen carrying only list-change opt-ins is always acknowledged — including the one the Inspector opens at connect time ([#1920](https://github.com/modelcontextprotocol/inspector/issues/1920)), which is why the counting is per *resource-subscription* listen: otherwise that connect-time listen spends the allowance before you have clicked anything and the very first Subscribe is refused. Connect with **Protocol Era = Modern**. diff --git a/test-servers/configs/subscriptions-broadcast-http.json b/test-servers/configs/subscriptions-broadcast-http.json new file mode 100644 index 0000000000..eaaaa57867 --- /dev/null +++ b/test-servers/configs/subscriptions-broadcast-http.json @@ -0,0 +1,14 @@ +{ + "serverInfo": { + "name": "subscriptions-broadcast", + "version": "1.0.0" + }, + "tools": [{ "preset": "update_resource" }], + "resources": [{ "preset": "numbered_resources", "params": { "count": 3 } }], + "subscriptions": true, + "broadcastResourceUpdates": true, + "transport": { + "type": "streamable-http", + "port": 3225 + } +} diff --git a/test-servers/src/composable-test-server.ts b/test-servers/src/composable-test-server.ts index fdd404c5d2..6aeaf26536 100644 --- a/test-servers/src/composable-test-server.ts +++ b/test-servers/src/composable-test-server.ts @@ -13,6 +13,10 @@ import { isCompletable, isInputRequiredResult, } from "@modelcontextprotocol/server"; +import { + setResourceText, + type ResourceUpdateBroadcast, +} from "./resource-update-broadcast.js"; import type { Implementation, Tool, @@ -199,6 +203,8 @@ export interface TestServerContext { server: McpServer; state: ServerState; serverControl?: { isClosing(): boolean }; + /** The cross-session hub, when {@link ServerConfig.resourceUpdateBroadcast} is set. */ + resourceUpdateBroadcast?: ResourceUpdateBroadcast; } // --------------------------------------------------------------------------- @@ -898,6 +904,15 @@ export interface ServerConfig { * collects the contexts here and sends to each. */ onResourceSubscribe?: (uri: string, context: TestServerContext) => void; + /** + * Share resource updates across every session this config builds (#2964). + * Each session joins the hub and leaves when its connection closes, and + * `update_resource` publishes through it — so one call changes the resource + * in every session and notifies each one subscribed, the shape a + * `--load-subscribe` fan-out run needs. Pass one hub per server: a config + * file sets `broadcastResourceUpdates: true` and `resolveConfig` creates it. + */ + resourceUpdateBroadcast?: ResourceUpdateBroadcast; } /** @@ -1033,6 +1048,9 @@ export function createMcpServer(config: ServerConfig): McpServer { server: mcpServer, state, ...(config.serverControl && { serverControl: config.serverControl }), + ...(config.resourceUpdateBroadcast && { + resourceUpdateBroadcast: config.resourceUpdateBroadcast, + }), }; // Set up logging handler if logging is enabled. `logging/setLevel` is a spec @@ -1737,9 +1755,59 @@ export function createMcpServer(config: ServerConfig): McpServer { wireExtensionGatedTools(mcpServer, state, config.extensionGatedTools); } + // Cross-session resource updates (#2964). Wired after the resources are + // registered, so the hub's replay of earlier updates has copies to apply to. + if (config.resourceUpdateBroadcast) { + wireResourceUpdateBroadcast( + mcpServer, + state, + config.resourceUpdateBroadcast, + ); + } + return mcpServer; } +/** + * Join `broadcast` as this session once the client completes initialization, + * and leave it when the connection closes. Joining on `initialized` rather + * than at construction keeps out a server that never becomes a session — a + * sessionless HTTP POST the transport rejects builds one, registers it + * nowhere and never closes it, so a hub joined at construction would hold it + * forever. Nothing can subscribe before `initialized`, and the hub replays + * the latest updates on join, so no update is lost by waiting. Chains any + * `oninitialized` / `onclose` already set rather than replacing them. + */ +function wireResourceUpdateBroadcast( + mcpServer: McpServer, + state: ServerState, + broadcast: ResourceUpdateBroadcast, +): void { + let leave: (() => void) | undefined; + const previousOnInitialized = mcpServer.server.oninitialized; + mcpServer.server.oninitialized = () => { + leave ??= broadcast.join({ + apply: (uri, text) => { + setResourceText(state.registeredResources, uri, text); + }, + notify: async (uri) => { + if (!state.resourceSubscriptions.has(uri)) { + return false; + } + await mcpServer.server.sendResourceUpdated({ uri }); + return true; + }, + }); + previousOnInitialized?.(); + }; + const previousOnClose = mcpServer.server.onclose; + mcpServer.server.onclose = () => { + leave?.(); + leave = undefined; + previousOnClose?.(); + }; +} + /** * Gate the named tools on a client-declared extension. Each gated tool is * disabled up front (so it is absent from `tools/list` by default), and the diff --git a/test-servers/src/index.ts b/test-servers/src/index.ts index e63eab10fd..f720345b6e 100644 --- a/test-servers/src/index.ts +++ b/test-servers/src/index.ts @@ -12,3 +12,4 @@ export * from "./test-server-oauth-jwt.js"; export * from "./test-helpers.js"; export * from "./load-config.js"; export * from "./resolve-config.js"; +export * from "./resource-update-broadcast.js"; diff --git a/test-servers/src/load-config.ts b/test-servers/src/load-config.ts index 39599c5d67..b5e878539c 100644 --- a/test-servers/src/load-config.ts +++ b/test-servers/src/load-config.ts @@ -91,6 +91,12 @@ export interface ConfigFile { prompts?: boolean; }; subscriptions?: boolean; + /** + * Share resource updates across sessions, so `update_resource` changes and + * notifies every session rather than only its caller — see + * {@link ServerConfig.resourceUpdateBroadcast} (#2964). + */ + broadcastResourceUpdates?: boolean; tasks?: { list?: boolean; cancel?: boolean; diff --git a/test-servers/src/resolve-config.ts b/test-servers/src/resolve-config.ts index e832cbda25..37f0f2be29 100644 --- a/test-servers/src/resolve-config.ts +++ b/test-servers/src/resolve-config.ts @@ -12,6 +12,7 @@ import type { } from "./composable-test-server.js"; import { createTestServerInfo } from "./test-server-fixtures.js"; import { resolvePreset } from "./preset-registry.js"; +import { createResourceUpdateBroadcast } from "./resource-update-broadcast.js"; import type { ConfigFile, PresetRef } from "./load-config.js"; function mergeRequiredScopes( @@ -88,6 +89,11 @@ export function resolveConfig(config: ConfigFile): ServerConfig { logging: config.logging, listChanged: config.listChanged, subscriptions: config.subscriptions, + // One hub per resolved config: every session is built from this object, + // so they all share it (#2964). + ...(config.broadcastResourceUpdates && { + resourceUpdateBroadcast: createResourceUpdateBroadcast(), + }), tasks: config.tasks, tasksExtension: config.tasksExtension, skills: config.skills, diff --git a/test-servers/src/resource-update-broadcast.ts b/test-servers/src/resource-update-broadcast.ts new file mode 100644 index 0000000000..74c2c2989e --- /dev/null +++ b/test-servers/src/resource-update-broadcast.ts @@ -0,0 +1,98 @@ +/** + * Resource-update broadcast — one resource update reaching every session (#2964). + * + * Each session of the composable HTTP server is its own `McpServer` with its + * own resource copies and its own subscription set, so the `update_resource` + * tool on its own changes and notifies **only the session that called it**. + * That is a faithful per-session server, and exactly wrong for a fan-out run: + * `--load-subscribe` against it reports every other subscriber as missed + * (`Delivered: 10 of 50`) because no other session was ever told. + * + * A broadcast is the cross-session half a real server has and the fixture did + * not: a hub shared by every session the server opens. Each session joins it + * once initialized and leaves when its connection closes; `publish` applies the new + * text to every member's copy of the resource and sends + * `notifications/resources/updated` to each member subscribed to it. It is + * opted into by `broadcastResourceUpdates: true` in a config file, which + * `resolveConfig` turns into one hub on the resolved `ServerConfig` — and + * since every session is built from that one config object, they all share it. + * + * The latest text per URI is kept, so a session opened after an update serves + * the updated content rather than the preset's original — a subscriber that + * re-reads on notification sees the same text whichever session it is. + */ + +import type { RegisteredResource } from "@modelcontextprotocol/server"; + +/** One session's view of the broadcast. */ +export interface ResourceUpdateMember { + /** Replace this session's copy of `uri`'s text; a no-op where it serves none. */ + apply(uri: string, text: string): void; + /** + * Send `notifications/resources/updated` for `uri` if this session is + * subscribed to it. Resolves `true` when a notification was sent. + */ + notify(uri: string): Promise; +} + +/** A hub shared by every session of one server. */ +export interface ResourceUpdateBroadcast { + /** Add a session; replays the latest update of each URI. Returns `leave`. */ + join(member: ResourceUpdateMember): () => void; + /** + * Apply `text` to every session's copy of `uri` and notify each subscribed + * session. A session whose send fails (its stream is gone) is skipped, not + * fatal — the others still receive theirs. Resolves to the count notified. + */ + publish(uri: string, text: string): Promise; + /** Sessions currently joined. */ + readonly size: number; +} + +/** Create a broadcast hub; share one per server across all its sessions. */ +export function createResourceUpdateBroadcast(): ResourceUpdateBroadcast { + const members = new Set(); + const latest = new Map(); + return { + join(member) { + for (const [uri, text] of latest) member.apply(uri, text); + members.add(member); + return () => { + members.delete(member); + }; + }, + async publish(uri, text) { + latest.set(uri, text); + const current = [...members]; + for (const member of current) member.apply(uri, text); + const sent = await Promise.all( + current.map((member) => member.notify(uri).catch(() => false)), + ); + return sent.filter(Boolean).length; + }, + get size() { + return members.size; + }, + }; +} + +/** + * Point a registered resource at new text, keeping its MIME type. Returns + * `false` when `uri` is not registered. Shared by `update_resource` (one + * session) and the broadcast (every session), so both update identically. + */ +export function setResourceText( + registered: Map, + uri: string, + text: string, +): boolean { + const resource = registered.get(uri); + if (!resource) { + return false; + } + const mimeType = resource.metadata?.mimeType || "text/plain"; + resource.update({ + callback: async () => ({ contents: [{ uri, mimeType, text }] }), + }); + return true; +} diff --git a/test-servers/src/test-server-fixtures.ts b/test-servers/src/test-server-fixtures.ts index 3fe73a35c2..579f39cd6e 100644 --- a/test-servers/src/test-server-fixtures.ts +++ b/test-servers/src/test-server-fixtures.ts @@ -35,6 +35,7 @@ import type { ShapeOutput, } from "./composable-test-server.js"; import { getTestServerControl } from "./test-server-control.js"; +import { setResourceText } from "./resource-update-broadcast.js"; /** Build a CallToolResult from a text message (and optional isError). */ function toToolResult(text: string, isError?: boolean): CallToolResult { @@ -2378,37 +2379,26 @@ export function createUpdateResourceTool(): ToolDefinition { throw new Error("Server context not available"); } - const { server, state } = context; + const { server, state, resourceUpdateBroadcast } = context; + const uri = params.uri as string; + const text = params.text as string; - // Find registered resource by URI - const resource = state.registeredResources.get(params.uri as string); - if (!resource) { - throw new Error(`Resource with URI ${params.uri} not found`); + if (!state.registeredResources.has(uri)) { + throw new Error(`Resource with URI ${uri} not found`); } - // Get the current resource metadata to preserve mimeType - const currentResource = state.registeredResources.get( - params.uri as string, - ); - const mimeType = currentResource?.metadata?.mimeType || "text/plain"; + // Broadcasting (#2964): every session's copy changes and every + // subscribed session is notified, not just this one. + if (resourceUpdateBroadcast) { + const notified = await resourceUpdateBroadcast.publish(uri, text); + return toToolResult( + `Resource ${uri} updated; notified ${notified} subscribed session(s)`, + ); + } - // Update the resource's callback to return new content - resource.update({ - callback: async () => { - return { - contents: [ - { - uri: params.uri as string, - mimeType, - text: params.text as string, - }, - ], - }; - }, - }); + setResourceText(state.registeredResources, uri, text); // Send resource updated notification only if subscribed - const uri = params.uri as string; if (state.resourceSubscriptions.has(uri)) { await server.server.sendResourceUpdated({ uri,