Skip to content

Commit f5186f6

Browse files
piyushbagPiyush Bag
authored andcommitted
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.
1 parent 827f90b commit f5186f6

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

@@ -251,27 +252,30 @@ func NewStreamableHTTPHandler(getServer func(*http.Request) *Server, opts *Strea
251252
return h
252253
}
253254

254-
// closeAll closes all ongoing sessions, for tests.
255-
//
256-
// TODO(rfindley): investigate the best API for callers to configure their
257-
// session lifecycle. (?)
255+
// Close closes the handler, closing and removing all connected sessions,
256+
// and preventing new sessions from being added.
258257
//
259-
// Should we allow passing in a session store? That would allow the handler to
260-
// be stateless.
261-
func (h *StreamableHTTPHandler) closeAll() {
262-
// TODO: if we ever expose this outside of tests, we'll need to do better
263-
// than simply collecting sessions while holding the lock: we need to prevent
264-
// new sessions from being added.
265-
//
266-
// Currently, sessions remove themselves from h.sessions when closed, so we
267-
// can't call Close while holding the lock.
258+
// Close is idempotent.
259+
func (h *StreamableHTTPHandler) Close() error {
268260
h.mu.Lock()
261+
if h.closed {
262+
h.mu.Unlock()
263+
return nil
264+
}
265+
h.closed = true
269266
sessionInfos := slices.Collect(maps.Values(h.sessions))
270-
h.sessions = nil
267+
h.sessions = make(map[string]*sessionInfo)
271268
h.mu.Unlock()
272-
for _, s := range sessionInfos {
273-
s.session.Close()
269+
270+
for _, info := range sessionInfos {
271+
info.session.Close()
274272
}
273+
return nil
274+
}
275+
276+
// closeAll closes all ongoing sessions, for tests.
277+
func (h *StreamableHTTPHandler) closeAll() {
278+
_ = h.Close()
275279
}
276280

277281
// disablelocalhostprotection is a compatibility parameter that allows to disable
@@ -340,6 +344,14 @@ func (h *StreamableHTTPHandler) ServeHTTP(w http.ResponseWriter, req *http.Reque
340344
}
341345
}
342346

347+
h.mu.Lock()
348+
closed := h.closed
349+
h.mu.Unlock()
350+
if closed {
351+
http.Error(w, "handler closed", http.StatusServiceUnavailable)
352+
return
353+
}
354+
343355
// Bound the request body to protect against OOM attacks.
344356
if req.Body != nil && h.opts.MaxRequestBodyBytes > 0 {
345357
req.Body = http.MaxBytesReader(w, req.Body, h.opts.MaxRequestBodyBytes)

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)