Client implementation of Streamable HTTP transport #290

ihrpr · 2025-04-08T10:11:46Z

Introduces support for streamable http transport on client. Server was implemented in #266

TO DO

Tests with examples and integration in the stacked PRs

Follow ups

Session management
Refactor auth and make it work with session management
Examples for client and server
Integration tests using examples
Resumability
Server and client to handle requests returning not only streams but also JSON
Cancellation for streams
Client to handle closing stream when all responses are done (bad server case)

jspahrsummers

Can you please refactor to use the eventsource package (which we already depend upon)? A lot of this code will be able to be deleted, and I expect that, e.g., resumability will become easier.

jspahrsummers · 2025-04-08T16:02:07Z

src/client/streamableHttp.ts

+      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;
+      }


Why are we permitting 404 here? It might be OK, just curious for the rationale.

intentional as we still need to add support for GET and Auth and the order it should work. Right now it will be rejected for clients that provide session id as it's pre-initialization. Will be addressing it in the follow up.

this was refactored, the implementation was wrong for GET, we need to have a separate method, unlike SSE implementation

jspahrsummers · 2025-04-08T16:04:02Z

src/client/streamableHttp.ts

+    // If we have a session ID, send a DELETE request to explicitly terminate the session
+    if (this._sessionId) {


I don't think we want to automatically do this. What if the client process is restarting, for example? We should definitely offer a way to do this, but leave it up to the user when to trigger.

oh, that's right! I'll remove automatic handling and add to follow ups to implemenet a way for a user to do it

jspahrsummers · 2025-04-08T16:04:30Z

src/client/streamableHttp.ts

+          // Server might respond with 405 if it doesn't support explicit session termination
+          // We don't throw an error in that case


You might also see a 404 if the session ID has already been deleted on the server side (because the server could clear it at any time).

jspahrsummers · 2025-04-08T16:07:30Z

src/client/streamableHttp.ts

+    this.onclose?.();
+  }
+
+  async send(message: JSONRPCMessage | JSONRPCMessage[]): Promise<void> {


Let's leave the batching to another PR?

ah, it's already there, I'd just leave it, the full batch support agree, will be a separate PR

jspahrsummers · 2025-04-08T16:19:34Z

src/client/streamableHttp.ts

+      }
+
+      // Successful connection, handle the SSE stream as a standalone listener
+      const streamId = `initial-${Date.now()}`;


Why do we need these stream IDs? I can't find any spot where we locate a stream by its ID.

we use it in the clean up to close all of the sessions

jspahrsummers · 2025-04-08T16:20:07Z

src/client/streamableHttp.ts

+    // Close all active streams
+    for (const reader of this._activeStreams.values()) {
+      try {
+        reader.cancel();
+      } catch (error) {
+        this.onerror?.(error as Error);
+      }
+    }
+    this._activeStreams.clear();


It might be easier to use a single AbortController for the whole object, then pass its AbortSignal into all the fetch/SSE calls.

jspahrsummers · 2025-04-08T16:22:39Z

src/client/streamableHttp.ts

+      // Extract IDs from request messages for tracking responses
+      const requestIds = messages.filter(msg => 'method' in msg && 'id' in msg)
+        .map(msg => 'id' in msg ? msg.id : undefined)
+        .filter(id => id !== undefined);


We may want to close the SSE stream after receiving responses to all these request IDs.

Also, we may want to figure out how to automatically close the stream if all the requests are cancelled (by the client).

we are doing closing on the server side and as we are reading the stream response it's done when the stream reader is done:

const { done, value: event } = await reader.read(); if (done) { this._activeStreams.delete(streamId); break; }

Also, we may want to figure out how to automatically close the stream if all the requests are cancelled (by the client).

Adding to the follow ups

This assumes a well-behaving server, which might not always be the case (e.g., the server might not be using an official SDK, or one of the SDKs may behave differently).

Basically, the client should not assume that the server will close the connection on its own.

good point! adding to follow ups

beaulac · 2025-04-08T17:43:07Z

src/client/streamableHttp.ts

+    }
+  }
+
+  private _handleSseStream(stream: ReadableStream<Uint8Array> | null, streamId: string): void {


NB:
import { EventSourceParserStream } from 'eventsource-parser/stream';
would take care of a lot of this parsing logic and eventsource-parser is already a transitive dependency of existing eventsource dependency

@beaulac this is great, thank you!

beaulac · 2025-04-08T17:51:28Z

@jspahrsummers

refactor to use the eventsource package

I tried to use eventsource in my own attempts at this, but EventSource can only be initialized with GET
Since we are potentially starting a SSE stream from an existing Response stream (since at time of request it's unknown whether it is or isn't an SSE stream) we can't use EventSource everywhere directly

jspahrsummers

jspahrsummers · 2025-04-09T09:47:28Z

src/client/streamableHttp.ts

@@ -121,32 +119,76 @@ export class StreamableHTTPClientTransport implements Transport {
        signal: this._abortController?.signal,
      });

-      if (response.status === 405 || response.status === 404) {


We still need to handle 405s here, no?

ALSoryu · 2025-04-10T12:01:52Z

How to implement session management without broker like redis in a distributed scenario

Fix Reconnect handler

ihrpr added 3 commits April 8, 2025 09:43

initial client implementation

640ae5a

multiple streams and initial stream with get

c3b446d

handle not implemented get

f1827ce

ihrpr requested a review from jspahrsummers April 8, 2025 15:18

ihrpr marked this pull request as ready for review April 8, 2025 15:18

jspahrsummers linked an issue Apr 8, 2025 that may be closed by this pull request

Client implementation of Streamable HTTP transport #274

Closed

jspahrsummers requested changes Apr 8, 2025

View reviewed changes

beaulac reviewed Apr 8, 2025

View reviewed changes

ihrpr added 4 commits April 8, 2025 21:39

use EventSourceParserStream

d4ad976

fix lint

15c10a5

extract standalone get into a method which can be called by a client

1d18a57

Merge branch 'main' into ihrpr/streamable-http-client

368d666

ihrpr requested a review from jspahrsummers April 9, 2025 09:38

ihrpr added 2 commits April 9, 2025 11:13

remove the _activeStreams map

1867400

fix lint

26929fa

jspahrsummers approved these changes Apr 9, 2025

View reviewed changes

ihrpr mentioned this pull request Apr 9, 2025

Server implementation of Streamable HTTP transport #266

Merged

1 task

ihrpr merged commit 0445095 into main Apr 9, 2025
4 checks passed

ihrpr deleted the ihrpr/streamable-http-client branch April 9, 2025 11:57

dtinth mentioned this pull request Apr 10, 2025

Support HTTP streaming transport supercorp-ai/supergateway#38

Open

Pizzaface pushed a commit to RewstApp/mcp-inspector that referenced this pull request May 2, 2025

Merge pull request modelcontextprotocol#290 from cliffhall/fix-reconnect

4a2ba7c

Fix Reconnect handler

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Client implementation of Streamable HTTP transport #290

Client implementation of Streamable HTTP transport #290

ihrpr commented Apr 8, 2025 •

edited

Loading

jspahrsummers left a comment

jspahrsummers Apr 8, 2025

ihrpr Apr 8, 2025

ihrpr Apr 9, 2025

jspahrsummers Apr 8, 2025

ihrpr Apr 8, 2025

jspahrsummers Apr 8, 2025

jspahrsummers Apr 8, 2025

ihrpr Apr 8, 2025

jspahrsummers Apr 8, 2025

ihrpr Apr 8, 2025

jspahrsummers Apr 8, 2025

jspahrsummers Apr 8, 2025

ihrpr Apr 8, 2025

jspahrsummers Apr 9, 2025 •

edited

Loading

ihrpr Apr 9, 2025

beaulac Apr 8, 2025 •

edited

Loading

ihrpr Apr 8, 2025

beaulac commented Apr 8, 2025 •

edited

Loading

jspahrsummers left a comment

jspahrsummers Apr 9, 2025

ALSoryu commented Apr 10, 2025

		// If we have a session ID, send a DELETE request to explicitly terminate the session
		if (this._sessionId) {

		// Server might respond with 405 if it doesn't support explicit session termination
		// We don't throw an error in that case

Client implementation of Streamable HTTP transport #290

Client implementation of Streamable HTTP transport #290

Conversation

ihrpr commented Apr 8, 2025 • edited Loading

TO DO

Follow ups

jspahrsummers left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jspahrsummers Apr 9, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

beaulac Apr 8, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

beaulac commented Apr 8, 2025 • edited Loading

jspahrsummers left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ALSoryu commented Apr 10, 2025

ihrpr commented Apr 8, 2025 •

edited

Loading

jspahrsummers Apr 9, 2025 •

edited

Loading

beaulac Apr 8, 2025 •

edited

Loading

beaulac commented Apr 8, 2025 •

edited

Loading