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/fix-streamable-connection-close' into ihrpr/examples

Browse source
This commit is contained in:
ihrpr 2025-04-09 10:15:07 +01:00
commit 6436c0756c
No known key found for this signature in database
3 changed files with 86 additions and 72 deletions
Download patch file Download diff file Expand all files Collapse all files

28
src/client/streamableHttp.test.ts
View file

@ -161,7 +161,10 @@ describe("StreamableHTTPClientTransport", () => {
statusText: "Method Not Allowed"
});
// We expect the 405 error to be caught and handled gracefully
// This should not throw an error that breaks the transport
await transport.start();
await expect(transport.openSseStream()).rejects.toThrow('Failed to open SSE stream: Method Not Allowed');
// Check that GET was attempted
expect(global.fetch).toHaveBeenCalledWith(
@ -206,6 +209,7 @@ describe("StreamableHTTPClientTransport", () => {
transport.onmessage = messageSpy;
await transport.start();
await transport.openSseStream();
// Give time for the SSE event to be processed
await new Promise(resolve => setTimeout(resolve, 50));
@ -233,7 +237,7 @@ describe("StreamableHTTPClientTransport", () => {
(global.fetch as jest.Mock)
.mockResolvedValueOnce({
ok: true,
ok: true,
status: 200,
headers: new Headers({ "content-type": "text/event-stream" }),
body: makeStream("request1")
@ -255,16 +259,21 @@ describe("StreamableHTTPClientTransport", () => {
]);
// Give time for SSE processing
await new Promise(resolve => setTimeout(resolve, 50));
await new Promise(resolve => setTimeout(resolve, 100));
// Both streams should have delivered their messages
expect(messageSpy).toHaveBeenCalledTimes(2);
expect(messageSpy).toHaveBeenCalledWith(
expect.objectContaining({ result: { id: "request1" }, id: "request1" })
);
expect(messageSpy).toHaveBeenCalledWith(
expect.objectContaining({ result: { id: "request2" }, id: "request2" })
);
// Verify received messages without assuming specific order
expect(messageSpy.mock.calls.some(call => {
const msg = call[0];
return msg.id === "request1" && msg.result?.id === "request1";
})).toBe(true);
expect(messageSpy.mock.calls.some(call => {
const msg = call[0];
return msg.id === "request2" && msg.result?.id === "request2";
})).toBe(true);
});
it("should include last-event-id header when resuming a broken connection", async () => {
@ -286,6 +295,7 @@ describe("StreamableHTTPClientTransport", () => {
});
await transport.start();
await transport.openSseStream();
await new Promise(resolve => setTimeout(resolve, 50));
// Now simulate attempting to reconnect
@ -296,7 +306,7 @@ describe("StreamableHTTPClientTransport", () => {
body: null
});
await transport.start();
await transport.openSseStream();
// Check that Last-Event-ID was included
const calls = (global.fetch as jest.Mock).mock.calls;

122
src/client/streamableHttp.ts
View file

@ -1,13 +1,11 @@
import { Transport } from "../shared/transport.js";
import { JSONRPCMessage, JSONRPCMessageSchema } from "../types.js";
import { auth, AuthResult, OAuthClientProvider, UnauthorizedError } from "./auth.js";
import { type ErrorEvent } from "eventsource";
import { EventSourceMessage, EventSourceParserStream } from 'eventsource-parser/stream';
export class StreamableHTTPError extends Error {
constructor(
public readonly code: number | undefined,
message: string | undefined,
public readonly event: ErrorEvent,
) {
super(`Streamable HTTP error: ${message}`);
}
@ -83,7 +81,7 @@ export class StreamableHTTPClientTransport implements Transport {
throw new UnauthorizedError();
}
return await this._startOrAuth();
return await this._startOrAuthStandaloneSSE();
}
private async _commonHeaders(): Promise<HeadersInit> {
@ -102,7 +100,7 @@ export class StreamableHTTPClientTransport implements Transport {
return headers;
}
private async _startOrAuth(): Promise<void> {
private async _startOrAuthStandaloneSSE(): Promise<void> {
try {
// Try to open an initial SSE stream with GET to listen for server messages
// This is optional according to the spec - server may not support it
@ -121,25 +119,22 @@ export class StreamableHTTPClientTransport implements Transport {
signal: this._abortController?.signal,
});
if (response.status === 405 || response.status === 404) {
// Server doesn't support GET for SSE, which is allowed by the spec
// We'll rely on SSE responses to POST requests for communication
return;
}
if (!response.ok) {
if (response.status === 401 && this._authProvider) {
// Need to authenticate
return await this._authThenStart();
}
const error = new Error(`Failed to open SSE stream: ${response.status} ${response.statusText}`);
const error = new StreamableHTTPError(
response.status,
`Failed to open SSE stream: ${response.statusText}`,
);
this.onerror?.(error);
throw error;
}
// Successful connection, handle the SSE stream as a standalone listener
const streamId = `initial-${Date.now()}`;
const streamId = `standalone-sse-${Date.now()}`;
this._handleSseStream(response.body, streamId);
} catch (error) {
this.onerror?.(error as Error);
@ -147,6 +142,53 @@ export class StreamableHTTPClientTransport implements Transport {
}
}
private _handleSseStream(stream: ReadableStream<Uint8Array> | null, streamId: string): 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;
}
// Update last event ID if provided
if (event.id) {
this._lastEventId = event.id;
}
// Handle message events (default event type is undefined per docs)
// or explicit 'message' event type
if (!event.event || event.event === 'message') {
try {
const message = JSONRPCMessageSchema.parse(JSON.parse(event.data));
this.onmessage?.(message);
} catch (error) {
this.onerror?.(error as Error);
}
}
}
} catch (error) {
this._activeStreams.delete(streamId);
this.onerror?.(error as Error);
}
};
processStream();
}
async start() {
if (this._activeStreams.size > 0) {
throw new Error(
@ -155,7 +197,6 @@ export class StreamableHTTPClientTransport implements Transport {
}
this._abortController = new AbortController();
return await this._startOrAuth();
}
/**
@ -271,50 +312,17 @@ export class StreamableHTTPClientTransport implements Transport {
}
}
private _handleSseStream(stream: ReadableStream<Uint8Array> | null, streamId: string): void {
if (!stream) {
return;
/**
* Opens SSE stream to receive messages from the server.
*
* This allows the server to push messages to the client without requiring the client
* to first send a request via HTTP POST. Some servers may not support this feature.
* If authentication is required but fails, this method will throw an UnauthorizedError.
*/
async openSseStream(): Promise<void> {
if (!this._abortController) {
this._abortController = new AbortController();
}
// 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;
}
// Update last event ID if provided
if (event.id) {
this._lastEventId = event.id;
}
// Handle message events (default event type is undefined per docs)
// or explicit 'message' event type
if (!event.event || event.event === 'message') {
try {
const message = JSONRPCMessageSchema.parse(JSON.parse(event.data));
this.onmessage?.(message);
} catch (error) {
this.onerror?.(error as Error);
}
}
}
} catch (error) {
this._activeStreams.delete(streamId);
this.onerror?.(error as Error);
}
};
processStream();
await this._startOrAuthStandaloneSSE();
}
}

8
src/server/streamableHttp.ts
View file

@ -359,13 +359,9 @@ export class StreamableHTTPServerTransport implements Transport {
async send(message: JSONRPCMessage, options?: { relatedRequestId?: RequestId }): Promise<void> {
let requestId = options?.relatedRequestId;
let shouldCloseConnection = false;
if ('result' in message || 'error' in message) {
// If the message is a response, use the request ID from the message
requestId = message.id;
// This is a response to the original request, we can close the stream
// after sending all related responses
shouldCloseConnection = true;
}
if (requestId === undefined) {
throw new Error("No request ID provided for the message");
@ -380,8 +376,8 @@ export class StreamableHTTPServerTransport implements Transport {
sseResponse.write(
`event: message\ndata: ${JSON.stringify(message)}\n\n`,
);
if (shouldCloseConnection) {
// After all JSON-RPC responses have been sent, the server SHOULD close the SSE stream.
if ('result' in message || 'error' in message) {
this._sseResponseMapping.delete(requestId);
// Only close the connection if it's not needed by other requests
const canCloseConnection = ![...this._sseResponseMapping.entries()].some(([id, res]) => res === sseResponse && id !== requestId);