Skip to content

Commit ad03c7f

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.
1 parent 9576afc commit ad03c7f

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

@@ -240,27 +241,30 @@ func NewStreamableHTTPHandler(getServer func(*http.Request) *Server, opts *Strea
240241
return h
241242
}
242243

243-
// closeAll closes all ongoing sessions, for tests.
244-
//
245-
// TODO(rfindley): investigate the best API for callers to configure their
246-
// session lifecycle. (?)
244+
// Close closes the handler, closing and removing all connected sessions,
245+
// and preventing new sessions from being added.
247246
//
248-
// Should we allow passing in a session store? That would allow the handler to
249-
// be stateless.
250-
func (h *StreamableHTTPHandler) closeAll() {
251-
// TODO: if we ever expose this outside of tests, we'll need to do better
252-
// than simply collecting sessions while holding the lock: we need to prevent
253-
// new sessions from being added.
254-
//
255-
// Currently, sessions remove themselves from h.sessions when closed, so we
256-
// can't call Close while holding the lock.
247+
// Close is idempotent.
248+
func (h *StreamableHTTPHandler) Close() error {
257249
h.mu.Lock()
250+
if h.closed {
251+
h.mu.Unlock()
252+
return nil
253+
}
254+
h.closed = true
258255
sessionInfos := slices.Collect(maps.Values(h.sessions))
259-
h.sessions = nil
256+
h.sessions = make(map[string]*sessionInfo)
260257
h.mu.Unlock()
261-
for _, s := range sessionInfos {
262-
s.session.Close()
258+
259+
for _, info := range sessionInfos {
260+
info.session.Close()
263261
}
262+
return nil
263+
}
264+
265+
// closeAll closes all ongoing sessions, for tests.
266+
func (h *StreamableHTTPHandler) closeAll() {
267+
_ = h.Close()
264268
}
265269

266270
// disablelocalhostprotection is a compatibility parameter that allows to disable
@@ -329,6 +333,14 @@ func (h *StreamableHTTPHandler) ServeHTTP(w http.ResponseWriter, req *http.Reque
329333
}
330334
}
331335

336+
h.mu.Lock()
337+
closed := h.closed
338+
h.mu.Unlock()
339+
if closed {
340+
http.Error(w, "handler closed", http.StatusServiceUnavailable)
341+
return
342+
}
343+
332344
// Bound the request body to protect against OOM attacks.
333345
if req.Body != nil && h.opts.MaxRequestBodyBytes > 0 {
334346
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)