Skip to content

Commit 059feaf

Browse files
committed
mcp: add StreamableHTTPHandler.Close for graceful shutdown
Expose public Close() to tear down all sessions and reject new requests with 503, matching the API proposed in #440. Co-authored-by: Piyush Bag <piyushbag4@gmail.com>
1 parent 65ece59 commit 059feaf

2 files changed

Lines changed: 82 additions & 16 deletions

File tree

mcp/streamable.go

Lines changed: 28 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -51,6 +51,7 @@ type StreamableHTTPHandler struct {
5151
onTransportDeletion func(sessionID string) // for testing
5252

5353
mu sync.Mutex
54+
closed bool
5455
sessions map[string]*sessionInfo // keyed by session ID
5556
}
5657

@@ -219,27 +220,30 @@ func NewStreamableHTTPHandler(getServer func(*http.Request) *Server, opts *Strea
219220
return h
220221
}
221222

222-
// closeAll closes all ongoing sessions, for tests.
223-
//
224-
// TODO(rfindley): investigate the best API for callers to configure their
225-
// session lifecycle. (?)
223+
// Close closes the handler, closing and removing all connected sessions,
224+
// and preventing new sessions from being added.
226225
//
227-
// Should we allow passing in a session store? That would allow the handler to
228-
// be stateless.
229-
func (h *StreamableHTTPHandler) closeAll() {
230-
// TODO: if we ever expose this outside of tests, we'll need to do better
231-
// than simply collecting sessions while holding the lock: we need to prevent
232-
// new sessions from being added.
233-
//
234-
// Currently, sessions remove themselves from h.sessions when closed, so we
235-
// can't call Close while holding the lock.
226+
// Close is idempotent.
227+
func (h *StreamableHTTPHandler) Close() error {
236228
h.mu.Lock()
229+
if h.closed {
230+
h.mu.Unlock()
231+
return nil
232+
}
233+
h.closed = true
237234
sessionInfos := slices.Collect(maps.Values(h.sessions))
238-
h.sessions = nil
235+
h.sessions = make(map[string]*sessionInfo)
239236
h.mu.Unlock()
240-
for _, s := range sessionInfos {
241-
s.session.Close()
237+
238+
for _, info := range sessionInfos {
239+
info.session.Close()
242240
}
241+
return nil
242+
}
243+
244+
// closeAll closes all ongoing sessions, for tests.
245+
func (h *StreamableHTTPHandler) closeAll() {
246+
_ = h.Close()
243247
}
244248

245249
// disablelocalhostprotection is a compatibility parameter that allows to disable
@@ -308,6 +312,14 @@ func (h *StreamableHTTPHandler) ServeHTTP(w http.ResponseWriter, req *http.Reque
308312
}
309313
}
310314

315+
h.mu.Lock()
316+
closed := h.closed
317+
h.mu.Unlock()
318+
if closed {
319+
http.Error(w, "handler closed", http.StatusServiceUnavailable)
320+
return
321+
}
322+
311323
// [§2.7] of the spec (2025-06-18): validate the MCP-Protocol-Version
312324
// header. If provided, it must be a supported version. If absent, the
313325
// version is unknown (the request may be an initialize for any version).

mcp/streamable_test.go

Lines changed: 54 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -641,6 +641,60 @@ func TestStreamableServerDisconnect(t *testing.T) {
641641
}
642642
}
643643

644+
func TestStreamableHTTPHandlerClose(t *testing.T) {
645+
server := NewServer(testImpl, nil)
646+
handler := NewStreamableHTTPHandler(func(*http.Request) *Server { return server }, nil)
647+
httpServer := httptest.NewServer(mustNotPanic(t, handler))
648+
defer httpServer.Close()
649+
650+
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
651+
defer cancel()
652+
653+
client := NewClient(testImpl, nil)
654+
// Pin a pre-SEP-2575 protocol version so Connect performs a single
655+
// legacy initialize handshake (one session). Under the modern protocol the
656+
// client first issues a stateless server/discover probe, which—when the
657+
// server falls back to the legacy handshake—transiently leaves an extra
658+
// session and would make the exact count below nondeterministic. This test
659+
// exercises Close, not protocol negotiation.
660+
clientSession, err := client.Connect(ctx, &StreamableClientTransport{Endpoint: httpServer.URL}, &ClientSessionOptions{protocolVersion: protocolVersion20251125})
661+
if err != nil {
662+
t.Fatalf("client.Connect() failed: %v", err)
663+
}
664+
t.Cleanup(func() { _ = clientSession.Close() })
665+
666+
handler.mu.Lock()
667+
if len(handler.sessions) != 1 {
668+
t.Fatalf("want 1 session before Close, got %d", len(handler.sessions))
669+
}
670+
handler.mu.Unlock()
671+
672+
if err := handler.Close(); err != nil {
673+
t.Fatalf("Close() failed: %v", err)
674+
}
675+
if err := handler.Close(); err != nil {
676+
t.Fatalf("second Close() failed: %v", err)
677+
}
678+
679+
handler.mu.Lock()
680+
if len(handler.sessions) != 0 {
681+
t.Fatalf("want 0 sessions after Close, got %d", len(handler.sessions))
682+
}
683+
if !handler.closed {
684+
t.Fatal("want handler.closed true after Close")
685+
}
686+
handler.mu.Unlock()
687+
688+
resp, err := http.Get(httpServer.URL)
689+
if err != nil {
690+
t.Fatalf("http.Get after Close failed: %v", err)
691+
}
692+
defer resp.Body.Close()
693+
if resp.StatusCode != http.StatusServiceUnavailable {
694+
t.Fatalf("got status %d after Close, want %d", resp.StatusCode, http.StatusServiceUnavailable)
695+
}
696+
}
697+
644698
func TestServerTransportCleanup(t *testing.T) {
645699
nClient := 3
646700

0 commit comments

Comments
 (0)