feat(server): per-session subscription cap with trusted resolver on createMcpHandler

.changeset/max-subscriptions-per-session.md

@@ -0,0 +1,7 @@
+---
+'@modelcontextprotocol/server': minor
+---
+
+Add `maxSubscriptionsPerSession` and `getSessionId` options to `createMcpHandler` for per-session `subscriptions/listen` fairness. Open listen streams are counted per session key resolved by `getSessionId(options, request)` and compared against the cap before the SSE ack; a refusal returns the same `-32603 "Subscription limit reached"` error as the existing handler-wide `maxSubscriptions` guard. The SDK does not authenticate the resolver's returned value. Deployments must derive the key from a trusted source (for example `authInfo.extra`), not from raw inbound headers.
+
+Fail-closed by construction: setting `maxSubscriptionsPerSession` without `getSessionId` throws `TypeError` at `createMcpHandler` time; a resolver returning `undefined` for a given listen refuses the stream rather than silently falling back to the handler-wide cap. Closes #3036.

packages/server/src/server/createMcpHandler.ts

@@ -197,6 +197,41 @@ export interface CreateMcpHandlerOptions {
      * @default 1024
      */
     maxSubscriptions?: number;
+    /**
+     * Reject a new `subscriptions/listen` with `-32603` 'Subscription limit
+     * reached' (in-band, HTTP 200, before the ack) when this many subscription
+     * streams are already open under the same resolved session key (see
+     * {@linkcode getSessionId}). Enforced alongside the handler-wide
+     * {@linkcode maxSubscriptions} cap; whichever fires first wins and both
+     * return the same error. Requires {@linkcode getSessionId} to be
+     * configured. {@linkcode createMcpHandler} throws `TypeError` at
+     * construction otherwise, since without a resolver the SDK has no safe
+     * source for per-session attribution.
+     * @default no per-session cap
+     */
+    maxSubscriptionsPerSession?: number;
+    /**
+     * Resolver for the trusted session key used by
+     * {@linkcode maxSubscriptionsPerSession} to attribute a
+     * `subscriptions/listen` stream to a session bucket. Required when
+     * `maxSubscriptionsPerSession` is configured, ignored otherwise.
+     *
+     * Receives the pass-through {@linkcode McpHandlerRequestOptions} (whose
+     * `authInfo` reflects what trusted middleware established) and the inbound
+     * `Request`; returns the session key when identity is established,
+     * `undefined` when it is not. The SDK does **not** authenticate the
+     * returned value: it is used only as a bucketing key for the per-session
+     * cap, so the deployment must derive it from a trusted source (for
+     * example `authInfo.extra`), never from an unvalidated inbound header
+     * forwarded as-is.
+     *
+     * **Fail-closed contract:** when this resolver returns `undefined` for a
+     * listen request, the stream is refused pre-ack with
+     * `-32603 'Subscription limit reached'` rather than silently falling back
+     * to the handler-wide cap. A misconfigured deployment therefore cannot
+     * accidentally appear to be enforcing isolation that was never applied.
+     */
+    getSessionId?: (options: McpHandlerRequestOptions, request: Request) => string | undefined;
     /**
      * SSE comment-frame keepalive interval for every SSE stream this handler
      * serves. In modern `auto` mode it starts after SSE upgrade. Set to `0` to disable.
@@ -693,6 +728,17 @@ export function createMcpHandler(factory: McpServerFactory, options: CreateMcpHa
         );
     }
 
+    // Fail-closed configuration guard: per-session attribution has no safe
+    // default source. If the cap is set without a resolver, there is no honest
+    // way to bucket a listen request. Reading `Mcp-Session-Id` off raw headers
+    // would claim isolation the SDK cannot vouch for. Reject at construction.
+    if (options.maxSubscriptionsPerSession !== undefined && options.getSessionId === undefined) {
+        throw new TypeError(
+            "The 'maxSubscriptionsPerSession' option requires a 'getSessionId' resolver to supply the trusted session key. " +
+                'Without a resolver the SDK has no safe source for per-session attribution. The entry does not authenticate inbound headers.'
+        );
+    }
+
     /** Modern per-request instances with an exchange still in flight (close() tears these down). */
     const inflight = new Set<Server>();
     let closed = false;
@@ -710,6 +756,7 @@ export function createMcpHandler(factory: McpServerFactory, options: CreateMcpHa
     const listenRouter = createListenRouter({
         bus,
         maxSubscriptions: options.maxSubscriptions ?? DEFAULT_MAX_SUBSCRIPTIONS,
+        ...(options.maxSubscriptionsPerSession !== undefined && { maxSubscriptionsPerSession: options.maxSubscriptionsPerSession }),
         keepAliveMs: options.keepAliveMs ?? DEFAULT_SSE_KEEP_ALIVE_MS,
         onerror: reportError
     });
@@ -726,7 +773,12 @@ export function createMcpHandler(factory: McpServerFactory, options: CreateMcpHa
     const legacyHandler: LegacyHttpHandler | undefined =
         legacy === 'reject' ? undefined : createLegacyStatelessFallback(factory, reportError, options.keepAliveMs, maxRequestBodySize);
 
-    async function serveModern(route: InboundModernRoute, request: Request, authInfo: AuthInfo | undefined): Promise<Response> {
+    async function serveModern(
+        route: InboundModernRoute,
+        request: Request,
+        requestOptions: McpHandlerRequestOptions | undefined,
+        authInfo: AuthInfo | undefined
+    ): Promise<Response> {
         const claimedRevision = route.classification.revision;
         if (claimedRevision === undefined || !SUPPORTED_MODERN_PROTOCOL_VERSIONS.includes(claimedRevision)) {
             // The claim names a revision this endpoint does not serve (an
@@ -811,7 +863,18 @@ export function createMcpHandler(factory: McpServerFactory, options: CreateMcpHa
             const capabilities = server.getCapabilities();
             const serverInfo = serverIdentityOf(server);
             void product.close().catch(reportError);
-            return listenRouter.serve(route.message, request.signal, capabilities, serverInfo);
+            // Resolve the per-session bucketing key through the caller-supplied
+            // getSessionId only when the per-session cap is actually configured.
+            // The resolver is documented as ignored otherwise, so skipping it
+            // avoids a user-supplied throw turning an otherwise valid listen
+            // into a 500. Deployments decide the trust source for identity
+            // (typically a field stamped onto authInfo by their auth
+            // middleware); the SDK never derives the key from request headers
+            // directly. Construction already rejected the "cap configured but
+            // resolver missing" misconfiguration.
+            const sessionId =
+                options.maxSubscriptionsPerSession === undefined ? undefined : options.getSessionId?.(requestOptions ?? {}, request);
+            return listenRouter.serve(route.message, request.signal, capabilities, serverInfo, sessionId);
         }
 
         // SEP-2243 `Mcp-Param-*` server-side validation (pre-dispatch ladder
@@ -984,7 +1047,7 @@ export function createMcpHandler(factory: McpServerFactory, options: CreateMcpHa
                     return rejectionResponse(outcome, echoableRequestId(body));
                 }
                 case 'modern': {
-                    return await serveModern(outcome, request, authInfo);
+                    return await serveModern(outcome, request, requestOptions, authInfo);
                 }
                 case 'legacy': {
                     return await serveLegacyRoute(outcome, forwardRequest, authInfo, parsedBody);

packages/server/src/server/listenRouter.ts

@@ -45,6 +45,22 @@ export interface ListenRouterOptions {
     bus: ServerEventBus;
     /** Reject a new listen with `-32603` when this many subscriptions are already open (default 1024). */
     maxSubscriptions?: number;
+    /**
+     * Reject a new listen with `-32603` when this many subscriptions are already
+     * open under the same resolved session key. The caller (the HTTP entry) is
+     * responsible for resolving the session key through a trusted path and
+     * passing it to {@linkcode serve}; this router does not read HTTP headers.
+     *
+     * Fail-closed: when `maxSubscriptionsPerSession` is set and the caller
+     * passes an undefined `sessionId` for a given listen, the stream is
+     * refused with `-32603` rather than silently falling back to the
+     * handler-wide cap. A misconfigured deployment cannot accidentally
+     * appear to be enforcing per-session isolation that was never applied.
+     *
+     * No default: when omitted the per-session guard is off and only the
+     * handler-wide {@linkcode maxSubscriptions} cap applies.
+     */
+    maxSubscriptionsPerSession?: number;
     /** SSE comment-frame keepalive interval; `0` disables keepalive (default 15000). */
     keepAliveMs?: number;
     /** Out-of-band error reporting (never alters the response). */
@@ -108,7 +124,13 @@ export interface ListenRouter {
      * graceful-close result's `_meta` (the spec's `SubscriptionsListenResultMetaObject`
      * extends `ResultMetaObject`, so the serverInfo SHOULD applies there too).
      */
-    serve(message: JSONRPCRequest, signal: AbortSignal | undefined, capabilities: ServerCapabilities, serverInfo: Implementation): Response;
+    serve(
+        message: JSONRPCRequest,
+        signal: AbortSignal | undefined,
+        capabilities: ServerCapabilities,
+        serverInfo: Implementation,
+        sessionId?: string
+    ): Response;
     /**
      * Gracefully close every open subscription stream: emits the empty
      * `subscriptions/listen` JSON-RPC result (the spec's graceful-close
@@ -122,21 +144,51 @@ export interface ListenRouter {
 export function createListenRouter(options: ListenRouterOptions): ListenRouter {
     const { bus, onerror } = options;
     const maxSubscriptions = options.maxSubscriptions ?? DEFAULT_MAX_SUBSCRIPTIONS;
+    const maxSubscriptionsPerSession = options.maxSubscriptionsPerSession;
     const keepAliveMs = options.keepAliveMs ?? DEFAULT_SSE_KEEP_ALIVE_MS;
 
     const open = new Set<(graceful: boolean) => void>();
+    // Per-session teardown handles, keyed on the transport-managed session id.
+    // Grown alongside `open` so the per-session cap can be enforced without a
+    // scan. A session's entry is removed when its last stream tears down.
+    const openPerSession = new Map<string, Set<(graceful: boolean) => void>>();
 
     function serve(
         message: JSONRPCRequest,
         signal: AbortSignal | undefined,
         capabilities: ServerCapabilities,
-        serverInfo: Implementation
+        serverInfo: Implementation,
+        sessionId?: string
     ): Response {
         // Capacity guard, pre-ack: in-band -32603 on HTTP 200.
         if (open.size >= maxSubscriptions) {
             onerror?.(new Error(`subscriptions/listen refused: subscription limit reached (${maxSubscriptions})`));
             return jsonRpcError(message.id, -32_603, 'Subscription limit reached');
         }
+        if (maxSubscriptionsPerSession !== undefined) {
+            // Fail-closed: the per-session cap is only honest with a resolved
+            // key. The caller's resolver returned undefined for this listen,
+            // so there is no bucket to attribute it to. Reject rather than
+            // silently fall through to the handler-wide cap and misreport the
+            // guarantee.
+            if (sessionId === undefined) {
+                onerror?.(
+                    new Error(
+                        'subscriptions/listen refused: per-session cap configured but no trusted session key was resolved for this request'
+                    )
+                );
+                return jsonRpcError(message.id, -32_603, 'Subscription limit reached');
+            }
+            const sessionOpen = openPerSession.get(sessionId);
+            if (sessionOpen !== undefined && sessionOpen.size >= maxSubscriptionsPerSession) {
+                onerror?.(
+                    new Error(
+                        `subscriptions/listen refused: per-session subscription limit reached (${maxSubscriptionsPerSession}, session ${sessionId})`
+                    )
+                );
+                return jsonRpcError(message.id, -32_603, 'Subscription limit reached');
+            }
+        }
         const filter = parseListenFilter(message);
         if (filter === undefined) {
             return jsonRpcError(message.id, -32_602, "Invalid params: 'notifications' is required and must be a valid SubscriptionFilter");
@@ -194,6 +246,13 @@ export function createListenRouter(options: ListenRouterOptions): ListenRouter {
             if (keepAliveTimer !== undefined) clearInterval(keepAliveTimer);
             abortCleanup?.();
             open.delete(teardown);
+            if (sessionId !== undefined) {
+                const sessionOpen = openPerSession.get(sessionId);
+                if (sessionOpen !== undefined) {
+                    sessionOpen.delete(teardown);
+                    if (sessionOpen.size === 0) openPerSession.delete(sessionId);
+                }
+            }
             try {
                 controller.close();
             } catch {
@@ -229,6 +288,14 @@ export function createListenRouter(options: ListenRouterOptions): ListenRouter {
                 keepAliveTimer = armSseKeepAlive(keepAliveMs, () => writeFrame(': keepalive\n\n'));
 
                 open.add(teardown);
+                if (sessionId !== undefined) {
+                    let sessionOpen = openPerSession.get(sessionId);
+                    if (sessionOpen === undefined) {
+                        sessionOpen = new Set();
+                        openPerSession.set(sessionId, sessionOpen);
+                    }
+                    sessionOpen.add(teardown);
+                }
             },
             cancel() {
                 // The client closed the SSE stream — the spec's HTTP cancel