move get sse stream from a separate method to automatic connection when notifications/initialized received

This commit is contained in:
ihrpr 2025-04-11 14:06:51 +01:00
commit 18e4981bd3
No known key found for this signature in database
4 changed files with 23 additions and 41 deletions

View file

@ -164,8 +164,7 @@ describe("StreamableHTTPClientTransport", () => {
// 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");
await expect(transport["_startOrAuthStandaloneSSE"]()).resolves.not.toThrow("Failed to open SSE stream: Method Not Allowed");
// Check that GET was attempted
expect(global.fetch).toHaveBeenCalledWith(
expect.anything(),
@ -209,7 +208,7 @@ describe("StreamableHTTPClientTransport", () => {
transport.onmessage = messageSpy;
await transport.start();
await transport.openSseStream();
await transport["_startOrAuthStandaloneSSE"]();
// Give time for the SSE event to be processed
await new Promise(resolve => setTimeout(resolve, 50));
@ -295,7 +294,7 @@ describe("StreamableHTTPClientTransport", () => {
});
await transport.start();
await transport.openSseStream();
await transport["_startOrAuthStandaloneSSE"]();
await new Promise(resolve => setTimeout(resolve, 50));
// Now simulate attempting to reconnect
@ -306,7 +305,7 @@ describe("StreamableHTTPClientTransport", () => {
body: null
});
await transport.openSseStream();
await transport["_startOrAuthStandaloneSSE"]();
// Check that Last-Event-ID was included
const calls = (global.fetch as jest.Mock).mock.calls;
@ -366,7 +365,7 @@ describe("StreamableHTTPClientTransport", () => {
await transport.start();
await transport.openSseStream();
await transport["_startOrAuthStandaloneSSE"]();
expect((actualReqInit.headers as Headers).get("x-custom-header")).toBe("CustomValue");
requestInit.headers["X-Custom-Header"] = "SecondCustomValue";

View file

@ -1,5 +1,5 @@
import { Transport } from "../shared/transport.js";
import { JSONRPCMessage, JSONRPCMessageSchema } from "../types.js";
import { isJSONRPCNotification, JSONRPCMessage, JSONRPCMessageSchema } from "../types.js";
import { auth, AuthResult, OAuthClientProvider, UnauthorizedError } from "./auth.js";
import { EventSourceParserStream } from "eventsource-parser/stream";
@ -126,6 +126,12 @@ export class StreamableHTTPClientTransport implements Transport {
return await this._authThenStart();
}
// 405 indicates that the server does not offer an SSE stream at GET endpoint
// This is an expected case that should not trigger an error
if (response.status === 405) {
return;
}
throw new StreamableHTTPError(
response.status,
`Failed to open SSE stream: ${response.statusText}`,
@ -243,6 +249,12 @@ export class StreamableHTTPClientTransport implements Transport {
// If the response is 202 Accepted, there's no body to process
if (response.status === 202) {
// if the accepted notification is initialized, we start the SSE stream
// if it's supported by the server
if (isJSONRPCNotification(message) && message.method === "notifications/initialized") {
// We don't need to handle 405 here anymore as it's handled in _startOrAuthStandaloneSSE
this._startOrAuthStandaloneSSE().catch(err => this.onerror?.(err));
}
return;
}
@ -279,20 +291,4 @@ export class StreamableHTTPClientTransport implements Transport {
throw error;
}
}
/**
* 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) {
throw new Error(
"StreamableHTTPClientTransport not started! Call connect() before openSseStream().",
);
}
await this._startOrAuthStandaloneSSE();
}
}

View file

@ -25,20 +25,10 @@ async function main(): Promise<void> {
const transport = new StreamableHTTPClientTransport(
new URL('http://localhost:3000/mcp')
);
let supportsStandaloneSse = false;
// Connect the client using the transport and initialize the server
// Connect the client using the transport and initialize the server
await client.connect(transport);
console.log('Connected to MCP server');
console.log('Opening SSE stream to receive server notifications...');
try {
await transport.openSseStream();
supportsStandaloneSse = true;
console.log('SSE stream established successfully. Waiting for notifications...');
}
catch (error) {
console.error('Failed to open SSE stream:', error);
}
// Set up notification handlers for server-initiated messages
client.setNotificationHandler(LoggingMessageNotificationSchema, (notification) => {
@ -136,11 +126,8 @@ async function main(): Promise<void> {
} catch (error) {
console.log(`Resources not supported by this server (${error})`);
}
if (supportsStandaloneSse) {
// Instead of closing immediately, keep the connection open to receive notifications
console.log('\nKeeping connection open to receive notifications. Press Ctrl+C to exit.');
}
// Keep the connection open to receive notifications
console.log('\nKeeping connection open to receive notifications. Press Ctrl+C to exit.');
}
main().catch((error: unknown) => {

View file

@ -578,7 +578,7 @@ describe("StreamableHTTPServerTransport", () => {
expect(mockResponse.writeHead).toHaveBeenCalledWith(200, expect.objectContaining({
"Content-Type": "text/event-stream",
"Cache-Control": "no-cache",
"Cache-Control": "no-cache, no-transform",
"Connection": "keep-alive",
"mcp-session-id": transport.sessionId,
}));