[SEP-2567] core: serverStatelessRouter + stdio/InMemory per-message routers

NEW core/shared/serverStatelessRouter.ts: routeServerStateless() —
per-message branch on isStatelessRequest to StatelessHandlers
{dispatch,listen}; notifications/cancelled aborts matching _inflight
controller.

stdio + InMemory server-side: receive paths route via
routeServerStateless; close() aborts all in-flight.

stdio + InMemory client-side: sendAndReceive threads opts?.signal to
StreamDriver.

Satisfies: 2567-R1 (pipe), 2567-R2

Author: felixweinberger

packages/client/src/client/stdio.ts

@@ -107,8 +107,8 @@ export class StdioClientTransport implements Transport {
      * Sends one request and returns the server's messages for it. Backed by
      * {@linkcode StreamDriver}; bypasses `Protocol.request()`.
      */
-    sendAndReceive(request: Omit<JSONRPCRequest, 'jsonrpc' | 'id'>): AsyncIterable<JSONRPCMessage> {
-        return this._driver.sendAndReceive(request);
+    sendAndReceive(request: Omit<JSONRPCRequest, 'jsonrpc' | 'id'>, opts?: { signal?: AbortSignal }): AsyncIterable<JSONRPCMessage> {
+        return this._driver.sendAndReceive(request, opts);
     }
 
     setProtocolVersion(version: string): void {

packages/core/src/index.ts

@@ -5,6 +5,7 @@ export * from './shared/authUtils.js';
 export * from './shared/dispatcher.js';
 export * from './shared/metadataUtils.js';
 export * from './shared/protocol.js';
+export * from './shared/serverStatelessRouter.js';
 export * from './shared/stateless.js';
 export * from './shared/stdio.js';
 export * from './shared/streamDriver.js';

packages/core/src/shared/serverStatelessRouter.ts

@@ -0,0 +1,74 @@
+import type { JSONRPCMessage, JSONRPCRequest, RequestId } from '../types/index.js';
+import { isJSONRPCNotification, ProtocolErrorCode } from '../types/index.js';
+import { errorResponse } from './dispatcher.js';
+import type { ListenContext, StatelessHandlers } from './stateless.js';
+import { isStatelessRequest } from './stateless.js';
+
+/**
+ * Per-message router for pipe-shaped server transports (stdio, in-memory).
+ * Call once per inbound message. Returns `true` if the message was claimed by
+ * the stateless path (so the caller should NOT pass it to legacy `onmessage`).
+ *
+ * Stateless requests are dispatched via {@linkcode StatelessHandlers};
+ * `notifications/cancelled` aborts a tracked in-flight request.
+ */
+export function routeServerStateless(
+    message: JSONRPCMessage,
+    handlers: StatelessHandlers,
+    inflight: Map<RequestId, AbortController>,
+    write: (m: JSONRPCMessage) => void,
+    ctx: ListenContext,
+    onerror?: (e: Error) => void
+): boolean {
+    if (isStatelessRequest(message)) {
+        const ac = new AbortController();
+        inflight.set(message.id, ac);
+        void handleOne(handlers, message, ac, write, ctx)
+            .catch(error => onerror?.(error instanceof Error ? error : new Error(String(error))))
+            .finally(() => inflight.delete(message.id));
+        return true;
+    }
+    if (isJSONRPCNotification(message) && message.method === 'notifications/cancelled') {
+        const requestId = (message.params as { requestId?: RequestId } | undefined)?.requestId;
+        const ac = requestId === undefined ? undefined : inflight.get(requestId);
+        if (ac) {
+            ac.abort();
+            return true;
+        }
+    }
+    return false;
+}
+
+async function handleOne(
+    handlers: StatelessHandlers,
+    req: JSONRPCRequest,
+    ac: AbortController,
+    write: (m: JSONRPCMessage) => void,
+    ctx: ListenContext
+): Promise<void> {
+    if (req.method === 'subscriptions/listen') {
+        let listenStream;
+        try {
+            listenStream = handlers.listen(req, ctx);
+        } catch (error) {
+            write(
+                errorResponse(req.id, ProtocolErrorCode.InvalidParams, error instanceof Error ? error.message : 'Invalid listen request')
+            );
+            return;
+        }
+        const { stream, close } = listenStream;
+        ac.signal.addEventListener('abort', close, { once: true });
+        try {
+            for await (const m of stream) {
+                if (ac.signal.aborted) break;
+                write(m);
+            }
+        } finally {
+            ac.signal.removeEventListener('abort', close);
+            close();
+        }
+    } else {
+        const response = await handlers.dispatch(req, { signal: ac.signal, authInfo: ctx.authInfo, notify: write });
+        write(response);
+    }
+}

packages/core/src/util/inMemory.ts

@@ -1,4 +1,6 @@
 import { SdkError, SdkErrorCode } from '../errors/sdkErrors.js';
+import { routeServerStateless } from '../shared/serverStatelessRouter.js';
+import type { StatelessHandlers } from '../shared/stateless.js';
 import { StreamDriver } from '../shared/streamDriver.js';
 import type { Transport } from '../shared/transport.js';
 import type { AuthInfo, JSONRPCMessage, JSONRPCRequest, RequestId } from '../types/index.js';
@@ -19,6 +21,9 @@ export class InMemoryTransport implements Transport {
     private _messageQueue: QueuedMessage[] = [];
     private _closed = false;
 
+    private _statelessHandlers?: StatelessHandlers;
+    private readonly _inflight = new Map<RequestId, AbortController>();
+
     /* eslint-disable-next-line unicorn/consistent-function-scoping */
     private readonly _driver = new StreamDriver(m => this.send(m));
 
@@ -28,8 +33,13 @@ export class InMemoryTransport implements Transport {
     sessionId?: string;
 
     /** Client-side: backed by {@linkcode StreamDriver}. */
-    sendAndReceive(request: Omit<JSONRPCRequest, 'jsonrpc' | 'id'>): AsyncIterable<JSONRPCMessage> {
-        return this._driver.sendAndReceive(request);
+    sendAndReceive(request: Omit<JSONRPCRequest, 'jsonrpc' | 'id'>, opts?: { signal?: AbortSignal }): AsyncIterable<JSONRPCMessage> {
+        return this._driver.sendAndReceive(request, opts);
+    }
+
+    /** Server-side: installed by `Server.connect()`. */
+    setStatelessHandlers(h: StatelessHandlers): void {
+        this._statelessHandlers = h;
     }
 
     /**
@@ -51,15 +61,37 @@ export class InMemoryTransport implements Transport {
         }
     }
 
-    /** Receive path: route to the StreamDriver first; fall through to `onmessage` for unclaimed. */
+    /**
+     * Receive path. Per-message router (mirrors stdio): server-side stateless
+     * requests go to {@linkcode StatelessHandlers}; client-side
+     * {@linkcode StreamDriver} claims responses for pending iterators; unclaimed
+     * messages fall through to legacy `onmessage`.
+     */
     private _receive(message: JSONRPCMessage, extra?: { authInfo?: AuthInfo }): void {
+        if (
+            this._statelessHandlers &&
+            routeServerStateless(
+                message,
+                this._statelessHandlers,
+                this._inflight,
+                m => {
+                    this.send(m).catch(error => this.onerror?.(error as Error));
+                },
+                { authInfo: extra?.authInfo },
+                e => this.onerror?.(e)
+            )
+        ) {
+            return;
+        }
         if (this._driver.onMessage(message)) return;
         this.onmessage?.(message, extra);
     }
 
     async close(): Promise<void> {
         if (this._closed) return;
         this._closed = true;
+        for (const ac of this._inflight.values()) ac.abort();
+        this._inflight.clear();
         this._driver.close();
 
         const other = this._otherTransport;

packages/server/src/server/stdio.ts

@@ -1,7 +1,7 @@
 import type { Readable, Writable } from 'node:stream';
 
-import type { JSONRPCMessage, Transport } from '@modelcontextprotocol/core';
-import { ReadBuffer, serializeMessage } from '@modelcontextprotocol/core';
+import type { JSONRPCMessage, RequestId, StatelessHandlers, Transport } from '@modelcontextprotocol/core';
+import { ReadBuffer, routeServerStateless, serializeMessage } from '@modelcontextprotocol/core';
 import { process } from '@modelcontextprotocol/server/_shims';
 
 /**
@@ -21,6 +21,9 @@ export class StdioServerTransport implements Transport {
     private _started = false;
     private _closed = false;
 
+    private _statelessHandlers?: StatelessHandlers;
+    private readonly _inflight = new Map<RequestId, AbortController>();
+
     constructor(
         private _stdin: Readable = process.stdin,
         private _stdout: Writable = process.stdout
@@ -30,6 +33,15 @@ export class StdioServerTransport implements Transport {
     onerror?: (error: Error) => void;
     onmessage?: (message: JSONRPCMessage) => void;
 
+    /**
+     * Installed by `Server.connect()`. When present, {@linkcode processReadBuffer}
+     * routes 2026-06 requests via {@linkcode StatelessHandlers}; everything else
+     * falls through to legacy `onmessage`.
+     */
+    setStatelessHandlers(h: StatelessHandlers): void {
+        this._statelessHandlers = h;
+    }
+
     // Arrow functions to bind `this` properly, while maintaining function identity.
     _ondata = (chunk: Buffer) => {
         this._readBuffer.append(chunk);
@@ -69,6 +81,25 @@ export class StdioServerTransport implements Transport {
                     break;
                 }
 
+                // Per-message router. Stateless requests (carrying _meta.protocolVersion
+                // for a 2026-06 version) go to the dispatch/listen handlers; everything
+                // else flows to the legacy onmessage path (Protocol._onrequest).
+                if (
+                    this._statelessHandlers &&
+                    routeServerStateless(
+                        message,
+                        this._statelessHandlers,
+                        this._inflight,
+                        m => {
+                            this.send(m).catch(error => this.onerror?.(error as Error));
+                        },
+                        {},
+                        e => this.onerror?.(e)
+                    )
+                ) {
+                    continue;
+                }
+
                 this.onmessage?.(message);
             } catch (error) {
                 this.onerror?.(error as Error);
@@ -81,6 +112,8 @@ export class StdioServerTransport implements Transport {
             return;
         }
         this._closed = true;
+        for (const ac of this._inflight.values()) ac.abort();
+        this._inflight.clear();
 
         // Remove our event listeners first
         this._stdin.off('data', this._ondata);