simplify: remove OutboundMiddleware/useOutbound; StreamDriver calls TaskManager directly

OutboundMiddleware was a 4-hook record (request/notification/response/close)
with first-claim-wins composition, registered via useOutbound() on McpServer/
Client/Protocol. Its only consumer was TaskManager, and it created a second,
incompatible middleware pattern alongside Dispatcher.use().

Now: StreamDriver takes `taskManager?: TaskManager` and calls processOutbound*/
processInboundResponse/onClose at explicit call sites. McpServer/Client/Protocol
pass their TaskManager via attach options. TaskManager.attachTo() returns void
(inbound use() registration + handler install only). Dispatcher.use() (the
standard (next)=>fn inbound pattern) is unchanged.

Author: felixweinberger

packages/client/src/client/client.ts

@@ -32,7 +32,6 @@ import type {
     Notification,
     NotificationMethod,
     NotificationOptions,
-    OutboundMiddleware,
     ProtocolOptions,
     ReadResourceRequest,
     Request,
@@ -55,7 +54,6 @@ import {
     CallToolResultSchema,
     CancelTaskResultSchema,
     CompleteResultSchema,
-    composeOutboundMiddleware,
     CreateMessageRequestSchema,
     CreateMessageResultSchema,
     CreateMessageResultWithToolsSchema,
@@ -230,7 +228,6 @@ export class Client extends Dispatcher<ClientContext> {
     private _experimental?: { tasks: ExperimentalClientTasks };
     private _listChangedDebounceTimers: Map<string, ReturnType<typeof setTimeout>> = new Map();
     private _taskManager: TaskManager;
-    private readonly _outboundMw: OutboundMiddleware[] = [];
 
     onclose?: () => void;
     onerror?: (error: Error) => void;
@@ -249,7 +246,7 @@ export class Client extends Dispatcher<ClientContext> {
 
         const tasksOpts = extractTaskManagerOptions(_options?.capabilities?.tasks);
         this._taskManager = tasksOpts ? new TaskManager(tasksOpts) : new NullTaskManager();
-        const tasksOutbound = this._taskManager.attachTo(this, {
+        this._taskManager.attachTo(this, {
             channel: () =>
                 this._ct
                     ? {
@@ -264,7 +261,6 @@ export class Client extends Dispatcher<ClientContext> {
             assertTaskCapability: () => {},
             assertTaskHandlerCapability: () => {}
         });
-        this.useOutbound(tasksOutbound);
 
         // Strip runtime-only fields from advertised capabilities
         if (_options?.capabilities?.tasks) {
@@ -277,14 +273,6 @@ export class Client extends Dispatcher<ClientContext> {
         super.setRequestHandler('ping', async () => ({}));
     }
 
-    /**
-     * Register an {@linkcode OutboundMiddleware} applied at the request-correlation seam.
-     */
-    useOutbound(mw: OutboundMiddleware): this {
-        this._outboundMw.push(mw);
-        return this;
-    }
-
     /**
      * Connects to a server. Accepts either a {@linkcode ClientTransport}
      * (2026-06-native, request-shaped) or a legacy pipe {@linkcode Transport}
@@ -297,7 +285,7 @@ export class Client extends Dispatcher<ClientContext> {
             const driverOpts: StreamDriverOptions = {
                 supportedProtocolVersions: this._supportedProtocolVersions,
                 debouncedNotificationMethods: this._options?.debouncedNotificationMethods,
-                outboundMw: this._outboundMw
+                taskManager: this._taskManager
             };
             this._ct = channelAsClientTransport(transport, this, driverOpts);
             this._ct.driver!.onclose = () => this.onclose?.();
@@ -354,8 +342,7 @@ export class Client extends Dispatcher<ClientContext> {
                         return resp ?? { jsonrpc: '2.0', id: r.id, error: { code: -32_601, message: 'Method not found' } };
                     },
                     onresponse: r => {
-                        const mw = composeOutboundMiddleware(this._outboundMw);
-                        const consumed = mw.response?.(r, Number(r.id))?.consumed ?? false;
+                        const consumed = this._taskManager.processInboundResponse(r, Number(r.id)).consumed;
                         if (!consumed) this.onerror?.(new Error(`Unmatched response on standalone stream: ${JSON.stringify(r)}`));
                     }
                 });
@@ -699,8 +686,7 @@ export class Client extends Dispatcher<ClientContext> {
                 onresumptiontoken: options?.onresumptiontoken,
                 onnotification: n => void this.dispatchNotification(n).catch(error => this.onerror?.(error)),
                 onresponse: r => {
-                    const mw = composeOutboundMiddleware(this._outboundMw);
-                    const consumed = mw.response?.(r, Number(r.id))?.consumed ?? false;
+                    const consumed = this._taskManager.processInboundResponse(r, Number(r.id)).consumed;
                     if (!consumed) this.onerror?.(new Error(`Unmatched response on stream: ${JSON.stringify(r)}`));
                 },
                 onrequest: async r => {

packages/core/src/shared/context.ts

@@ -7,12 +7,7 @@ import type {
     ElicitRequestFormParams,
     ElicitRequestURLParams,
     ElicitResult,
-    JSONRPCErrorResponse,
     JSONRPCMessage,
-    JSONRPCNotification,
-    JSONRPCRequest,
-    JSONRPCResponse,
-    JSONRPCResultResponse,
     LoggingLevel,
     Notification,
     Progress,
@@ -194,76 +189,6 @@ export interface Outbound {
     sendRaw?(message: JSONRPCMessage, options?: { relatedRequestId?: RequestId }): Promise<void>;
 }
 
-/**
- * Middleware around the request-correlation seam of an {@linkcode Outbound}.
- * Registered via `useOutbound()` on {@linkcode McpServer} / {@linkcode Client};
- * the transport adapter (e.g. {@linkcode StreamDriver}) calls each hook without
- * knowing why a message is queued or consumed.
- *
- * Unlike {@linkcode DispatchMiddleware} (a single function wrapping `next`), the
- * outbound seam has four distinct call sites, so this is a record of optional hooks.
- * Multiple middleware are composed with {@linkcode composeOutboundMiddleware} —
- * first to claim wins for `request`/`notification`/`response`; all `close` run.
- */
-export interface OutboundMiddleware {
-    /** Called before each outbound request hits the wire. Return `queued: true` to suppress the send (caller resolves via `settle`). */
-    request?(
-        jr: JSONRPCRequest,
-        options: RequestOptions | undefined,
-        messageId: number,
-        settle: (r: JSONRPCResultResponse | Error) => void,
-        reject: (e: unknown) => void
-    ): { queued: boolean };
-    /** Called before each outbound notification. May suppress and/or rewrite. */
-    notification?(
-        n: Notification,
-        options: NotificationOptions | undefined
-    ): Promise<{ queued: boolean; jsonrpcNotification?: JSONRPCNotification }>;
-    /** Called for each inbound response before correlation. `consumed: true` swallows it. */
-    response?(r: JSONRPCResponse | JSONRPCErrorResponse, messageId: number): { consumed: boolean; preserveProgress?: boolean };
-    /** Called on connection close. */
-    close?(): void;
-}
-
-/**
- * Composes a list of {@linkcode OutboundMiddleware} into one, registration-order.
- * For `request`/`notification`/`response` the first middleware to claim (queued/consumed)
- * short-circuits the rest; `close` runs all.
- */
-export function composeOutboundMiddleware(mws: OutboundMiddleware[]): OutboundMiddleware {
-    if (mws.length <= 1) return mws[0] ?? {};
-    return {
-        request(jr, opts, id, settle, reject) {
-            for (const mw of mws) {
-                const r = mw.request?.(jr, opts, id, settle, reject);
-                if (r?.queued) return r;
-            }
-            return { queued: false };
-        },
-        async notification(n, opts) {
-            let rewritten: JSONRPCNotification | undefined;
-            for (const mw of mws) {
-                const r = await mw.notification?.(n, opts);
-                if (r?.queued) return r;
-                if (r?.jsonrpcNotification) rewritten = r.jsonrpcNotification;
-            }
-            return { queued: false, jsonrpcNotification: rewritten };
-        },
-        response(r, id) {
-            let preserveProgress = false;
-            for (const mw of mws) {
-                const out = mw.response?.(r, id);
-                if (out?.consumed) return out;
-                if (out?.preserveProgress) preserveProgress = true;
-            }
-            return { consumed: false, preserveProgress };
-        },
-        close() {
-            for (const mw of mws) mw.close?.();
-        }
-    };
-}
-
 /**
  * Base context provided to all request handlers.
  */

packages/core/src/shared/protocol.ts

@@ -20,15 +20,7 @@ import type {
 } from '../types/index.js';
 import { getResultSchema, SUPPORTED_PROTOCOL_VERSIONS } from '../types/index.js';
 import type { AnySchema, SchemaOutput } from '../util/schema.js';
-import type {
-    BaseContext,
-    NotificationOptions,
-    Outbound,
-    OutboundMiddleware,
-    ProtocolOptions,
-    RequestEnv,
-    RequestOptions
-} from './context.js';
+import type { BaseContext, NotificationOptions, Outbound, ProtocolOptions, RequestEnv, RequestOptions } from './context.js';
 import type { DispatchMiddleware } from './dispatcher.js';
 import { Dispatcher } from './dispatcher.js';
 import { StreamDriver } from './streamDriver.js';
@@ -73,31 +65,23 @@ export abstract class Protocol<ContextT extends BaseContext> {
         })();
         this._supportedProtocolVersions = _options?.supportedProtocolVersions ?? SUPPORTED_PROTOCOL_VERSIONS;
         this._ownTaskManager = _options?.tasks ? new TaskManager(_options.tasks) : new NullTaskManager();
-        const omw = this._ownTaskManager.attachTo(this._dispatcher, {
+        this._ownTaskManager.attachTo(this._dispatcher, {
             channel: () => this._outbound,
             reportError: e => this.onerror?.(e),
             enforceStrictCapabilities: this._options?.enforceStrictCapabilities === true,
             assertTaskCapability: m => this.assertTaskCapability(m),
             assertTaskHandlerCapability: m => this.assertTaskHandlerCapability(m)
         });
-        this._outboundMw.push(omw);
     }
 
     private readonly _ownTaskManager: TaskManager;
-    private readonly _outboundMw: OutboundMiddleware[] = [];
 
     /** Register a {@linkcode DispatchMiddleware} on the inner dispatcher. */
     use(mw: DispatchMiddleware): this {
         this._dispatcher.use(mw);
         return this;
     }
 
-    /** Register an {@linkcode OutboundMiddleware} applied at the request-correlation seam. */
-    useOutbound(mw: OutboundMiddleware): this {
-        this._outboundMw.push(mw);
-        return this;
-    }
-
     // ───────────────────────────────────────────────────────────────────────
     // Subclass hooks (v1 signatures)
     // ───────────────────────────────────────────────────────────────────────
@@ -179,7 +163,7 @@ export abstract class Protocol<ContextT extends BaseContext> {
             supportedProtocolVersions: this._supportedProtocolVersions,
             debouncedNotificationMethods: this._options?.debouncedNotificationMethods,
             buildEnv: (extra, base) => ({ ...base, _transportExtra: extra }),
-            outboundMw: this._outboundMw
+            taskManager: this._ownTaskManager
         });
         this._outbound = driver;
         driver.onclose = () => {

packages/core/src/shared/streamDriver.ts

@@ -25,9 +25,10 @@ import {
 } from '../types/index.js';
 import type { AnySchema, SchemaOutput } from '../util/schema.js';
 import { parseSchema } from '../util/schema.js';
-import type { NotificationOptions, Outbound, OutboundMiddleware, ProgressCallback, RequestEnv, RequestOptions } from './context.js';
-import { composeOutboundMiddleware, DEFAULT_REQUEST_TIMEOUT_MSEC } from './context.js';
+import type { NotificationOptions, Outbound, ProgressCallback, RequestEnv, RequestOptions } from './context.js';
+import { DEFAULT_REQUEST_TIMEOUT_MSEC } from './context.js';
 import type { Dispatcher } from './dispatcher.js';
+import type { TaskManager } from './taskManager.js';
 import type { AttachOptions, Transport } from './transport.js';
 
 type TimeoutInfo = {
@@ -48,10 +49,12 @@ export type StreamDriverOptions = {
      */
     buildEnv?: (extra: MessageExtraInfo | undefined, base: RequestEnv) => RequestEnv;
     /**
-     * {@linkcode OutboundMiddleware} hooks invoked at the request-correlation seam.
-     * Composed in registration order via {@linkcode composeOutboundMiddleware}.
+     * Optional {@linkcode TaskManager}. When provided, the driver calls its
+     * `processOutbound*` / `processInboundResponse` / `onClose` hooks at the
+     * request-correlation seam. Inbound task processing happens via the
+     * TaskManager's dispatch middleware (registered by the caller), not here.
      */
-    outboundMw?: OutboundMiddleware[];
+    taskManager?: TaskManager;
 };
 
 /**
@@ -70,7 +73,7 @@ export class StreamDriver implements Outbound {
     private _pendingDebouncedNotifications = new Set<string>();
     private _closed = false;
     private _supportedProtocolVersions: string[];
-    private _mw: OutboundMiddleware;
+    private _taskManager?: TaskManager;
 
     onclose?: () => void;
     onerror?: (error: Error) => void;
@@ -82,7 +85,7 @@ export class StreamDriver implements Outbound {
         private _options: StreamDriverOptions = {}
     ) {
         this._supportedProtocolVersions = _options.supportedProtocolVersions ?? SUPPORTED_PROTOCOL_VERSIONS;
-        this._mw = composeOutboundMiddleware(_options.outboundMw ?? []);
+        this._taskManager = _options.taskManager;
     }
 
     /** {@linkcode Outbound.removeProgressHandler}. */
@@ -206,14 +209,14 @@ export class StreamDriver implements Outbound {
             );
 
             let queued = false;
-            if (this._mw.request) {
+            if (this._taskManager) {
                 const sideChannelResponse = (resp: JSONRPCResultResponse | Error) => {
                     const h = this._responseHandlers.get(messageId);
                     if (h) h(resp);
                     else this._onerror(new Error(`Response handler missing for side-channeled request ${messageId}`));
                 };
                 try {
-                    queued = this._mw.request(jsonrpcRequest, options, messageId, sideChannelResponse, error => {
+                    queued = this._taskManager.processOutboundRequest(jsonrpcRequest, options, messageId, sideChannelResponse, error => {
                         this._progressHandlers.delete(messageId);
                         reject(error);
                     }).queued;
@@ -243,9 +246,9 @@ export class StreamDriver implements Outbound {
      * Sends a notification over the pipe. Supports debouncing per the constructor option.
      */
     async notification(notification: Notification, options?: NotificationOptions): Promise<void> {
-        const intercepted = await this._mw.notification?.(notification, options);
-        if (intercepted?.queued || this._closed) return;
-        const jsonrpc: JSONRPCNotification = intercepted?.jsonrpcNotification ?? {
+        const taskResult = await this._taskManager?.processOutboundNotification(notification, options);
+        if (taskResult?.queued || this._closed) return;
+        const jsonrpc: JSONRPCNotification = taskResult?.jsonrpcNotification ?? {
             jsonrpc: '2.0',
             method: notification.method,
             params: notification.params
@@ -342,8 +345,8 @@ export class StreamDriver implements Outbound {
 
     private _onresponse(response: JSONRPCResponse | JSONRPCErrorResponse): void {
         const messageId = Number(response.id);
-        const intercepted = this._mw.response?.(response, messageId);
-        if (intercepted?.consumed) return;
+        const taskResult = this._taskManager?.processInboundResponse(response, messageId);
+        if (taskResult?.consumed) return;
 
         const handler = this._responseHandlers.get(messageId);
         if (handler === undefined) {
@@ -352,7 +355,7 @@ export class StreamDriver implements Outbound {
         }
         this._responseHandlers.delete(messageId);
         this._cleanupTimeout(messageId);
-        if (!intercepted?.preserveProgress) {
+        if (!taskResult?.preserveProgress) {
             this._progressHandlers.delete(messageId);
         }
         if (isJSONRPCResultResponse(response)) {
@@ -367,7 +370,7 @@ export class StreamDriver implements Outbound {
         const responseHandlers = this._responseHandlers;
         this._responseHandlers = new Map();
         this._progressHandlers.clear();
-        this._mw.close?.();
+        this._taskManager?.onClose();
         this._pendingDebouncedNotifications.clear();
         for (const info of this._timeoutInfo.values()) clearTimeout(info.timeoutId);
         this._timeoutInfo.clear();
@@ -436,7 +439,7 @@ export async function attachChannelTransport(
     const driver = new StreamDriver(dispatcher, pipe, {
         supportedProtocolVersions: options?.supportedProtocolVersions,
         debouncedNotificationMethods: options?.debouncedNotificationMethods,
-        outboundMw: options?.outboundMw,
+        taskManager: options?.taskManager,
         buildEnv: options?.buildEnv
     });
     if (options?.onclose || options?.onerror) {