feat(server): BackchannelCompat; Server.buildContext outbound via ctx.mcpReq.send

BackchannelCompat: opt-in 2025-11 server-to-client backchannel for handleHttp. A handler's elicitInput/requestSampling writes onto the open POST's SSE stream and the client's POSTed-back response resolves the await.

Server.buildContext factors elicitInput/requestSampling into _createMessageVia/_elicitInputVia so instance methods and ctx.mcpReq share the same validation and capability check (capability check reads instance _clientCapabilities; per-request flow in F1).

Author: felixweinberger

packages/server/src/index.ts

@@ -31,6 +31,7 @@ export { Server } from './server/server.js';
 // StdioServerTransport is exported from the './stdio' subpath — server stdio has only type-level Node
 // imports (erased at compile time), but matching the client's `./stdio` subpath gives consumers a
 // consistent shape across packages.
+export { BackchannelCompat } from './server/backchannelCompat.js';
 export type { Dispatchable, HandleHttpOptions, HandleHttpRequestExtra } from './server/handleHttp.js';
 export { handleHttp } from './server/handleHttp.js';
 export type { SessionCompatOptions, SessionValidation } from './server/sessionCompat.js';

packages/server/src/server/backchannelCompat.ts

@@ -0,0 +1,122 @@
+import type {
+    JSONRPCErrorResponse,
+    JSONRPCMessage,
+    JSONRPCRequest,
+    JSONRPCResultResponse,
+    Request,
+    RequestOptions,
+    Result
+} from '@modelcontextprotocol/core';
+import { DEFAULT_REQUEST_TIMEOUT_MSEC, isJSONRPCErrorResponse, ProtocolError, SdkError, SdkErrorCode } from '@modelcontextprotocol/core';
+
+/**
+ * Isolated 2025-11 server-to-client request backchannel for `handleHttp`.
+ *
+ * The 2025-11 protocol allows a server to send `elicitation/create` and
+ * `sampling/createMessage` requests to the client mid-tool-call by writing them as
+ * SSE events on the open POST response stream and waiting for the client to POST
+ * the response back. This class owns the per-session `{requestId -> resolver}`
+ * map that correlation requires.
+ *
+ * It exists so this stateful behaviour is in one removable file once MRTR
+ * (SEP-2322) is the protocol floor and `env.send` becomes a hard error in
+ * stateless paths.
+ */
+export class BackchannelCompat {
+    private _pending = new Map<string, Map<number, { resolve: (r: Result) => void; reject: (e: Error) => void }>>();
+    private _nextId = 0;
+
+    /**
+     * Returns an `env.send` implementation bound to the given session and POST-stream writer.
+     * The returned function writes the outbound JSON-RPC request to `writeSSE` and resolves when
+     * {@linkcode handleResponse} is called for the same id on the same session.
+     *
+     * `writeSSE` returns `false` when the underlying stream is closed; the returned promise then
+     * rejects immediately with `SendFailed` instead of waiting for the timeout.
+     *
+     * Backchannel writes are not persisted to the {@linkcode EventStore}, so a client that
+     * disconnects mid-elicitation and resumes via `Last-Event-ID` will not see the outbound
+     * request again; the awaiting handler will time out. This is a known limitation of the
+     * legacy backchannel path; SEP-2322 (`ContinuationCompat`) is the resumable alternative.
+     */
+    makeEnvSend(sessionId: string, writeSSE: (msg: JSONRPCMessage) => boolean): (req: Request, opts?: RequestOptions) => Promise<Result> {
+        return (req: Request, opts?: RequestOptions): Promise<Result> => {
+            return new Promise<Result>((resolve, reject) => {
+                if (opts?.signal?.aborted) {
+                    reject(opts.signal.reason instanceof Error ? opts.signal.reason : new Error(String(opts.signal.reason)));
+                    return;
+                }
+
+                const id = this._nextId++;
+                const sessionMap = this._pending.get(sessionId) ?? new Map();
+                this._pending.set(sessionId, sessionMap);
+
+                // eslint-disable-next-line prefer-const -- forward-referenced by cleanup() before assignment site
+                let timer: ReturnType<typeof setTimeout> | undefined;
+                const onAbort = () => {
+                    // Tell the client to stop processing, then reject locally.
+                    writeSSE({ jsonrpc: '2.0', method: 'notifications/cancelled', params: { requestId: id } });
+                    settle.reject(opts!.signal!.reason instanceof Error ? opts!.signal!.reason : new Error(String(opts!.signal!.reason)));
+                };
+                const cleanup = () => {
+                    if (timer !== undefined) clearTimeout(timer);
+                    sessionMap.delete(id);
+                    if (sessionMap.size === 0) this._pending.delete(sessionId);
+                    opts?.signal?.removeEventListener('abort', onAbort);
+                };
+                const settle = {
+                    resolve: (r: Result) => {
+                        cleanup();
+                        resolve(r);
+                    },
+                    reject: (e: Error) => {
+                        cleanup();
+                        reject(e);
+                    }
+                };
+                sessionMap.set(id, settle);
+
+                const timeoutMs = opts?.timeout ?? DEFAULT_REQUEST_TIMEOUT_MSEC;
+                timer = setTimeout(
+                    () => settle.reject(new SdkError(SdkErrorCode.RequestTimeout, 'Request timed out', { timeout: timeoutMs })),
+                    timeoutMs
+                );
+
+                opts?.signal?.addEventListener('abort', onAbort, { once: true });
+
+                const wire: JSONRPCRequest = { jsonrpc: '2.0', id, method: req.method, params: req.params };
+                if (!writeSSE(wire)) {
+                    settle.reject(new SdkError(SdkErrorCode.SendFailed, 'Backchannel stream closed'));
+                }
+            });
+        };
+    }
+
+    /**
+     * Routes an incoming JSON-RPC response (from a client POST) to the waiting `env.send` promise.
+     * @returns true if a pending request matched and was settled.
+     */
+    handleResponse(sessionId: string, response: JSONRPCResultResponse | JSONRPCErrorResponse): boolean {
+        // We only mint numeric ids; a non-numeric id cannot be ours, so do not coerce
+        // (string "0" must not claim numeric pending id 0).
+        if (typeof response.id !== 'number') return false;
+        const sessionMap = this._pending.get(sessionId);
+        const settle = sessionMap?.get(response.id);
+        if (!settle) return false;
+        if (isJSONRPCErrorResponse(response)) {
+            settle.reject(ProtocolError.fromError(response.error.code, response.error.message, response.error.data));
+        } else {
+            settle.resolve(response.result);
+        }
+        return true;
+    }
+
+    /** Rejects all pending requests for a session and forgets it. */
+    closeSession(sessionId: string): void {
+        const sessionMap = this._pending.get(sessionId);
+        if (!sessionMap) return;
+        const err = new SdkError(SdkErrorCode.ConnectionClosed, 'Session closed');
+        for (const s of sessionMap.values()) s.reject(err);
+        this._pending.delete(sessionId);
+    }
+}

packages/server/src/server/handleHttp.ts

@@ -33,7 +33,10 @@ export interface Dispatchable {
  *
  * `mcp.connect(transport)` is not called; each HTTP request flows through
  * `mcp.dispatch()` directly. Supply a `SessionCompat` via `options.session`
- * to serve clients that send `Mcp-Session-Id` (the pre-2026-06 stateful flow).
+ * to serve clients that send `Mcp-Session-Id` (the pre-2026-06 stateful flow),
+ * and a `BackchannelCompat` via `options.backchannel` to let handlers'
+ * `ctx.mcpReq.send` (e.g. `elicitInput`, `requestSampling`) reach those
+ * clients over the open POST SSE stream.
  */
 export function handleHttp(
     mcp: Dispatchable,

packages/server/src/server/server.ts

@@ -28,6 +28,7 @@ import type {
     Result,
     ServerCapabilities,
     ServerContext,
+    StandardSchemaV1,
     ToolResultContent,
     ToolUseContent
 } from '@modelcontextprotocol/core';
@@ -53,6 +54,13 @@ import { DefaultJsonSchemaValidator } from '@modelcontextprotocol/server/_shims'
 
 import { ExperimentalServerTasks } from '../experimental/tasks/server.js';
 
+/** Three-arg send used by `_createMessageVia`/`_elicitInputVia`; satisfied by both `_requestWithSchema` and `ctx.mcpReq.send`. */
+type SendWithSchema = <T extends StandardSchemaV1>(
+    request: { method: string; params?: Record<string, unknown> },
+    resultSchema: T,
+    options?: RequestOptions
+) => Promise<StandardSchemaV1.InferOutput<T>>;
+
 export type ServerOptions = ProtocolOptions & {
     /**
      * Capabilities to advertise as being supported by this server.
@@ -131,13 +139,18 @@ export class Server extends Protocol<ServerContext> {
     protected override buildContext(ctx: BaseContext, transportInfo?: MessageExtraInfo): ServerContext {
         // Only create http when there's actual HTTP transport info or auth info
         const hasHttpInfo = ctx.http || transportInfo?.request || transportInfo?.closeSSEStream || transportInfo?.closeStandaloneSSEStream;
+        const sendOpts = (options?: RequestOptions): RequestOptions => ({ ...options, relatedRequestId: ctx.mcpReq.id });
         return {
             ...ctx,
             mcpReq: {
                 ...ctx.mcpReq,
-                log: (level, data, logger) => this.sendLoggingMessage({ level, data, logger }),
-                elicitInput: (params, options) => this.elicitInput(params, options),
-                requestSampling: (params, options) => this.createMessage(params, options)
+                log: async (level, data, logger) => {
+                    if (this._capabilities.logging && !this.isMessageIgnored(level, ctx.sessionId)) {
+                        await ctx.mcpReq.notify({ method: 'notifications/message', params: { level, data, logger } });
+                    }
+                },
+                elicitInput: (params, options) => this._elicitInputVia(ctx.mcpReq.send, params, sendOpts(options)),
+                requestSampling: (params, options) => this._createMessageVia(ctx.mcpReq.send, params, sendOpts(options))
             },
             http: hasHttpInfo
                 ? {
@@ -441,14 +454,31 @@ export class Server extends Protocol<ServerContext> {
         params: CreateMessageRequest['params'],
         options?: RequestOptions
     ): Promise<CreateMessageResult | CreateMessageResultWithTools> {
-        // Capability check - only required when tools/toolChoice are provided
-        if ((params.tools || params.toolChoice) && !this._clientCapabilities?.sampling?.tools) {
+        return this._createMessageVia((r, schema, opts) => this._requestWithSchema(r, schema, opts), params, options);
+    }
+
+    /**
+     * Shared body for {@linkcode createMessage} and `ctx.mcpReq.requestSampling`: capability check,
+     * tool_use/tool_result pairing validation, and result-schema selection. The `send` argument
+     * routes to either the connected driver (instance method) or `ctx.mcpReq.send` (per-request).
+     *
+     * NOTE: the capability check below reads `this._clientCapabilities`, which is a singleton
+     * (set on the most recent `initialize`). For multi-session `handleHttp` deployments this
+     * can read a different session's caps. The per-request `_meta.clientCapabilities` flow
+     * fixes this in a follow-up; until then, do not rely on the cap gate for isolation.
+     */
+    private async _createMessageVia(
+        send: SendWithSchema,
+        params: CreateMessageRequest['params'],
+        options?: RequestOptions
+    ): Promise<CreateMessageResult | CreateMessageResultWithTools> {
+        if (!this._clientCapabilities?.sampling) {
+            throw new SdkError(SdkErrorCode.CapabilityNotSupported, 'Client does not support sampling capability.');
+        }
+        if ((params.tools || params.toolChoice) && !this._clientCapabilities.sampling.tools) {
             throw new SdkError(SdkErrorCode.CapabilityNotSupported, 'Client does not support sampling tools capability.');
         }
 
-        // Message structure validation - always validate tool_use/tool_result pairs.
-        // These may appear even without tools/toolChoice in the current request when
-        // a previous sampling request returned tool_use and this is a follow-up with results.
         if (params.messages.length > 0) {
             const lastMessage = params.messages.at(-1)!;
             const lastContent = Array.isArray(lastMessage.content) ? lastMessage.content : [lastMessage.content];
@@ -490,11 +520,10 @@ export class Server extends Protocol<ServerContext> {
             }
         }
 
-        // Use different schemas based on whether tools are provided
         if (params.tools) {
-            return this._requestWithSchema({ method: 'sampling/createMessage', params }, CreateMessageResultWithToolsSchema, options);
+            return send({ method: 'sampling/createMessage', params }, CreateMessageResultWithToolsSchema, options);
         }
-        return this._requestWithSchema({ method: 'sampling/createMessage', params }, CreateMessageResultSchema, options);
+        return send({ method: 'sampling/createMessage', params }, CreateMessageResultSchema, options);
     }
 
     /**
@@ -505,46 +534,49 @@ export class Server extends Protocol<ServerContext> {
      * @returns The result of the elicitation request.
      */
     async elicitInput(params: ElicitRequestFormParams | ElicitRequestURLParams, options?: RequestOptions): Promise<ElicitResult> {
+        return this._elicitInputVia((r, schema, opts) => this._requestWithSchema(r, schema, opts), params, options);
+    }
+
+    /**
+     * Shared body for {@linkcode elicitInput} and `ctx.mcpReq.elicitInput`: form/url capability
+     * sub-field check, mode defaulting, and post-receipt JSON-schema validation of `content`.
+     */
+    private async _elicitInputVia(
+        send: SendWithSchema,
+        params: ElicitRequestFormParams | ElicitRequestURLParams,
+        options?: RequestOptions
+    ): Promise<ElicitResult> {
         const mode = (params.mode ?? 'form') as 'form' | 'url';
 
         switch (mode) {
             case 'url': {
                 if (!this._clientCapabilities?.elicitation?.url) {
                     throw new SdkError(SdkErrorCode.CapabilityNotSupported, 'Client does not support url elicitation.');
                 }
-
                 const urlParams = params as ElicitRequestURLParams;
-                return this._requestWithSchema({ method: 'elicitation/create', params: urlParams }, ElicitResultSchema, options);
+                return send({ method: 'elicitation/create', params: urlParams }, ElicitResultSchema, options);
             }
             case 'form': {
                 if (!this._clientCapabilities?.elicitation?.form) {
                     throw new SdkError(SdkErrorCode.CapabilityNotSupported, 'Client does not support form elicitation.');
                 }
-
                 const formParams: ElicitRequestFormParams =
                     params.mode === 'form' ? (params as ElicitRequestFormParams) : { ...(params as ElicitRequestFormParams), mode: 'form' };
 
-                const result = await this._requestWithSchema(
-                    { method: 'elicitation/create', params: formParams },
-                    ElicitResultSchema,
-                    options
-                );
+                const result = await send({ method: 'elicitation/create', params: formParams }, ElicitResultSchema, options);
 
                 if (result.action === 'accept' && result.content && formParams.requestedSchema) {
                     try {
                         const validator = this._jsonSchemaValidator.getValidator(formParams.requestedSchema as JsonSchemaType);
                         const validationResult = validator(result.content);
-
                         if (!validationResult.valid) {
                             throw new ProtocolError(
                                 ProtocolErrorCode.InvalidParams,
                                 `Elicitation response content does not match requested schema: ${validationResult.errorMessage}`
                             );
                         }
                     } catch (error) {
-                        if (error instanceof ProtocolError) {
-                            throw error;
-                        }
+                        if (error instanceof ProtocolError) throw error;
                         throw new ProtocolError(
                             ProtocolErrorCode.InternalError,
                             `Error validating elicitation response: ${error instanceof Error ? error.message : String(error)}`

packages/server/src/server/shttpHandler.ts

@@ -19,6 +19,7 @@ import {
     SUPPORTED_PROTOCOL_VERSIONS
 } from '@modelcontextprotocol/core';
 
+import type { BackchannelCompat } from './backchannelCompat.js';
 import type { SessionCompat } from './sessionCompat.js';
 import type { EventId, EventStore } from './streamableHttp.js';
 
@@ -54,6 +55,15 @@ export interface ShttpHandlerOptions {
      */
     session?: SessionCompat;
 
+    /**
+     * Pre-2026-06 server-to-client request backchannel. When provided alongside `session`,
+     * a handler's `ctx.mcpReq.send` (e.g. `elicitInput`, `requestSampling`) writes the
+     * outbound request onto the open POST's SSE stream and the client's POSTed-back
+     * response resolves the awaited promise. When omitted, `ctx.mcpReq.send` rejects
+     * with `NotConnected` on this path.
+     */
+    backchannel?: BackchannelCompat;
+
     /**
      * Event store for SSE resumability via `Last-Event-ID`. When configured, every
      * outgoing SSE event is persisted and a priming event is sent at stream start.
@@ -145,6 +155,8 @@ const SSE_HEADERS: Record<string, string> = {
     Connection: 'keep-alive'
 };
 
+const wiredBackchannels = new WeakMap<SessionCompat, WeakSet<BackchannelCompat>>();
+
 /**
  * Creates a Web-standard `(Request) => Promise<Response>` handler for the MCP Streamable HTTP
  * transport, driven by {@linkcode ShttpCallbacks.onrequest} per request.
@@ -161,11 +173,24 @@ export function shttpHandler(
 ): (req: Request, extra?: ShttpRequestExtra) => Promise<Response> {
     const enableJsonResponse = options.enableJsonResponse ?? false;
     const session = options.session;
+    const backchannel = options.backchannel;
     const eventStore = options.eventStore;
     const retryInterval = options.retryInterval;
     const supportedProtocolVersions = options.supportedProtocolVersions ?? SUPPORTED_PROTOCOL_VERSIONS;
     const onerror = options.onerror;
 
+    // Reject any pending backchannel sends on session close (DELETE, idle/capacity eviction).
+    // Guard so repeated shttpHandler()/handleHttp() calls on the same session+backchannel
+    // pair do not accumulate listeners.
+    if (session && backchannel) {
+        let wired = wiredBackchannels.get(session);
+        if (!wired) wiredBackchannels.set(session, (wired = new WeakSet()));
+        if (!wired.has(backchannel)) {
+            wired.add(backchannel);
+            session.addCloseListener(sid => backchannel.closeSession(sid));
+        }
+    }
+
     /**
      * Per-request abort controllers for `notifications/cancelled`. Keyed by
      * `(sessionId, requestId)` so concurrent sessions reusing the same JSON-RPC id don't collide.
@@ -283,7 +308,8 @@ export function shttpHandler(
         }
 
         for (const r of responses) {
-            if (cb.onresponse?.(r) === false) {
+            const claimedByBackchannel = backchannel && sessionId !== undefined && backchannel.handleResponse(sessionId, r);
+            if (!claimedByBackchannel && cb.onresponse?.(r) === false) {
                 onerror?.(new Error(`Unclaimed JSON-RPC response (id=${String(r.id)}); no pending server-initiated request matched.`));
             }
         }
@@ -359,7 +385,14 @@ export function shttpHandler(
                     closeStandaloneSSEStream:
                         supportsPolling && sessionId !== undefined ? () => session?.closeStandaloneStream(sessionId) : undefined
                 };
-                const env: ShttpRequestEnv = { ...baseEnv, _transportExtra: transportExtra };
+                // Backchannel writes go straight to writeSSEEvent (synchronous boolean) so a closed
+                // stream surfaces as `false` immediately instead of hanging until the timeout.
+                const writeSSE = (msg: JSONRPCMessage): boolean => writeSSEEvent(controller, encoder, msg);
+                const env: ShttpRequestEnv = {
+                    ...baseEnv,
+                    _transportExtra: transportExtra,
+                    ...(backchannel && sessionId !== undefined ? { send: backchannel.makeEnvSend(sessionId, writeSSE) } : {})
+                };
                 void (async () => {
                     try {
                         await writePrimingEvent(controller, encoder, streamId, clientProtocolVersion);