feat(server): subscriptions/listen — entry-handled router and ServerEventBus (#2321)

Author: felixweinberger

.changeset/subscriptions-listen-server.md

@@ -0,0 +1,17 @@
+---
+'@modelcontextprotocol/core': minor
+'@modelcontextprotocol/server': minor
+---
+
+`subscriptions/listen` (SEP-1865) is served by both serving entries on protocol revision 2026-07-28. The entry owns ack-first, per-stream filtering, subscription-id stamping, keepalive (HTTP), the pre-ack `-32603` capacity guard, and teardown (HTTP stream close; one
+`notifications/cancelled` per subscription on stdio). `server/discover` now advertises `listChanged`/`subscribe` capability bits — the rider that suppressed them until listen was served is discharged.
+
+Under `createMcpHandler` the consumer's factory **is** constructed for `subscriptions/listen` (a capabilities-only probe so the acknowledged filter reflects what the server advertises; the instance is never connected and is closed immediately). Per-request authorization performed inside the factory therefore sees listen requests; token verification still belongs at the middleware layer mounted in front of the entry.
+
+New public surface:
+
+- `@modelcontextprotocol/server`: `ServerEventBus`, `ServerEvent`, `ServerNotifier` (types); `InMemoryServerEventBus` (class).
+- `McpHttpHandler` gains `.notify` (`ServerNotifier`: `toolsChanged()`, `promptsChanged()`, `resourcesChanged()`, `resourceUpdated(uri)`) and `.bus` (the `ServerEventBus` listen streams subscribe to).
+- `CreateMcpHandlerOptions` gains `bus?: ServerEventBus` (an in-process `InMemoryServerEventBus` is created when omitted), `maxSubscriptions?: number` (default 1024), and `keepAliveMs?: number` (default 15000).
+- `ServeStdioOptions` gains `maxSubscriptions?: number` (default 1024). On a modern-pinned connection `serveStdio` routes the pinned instance's existing `send*ListChanged()` calls onto active subscriptions; legacy connections are unchanged.
+- `@modelcontextprotocol/core`: `SUBSCRIPTION_ID_META_KEY` (const); `SubscriptionFilter`, `SubscriptionsListenRequest`, `SubscriptionsListenRequestParams`, `SubscriptionsAcknowledgedNotification`, `SubscriptionsAcknowledgedNotificationParams` (types).

packages/codemod/src/generated/specSchemaMap.ts

@@ -133,6 +133,11 @@ export const SPEC_SCHEMA_NAMES: ReadonlySet<string> = new Set([
     'StringSchemaSchema',
     'SubscribeRequestParamsSchema',
     'SubscribeRequestSchema',
+    'SubscriptionFilterSchema',
+    'SubscriptionsAcknowledgedNotificationParamsSchema',
+    'SubscriptionsAcknowledgedNotificationSchema',
+    'SubscriptionsListenRequestParamsSchema',
+    'SubscriptionsListenRequestSchema',
     'TaskAugmentedRequestParamsSchema',
     'TaskMetadataSchema',
     'TextContentSchema',

packages/core/src/exports/public/index.ts

@@ -85,6 +85,7 @@ export {
     PARSE_ERROR,
     PROTOCOL_VERSION_META_KEY,
     RELATED_TASK_META_KEY,
+    SUBSCRIPTION_ID_META_KEY,
     SUPPORTED_PROTOCOL_VERSIONS,
     TRACEPARENT_META_KEY,
     TRACESTATE_META_KEY

packages/core/src/shared/inboundClassification.ts

@@ -67,7 +67,7 @@ import { PROTOCOL_VERSION_META_KEY } from '../types/constants.js';
 import { ProtocolErrorCode } from '../types/enums.js';
 import { ProtocolError, UnsupportedProtocolVersionError } from '../types/errors.js';
 import { isJSONRPCErrorResponse, isJSONRPCNotification, isJSONRPCRequest, isJSONRPCResultResponse } from '../types/guards.js';
-import type { MessageClassification } from '../types/types.js';
+import type { JSONRPCNotification, JSONRPCRequest, MessageClassification } from '../types/types.js';
 import { envelopeClaimVersion, hasEnvelopeClaim, requestMetaOf, validateEnvelopeMeta } from './envelope.js';
 import { isModernProtocolVersion } from './protocolEras.js';
 
@@ -126,17 +126,32 @@ export interface InboundLegacyRoute {
     requestedVersion?: string;
 }
 
-/** The request claims the per-request envelope mechanism and is served on the modern path. */
-export interface InboundModernRoute {
-    kind: 'modern';
-    /** Whether the classified message is a request or a notification. */
-    messageKind: 'request' | 'notification';
-    /**
-     * The classification handed to the per-request transport and validated by
-     * the protocol layer against the serving instance's negotiated era.
-     */
-    classification: MessageClassification;
-}
+/**
+ * The request claims the per-request envelope mechanism and is served on the
+ * modern path. Discriminated by `messageKind` so the typed `message` narrows
+ * with it — the classifier has already proved the JSON-RPC shape via the
+ * `isJSONRPCRequest` / `isJSONRPCNotification` guards, so consumers never
+ * cast the body again.
+ */
+export type InboundModernRoute =
+    | {
+          kind: 'modern';
+          messageKind: 'request';
+          /** The classified body — guard-proved {@linkcode JSONRPCRequest} shape. */
+          message: JSONRPCRequest;
+          /**
+           * The classification handed to the per-request transport and validated by
+           * the protocol layer against the serving instance's negotiated era.
+           */
+          classification: MessageClassification;
+      }
+    | {
+          kind: 'modern';
+          messageKind: 'notification';
+          /** The classified body — guard-proved {@linkcode JSONRPCNotification} shape. */
+          message: JSONRPCNotification;
+          classification: MessageClassification;
+      };
 
 /** The named steps of the inbound validation ladder, in evaluation order. */
 export type InboundValidationRung =
@@ -461,9 +476,9 @@ function classifyBatch(body: readonly unknown[]): InboundClassificationOutcome {
     return { kind: 'legacy', reason: 'batch' };
 }
 
-function classifyRequestBody(request: InboundHttpRequest, body: Record<string, unknown>): InboundClassificationOutcome {
-    const params = body['params'];
-    const method = body['method'] as string;
+function classifyRequestBody(request: InboundHttpRequest, body: JSONRPCRequest): InboundClassificationOutcome {
+    const params = body.params;
+    const method = body.method;
     const headerVersion = request.protocolVersionHeader;
     const headerNamesModern = headerVersion !== undefined && isModernProtocolVersion(headerVersion);
 
@@ -523,7 +538,7 @@ function classifyRequestBody(request: InboundHttpRequest, body: Record<string, u
                 `the body names method ${method} but the Mcp-Method header names ${request.mcpMethodHeader}`
             );
         }
-        return { kind: 'modern', messageKind: 'request', classification: classificationForClaim(claimedVersion) };
+        return { kind: 'modern', messageKind: 'request', message: body, classification: classificationForClaim(claimedVersion) };
     }
 
     // No claim: legacy-era traffic — unless the protocol-version header names a
@@ -554,9 +569,9 @@ function classifyRequestBody(request: InboundHttpRequest, body: Record<string, u
     return { kind: 'legacy', reason: 'no-claim', ...(headerVersion !== undefined && { requestedVersion: headerVersion }) };
 }
 
-function classifyNotificationBody(request: InboundHttpRequest, body: Record<string, unknown>): InboundClassificationOutcome {
-    const params = body['params'];
-    const method = body['method'] as string;
+function classifyNotificationBody(request: InboundHttpRequest, body: JSONRPCNotification): InboundClassificationOutcome {
+    const params = body.params;
+    const method = body.method;
     const headerVersion = request.protocolVersionHeader;
     const headerNamesModern = headerVersion !== undefined && isModernProtocolVersion(headerVersion);
 
@@ -603,7 +618,7 @@ function classifyNotificationBody(request: InboundHttpRequest, body: Record<stri
                 `the notification body names method ${method} but the Mcp-Method header names ${request.mcpMethodHeader}`
             );
         }
-        return { kind: 'modern', messageKind: 'notification', classification };
+        return { kind: 'modern', messageKind: 'notification', message: body, classification };
     }
 
     // Notifications carry no body claim under the current spec, so the
@@ -623,6 +638,7 @@ function classifyNotificationBody(request: InboundHttpRequest, body: Record<stri
         return {
             kind: 'modern',
             messageKind: 'notification',
+            message: body,
             classification: { era: 'modern', revision: headerVersion }
         };
     }

packages/core/src/types/constants.ts

@@ -31,6 +31,19 @@ export const CLIENT_INFO_META_KEY = 'io.modelcontextprotocol/clientInfo';
  */
 export const CLIENT_CAPABILITIES_META_KEY = 'io.modelcontextprotocol/clientCapabilities';
 
+/**
+ * `_meta` key carrying the JSON-RPC ID of the `subscriptions/listen` request
+ * that opened the stream a notification was delivered on.
+ *
+ * Stamped by the server on every notification delivered via a
+ * `subscriptions/listen` stream (including the leading
+ * `notifications/subscriptions/acknowledged`); on stdio, where all messages
+ * share one channel, clients use it to correlate notifications with their
+ * originating subscription. The value is the listen request's JSON-RPC ID
+ * verbatim.
+ */
+export const SUBSCRIPTION_ID_META_KEY = 'io.modelcontextprotocol/subscriptionId';
+
 /**
  * `_meta` key carrying the desired log level for a request.
  *

packages/core/src/types/schemas.ts

@@ -898,6 +898,67 @@ export const UnsubscribeRequestSchema = RequestSchema.extend({
     params: UnsubscribeRequestParamsSchema
 });
 
+/* Subscriptions (protocol revision 2026-07-28) */
+/**
+ * The set of notification types a client opts in to on a `subscriptions/listen`
+ * request. Each type is opt-in; the server MUST NOT send a notification type
+ * the client has not explicitly requested here.
+ */
+export const SubscriptionFilterSchema = z.object({
+    /**
+     * If true, receive `notifications/tools/list_changed`.
+     */
+    toolsListChanged: z.boolean().optional(),
+    /**
+     * If true, receive `notifications/prompts/list_changed`.
+     */
+    promptsListChanged: z.boolean().optional(),
+    /**
+     * If true, receive `notifications/resources/list_changed`.
+     */
+    resourcesListChanged: z.boolean().optional(),
+    /**
+     * Subscribe to `notifications/resources/updated` for these resource URIs.
+     * Replaces the former `resources/subscribe` RPC on the 2026-07-28 revision.
+     */
+    resourceSubscriptions: z.array(z.string()).optional()
+});
+
+export const SubscriptionsListenRequestParamsSchema = BaseRequestParamsSchema.extend({
+    /**
+     * The notifications the client opts in to on this stream. The server MUST
+     * NOT send notification types the client has not explicitly requested.
+     */
+    notifications: SubscriptionFilterSchema
+});
+
+/**
+ * Sent from the client to open a long-lived channel for receiving notifications
+ * outside the context of a specific request (protocol revision 2026-07-28).
+ * Replaces the previous HTTP GET endpoint and `resources/subscribe`.
+ */
+export const SubscriptionsListenRequestSchema = RequestSchema.extend({
+    method: z.literal('subscriptions/listen'),
+    params: SubscriptionsListenRequestParamsSchema
+});
+
+export const SubscriptionsAcknowledgedNotificationParamsSchema = NotificationsParamsSchema.extend({
+    /**
+     * The subset of requested notification types the server agreed to honor.
+     */
+    notifications: SubscriptionFilterSchema
+});
+
+/**
+ * Sent by the server as the first message on a `subscriptions/listen` stream
+ * to acknowledge that the subscription has been established and report which
+ * notification types it agreed to honor (protocol revision 2026-07-28).
+ */
+export const SubscriptionsAcknowledgedNotificationSchema = NotificationSchema.extend({
+    method: z.literal('notifications/subscriptions/acknowledged'),
+    params: SubscriptionsAcknowledgedNotificationParamsSchema
+});
+
 /**
  * Parameters for a {@linkcode ResourceUpdatedNotification | notifications/resources/updated} notification.
  */
@@ -2025,6 +2086,7 @@ export const ClientRequestSchema = z.union([
     ReadResourceRequestSchema,
     SubscribeRequestSchema,
     UnsubscribeRequestSchema,
+    SubscriptionsListenRequestSchema,
     CallToolRequestSchema,
     ListToolsRequestSchema
 ]);
@@ -2055,6 +2117,7 @@ export const ServerNotificationSchema = z.union([
     ResourceListChangedNotificationSchema,
     ToolListChangedNotificationSchema,
     PromptListChangedNotificationSchema,
+    SubscriptionsAcknowledgedNotificationSchema,
     ElicitationCompleteNotificationSchema
 ]);
 

packages/core/src/types/specTypeSchema.ts

@@ -153,6 +153,11 @@ const SPEC_SCHEMA_KEYS = [
     'StringSchemaSchema',
     'SubscribeRequestSchema',
     'SubscribeRequestParamsSchema',
+    'SubscriptionFilterSchema',
+    'SubscriptionsAcknowledgedNotificationSchema',
+    'SubscriptionsAcknowledgedNotificationParamsSchema',
+    'SubscriptionsListenRequestSchema',
+    'SubscriptionsListenRequestParamsSchema',
     'TaskAugmentedRequestParamsSchema',
     'TaskMetadataSchema',
     'TextContentSchema',

packages/core/src/types/types.ts

@@ -146,6 +146,11 @@ import type {
     StringSchemaSchema,
     SubscribeRequestParamsSchema,
     SubscribeRequestSchema,
+    SubscriptionFilterSchema,
+    SubscriptionsAcknowledgedNotificationParamsSchema,
+    SubscriptionsAcknowledgedNotificationSchema,
+    SubscriptionsListenRequestParamsSchema,
+    SubscriptionsListenRequestSchema,
     TaskAugmentedRequestParamsSchema,
     TaskMetadataSchema,
     TextContentSchema,
@@ -352,6 +357,13 @@ export type UnsubscribeRequest = Infer<typeof UnsubscribeRequestSchema>;
 export type ResourceUpdatedNotificationParams = Infer<typeof ResourceUpdatedNotificationParamsSchema>;
 export type ResourceUpdatedNotification = Infer<typeof ResourceUpdatedNotificationSchema>;
 
+/* Subscriptions (protocol revision 2026-07-28) */
+export type SubscriptionFilter = Infer<typeof SubscriptionFilterSchema>;
+export type SubscriptionsListenRequestParams = Infer<typeof SubscriptionsListenRequestParamsSchema>;
+export type SubscriptionsListenRequest = Infer<typeof SubscriptionsListenRequestSchema>;
+export type SubscriptionsAcknowledgedNotificationParams = Infer<typeof SubscriptionsAcknowledgedNotificationParamsSchema>;
+export type SubscriptionsAcknowledgedNotification = Infer<typeof SubscriptionsAcknowledgedNotificationSchema>;
+
 /* Prompts */
 export type PromptArgument = Infer<typeof PromptArgumentSchema>;
 export type Prompt = Infer<typeof PromptSchema>;
@@ -551,6 +563,11 @@ export type ResultTypeMap = {
     'resources/read': ReadResourceResult;
     'resources/subscribe': EmptyResult;
     'resources/unsubscribe': EmptyResult;
+    // `subscriptions/listen` never receives a JSON-RPC result on the wire:
+    // termination is stream close (HTTP) or `notifications/cancelled` (stdio).
+    // The `EmptyResult` entry exists only to keep the mapped types total —
+    // see the serving entries' listen routers.
+    'subscriptions/listen': EmptyResult;
     'tools/call': CallToolResult;
     'tools/list': ListToolsResult;
     'sampling/createMessage': CreateMessageResult | CreateMessageResultWithTools;

packages/core/src/wire/rev2026-07-28/encodeContract.ts

@@ -41,9 +41,9 @@ export const DEFAULT_CACHE_SCOPE = 'private';
  * Request methods whose spec result vocabulary goes beyond `'complete'` on the
  * 2026-07-28 revision: their results may be `input_required` (multi
  * round-trip requests), so a handler-provided `resultType` passes through the
- * stamp untouched. `subscriptions/listen` joins this set when the
- * subscriptions feature is served (its terminal result uses the same
- * mechanism).
+ * stamp untouched. `subscriptions/listen` is NOT in this set: it never emits
+ * a JSON-RPC result — termination is stream close (HTTP) or
+ * `notifications/cancelled` (stdio) per the spec.
  */
 export const EXTENDED_RESULT_TYPE_METHODS: readonly string[] = ['tools/call', 'prompts/get', 'resources/read'];
 

packages/core/src/wire/rev2026-07-28/registry.ts

@@ -20,10 +20,10 @@
  *   (the ATK-D flavor-b trap); this hand registry excludes them by
  *   construction. Their in-band role lands with the MRTR driver (#13).
  * - `subscriptions/listen` + `notifications/subscriptions/acknowledged`
- *   (SEP-1865): 2026-only vocabulary whose SHELLS land with the
- *   subscriptions feature (#14). Until then they are absent here — inbound
- *   listen gets −32601 (capability not yet served), which is protocol-legal
- *   for a server that does not implement subscriptions.
+ *   (SEP-1865): 2026-only vocabulary, present here as registry shells.
+ *   Dispatch never reaches a registered handler — the serving entries
+ *   (`createMcpHandler`, `serveStdio`) recognize listen at the entry layer
+ *   and own ack/filter/stamp/teardown themselves.
  */
 import type * as z from 'zod/v4';