diff --git a/.changeset/quiet-stream-provenance.md b/.changeset/quiet-stream-provenance.md new file mode 100644 index 0000000000..b04cb9c2f6 --- /dev/null +++ b/.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. diff --git a/docs/clients/server-requests.md b/docs/clients/server-requests.md index 8d3c75a45b..6b2ea8f721 100644 --- a/docs/clients/server-requests.md +++ b/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. diff --git a/packages/client/src/client/streamableHttp.ts b/packages/client/src/client/streamableHttp.ts index ace0663158..390f6870bf 100644 --- a/packages/client/src/client/streamableHttp.ts +++ b/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,7 +719,7 @@ 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, @@ -721,7 +727,7 @@ export class StreamableHTTPClientTransport implements Transport { // 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 }, diff --git a/packages/client/test/client/streamProvenanceContext.test.ts b/packages/client/test/client/streamProvenanceContext.test.ts new file mode 100644 index 0000000000..f96b9c71b2 --- /dev/null +++ b/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 {} + + async close(): Promise { + this.onclose?.(); + } + + async send(message: JSONRPCMessage): Promise { + 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(); + }); +}); diff --git a/packages/client/test/client/streamableHttp.test.ts b/packages/client/test/client/streamableHttp.test.ts index a36bbc0ad3..e5025df910 100644 --- a/packages/client/test/client/streamableHttp.test.ts +++ b/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; + }; + 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 diff --git a/packages/core-internal/src/shared/protocol.ts b/packages/core-internal/src/shared/protocol.ts index 637be389aa..59f2fb99f5 100644 --- a/packages/core-internal/src/shared/protocol.ts +++ b/packages/core-internal/src/shared/protocol.ts @@ -339,6 +339,13 @@ export type BaseContext = { */ id: RequestId; + /** + * The client request whose Streamable HTTP response stream carried + * this server-initiated request, when the transport supplied one. + * Standalone GET streams and non-HTTP transports leave this unset. + */ + relatedRequestId?: RequestId; + /** * The method name of the request (e.g., 'tools/call', 'ping'). */ @@ -1056,6 +1063,7 @@ export abstract class Protocol { mcpReq: { id: request.id, method: request.method, + ...(extra?.relatedRequestId !== undefined && { relatedRequestId: extra.relatedRequestId }), _meta: request.params?._meta, ...(lifted.envelope !== undefined && { envelope: lifted.envelope }), ...(partitionedInputResponses !== undefined && { inputResponses: partitionedInputResponses.accepted }), diff --git a/packages/core-internal/src/types/types.ts b/packages/core-internal/src/types/types.ts index f2bc9d67fc..6ac0ecc32b 100644 --- a/packages/core-internal/src/types/types.ts +++ b/packages/core-internal/src/types/types.ts @@ -888,6 +888,13 @@ export interface MessageClassification { * Extra information about a message. */ export interface MessageExtraInfo { + /** + * The client request whose Streamable HTTP response stream carried this message. + * Set by the client transport for server-initiated requests received on a + * per-request SSE stream; absent for the standalone GET stream. + */ + relatedRequestId?: RequestId; + /** * The original HTTP request. */