pull transport send options into own named type

This commit is contained in:
ihrpr 2025-04-15 11:42:16 +01:00
commit 1359e9fe6a
No known key found for this signature in database
5 changed files with 43 additions and 17 deletions

View file

@ -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<void> {
async send(message: JSONRPCMessage | JSONRPCMessage[], options?: { resumptionToken?: string, onresumptiontoken?: (event: string) => void }): Promise<void> {
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));

View file

@ -302,7 +302,7 @@ async function callTool(name: string, args: Record<string, unknown>): Promise<vo
notificationsToolLastEventId = event;
};
const result = await client.request(request, CallToolResultSchema, {
lastEventId: notificationsToolLastEventId, onLastEventIdUpdate
resumptionToken: notificationsToolLastEventId, onresumptiontoken: onLastEventIdUpdate
});
console.log('Tool result:');

View file

@ -247,8 +247,8 @@ describe('Transport resumability', () => {
}
}
}, 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

View file

@ -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<z.infer<T>> {
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);
});

View file

@ -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<void>;
send(message: JSONRPCMessage, options?: TransportSendOptions): Promise<void>;
/**
* Closes the connection.