From 1359e9fe6a69913a450a5cc88ebb480af8551f51 Mon Sep 17 00:00:00 2001 From: ihrpr Date: Tue, 15 Apr 2025 11:42:16 +0100 Subject: [PATCH] pull transport send options into own named type --- src/client/streamableHttp.ts | 6 +++-- src/examples/client/simpleStreamableHttp.ts | 2 +- .../taskResumability.test.ts | 8 +++--- src/shared/protocol.ts | 25 +++++++++++-------- src/shared/transport.ts | 23 ++++++++++++++++- 5 files changed, 45 insertions(+), 19 deletions(-) diff --git a/src/client/streamableHttp.ts b/src/client/streamableHttp.ts index 63677d55..c1c93ebe 100644 --- a/src/client/streamableHttp.ts +++ b/src/client/streamableHttp.ts @@ -88,6 +88,7 @@ export type StreamableHTTPClientTransportOptions = { * Options to configure the reconnection behavior. */ reconnectionOptions?: StreamableHTTPReconnectionOptions; + /** * Session ID for the connection. This is used to identify the session on the server. * When not provided and connecting to a server that supports session IDs, the server will generate a new session ID. @@ -345,10 +346,11 @@ export class StreamableHTTPClientTransport implements Transport { this.onclose?.(); } - async send(message: JSONRPCMessage | JSONRPCMessage[], options?: { lastEventId?: string, onLastEventIdUpdate?: (event: string) => void }): Promise { + async send(message: JSONRPCMessage | JSONRPCMessage[], options?: { resumptionToken?: string, onresumptiontoken?: (event: string) => void }): Promise { try { // If client passes in a lastEventId in the request options, we need to reconnect the SSE stream - const { lastEventId, onLastEventIdUpdate } = options ?? {}; + const lastEventId = options?.resumptionToken + const onLastEventIdUpdate = options?.onresumptiontoken; if (lastEventId) { // If we have at last event ID, we need to reconnect the SSE stream this._startOrAuthSse({ lastEventId }).catch(err => this.onerror?.(err)); diff --git a/src/examples/client/simpleStreamableHttp.ts b/src/examples/client/simpleStreamableHttp.ts index 922aa0ce..6cfad987 100644 --- a/src/examples/client/simpleStreamableHttp.ts +++ b/src/examples/client/simpleStreamableHttp.ts @@ -302,7 +302,7 @@ async function callTool(name: string, args: Record): Promise { } } }, CallToolResultSchema, { - lastEventId, - onLastEventIdUpdate + resumptionToken: lastEventId, + onresumptiontoken: onLastEventIdUpdate }); // Wait for some notifications to arrive (not all) @@ -311,8 +311,8 @@ describe('Transport resumability', () => { } } }, CallToolResultSchema, { - lastEventId, // Pass the lastEventId from the previous session - onLastEventIdUpdate + resumptionToken: lastEventId, // Pass the lastEventId from the previous session + onresumptiontoken: onLastEventIdUpdate }); // Verify we eventually received at leaset a few motifications diff --git a/src/shared/protocol.ts b/src/shared/protocol.ts index bc7882e4..3b098193 100644 --- a/src/shared/protocol.ts +++ b/src/shared/protocol.ts @@ -87,17 +87,20 @@ export type RequestOptions = { * May be used to indicate to the transport which incoming request to associate this outgoing request with. */ relatedRequestId?: RequestId; - /** - * The last event ID to associate with the request. - * Used to resume long-running requests that were interrupted when creating a new client instance. - */ - lastEventId?: string; /** - * A callback that is invoked when the last event ID is updated. - * This is used to notify the client that the last event ID has changed, so the client can update its state accordingly. + * The resumption token used to continue long-running requests that were interrupted. + * + * This allows clients to reconnect and continue from where they left off, if supported by the transport. */ - onLastEventIdUpdate?: (event: string) => void; + resumptionToken?: string; + + /** + * A callback that is invoked when the resumption token changes, if supported by the transport. + * + * This allows clients to persist the latest token for potential reconnection. + */ + onresumptiontoken?: (token: string) => void; }; /** @@ -512,7 +515,7 @@ export abstract class Protocol< resultSchema: T, options?: RequestOptions, ): Promise> { - const { relatedRequestId, lastEventId, onLastEventIdUpdate } = options ?? {}; + const { relatedRequestId, resumptionToken, onresumptiontoken } = options ?? {}; return new Promise((resolve, reject) => { if (!this._transport) { @@ -554,7 +557,7 @@ export abstract class Protocol< requestId: messageId, reason: String(reason), }, - }, { relatedRequestId, lastEventId, onLastEventIdUpdate }) + }, { relatedRequestId, resumptionToken, onresumptiontoken }) .catch((error) => this._onerror(new Error(`Failed to send cancellation: ${error}`)), ); @@ -592,7 +595,7 @@ export abstract class Protocol< this._setupTimeout(messageId, timeout, options?.maxTotalTimeout, timeoutHandler, options?.resetTimeoutOnProgress ?? false); - this._transport.send(jsonrpcRequest, { relatedRequestId, lastEventId, onLastEventIdUpdate }).catch((error) => { + this._transport.send(jsonrpcRequest, { relatedRequestId, resumptionToken, onresumptiontoken }).catch((error) => { this._cleanupTimeout(messageId); reject(error); }); diff --git a/src/shared/transport.ts b/src/shared/transport.ts index cfbe8461..601a86af 100644 --- a/src/shared/transport.ts +++ b/src/shared/transport.ts @@ -1,5 +1,26 @@ import { JSONRPCMessage, RequestId } from "../types.js"; +/** + * Options for sending a JSON-RPC message. + */ +export interface TransportSendOptions { + /** + * If present, `relatedRequestId` is used to indicate to the transport which incoming request to associate this outgoing message with. + */ + relatedRequestId?: RequestId; + /** + * The resumption token used to continue long-running requests that were interrupted. + * + * This allows clients to reconnect and continue from where they left off, if supported by the transport. + */ + resumptionToken?: string; + /** + * A callback that is invoked when the resumption token changes, if supported by the transport. + * + * This allows clients to persist the latest token for potential reconnection. + */ + onresumptiontoken?: (token: string) => void; +} /** * Describes the minimal contract for a MCP transport that a client or server can communicate over. */ @@ -18,7 +39,7 @@ export interface Transport { * * If present, `relatedRequestId` is used to indicate to the transport which incoming request to associate this outgoing message with. */ - send(message: JSONRPCMessage, options?: { relatedRequestId?: RequestId, lastEventId?: string, onLastEventIdUpdate?: (event: string) => void }): Promise; + send(message: JSONRPCMessage, options?: TransportSendOptions): Promise; /** * Closes the connection.