mcp: add DisableListening option to StreamableClientTransport (#729)

mcp: add DisableStandaloneSSE option to control standalone SSE stream

Currently, StreamableClientTransport always establishes a standalone SSE
connection after initialization to receive server-initiated messages.
While this is useful for receiving notifications like
ToolListChangedNotification, it causes problems in several real-world
scenarios:

1. **Unnecessary resource usage**: Many use cases (for me), especially
in scientific computing scenarios, only require simple request-response
communication. Clients don't need server-initiated notifications, making
the persistent SSE connection wasteful.

2. **Server compatibility issues**: Some third-party MCP servers don't
properly handle GET requests for SSE streams. When the client
automatically tries to establish the SSE connection, it fails or hangs,
blocking the entire connection process. This is particularly problematic
when using third-party servers that cannot be modified. This issue is
similar to what was discussed in issue #634.

3. **Lack of user control**: The MCP specification states that the
standalone SSE stream is optional ("The client MAY issue an HTTP GET"),
but the current SDK implementation doesn't provide a way to opt out.
This implementation is also inconsistent with other MCP SDKs like
github.com/mark3labs/mcp-go, which provides getListeningEnabled control
(defaults to false, requires explicit enablement).

Fixes #728

Co-authored-by: zhaoxin <[email protected]>

Author: zxxf18

mcp/streamable.go

@@ -1389,6 +1389,26 @@ type StreamableClientTransport struct {
 	// It defaults to 5. To disable retries, use a negative number.
 	MaxRetries int
 
+	// DisableStandaloneSSE controls whether the client establishes a standalone SSE stream
+	// for receiving server-initiated messages.
+	//
+	// When false (the default), after initialization the client sends an HTTP GET request
+	// to establish a persistent server-sent events (SSE) connection. This allows the server
+	// to send messages to the client at any time, such as ToolListChangedNotification or
+	// other server-initiated requests and notifications. The connection persists for the
+	// lifetime of the session and automatically reconnects if interrupted.
+	//
+	// When true, the client does not establish the standalone SSE stream. The client will
+	// only receive responses to its own POST requests. Server-initiated messages will not
+	// be received.
+	//
+	// According to the MCP specification, the standalone SSE stream is optional.
+	// Setting DisableStandaloneSSE to true is useful when:
+	//   - You only need request-response communication and don't need server-initiated notifications
+	//   - The server doesn't properly handle GET requests for SSE streams
+	//   - You want to avoid maintaining a persistent connection
+	DisableStandaloneSSE bool
+
 	// TODO(rfindley): propose exporting these.
 	// If strict is set, the transport is in 'strict mode', where any violation
 	// of the MCP spec causes a failure.
@@ -1453,16 +1473,17 @@ func (t *StreamableClientTransport) Connect(ctx context.Context) (Connection, er
 	// middleware), yet only cancel the standalone stream when the connection is closed.
 	connCtx, cancel := context.WithCancel(xcontext.Detach(ctx))
 	conn := &streamableClientConn{
-		url:        t.Endpoint,
-		client:     client,
-		incoming:   make(chan jsonrpc.Message, 10),
-		done:       make(chan struct{}),
-		maxRetries: maxRetries,
-		strict:     t.strict,
-		logger:     ensureLogger(t.logger), // must be non-nil for safe logging
-		ctx:        connCtx,
-		cancel:     cancel,
-		failed:     make(chan struct{}),
+		url:                  t.Endpoint,
+		client:               client,
+		incoming:             make(chan jsonrpc.Message, 10),
+		done:                 make(chan struct{}),
+		maxRetries:           maxRetries,
+		strict:               t.strict,
+		logger:               ensureLogger(t.logger), // must be non-nil for safe logging
+		ctx:                  connCtx,
+		cancel:               cancel,
+		failed:               make(chan struct{}),
+		disableStandaloneSSE: t.DisableStandaloneSSE,
 	}
 	return conn, nil
 }
@@ -1477,6 +1498,10 @@ type streamableClientConn struct {
 	strict     bool         // from [StreamableClientTransport.strict]
 	logger     *slog.Logger // from [StreamableClientTransport.logger]
 
+	// disableStandaloneSSE controls whether to disable the standalone SSE stream
+	// for receiving server-to-client notifications when no request is in flight.
+	disableStandaloneSSE bool // from [StreamableClientTransport.DisableStandaloneSSE]
+
 	// Guard calls to Close, as it may be called multiple times.
 	closeOnce sync.Once
 	closeErr  error
@@ -1518,7 +1543,7 @@ func (c *streamableClientConn) sessionUpdated(state clientSessionState) {
 	c.mu.Unlock()
 
 	// Start the standalone SSE stream as soon as we have the initialized
-	// result.
+	// result, if continuous listening is enabled.
 	//
 	// § 2.2: The client MAY issue an HTTP GET to the MCP endpoint. This can be
 	// used to open an SSE stream, allowing the server to communicate to the
@@ -1528,9 +1553,11 @@ func (c *streamableClientConn) sessionUpdated(state clientSessionState) {
 	// initialized, we don't know whether the server requires a sessionID.
 	//
 	// § 2.5: A server using the Streamable HTTP transport MAY assign a session
-	// ID at initialization time, by including it in an Mcp-Session-Id header
+	// ID at initialization time, by including it in a Mcp-Session-Id header
 	// on the HTTP response containing the InitializeResult.
-	c.connectStandaloneSSE()
+	if !c.disableStandaloneSSE {
+		c.connectStandaloneSSE()
+	}
 }
 
 func (c *streamableClientConn) connectStandaloneSSE() {

mcp/streamable_client_test.go

@@ -696,3 +696,121 @@ func TestStreamableClientTransientErrors(t *testing.T) {
 		})
 	}
 }
+
+func TestStreamableClientDisableStandaloneSSE(t *testing.T) {
+	ctx := context.Background()
+
+	tests := []struct {
+		name                 string
+		disableStandaloneSSE bool
+		expectGETRequest     bool
+	}{
+		{
+			name:                 "default behavior (standalone SSE enabled)",
+			disableStandaloneSSE: false,
+			expectGETRequest:     true,
+		},
+		{
+			name:                 "standalone SSE disabled",
+			disableStandaloneSSE: true,
+			expectGETRequest:     false,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			getRequestKey := streamableRequestKey{"GET", "123", "", ""}
+
+			fake := &fakeStreamableServer{
+				t: t,
+				responses: fakeResponses{
+					{"POST", "", methodInitialize, ""}: {
+						header: header{
+							"Content-Type":  "application/json",
+							sessionIDHeader: "123",
+						},
+						body: jsonBody(t, initResp),
+					},
+					{"POST", "123", notificationInitialized, ""}: {
+						status:              http.StatusAccepted,
+						wantProtocolVersion: latestProtocolVersion,
+					},
+					getRequestKey: {
+						header: header{
+							"Content-Type": "text/event-stream",
+						},
+						wantProtocolVersion: latestProtocolVersion,
+						optional:            !test.expectGETRequest,
+					},
+					{"DELETE", "123", "", ""}: {
+						optional: true,
+					},
+				},
+			}
+
+			httpServer := httptest.NewServer(fake)
+			defer httpServer.Close()
+
+			transport := &StreamableClientTransport{
+				Endpoint:             httpServer.URL,
+				DisableStandaloneSSE: test.disableStandaloneSSE,
+			}
+			client := NewClient(testImpl, nil)
+			session, err := client.Connect(ctx, transport, nil)
+			if err != nil {
+				t.Fatalf("client.Connect() failed: %v", err)
+			}
+
+			// Give some time for the standalone SSE connection to be established (if enabled)
+			time.Sleep(100 * time.Millisecond)
+
+			// Verify the connection state
+			streamableConn, ok := session.mcpConn.(*streamableClientConn)
+			if !ok {
+				t.Fatalf("Expected *streamableClientConn, got %T", session.mcpConn)
+			}
+
+			if got, want := streamableConn.disableStandaloneSSE, test.disableStandaloneSSE; got != want {
+				t.Errorf("disableStandaloneSSE field: got %v, want %v", got, want)
+			}
+
+			// Clean up
+			if err := session.Close(); err != nil {
+				t.Errorf("closing session: %v", err)
+			}
+
+			// Check if GET request was received
+			fake.calledMu.Lock()
+			getRequestReceived := false
+			if fake.called != nil {
+				getRequestReceived = fake.called[getRequestKey]
+			}
+			fake.calledMu.Unlock()
+
+			if got, want := getRequestReceived, test.expectGETRequest; got != want {
+				t.Errorf("GET request received: got %v, want %v", got, want)
+			}
+
+			// If we expected a GET request, verify it was actually received
+			if test.expectGETRequest {
+				if missing := fake.missingRequests(); len(missing) > 0 {
+					// Filter out optional requests
+					var requiredMissing []streamableRequestKey
+					for _, key := range missing {
+						if resp, ok := fake.responses[key]; ok && !resp.optional {
+							requiredMissing = append(requiredMissing, key)
+						}
+					}
+					if len(requiredMissing) > 0 {
+						t.Errorf("did not receive expected requests: %v", requiredMissing)
+					}
+				}
+			} else {
+				// If we didn't expect a GET request, verify it wasn't sent
+				if getRequestReceived {
+					t.Error("GET request was sent unexpectedly when DisableStandaloneSSE is true")
+				}
+			}
+		})
+	}
+}