Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions .claude/skills/test-servers/SKILL.md
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
8 changes: 8 additions & 0 deletions clients/cli/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
197 changes: 197 additions & 0 deletions clients/cli/__tests__/subscriptions-broadcast.test.ts
Original file line number Diff line number Diff line change
@@ -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);
});
});
23 changes: 23 additions & 0 deletions docs/test-servers.md
Original file line number Diff line number Diff line change
Expand Up @@ -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) |
Expand Down Expand Up @@ -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**.
Expand Down
14 changes: 14 additions & 0 deletions test-servers/configs/subscriptions-broadcast-http.json
Original file line number Diff line number Diff line change
@@ -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
}
}
68 changes: 68 additions & 0 deletions test-servers/src/composable-test-server.ts
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,10 @@ import {
isCompletable,
isInputRequiredResult,
} from "@modelcontextprotocol/server";
import {
setResourceText,
type ResourceUpdateBroadcast,
} from "./resource-update-broadcast.js";
import type {
Implementation,
Tool,
Expand Down Expand Up @@ -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;
}

// ---------------------------------------------------------------------------
Expand Down Expand Up @@ -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;
}

/**
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down
1 change: 1 addition & 0 deletions test-servers/src/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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";
6 changes: 6 additions & 0 deletions test-servers/src/load-config.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
Loading
Loading