modelcontextprotocol/typescript-sdk
Watch
1
0
Fork
You've already forked typescript-sdk
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

Merge branch 'ihrpr/streamable-http-client' into ihrpr/fix-streamable-connection-close

Browse source
This commit is contained in:
ihrpr 2025-04-09 11:15:08 +01:00
commit 25844aca8e
No known key found for this signature in database
1 changed files with 7 additions and 23 deletions
Download patch file Download diff file Expand all files Collapse all files

30
src/client/streamableHttp.ts
View file

@ -43,7 +43,6 @@ export type StreamableHTTPClientTransportOptions = {
* for receiving messages.
*/
export class StreamableHTTPClientTransport implements Transport {
private _activeStreams: Map<string, ReadableStreamDefaultReader<EventSourceMessage>> = new Map();
private _abortController?: AbortController;
private _url: URL;
private _requestInit?: RequestInit;
@ -134,33 +133,28 @@ export class StreamableHTTPClientTransport implements Transport {
}
// Successful connection, handle the SSE stream as a standalone listener
const streamId = `standalone-sse-${Date.now()}`;
this._handleSseStream(response.body, streamId);
this._handleSseStream(response.body);
} catch (error) {
this.onerror?.(error as Error);
throw error;
}
}
private _handleSseStream(stream: ReadableStream<Uint8Array> | null, streamId: string): void {
private _handleSseStream(stream: ReadableStream<Uint8Array> | null): void {
if (!stream) {
return;
}
// Create a pipeline: binary stream -> text decoder -> SSE parser
const eventStream = stream
.pipeThrough(new TextDecoderStream())
.pipeThrough(new EventSourceParserStream());
const reader = eventStream.getReader();
this._activeStreams.set(streamId, reader);
const processStream = async () => {
try {
while (true) {
const { done, value: event } = await reader.read();
if (done) {
this._activeStreams.delete(streamId);
break;
}
@ -181,7 +175,6 @@ export class StreamableHTTPClientTransport implements Transport {
}
}
} catch (error) {
this._activeStreams.delete(streamId);
this.onerror?.(error as Error);
}
};
@ -190,7 +183,7 @@ export class StreamableHTTPClientTransport implements Transport {
}
async start() {
if (this._activeStreams.size > 0) {
if (this._abortController) {
throw new Error(
"StreamableHTTPClientTransport already started! If using Client class, note that connect() calls start() automatically.",
);
@ -214,16 +207,6 @@ export class StreamableHTTPClientTransport implements Transport {
}
async close(): Promise<void> {
// Close all active streams
for (const reader of this._activeStreams.values()) {
try {
reader.cancel();
} catch (error) {
this.onerror?.(error as Error);
}
}
this._activeStreams.clear();
// Abort any pending requests
this._abortController?.abort();
@ -292,8 +275,7 @@ export class StreamableHTTPClientTransport implements Transport {
if (hasRequests) {
if (contentType?.includes("text/event-stream")) {
// For streaming responses, create a unique stream ID based on request IDs
const streamId = `req-${requestIds.join('-')}-${Date.now()}`;
this._handleSseStream(response.body, streamId);
this._handleSseStream(response.body);
} else if (contentType?.includes("application/json")) {
// For non-streaming servers, we might get direct JSON responses
const data = await response.json();
@ -321,7 +303,9 @@ export class StreamableHTTPClientTransport implements Transport {
*/
async openSseStream(): Promise<void> {
if (!this._abortController) {
this._abortController = new AbortController();
throw new Error(
"StreamableHTTPClientTransport not started! Call connect() before openSseStream().",
);
}
await this._startOrAuthStandaloneSSE();
}