[SEP-2575] core: StreamDriver + stdio/InMemory sendAndReceive

NEW core/shared/streamDriver.ts: minimal request->response correlator for
pipe-shaped client transports (one _pending map keyed by RequestId AND
String(id) for _meta.subscriptionId routing). sendAndReceive yields
notifications then one response (or indefinite for subscriptions/listen);
break/return() sends notifications/cancelled. Send-failure -> synthetic
error response (no hang). onMessage routes; close() ends all pending.

InMemoryTransport: composes StreamDriver. send() routes via _receive()
(driver claims first, falls through to onmessage). close() closes driver.

StdioClientTransport: composes StreamDriver. processReadBuffer routes to
driver when protocolVersion is unset or stateless. setProtocolVersion()
gates back to onmessage for legacy.

Satisfies: 2575-R12 (client sendAndReceive contract, pipe transports)

Author: felixweinberger

packages/client/src/client/stdio.ts

@@ -3,8 +3,8 @@ import process from 'node:process';
 import type { Stream } from 'node:stream';
 import { PassThrough } from 'node:stream';
 
-import type { JSONRPCMessage, Transport } from '@modelcontextprotocol/core';
-import { ReadBuffer, SdkError, SdkErrorCode, serializeMessage } from '@modelcontextprotocol/core';
+import type { JSONRPCMessage, JSONRPCRequest, Transport } from '@modelcontextprotocol/core';
+import { isStatelessProtocolVersion, ReadBuffer, SdkError, SdkErrorCode, serializeMessage, StreamDriver } from '@modelcontextprotocol/core';
 import spawn from 'cross-spawn';
 
 export type StdioServerParameters = {
@@ -95,11 +95,26 @@ export class StdioClientTransport implements Transport {
     private _readBuffer: ReadBuffer = new ReadBuffer();
     private _serverParams: StdioServerParameters;
     private _stderrStream: PassThrough | null = null;
+    private _protocolVersion?: string;
+    /* eslint-disable-next-line unicorn/consistent-function-scoping */
+    private readonly _driver = new StreamDriver(m => this.send(m));
 
     onclose?: () => void;
     onerror?: (error: Error) => void;
     onmessage?: (message: JSONRPCMessage) => void;
 
+    /**
+     * Sends one request and returns the server's messages for it. Backed by
+     * `StreamDriver`; bypasses `Protocol.request()`.
+     */
+    sendAndReceive(request: Omit<JSONRPCRequest, 'jsonrpc' | 'id'>): AsyncIterable<JSONRPCMessage> {
+        return this._driver.sendAndReceive(request);
+    }
+
+    setProtocolVersion(version: string): void {
+        this._protocolVersion = version;
+    }
+
     constructor(server: StdioServerParameters) {
         this._serverParams = server;
         if (server.stderr === 'pipe' || server.stderr === 'overlapped') {
@@ -195,6 +210,16 @@ export class StdioClientTransport implements Transport {
                     break;
                 }
 
+                // Default to StreamDriver until setProtocolVersion is called with
+                // a pre-2026 version. The discover/initialize probe goes via
+                // sendAndReceive, so the driver claims those.
+                if (
+                    (this._protocolVersion === undefined || isStatelessProtocolVersion(this._protocolVersion)) &&
+                    this._driver.onMessage(message)
+                ) {
+                    continue;
+                }
+
                 this.onmessage?.(message);
             } catch (error) {
                 this.onerror?.(error as Error);
@@ -203,6 +228,7 @@ export class StdioClientTransport implements Transport {
     }
 
     async close(): Promise<void> {
+        this._driver.close();
         if (this._process) {
             const processToClose = this._process;
             this._process = undefined;

packages/core/src/index.ts

@@ -7,6 +7,7 @@ export * from './shared/metadataUtils.js';
 export * from './shared/protocol.js';
 export * from './shared/stateless.js';
 export * from './shared/stdio.js';
+export * from './shared/streamDriver.js';
 export * from './shared/toolNameValidation.js';
 export * from './shared/transport.js';
 export * from './shared/uriTemplate.js';

packages/core/src/shared/streamDriver.ts

@@ -0,0 +1,132 @@
+import { isJSONRPCNotification, isJSONRPCResponse } from '../types/guards.js';
+import type { JSONRPCMessage, JSONRPCRequest, RequestId } from '../types/index.js';
+import { JSONRPC_VERSION, ProtocolErrorCode } from '../types/index.js';
+import { AsyncQueue } from '../util/asyncQueue.js';
+import { META_KEYS } from './stateless.js';
+
+/**
+ * Minimal request→response correlator for pipe-shaped client transports
+ * (stdio, in-memory) under the 2026-06 stateless model. Provides
+ * `sendAndReceive(request) → AsyncIterable<JSONRPCMessage>` so the Client can
+ * make stateless calls without going through `Protocol.request()` and its
+ * `_responseHandlers` map.
+ *
+ * The transport feeds every inbound message to {@linkcode onMessage}; the
+ * driver routes responses by `id` and notifications by `_meta.subscriptionId`
+ * (which is the originating request's id, per SEP-2575) to the matching
+ * iterator. Closing/breaking the iterator sends `notifications/cancelled`.
+ */
+export class StreamDriver {
+    // Seed in a range Protocol's `_requestMessageId` (which starts at 0) will
+    // not reach, so a stdio fallback that mixes a discover-probe (StreamDriver)
+    // with a legacy initialize (Protocol.request) on the same pipe cannot
+    // collide on id 0.
+    private _nextId = 0x40_00_00_00;
+    private readonly _pending = new Map<RequestId, AsyncQueue<JSONRPCMessage>>();
+
+    constructor(private readonly _send: (m: JSONRPCMessage) => Promise<void>) {}
+
+    /**
+     * Sends one request and returns an async-iterable of the messages the server
+     * emits for it: zero or more notifications, then exactly one response (which
+     * ends the iteration). For `subscriptions/listen`, the iteration continues
+     * until `break`/`return()` (which sends `notifications/cancelled`).
+     *
+     * The request is dispatched and registered in `_pending` immediately, before
+     * the first `next()`. Callers MUST consume the iterable (`for await` is
+     * sufficient: it calls `return()` on break/throw); obtaining it and never
+     * iterating leaks the `_pending` entry until {@linkcode close}.
+     */
+    sendAndReceive(request: Omit<JSONRPCRequest, 'jsonrpc' | 'id'>, opts?: { signal?: AbortSignal }): AsyncIterable<JSONRPCMessage> {
+        const id = this._nextId++;
+        const isListen = request.method === 'subscriptions/listen';
+        const queue = new AsyncQueue<JSONRPCMessage>(256);
+
+        const cleanup = () => {
+            this._pending.delete(id);
+            this._pending.delete(String(id));
+            opts?.signal?.removeEventListener('abort', onAbort);
+        };
+        const cancel = () => {
+            if (queue.closed) return;
+            this._send({ jsonrpc: JSONRPC_VERSION, method: 'notifications/cancelled', params: { requestId: id } }).catch(() => {});
+            queue.close();
+            cleanup();
+        };
+        const onAbort = () => cancel();
+        opts?.signal?.addEventListener('abort', onAbort, { once: true });
+
+        this._pending.set(id, queue);
+        // `_meta.subscriptionId` on inbound notifications equals the string form
+        // of the request id (SEP-2575). Register the same queue under that key
+        // so {@linkcode onMessage} can route notifications without a second map.
+        this._pending.set(String(id), queue);
+
+        this._send({ jsonrpc: JSONRPC_VERSION, id, ...request }).catch(error => {
+            // Surface send failure to the iterator instead of hanging forever.
+            queue.push({
+                jsonrpc: JSONRPC_VERSION,
+                id,
+                error: {
+                    code: ProtocolErrorCode.InternalError,
+                    message: `Transport send failed: ${error instanceof Error ? error.message : String(error)}`
+                }
+            });
+            queue.close();
+            cleanup();
+        });
+
+        const inner = queue.iterate();
+        return {
+            [Symbol.asyncIterator]: () => ({
+                async next(): Promise<IteratorResult<JSONRPCMessage>> {
+                    const r = await inner.next();
+                    if (r.done) {
+                        cleanup();
+                    } else if (!isListen && isJSONRPCResponse(r.value)) {
+                        // Non-listen: end after the response.
+                        queue.close();
+                        cleanup();
+                    }
+                    return r;
+                },
+                async return(): Promise<IteratorResult<JSONRPCMessage>> {
+                    cancel();
+                    return { value: undefined, done: true };
+                }
+            })
+        };
+    }
+
+    /**
+     * Feeds one inbound message to the driver. The transport calls this for
+     * every message received while in stateless mode. Returns `true` if the
+     * message was claimed (routed to a pending iterator).
+     */
+    onMessage(m: JSONRPCMessage): boolean {
+        if ('id' in m && m.id !== null && m.id !== undefined) {
+            const q = this._pending.get(m.id);
+            if (q) {
+                q.push(m);
+                return true;
+            }
+        }
+        if (isJSONRPCNotification(m)) {
+            const sid = (m.params?._meta as Record<string, unknown> | undefined)?.[META_KEYS.subscriptionId];
+            if (typeof sid === 'string') {
+                const q = this._pending.get(sid);
+                if (q) {
+                    q.push(m);
+                    return true;
+                }
+            }
+        }
+        return false;
+    }
+
+    /** Ends every pending iterator (e.g., on transport close). */
+    close(): void {
+        for (const q of this._pending.values()) q.close();
+        this._pending.clear();
+    }
+}

packages/core/src/util/inMemory.ts

@@ -1,6 +1,7 @@
 import { SdkError, SdkErrorCode } from '../errors/sdkErrors.js';
+import { StreamDriver } from '../shared/streamDriver.js';
 import type { Transport } from '../shared/transport.js';
-import type { AuthInfo, JSONRPCMessage, RequestId } from '../types/index.js';
+import type { AuthInfo, JSONRPCMessage, JSONRPCRequest, RequestId } from '../types/index.js';
 
 interface QueuedMessage {
     message: JSONRPCMessage;
@@ -18,11 +19,19 @@ export class InMemoryTransport implements Transport {
     private _messageQueue: QueuedMessage[] = [];
     private _closed = false;
 
+    /* eslint-disable-next-line unicorn/consistent-function-scoping */
+    private readonly _driver = new StreamDriver(m => this.send(m));
+
     onclose?: () => void;
     onerror?: (error: Error) => void;
     onmessage?: (message: JSONRPCMessage, extra?: { authInfo?: AuthInfo }) => void;
     sessionId?: string;
 
+    /** Client-side: backed by `StreamDriver`. */
+    sendAndReceive(request: Omit<JSONRPCRequest, 'jsonrpc' | 'id'>): AsyncIterable<JSONRPCMessage> {
+        return this._driver.sendAndReceive(request);
+    }
+
     /**
      * Creates a pair of linked in-memory transports that can communicate with each other. One should be passed to a {@linkcode @modelcontextprotocol/client!client/client.Client | Client} and one to a {@linkcode @modelcontextprotocol/server!server/server.Server | Server}.
      */
@@ -38,13 +47,20 @@ export class InMemoryTransport implements Transport {
         // Process any messages that were queued before start was called
         while (this._messageQueue.length > 0) {
             const queuedMessage = this._messageQueue.shift()!;
-            this.onmessage?.(queuedMessage.message, queuedMessage.extra);
+            this._receive(queuedMessage.message, queuedMessage.extra);
         }
     }
 
+    /** Receive path: route to the StreamDriver first; fall through to `onmessage` for unclaimed. */
+    private _receive(message: JSONRPCMessage, extra?: { authInfo?: AuthInfo }): void {
+        if (this._driver.onMessage(message)) return;
+        this.onmessage?.(message, extra);
+    }
+
     async close(): Promise<void> {
         if (this._closed) return;
         this._closed = true;
+        this._driver.close();
 
         const other = this._otherTransport;
         this._otherTransport = undefined;
@@ -65,7 +81,7 @@ export class InMemoryTransport implements Transport {
         }
 
         if (this._otherTransport.onmessage) {
-            this._otherTransport.onmessage(message, { authInfo: options?.authInfo });
+            this._otherTransport._receive(message, { authInfo: options?.authInfo });
         } else {
             this._otherTransport._messageQueue.push({ message, extra: { authInfo: options?.authInfo } });
         }