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

mcp: add DisableListening option to StreamableClientTransport (#729)

Browse source 
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 <zhaoxin@pjlab.org.cn>
This commit is contained in:
zhaoxin 2026-01-11 20:56:47 +08:00 • committed by GitHub
commit b4f957ff3c
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 158 additions and 13 deletions
Download patch file Download diff file Expand all files Collapse all files

53
mcp/streamable.go
View file

@ -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() {

118
mcp/streamable_client_test.go
View file

@ -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")
}
}
})
}
}