diff --git a/.changeset/forwarded-notification-params.md b/.changeset/forwarded-notification-params.md new file mode 100644 index 0000000000..9ead2b1951 --- /dev/null +++ b/.changeset/forwarded-notification-params.md @@ -0,0 +1,5 @@ +--- +'@modelcontextprotocol/server': minor +--- + +`ServerEvent` takes optional `params`, which the `subscriptions/listen` router delivers with its own subscription ID, so a gateway can forward change notifications with their `_meta` and extension params. diff --git a/.changeset/per-request-transport-revision.md b/.changeset/per-request-transport-revision.md new file mode 100644 index 0000000000..291c9a9e1d --- /dev/null +++ b/.changeset/per-request-transport-revision.md @@ -0,0 +1,5 @@ +--- +'@modelcontextprotocol/server': patch +--- + +Connecting a `Server` or `McpServer` to a `PerRequestHTTPServerTransport` classified at a modern revision that `supportedProtocolVersions` lists and the SDK implements now binds that revision, instead of answering `-32022`. diff --git a/docs/advanced/low-level-server.md b/docs/advanced/low-level-server.md index eb04933fee..b815379ec9 100644 --- a/docs/advanced/low-level-server.md +++ b/docs/advanced/low-level-server.md @@ -120,6 +120,8 @@ createMcpHandler(buildServer); Every serving recipe — [stdio](../serving/stdio.md), [HTTP](../serving/http.md) — applies to this server unchanged. +For a custom HTTP entry built from `classifyInboundRequest` and `PerRequestHTTPServerTransport`, list the modern revision in the server's `supportedProtocolVersions`: `server.connect(transport)` then binds the instance to the revision the transport was classified at, and a modern revision the server does not list, or the SDK does not implement, is answered with `-32022`. Read the client's identity from `ctx.mcpReq.envelope`; the deprecated `getClientVersion()` and `getClientCapabilities()` stay empty on such an instance. Use a fresh instance for each per-request exchange, as the built-in HTTP entry does. + ## 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. diff --git a/docs/migration/support-2026-07-28.md b/docs/migration/support-2026-07-28.md index 449fa6a177..714564f803 100644 --- a/docs/migration/support-2026-07-28.md +++ b/docs/migration/support-2026-07-28.md @@ -12,7 +12,8 @@ wire-only members directly. If you are on `@modelcontextprotocol/sdk` (v1.x), st Nothing in v2 puts a 2026-07-28 byte on the wire by default: a hand-constructed `Client` / `Server` / `McpServer` keeps speaking the 2025-era protocol it was written for. Serving or speaking 2026-07-28 is always an explicit opt-in via one of the entries -below. +below, or by listing the revision on a hand-built server connected to a +`PerRequestHTTPServerTransport` (see [the low-level server guide](../advanced/low-level-server.md)). ## Contents diff --git a/docs/servers/notifications.md b/docs/servers/notifications.md index 03bada8ef5..0d9fc16c4d 100644 --- a/docs/servers/notifications.md +++ b/docs/servers/notifications.md @@ -102,6 +102,8 @@ const shared = createMcpHandler(() => buildJobsServer(), { bus }); The in-memory bus never leaves the process, so one process needs nothing more. Run more than one and you implement the two-method `ServerEventBus` interface — `publish` and `subscribe` — over your own pub/sub backend, then pass the same instance to every handler; see [Sessions, state, and scaling](../serving/sessions-state-scaling.md). +When forwarding another server's notifications, include their params on the event: `bus.publish({ kind: 'tools_list_changed', params: notification.params })`. The router preserves custom `_meta` and extension params while replacing the subscription ID with the receiving stream's ID. For `resource_updated`, the event's `uri` controls both filtering and the delivered URI. + ## Recap - `sendToolListChanged()`, `sendPromptListChanged()`, and `sendResourceListChanged()` push a list-changed notification to connected clients. diff --git a/packages/server/src/server/perRequestTransport.ts b/packages/server/src/server/perRequestTransport.ts index 5c17f9c455..971697d2e1 100644 --- a/packages/server/src/server/perRequestTransport.ts +++ b/packages/server/src/server/perRequestTransport.ts @@ -114,12 +114,21 @@ interface SseSink { keepAliveTimer?: ReturnType; } +let readClassification: (transport: PerRequestHTTPServerTransport) => MessageClassification; + /** * The per-request micro-transport: a real, connected `Transport` whose whole * lifetime is one HTTP exchange. See the module documentation for the * response shapes it produces. + * + * Connecting a `Server` binds the instance to the classified modern revision + * when its `supportedProtocolVersions` lists it and the SDK implements it. */ export class PerRequestHTTPServerTransport implements Transport { + static { + readClassification = transport => transport._classification; + } + onclose?: () => void; onerror?: (error: Error) => void; onmessage?: (message: T, extra?: MessageExtraInfo) => void; @@ -439,3 +448,8 @@ export class PerRequestHTTPServerTransport implements Transport { } } } + +/** Package-internal: the classification a per-request transport serves; `undefined` for any other transport. Not public API. */ +export function perRequestClassificationOf(transport: Transport): MessageClassification | undefined { + return transport instanceof PerRequestHTTPServerTransport ? readClassification(transport) : undefined; +} diff --git a/packages/server/src/server/server.ts b/packages/server/src/server/server.ts index 032fdc9c7f..441857f646 100644 --- a/packages/server/src/server/server.ts +++ b/packages/server/src/server/server.ts @@ -60,11 +60,13 @@ import { ProtocolErrorCode, SdkError, SdkErrorCode, + SUPPORTED_MODERN_PROTOCOL_VERSIONS, withRequestStateValue } from '@modelcontextprotocol/core-internal'; import { DefaultJsonSchemaValidator } from '@modelcontextprotocol/server/_shims'; import { coerceEmbeddedInputRequest, LegacyInputRequiredShim, resolveLegacyShimOptions } from './legacyInputRequiredShim'; +import { perRequestClassificationOf } from './perRequestTransport'; /** * The request methods whose 2026-07-28 result vocabulary includes @@ -290,6 +292,7 @@ export class Server extends Protocol { private _requestStateVerify?: (state: string, ctx: ServerContext) => unknown | Promise; private _inputRequiredServing: { maxRounds: number; roundTimeoutMs: number; legacyShim: boolean }; private _legacyShim?: LegacyInputRequiredShim; + private _boundFromTransport = false; /** Lazily-built legacy shim; the loop lives in legacyInputRequiredShim.ts behind a narrow host contract. */ private _legacyInputRequiredShim(): LegacyInputRequiredShim { @@ -340,9 +343,7 @@ export class Server extends Protocol { // server/discover is installed only when the supported-versions list // carries a modern revision: a legacy-only server keeps answering -32601. - // A hand-constructed instance is never era-bound, so the handler stays - // unreachable behind the era gate until a serving entry (createMcpHandler, - // serveStdio) marks the instance as serving the 2026-07-28 era. + // The handler becomes reachable when a serving entry or connect() binds a modern revision. if (modernProtocolVersions(this._supportedProtocolVersions).length > 0) { this.setRequestHandler('server/discover', () => this._ondiscover()); } @@ -360,6 +361,22 @@ export class Server extends Protocol { 'Use a separate Server instance per connection: this instance is already connected to a transport. To connect it to a new transport, call close() first.' ); } + // A new connection never inherits a binding taken from an earlier per-request transport. + if (this._boundFromTransport) { + this._boundFromTransport = false; + this._negotiatedProtocolVersion = undefined; + } + // A per-request transport classified at a modern revision this instance lists and the SDK implements binds that revision. + const revision = perRequestClassificationOf(transport)?.revision; + if ( + revision !== undefined && + !this._servedModernEra() && + SUPPORTED_MODERN_PROTOCOL_VERSIONS.includes(revision) && + modernProtocolVersions(this._supportedProtocolVersions).includes(revision) + ) { + this._negotiatedProtocolVersion = revision; + this._boundFromTransport = true; + } await super.connect(transport); } @@ -966,8 +983,8 @@ export class Server extends Protocol { * @deprecated Read client identity from the per-request handler context instead: on * 2026-07-28 (per-request envelope) requests `ctx.mcpReq.envelope` carries the client's * declared capabilities, while on 2025-era connections this accessor keeps returning the - * `initialize`-scoped value. The accessor remains functional — instances serving the - * 2026-07-28 era are backfilled per request from the validated envelope. + * `initialize`-scoped value. The accessor remains functional where `createMcpHandler` + * serves the 2026-07-28 era: it backfills it per request from the validated envelope. */ getClientCapabilities(): ClientCapabilities | undefined { return this._clientCapabilities; @@ -979,8 +996,8 @@ export class Server extends Protocol { * @deprecated Read client identity from the per-request handler context instead: on * 2026-07-28 (per-request envelope) requests `ctx.mcpReq.envelope` carries the client's * name and version, while on 2025-era connections this accessor keeps returning the - * `initialize`-scoped value. The accessor remains functional — instances serving the - * 2026-07-28 era are backfilled per request from the validated envelope. + * `initialize`-scoped value. The accessor remains functional where `createMcpHandler` + * serves the 2026-07-28 era: it backfills it per request from the validated envelope. */ getClientVersion(): Implementation | undefined { return this._clientVersion; diff --git a/packages/server/src/server/serverEventBus.ts b/packages/server/src/server/serverEventBus.ts index 3f97f62309..580cd8bc86 100644 --- a/packages/server/src/server/serverEventBus.ts +++ b/packages/server/src/server/serverEventBus.ts @@ -1,4 +1,4 @@ -import type { ServerCapabilities, SubscriptionFilter } from '@modelcontextprotocol/core-internal'; +import type { JSONRPCNotification, ServerCapabilities, SubscriptionFilter } from '@modelcontextprotocol/core-internal'; /** * A change event a server publishes for delivery on open `subscriptions/listen` @@ -9,14 +9,18 @@ import type { ServerCapabilities, SubscriptionFilter } from '@modelcontextprotoc * - `resources_list_changed` → `notifications/resources/list_changed` * - `resource_updated` → `notifications/resources/updated` (carries the URI) * - * The bus carries the EVENT, not the wire shape — the entry's listen router - * owns subscription-id stamping and per-stream filtering. + * The bus carries the event and any forwarded notification params — the + * entry's listen router owns subscription-id stamping and per-stream filtering. */ -export type ServerEvent = +export type ServerEvent = ( | { kind: 'tools_list_changed' } | { kind: 'prompts_list_changed' } | { kind: 'resources_list_changed' } - | { kind: 'resource_updated'; uri: string }; + | { kind: 'resource_updated'; uri: string } +) & { + /** Forwarded notification params; the router stamps its own subscription ID and the event's resource URI. */ + params?: JSONRPCNotification['params']; +}; /** * The server-side change-event seam for `subscriptions/listen`. @@ -176,19 +180,19 @@ export function honoredSubset(requested: SubscriptionFilter, capabilities?: Serv } /** Map a {@linkcode ServerEvent} onto its wire notification `{method, params}`. */ -export function serverEventToNotification(event: ServerEvent): { method: string; params?: { uri: string } } { +export function serverEventToNotification(event: ServerEvent): { method: string; params?: JSONRPCNotification['params'] } { switch (event.kind) { case 'tools_list_changed': { - return { method: 'notifications/tools/list_changed' }; + return { method: 'notifications/tools/list_changed', ...(event.params !== undefined && { params: event.params }) }; } case 'prompts_list_changed': { - return { method: 'notifications/prompts/list_changed' }; + return { method: 'notifications/prompts/list_changed', ...(event.params !== undefined && { params: event.params }) }; } case 'resources_list_changed': { - return { method: 'notifications/resources/list_changed' }; + return { method: 'notifications/resources/list_changed', ...(event.params !== undefined && { params: event.params }) }; } case 'resource_updated': { - return { method: 'notifications/resources/updated', params: { uri: event.uri } }; + return { method: 'notifications/resources/updated', params: { ...event.params, uri: event.uri } }; } } } diff --git a/test/integration/test/server/forwardNotifications.test.ts b/test/integration/test/server/forwardNotifications.test.ts new file mode 100644 index 0000000000..ca46e68468 --- /dev/null +++ b/test/integration/test/server/forwardNotifications.test.ts @@ -0,0 +1,237 @@ +import { createServer } from 'node:http'; + +import { Client, InMemoryTransport } from '@modelcontextprotocol/client'; +import { toNodeHandler } from '@modelcontextprotocol/node'; +import type { ServerEvent } from '@modelcontextprotocol/server'; +import { + classifyInboundRequest, + CLIENT_CAPABILITIES_META_KEY, + CLIENT_INFO_META_KEY, + createMcpHandler, + InMemoryServerEventBus, + McpServer, + PerRequestHTTPServerTransport, + PROTOCOL_VERSION_META_KEY, + Server, + SUBSCRIPTION_ID_META_KEY +} from '@modelcontextprotocol/server'; +import { listenOnRandomPort } from '@modelcontextprotocol/test-helpers'; +import { describe, expect, it } from 'vitest'; + +const revision = '2026-07-28'; +const envelope = { + [PROTOCOL_VERSION_META_KEY]: revision, + [CLIENT_INFO_META_KEY]: { name: 'forwarding-client', version: '1' }, + [CLIENT_CAPABILITIES_META_KEY]: {} +}; + +describe('public gateway serving APIs', () => { + it.each([ + ['tools_list_changed', 'notifications/tools/list_changed'], + ['prompts_list_changed', 'notifications/prompts/list_changed'], + ['resources_list_changed', 'notifications/resources/list_changed'], + ['resource_updated', 'notifications/resources/updated'] + ] as const)('preserves forwarded %s params on the subscription stream', async (kind, method) => { + const bus = new InMemoryServerEventBus(); + const handler = createMcpHandler( + () => + new Server( + { name: 'gateway', version: '1' }, + { + capabilities: { + tools: { listChanged: true }, + prompts: { listChanged: true }, + resources: { listChanged: true, subscribe: true } + } + } + ), + { bus, keepAliveMs: 0 } + ); + const http = createServer(toNodeHandler(handler)); + const url = await listenOnRandomPort(http); + const abort = new AbortController(); + try { + const response = await fetch(url, { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + Accept: 'application/json, text/event-stream', + 'MCP-Protocol-Version': revision, + 'Mcp-Method': 'subscriptions/listen' + }, + signal: abort.signal, + body: JSON.stringify({ + jsonrpc: '2.0', + id: 'downstream', + method: 'subscriptions/listen', + params: { + _meta: envelope, + notifications: { + toolsListChanged: true, + promptsListChanged: true, + resourcesListChanged: true, + resourceSubscriptions: ['file:///config'] + } + } + }) + }); + expect(response.status).toBe(200); + const reader = response.body!.getReader(); + const decoder = new TextDecoder(); + let buffer = ''; + const next = async () => { + while (!buffer.includes('\n\n')) { + const chunk = await reader.read(); + if (chunk.done) throw new Error('Subscription ended before notification'); + buffer += decoder.decode(chunk.value, { stream: true }); + } + const end = buffer.indexOf('\n\n'); + const frame = buffer.slice(0, end); + buffer = buffer.slice(end + 2); + return JSON.parse( + frame + .split('\n') + .find(line => line.startsWith('data: '))! + .slice(6) + ); + }; + const acknowledgement = await next(); + expect(acknowledgement.method).toBe('notifications/subscriptions/acknowledged'); + const params = { + _meta: { fixture: 'kept', nested: { value: 7 }, [SUBSCRIPTION_ID_META_KEY]: 'upstream' }, + extension: { changed: ['one', 'two'] }, + ...(kind === 'resource_updated' ? { uri: 'file:///upstream' } : {}) + }; + const event: ServerEvent = kind === 'resource_updated' ? { kind, uri: 'file:///config', params } : { kind, params }; + bus.publish(event); + expect(await next()).toEqual({ + jsonrpc: '2.0', + method, + params: { + ...params, + ...(kind === 'resource_updated' ? { uri: 'file:///config' } : {}), + _meta: { ...params._meta, [SUBSCRIPTION_ID_META_KEY]: 'downstream' } + } + }); + expect(params._meta[SUBSCRIPTION_ID_META_KEY]).toBe('upstream'); + await reader.cancel(); + } finally { + abort.abort(); + await handler.close(); + await new Promise((resolve, reject) => http.close(error => (error ? reject(error) : resolve()))); + } + }); + + it.each(['Server', 'McpServer'] as const)('admits a classified modern request on a manually connected %s', async kind => { + const message = { jsonrpc: '2.0' as const, id: 1, method: 'tools/list', params: { _meta: envelope } }; + const route = classifyInboundRequest({ + httpMethod: 'POST', + protocolVersionHeader: revision, + mcpMethodHeader: message.method, + body: message + }); + if (route.kind !== 'modern') throw new Error('Expected modern request'); + const server = + kind === 'Server' + ? new Server( + { name: 'gateway', version: '1' }, + { + supportedProtocolVersions: [revision], + capabilities: { tools: {} } + } + ) + : new McpServer( + { name: 'gateway', version: '1' }, + { + supportedProtocolVersions: [revision], + capabilities: { tools: {} } + } + ); + let called = false; + const protocol = server instanceof McpServer ? server.server : server; + protocol.setRequestHandler('tools/list', () => { + called = true; + return { tools: [] }; + }); + const transport = new PerRequestHTTPServerTransport({ classification: route.classification }); + try { + await server.connect(transport); + const response = await transport.handleMessage(route.message); + expect(response.status).toBe(200); + expect(called).toBe(true); + expect(await response.json()).toMatchObject({ id: 1, result: { tools: [] } }); + } finally { + await server.close(); + } + }); + + it.each([ + ['2030-01-01', ['2025-11-25']], + [revision, ['2025-11-25']], + ['2030-01-01', [revision, '2030-01-01']] + ])('answers a %s classification with -32022 when the server lists %j', async (classified, supportedProtocolVersions) => { + const server = new Server({ name: 'gateway', version: '1' }, { supportedProtocolVersions, capabilities: { tools: {} } }); + server.setRequestHandler('tools/list', () => ({ tools: [] })); + server.onerror = () => {}; + const transport = new PerRequestHTTPServerTransport({ classification: { era: 'modern', revision: classified } }); + try { + await server.connect(transport); + expect(server.getNegotiatedProtocolVersion()).toBeUndefined(); + const response = await transport.handleMessage({ jsonrpc: '2.0', id: 1, method: 'tools/list', params: { _meta: envelope } }); + expect(response.status).toBe(400); + expect(await response.json()).toMatchObject({ id: 1, error: { code: -32_022, data: { requested: classified } } }); + } finally { + await server.close(); + } + }); + + it('allows a legacy connection after closing a transport-bound modern connection', async () => { + const server = new Server( + { name: 'gateway', version: '1' }, + { + supportedProtocolVersions: [revision, '2025-11-25'] + } + ); + const modern = new PerRequestHTTPServerTransport({ classification: { era: 'modern', revision } }); + let versionInOnclose: string | undefined; + server.onclose = () => { + versionInOnclose = server.getNegotiatedProtocolVersion(); + }; + await server.connect(modern); + expect(server.getNegotiatedProtocolVersion()).toBe(revision); + await server.close(); + expect(versionInOnclose).toBe(revision); + server.onclose = undefined; + const [clientTransport, serverTransport] = InMemoryTransport.createLinkedPair(); + const client = new Client({ name: 'legacy-client', version: '1' }); + try { + await server.connect(serverTransport); + await client.connect(clientTransport); + expect(server.getNegotiatedProtocolVersion()).toBe('2025-11-25'); + } finally { + await client.close(); + await server.close(); + } + }); + + it('allows a legacy connection made from onclose of a transport-bound modern connection', async () => { + const server = new Server({ name: 'gateway', version: '1' }, { supportedProtocolVersions: [revision, '2025-11-25'] }); + await server.connect(new PerRequestHTTPServerTransport({ classification: { era: 'modern', revision } })); + const [clientTransport, serverTransport] = InMemoryTransport.createLinkedPair(); + let reconnected: Promise | undefined; + server.onclose = () => { + server.onclose = undefined; + reconnected = server.connect(serverTransport); + }; + await server.close(); + await reconnected; + const client = new Client({ name: 'legacy-client', version: '1' }); + try { + await client.connect(clientTransport); + expect(server.getNegotiatedProtocolVersion()).toBe('2025-11-25'); + } finally { + await client.close(); + await server.close(); + } + }); +});