fix(client): preserve Streamable HTTP request provenance

.changeset/quiet-stream-provenance.md

@@ -0,0 +1,5 @@
+---
+"@modelcontextprotocol/client": patch
+---
+
+Expose the originating client request ID for server-initiated requests received on a Streamable HTTP response stream.

docs/clients/server-requests.md

@@ -81,6 +81,22 @@ Sampling request: { type: 'text', text: 'Summarize this order: 1 Travel mug to L
 [ { type: 'text', text: 'host-model: One travel mug to Lisbon.' } ]
 ```
 
+## Associate a Streamable HTTP request with its parent
+
+When a server-initiated request arrives on a Streamable HTTP response stream, the handler context
+includes `ctx.mcpReq.relatedRequestId`: the JSON-RPC id of the client request whose stream carried
+it. Use it to associate an elicitation, sampling request, or roots request with the operation that
+started it. The field is absent for standalone GET messages and transports that do not provide a
+stream association.
+
+```ts
+client.setRequestHandler('elicitation/create', async (_request, ctx) => {
+    const parentRequestId = ctx.mcpReq.relatedRequestId;
+    console.log('Elicitation belongs to:', parentRequestId ?? 'no associated request');
+    return { action: 'accept' };
+});
+```
+
 ## Register each handler once
 
 Register each handler once, on the `Client` you construct. The same handler answers a request the server pushes to your client and a request the SDK fulfils for you inside a `callTool()` round — your code never sees the difference.

packages/client/src/client/streamableHttp.ts

@@ -1,6 +1,6 @@
 import type { ReadableWritablePair } from 'node:stream/web';
 
-import type { FetchLike, JSONRPCMessage, Transport } from '@modelcontextprotocol/core-internal';
+import type { FetchLike, JSONRPCMessage, MessageExtraInfo, RequestId, Transport } from '@modelcontextprotocol/core-internal';
 import {
     createFetchWithInit,
     encodeMcpParamValue,
@@ -54,6 +54,12 @@ const DEFAULT_STREAMABLE_HTTP_RECONNECTION_OPTIONS: StreamableHTTPReconnectionOp
  * Options for starting or authenticating an SSE connection
  */
 export interface StartSSEOptions {
+    /**
+     * The client request whose POST response stream carries this SSE stream.
+     * Standalone GET streams have no related request.
+     */
+    relatedRequestId?: RequestId;
+
     /**
      * The resumption token used to continue long-running requests that were interrupted.
      *
@@ -330,7 +336,7 @@ export class StreamableHTTPClientTransport implements Transport {
 
     onclose?: () => void;
     onerror?: (error: Error) => void;
-    onmessage?: (message: JSONRPCMessage) => void;
+    onmessage?: (message: JSONRPCMessage, extra?: MessageExtraInfo) => void;
 
     /**
      * Streamable HTTP opens one POST (and SSE response stream) per outbound
@@ -713,15 +719,15 @@ export class StreamableHTTPClientTransport implements Transport {
             options.onRequestStreamEnd?.();
             return;
         }
-        const { onresumptiontoken, replayMessageId, requestSignal, onRequestStreamEnd } = options;
+        const { onresumptiontoken, replayMessageId, relatedRequestId, requestSignal, onRequestStreamEnd } = options;
         // An intentional abort — transport-wide close OR a per-request abort
         // (McpSubscription.close() aborting its `requestSignal`) — must read as
         // a clean shutdown: no misleading "SSE stream disconnected" onerror,
         // and no GET+Last-Event-ID reconnect that would resurrect a stream the
         // caller just tore down.
         const isIntentionalAbort = (): boolean => this._abortController?.signal.aborted === true || requestSignal?.aborted === true;
 
-        let lastEventId: string | undefined;
+        let lastEventId: string | undefined = options.resumptionToken;
         // Track whether we've received a priming event (event with ID)
         // Per spec, server SHOULD send a priming event with ID before closing
         let hasPrimingEvent = false;
@@ -775,7 +781,11 @@ export class StreamableHTTPClientTransport implements Transport {
                                     message.id = replayMessageId;
                                 }
                             }
-                            this.onmessage?.(message);
+                            if (relatedRequestId === undefined || !isJSONRPCRequest(message)) {
+                                this.onmessage?.(message);
+                            } else {
+                                this.onmessage?.(message, { relatedRequestId });
+                            }
                         } catch (error) {
                             this.onerror?.(error as Error);
                         }
@@ -794,6 +804,7 @@ export class StreamableHTTPClientTransport implements Transport {
                             resumptionToken: lastEventId,
                             onresumptiontoken,
                             replayMessageId,
+                            relatedRequestId,
                             requestSignal,
                             onRequestStreamEnd
                         },
@@ -827,6 +838,7 @@ export class StreamableHTTPClientTransport implements Transport {
                                 resumptionToken: lastEventId,
                                 onresumptiontoken,
                                 replayMessageId,
+                                relatedRequestId,
                                 requestSignal,
                                 onRequestStreamEnd
                             },
@@ -954,10 +966,18 @@ export class StreamableHTTPClientTransport implements Transport {
                 // same per-request abort as the original POST — modern-era
                 // cancel-via-stream-close routes through `requestSignal`, and
                 // without it a resumed long-running request would not cancel.
+                // `relatedRequestId` rides along for the same reason: the
+                // resumed GET continues *this* request's stream, so a server
+                // request replayed on it keeps the provenance the original
+                // POST stream would have carried.
+                const resumedRequestId = isJSONRPCRequest(message) ? message.id : undefined;
                 this._startOrAuthSse({
                     resumptionToken,
-                    replayMessageId: isJSONRPCRequest(message) ? message.id : undefined,
-                    requestSignal: options?.requestSignal
+                    onresumptiontoken,
+                    replayMessageId: resumedRequestId,
+                    relatedRequestId: resumedRequestId,
+                    requestSignal: options?.requestSignal,
+                    onRequestStreamEnd: options?.onRequestStreamEnd
                 }).catch(error => this.onerror?.(error));
                 return;
             }
@@ -1120,7 +1140,9 @@ export class StreamableHTTPClientTransport implements Transport {
             // Get original message(s) for detecting request IDs
             const messages = Array.isArray(message) ? message : [message];
 
-            const hasRequests = messages.some(msg => 'method' in msg && 'id' in msg && msg.id !== undefined);
+            const requests = messages.filter(msg => isJSONRPCRequest(msg));
+            const hasRequests = requests.length > 0;
+            const relatedRequestId = messages.length === 1 && requests.length === 1 ? requests[0]!.id : undefined;
 
             // Check the response type (parsed media type — see mediaTypeEssence)
             const contentType = response.headers.get('content-type');
@@ -1135,6 +1157,7 @@ export class StreamableHTTPClientTransport implements Transport {
                         response.body,
                         {
                             onresumptiontoken,
+                            relatedRequestId,
                             requestSignal: options?.requestSignal,
                             onRequestStreamEnd: options?.onRequestStreamEnd
                         },

packages/client/test/client/streamProvenanceContext.test.ts

@@ -0,0 +1,59 @@
+import type { JSONRPCMessage, MessageExtraInfo, Transport } from '@modelcontextprotocol/core-internal';
+import { isJSONRPCRequest } from '@modelcontextprotocol/core-internal';
+import { describe, expect, it } from 'vitest';
+
+import { Client } from '../../src/client/client';
+
+class ScriptedTransport implements Transport {
+    onclose?: () => void;
+    onerror?: (error: Error) => void;
+    onmessage?: (message: JSONRPCMessage, extra?: MessageExtraInfo) => void;
+    sent: JSONRPCMessage[] = [];
+
+    async start(): Promise<void> {}
+
+    async close(): Promise<void> {
+        this.onclose?.();
+    }
+
+    async send(message: JSONRPCMessage): Promise<void> {
+        this.sent.push(message);
+        if (isJSONRPCRequest(message) && message.method === 'initialize') {
+            queueMicrotask(() =>
+                this.onmessage?.({
+                    jsonrpc: '2.0',
+                    id: message.id,
+                    result: {
+                        protocolVersion: '2025-11-25',
+                        capabilities: {},
+                        serverInfo: { name: 'scripted-server', version: '1.0.0' }
+                    }
+                })
+            );
+        }
+    }
+
+    emit(message: JSONRPCMessage, extra?: MessageExtraInfo): void {
+        this.onmessage?.(message, extra);
+    }
+}
+
+describe('Streamable HTTP provenance in client request context', () => {
+    it('exposes relatedRequestId to the client request handler', async () => {
+        const transport = new ScriptedTransport();
+        const client = new Client({ name: 'provenance-client', version: '1.0.0' }, { capabilities: { roots: { listChanged: false } } });
+        let relatedRequestId: string | number | undefined;
+
+        client.setRequestHandler('roots/list', async (_request, context) => {
+            relatedRequestId = context.mcpReq.relatedRequestId;
+            return { roots: [] };
+        });
+
+        await client.connect(transport);
+        transport.emit({ jsonrpc: '2.0', id: 'server-request-1', method: 'roots/list', params: {} }, { relatedRequestId: 'tool-call-1' });
+        await new Promise(resolve => setTimeout(resolve, 0));
+
+        expect(relatedRequestId).toBe('tool-call-1');
+        await client.close();
+    });
+});

packages/client/test/client/streamableHttp.test.ts

@@ -524,6 +524,188 @@ describe('StreamableHTTPClientTransport', () => {
         ).toBe(true);
     });
 
+    it('attributes server requests received on a POST SSE stream to the originating request', async () => {
+        const encoder = new TextEncoder();
+        const stream = new ReadableStream({
+            start(controller) {
+                controller.enqueue(
+                    encoder.encode(
+                        'event: message\ndata: {"jsonrpc":"2.0","id":"elicitation-1","method":"elicitation/create","params":{}}\n\n'
+                    )
+                );
+            }
+        });
+
+        (globalThis.fetch as Mock).mockResolvedValueOnce({
+            ok: true,
+            status: 200,
+            headers: new Headers({ 'content-type': 'text/event-stream' }),
+            body: stream
+        });
+
+        const messageSpy = vi.fn();
+        transport.onmessage = messageSpy;
+
+        await transport.send({ jsonrpc: '2.0', id: 'tool-call-1', method: 'tools/call', params: { name: 'route' } });
+        await new Promise(resolve => setTimeout(resolve, 50));
+
+        expect(messageSpy).toHaveBeenCalledWith(expect.objectContaining({ id: 'elicitation-1', method: 'elicitation/create' }), {
+            relatedRequestId: 'tool-call-1'
+        });
+    });
+
+    it('keeps that attribution when the request is resumed with a resumption token', async () => {
+        // Resuming an interrupted request replaces the POST response stream
+        // with a `Last-Event-ID` GET, but the stream still belongs to the same
+        // client request — so provenance has to survive the swap. This is the
+        // long-running-call case: the stream drops, the client resumes, and the
+        // elicitation arrives on the resumed stream.
+        const encoder = new TextEncoder();
+        const stream = new ReadableStream({
+            start(controller) {
+                controller.enqueue(
+                    encoder.encode(
+                        'id: event-43\nevent: message\ndata: {"jsonrpc":"2.0","id":"elicitation-1","method":"elicitation/create","params":{}}\n\n'
+                    )
+                );
+            }
+        });
+
+        (globalThis.fetch as Mock).mockResolvedValueOnce({
+            ok: true,
+            status: 200,
+            headers: new Headers({ 'content-type': 'text/event-stream' }),
+            body: stream
+        });
+
+        const messageSpy = vi.fn();
+        transport.onmessage = messageSpy;
+
+        const resumptionTokenSpy = vi.fn();
+        await transport.send(
+            { jsonrpc: '2.0', id: 'tool-call-1', method: 'tools/call', params: { name: 'route' } },
+            { resumptionToken: 'event-42', onresumptiontoken: resumptionTokenSpy }
+        );
+        await new Promise(resolve => setTimeout(resolve, 50));
+
+        expect(messageSpy).toHaveBeenCalledWith(expect.objectContaining({ id: 'elicitation-1', method: 'elicitation/create' }), {
+            relatedRequestId: 'tool-call-1'
+        });
+        expect(resumptionTokenSpy).toHaveBeenCalledWith('event-43');
+    });
+
+    it('keeps per-request stream callbacks when a resumption-token GET is started directly', async () => {
+        const fetchMock = globalThis.fetch as Mock;
+        const onresumptiontoken = vi.fn();
+        const onRequestStreamEnd = vi.fn();
+        fetchMock.mockResolvedValueOnce({ ok: false, status: 405, headers: new Headers() });
+
+        await transport.start();
+        await transport.send(
+            { jsonrpc: '2.0', id: 'tool-call-1', method: 'tools/call', params: { name: 'route' } },
+            { resumptionToken: 'event-42', onresumptiontoken, onRequestStreamEnd }
+        );
+        await vi.waitFor(() => expect(onRequestStreamEnd).toHaveBeenCalledTimes(1));
+
+        expect(fetchMock.mock.calls[0]![1]?.method).toBe('GET');
+        expect((fetchMock.mock.calls[0]![1]?.headers as Headers).get('last-event-id')).toBe('event-42');
+        expect(onresumptiontoken).not.toHaveBeenCalled();
+    });
+
+    it('keeps the original resumption token when a resumed GET closes before receiving a new event ID', async () => {
+        transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), {
+            reconnectionOptions: {
+                initialReconnectionDelay: 5,
+                maxReconnectionDelay: 100,
+                reconnectionDelayGrowFactor: 1,
+                maxRetries: 1
+            }
+        });
+        const fetchMock = globalThis.fetch as Mock;
+        fetchMock.mockResolvedValueOnce({
+            ok: true,
+            status: 200,
+            headers: new Headers({ 'content-type': 'text/event-stream' }),
+            body: new ReadableStream({
+                start(controller) {
+                    controller.close();
+                }
+            })
+        });
+        fetchMock.mockResolvedValueOnce({ ok: false, status: 405, headers: new Headers() });
+
+        await transport.start();
+        await transport['_startOrAuthSse']({ resumptionToken: 'event-42', relatedRequestId: 'tool-call-1' });
+        await vi.waitFor(() => expect(fetchMock).toHaveBeenCalledTimes(2), { timeout: 250 });
+
+        const reconnectHeaders = fetchMock.mock.calls[1]![1]?.headers as Headers;
+        expect(reconnectHeaders.get('last-event-id')).toBe('event-42');
+    });
+
+    it('does not attribute server requests received on the standalone GET stream', async () => {
+        // Transports spec (2025-03-26 … 2025-11-25) §Listening for Messages:
+        // messages on the standalone GET stream SHOULD be unrelated to any
+        // concurrently-running client request. Attributing one to whatever
+        // request happened to be in flight would be a fabricated relation.
+        const encoder = new TextEncoder();
+        const stream = new ReadableStream({
+            start(controller) {
+                controller.enqueue(
+                    encoder.encode(
+                        'event: message\ndata: {"jsonrpc":"2.0","id":"elicitation-1","method":"elicitation/create","params":{}}\n\n'
+                    )
+                );
+            }
+        });
+
+        (globalThis.fetch as Mock).mockResolvedValueOnce({
+            ok: true,
+            status: 200,
+            headers: new Headers({ 'content-type': 'text/event-stream' }),
+            body: stream
+        });
+
+        const messageSpy = vi.fn();
+        transport.onmessage = messageSpy;
+
+        const transportWithPrivateMethods = transport as unknown as {
+            _startOrAuthSse: (options: StartSSEOptions) => Promise<void>;
+        };
+        await transportWithPrivateMethods._startOrAuthSse({ resumptionToken: undefined });
+        await new Promise(resolve => setTimeout(resolve, 50));
+
+        expect(messageSpy).toHaveBeenCalledTimes(1);
+        expect(messageSpy.mock.calls[0]![1]).toBeUndefined();
+    });
+
+    it('does not attach provenance to the response that terminates a POST SSE stream', async () => {
+        // A response already carries its own correlation — its `id` IS the
+        // originating request. Only server-initiated requests, whose ids come
+        // from the server's own numbering, need the stream to supply it.
+        const encoder = new TextEncoder();
+        const stream = new ReadableStream({
+            start(controller) {
+                controller.enqueue(encoder.encode('event: message\ndata: {"jsonrpc":"2.0","id":"tool-call-1","result":{}}\n\n'));
+            }
+        });
+
+        (globalThis.fetch as Mock).mockResolvedValueOnce({
+            ok: true,
+            status: 200,
+            headers: new Headers({ 'content-type': 'text/event-stream' }),
+            body: stream
+        });
+
+        const messageSpy = vi.fn();
+        transport.onmessage = messageSpy;
+
+        await transport.send({ jsonrpc: '2.0', id: 'tool-call-1', method: 'tools/call', params: { name: 'route' } });
+        await new Promise(resolve => setTimeout(resolve, 50));
+
+        expect(messageSpy).toHaveBeenCalledTimes(1);
+        expect(messageSpy.mock.calls[0]![1]).toBeUndefined();
+    });
+
     it('declares hasPerRequestStream so the protocol layer routes 2026-era cancellation to stream-close', () => {
         // Spec basic/patterns/cancellation §Transport-Specific (2026-07-28):
         // closing the per-request SSE stream IS the cancel signal on