-
Notifications
You must be signed in to change notification settings - Fork 419
Expand file tree
/
Copy pathstreamable.go
More file actions
1836 lines (1676 loc) · 58.5 KB
/
streamable.go
File metadata and controls
1836 lines (1676 loc) · 58.5 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
// Copyright 2025 The Go MCP SDK Authors. All rights reserved.
// Use of this source code is governed by an MIT-style
// license that can be found in the LICENSE file.
package mcp
import (
"bytes"
"context"
"encoding/json"
"errors"
"fmt"
"io"
"log/slog"
"maps"
"math"
"math/rand/v2"
"net/http"
"slices"
"strconv"
"strings"
"sync"
"sync/atomic"
"time"
"github.com/modelcontextprotocol/go-sdk/auth"
"github.com/modelcontextprotocol/go-sdk/internal/jsonrpc2"
"github.com/modelcontextprotocol/go-sdk/jsonrpc"
)
const (
protocolVersionHeader = "Mcp-Protocol-Version"
sessionIDHeader = "Mcp-Session-Id"
)
// A StreamableHTTPHandler is an http.Handler that serves streamable MCP
// sessions, as defined by the [MCP spec].
//
// [MCP spec]: https://modelcontextprotocol.io/2025/03/26/streamable-http-transport.html
type StreamableHTTPHandler struct {
getServer func(*http.Request) *Server
opts StreamableHTTPOptions
onTransportDeletion func(sessionID string) // for testing
mu sync.Mutex
sessions map[string]*sessionInfo // keyed by session ID
}
type sessionInfo struct {
session *ServerSession
transport *StreamableServerTransport
// If timeout is set, automatically close the session after an idle period.
timeout time.Duration
timerMu sync.Mutex
refs int // reference count
timer *time.Timer
}
// startPOST signals that a POST request for this session is starting (which
// carries a client->server message), pausing the session timeout if it was
// running.
//
// TODO: we may want to also pause the timer when resuming non-standalone SSE
// streams, but that is tricy to implement. Clients should generally make
// keepalive pings if they want to keep the session live.
func (i *sessionInfo) startPOST() {
if i.timeout <= 0 {
return
}
i.timerMu.Lock()
defer i.timerMu.Unlock()
if i.timer == nil {
return // timer stopped permanently
}
if i.refs == 0 {
i.timer.Stop()
}
i.refs++
}
// endPOST sigals that a request for this session is ending, starting the
// timeout if there are no other requests running.
func (i *sessionInfo) endPOST() {
if i.timeout <= 0 {
return
}
i.timerMu.Lock()
defer i.timerMu.Unlock()
if i.timer == nil {
return // timer stopped permanently
}
i.refs--
assert(i.refs >= 0, "negative ref count")
if i.refs == 0 {
i.timer.Reset(i.timeout)
}
}
// stopTimer stops the inactivity timer permanently.
func (i *sessionInfo) stopTimer() {
i.timerMu.Lock()
defer i.timerMu.Unlock()
if i.timer != nil {
i.timer.Stop()
i.timer = nil
}
}
// StreamableHTTPOptions configures the StreamableHTTPHandler.
type StreamableHTTPOptions struct {
// Stateless controls whether the session is 'stateless'.
//
// A stateless server does not validate the Mcp-Session-Id header, and uses a
// temporary session with default initialization parameters. Any
// server->client request is rejected immediately as there's no way for the
// client to respond. Server->Client notifications may reach the client if
// they are made in the context of an incoming request, as described in the
// documentation for [StreamableServerTransport].
Stateless bool
// TODO(#148): support session retention (?)
// JSONResponse causes streamable responses to return application/json rather
// than text/event-stream ([§2.1.5] of the spec).
//
// [§2.1.5]: https://modelcontextprotocol.io/specification/2025-06-18/basic/transports#sending-messages-to-the-server
JSONResponse bool
// Logger specifies the logger to use.
// If nil, do not log.
Logger *slog.Logger
// EventStore enables stream resumption.
//
// If set, EventStore will be used to persist stream events and replay them
// upon stream resumption.
EventStore EventStore
// SessionTimeout configures a timeout for idle sessions.
//
// When sessions receive no new HTTP requests from the client for this
// duration, they are automatically closed.
//
// If SessionTimeout is the zero value, idle sessions are never closed.
SessionTimeout time.Duration
// SessionStateStore enables persisting session state across process
// restarts. When configured, the handler will attempt to restore server
// sessions whose identifiers are unknown to the current process, allowing
// serverless deployments that spin up per-request.
SessionStateStore ServerSessionStateStore
}
// NewStreamableHTTPHandler returns a new [StreamableHTTPHandler].
//
// The getServer function is used to create or look up servers for new
// sessions. It is OK for getServer to return the same server multiple times.
// If getServer returns nil, a 400 Bad Request will be served.
func NewStreamableHTTPHandler(getServer func(*http.Request) *Server, opts *StreamableHTTPOptions) *StreamableHTTPHandler {
h := &StreamableHTTPHandler{
getServer: getServer,
sessions: make(map[string]*sessionInfo),
}
if opts != nil {
h.opts = *opts
}
if h.opts.Logger == nil { // ensure we have a logger
h.opts.Logger = ensureLogger(nil)
}
return h
}
// closeAll closes all ongoing sessions, for tests.
//
// TODO(rfindley): investigate the best API for callers to configure their
// session lifecycle. (?)
//
// Should we allow passing in a session store? That would allow the handler to
// be stateless.
func (h *StreamableHTTPHandler) closeAll() {
// TODO: if we ever expose this outside of tests, we'll need to do better
// than simply collecting sessions while holding the lock: we need to prevent
// new sessions from being added.
//
// Currently, sessions remove themselves from h.sessions when closed, so we
// can't call Close while holding the lock.
h.mu.Lock()
sessionInfos := slices.Collect(maps.Values(h.sessions))
h.sessions = nil
h.mu.Unlock()
for _, s := range sessionInfos {
s.session.Close()
}
}
func (h *StreamableHTTPHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
// Allow multiple 'Accept' headers.
// https://developer.mozilla.org/en-US/docs/Web/HTTP/Reference/Headers/Accept#syntax
accept := strings.Split(strings.Join(req.Header.Values("Accept"), ","), ",")
var jsonOK, streamOK bool
for _, c := range accept {
switch strings.TrimSpace(c) {
case "application/json", "application/*":
jsonOK = true
case "text/event-stream", "text/*":
streamOK = true
case "*/*":
jsonOK = true
streamOK = true
}
}
if req.Method == http.MethodGet {
if !streamOK {
http.Error(w, "Accept must contain 'text/event-stream' for GET requests", http.StatusBadRequest)
return
}
} else if (!jsonOK || !streamOK) && req.Method != http.MethodDelete { // TODO: consolidate with handling of http method below.
http.Error(w, "Accept must contain both 'application/json' and 'text/event-stream'", http.StatusBadRequest)
return
}
logger := ensureLogger(h.opts.Logger)
sessionID := req.Header.Get(sessionIDHeader)
var (
sessInfo *sessionInfo
restoredState *ServerSessionState
)
if sessionID != "" {
h.mu.Lock()
sessInfo = h.sessions[sessionID]
h.mu.Unlock()
if sessInfo == nil && !h.opts.Stateless {
if store := h.opts.SessionStateStore; store != nil {
state, err := store.Load(req.Context(), sessionID)
if err != nil {
logger.Error("session state load failed", "session_id", sessionID, "error", err)
http.Error(w, "failed to load session state", http.StatusInternalServerError)
return
}
restoredState = state
if state == nil {
http.Error(w, "session not found", http.StatusNotFound)
return
}
} else {
// Unless we're in 'stateless' mode, which doesn't perform any Session-ID
// validation, we require that the session ID matches a known session.
//
// In stateless mode, a temporary transport is be created below.
http.Error(w, "session not found", http.StatusNotFound)
return
}
}
}
if req.Method == http.MethodDelete {
if sessionID == "" {
http.Error(w, "Bad Request: DELETE requires an Mcp-Session-Id header", http.StatusBadRequest)
return
}
if sessInfo != nil { // sessInfo may be nil in stateless mode
// Closing the session also removes it from h.sessions, due to the
// onClose callback.
sessInfo.session.Close()
} else if restoredState != nil {
// There is no running session, but persisted state exists. Delete it so
// that clients can terminate resumed sessions without an active server.
if store := h.opts.SessionStateStore; store != nil {
if err := store.Delete(req.Context(), sessionID); err != nil && !errors.Is(err, context.Canceled) {
logger.Error("session state delete failed", "session_id", sessionID, "error", err)
http.Error(w, "failed to delete session state", http.StatusInternalServerError)
return
}
}
}
w.WriteHeader(http.StatusNoContent)
return
}
switch req.Method {
case http.MethodPost, http.MethodGet:
if req.Method == http.MethodGet && (h.opts.Stateless || sessionID == "") {
http.Error(w, "GET requires an active session", http.StatusMethodNotAllowed)
return
}
default:
w.Header().Set("Allow", "GET, POST, DELETE")
http.Error(w, "Method Not Allowed: streamable MCP servers support GET, POST, and DELETE requests", http.StatusMethodNotAllowed)
return
}
// [§2.7] of the spec (2025-06-18) states:
//
// "If using HTTP, the client MUST include the MCP-Protocol-Version:
// <protocol-version> HTTP header on all subsequent requests to the MCP
// server, allowing the MCP server to respond based on the MCP protocol
// version.
//
// For example: MCP-Protocol-Version: 2025-06-18
// The protocol version sent by the client SHOULD be the one negotiated during
// initialization.
//
// For backwards compatibility, if the server does not receive an
// MCP-Protocol-Version header, and has no other way to identify the version -
// for example, by relying on the protocol version negotiated during
// initialization - the server SHOULD assume protocol version 2025-03-26.
//
// If the server receives a request with an invalid or unsupported
// MCP-Protocol-Version, it MUST respond with 400 Bad Request."
//
// Since this wasn't present in the 2025-03-26 version of the spec, this
// effectively means:
// 1. IF the client provides a version header, it must be a supported
// version.
// 2. In stateless mode, where we've lost the state of the initialize
// request, we assume that whatever the client tells us is the truth (or
// assume 2025-03-26 if the client doesn't say anything).
//
// This logic matches the typescript SDK.
//
// [§2.7]: https://modelcontextprotocol.io/specification/2025-06-18/basic/transports#protocol-version-header
protocolVersion := req.Header.Get(protocolVersionHeader)
if protocolVersion == "" {
protocolVersion = protocolVersion20250326
}
if !slices.Contains(supportedProtocolVersions, protocolVersion) {
http.Error(w, fmt.Sprintf("Bad Request: Unsupported protocol version (supported versions: %s)", strings.Join(supportedProtocolVersions, ",")), http.StatusBadRequest)
return
}
if sessInfo == nil {
server := h.getServer(req)
if server == nil {
// The getServer argument to NewStreamableHTTPHandler returned nil.
http.Error(w, "no server available", http.StatusBadRequest)
return
}
if sessionID == "" {
// In stateless mode, sessionID may be nonempty even if there's no
// existing transport.
sessionID = server.opts.GetSessionID()
}
transport := &StreamableServerTransport{
SessionID: sessionID,
Stateless: h.opts.Stateless,
EventStore: h.opts.EventStore,
jsonResponse: h.opts.JSONResponse,
logger: h.opts.Logger,
StateStore: h.opts.SessionStateStore,
}
// Sessions without a session ID are also stateless: there's no way to
// address them.
stateless := h.opts.Stateless || sessionID == ""
// To support stateless mode, we initialize the session with a default
// state, so that it doesn't reject subsequent requests.
connectOpts := &ServerSessionOptions{
State: restoredState,
}
if stateless {
// Peek at the body to see if it is initialize or initialized.
// We want those to be handled as usual.
var hasInitialize, hasInitialized bool
{
// TODO: verify that this allows protocol version negotiation for
// stateless servers.
body, err := io.ReadAll(req.Body)
if err != nil {
http.Error(w, "failed to read body", http.StatusInternalServerError)
return
}
req.Body.Close()
// Reset the body so that it can be read later.
req.Body = io.NopCloser(bytes.NewBuffer(body))
msgs, _, err := readBatch(body)
if err == nil {
for _, msg := range msgs {
if req, ok := msg.(*jsonrpc.Request); ok {
switch req.Method {
case methodInitialize:
hasInitialize = true
case notificationInitialized:
hasInitialized = true
}
}
}
}
}
// If we don't have InitializeParams or InitializedParams in the request,
// set the initial state to a default value.
state := new(ServerSessionState)
if !hasInitialize {
state.InitializeParams = &InitializeParams{
ProtocolVersion: protocolVersion,
}
}
if !hasInitialized {
state.InitializedParams = new(InitializedParams)
}
state.LogLevel = "info"
connectOpts.State = state
} else {
// Cleanup is only required in stateful mode, as transportation is
// not stored in the map otherwise.
connectOpts = &ServerSessionOptions{
State: restoredState,
onClose: func() {
h.mu.Lock()
defer h.mu.Unlock()
if info, ok := h.sessions[transport.SessionID]; ok {
info.stopTimer()
delete(h.sessions, transport.SessionID)
if h.onTransportDeletion != nil {
h.onTransportDeletion(transport.SessionID)
}
}
if store := h.opts.SessionStateStore; store != nil && transport.SessionID != "" {
if err := store.Delete(context.Background(), transport.SessionID); err != nil {
logger.Error("session state delete failed", "session_id", transport.SessionID, "error", err)
}
}
},
}
}
// Pass req.Context() here, to allow middleware to add context values.
// The context is detached in the jsonrpc2 library when handling the
// long-running stream.
session, err := server.Connect(req.Context(), transport, connectOpts)
if err != nil {
http.Error(w, "failed connection", http.StatusInternalServerError)
return
}
sessInfo = &sessionInfo{
session: session,
transport: transport,
}
if stateless {
// Stateless mode: close the session when the request exits.
defer session.Close() // close the fake session after handling the request
} else {
// Otherwise, save the transport so that it can be reused
// Clean up the session when it times out.
//
// Note that the timer here may fire multiple times, but
// sessInfo.session.Close is idempotent.
if h.opts.SessionTimeout > 0 {
sessInfo.timeout = h.opts.SessionTimeout
sessInfo.timer = time.AfterFunc(sessInfo.timeout, func() {
sessInfo.session.Close()
})
}
h.mu.Lock()
h.sessions[transport.SessionID] = sessInfo
h.mu.Unlock()
defer func() {
// If initialization failed, clean up the session (#578).
if session.InitializeParams() == nil {
// Initialization failed.
session.Close()
}
}()
}
}
if req.Method == http.MethodPost {
sessInfo.startPOST()
defer sessInfo.endPOST()
}
sessInfo.transport.ServeHTTP(w, req)
}
// A StreamableServerTransport implements the server side of the MCP streamable
// transport.
//
// Each StreamableServerTransport must be connected (via [Server.Connect]) at
// most once, since [StreamableServerTransport.ServeHTTP] serves messages to
// the connected session.
//
// Reads from the streamable server connection receive messages from http POST
// requests from the client. Writes to the streamable server connection are
// sent either to the related stream, or to the standalone SSE stream,
// according to the following rules:
// - JSON-RPC responses to incoming requests are always routed to the
// appropriate HTTP response.
// - Requests or notifications made with a context.Context value derived from
// an incoming request handler, are routed to the HTTP response
// corresponding to that request, unless it has already terminated, in
// which case they are routed to the standalone SSE stream.
// - Requests or notifications made with a detached context.Context value are
// routed to the standalone SSE stream.
type StreamableServerTransport struct {
// SessionID is the ID of this session.
//
// If SessionID is the empty string, this is a 'stateless' session, which has
// limited ability to communicate with the client. Otherwise, the session ID
// must be globally unique, that is, different from any other session ID
// anywhere, past and future. (We recommend using a crypto random number
// generator to produce one, as with [crypto/rand.Text].)
SessionID string
// Stateless controls whether the eventstore is 'Stateless'. Server sessions
// connected to a stateless transport are disallowed from making outgoing
// requests.
//
// See also [StreamableHTTPOptions.Stateless].
Stateless bool
// EventStore enables stream resumption.
//
// If set, EventStore will be used to persist stream events and replay them
// upon stream resumption.
EventStore EventStore
// StateStore receives session state updates so that callers can resume
// sessions across processes.
StateStore ServerSessionStateStore
// jsonResponse, if set, tells the server to prefer to respond to requests
// using application/json responses rather than text/event-stream.
//
// Specifically, responses will be application/json whenever incoming POST
// request contain only a single message. In this case, notifications or
// requests made within the context of a server request will be sent to the
// standalone SSE stream, if any.
//
// TODO(rfindley): jsonResponse should be exported, since
// StreamableHTTPOptions.JSONResponse is exported, and we want to allow users
// to write their own streamable HTTP handler.
jsonResponse bool
// optional logger provided through the [StreamableHTTPOptions.Logger].
//
// TODO(rfindley): logger should be exported, since we want to allow users
// to write their own streamable HTTP handler.
logger *slog.Logger
// connection is non-nil if and only if the transport has been connected.
connection *streamableServerConn
}
// Connect implements the [Transport] interface.
func (t *StreamableServerTransport) Connect(ctx context.Context) (Connection, error) {
if t.connection != nil {
return nil, fmt.Errorf("transport already connected")
}
t.connection = &streamableServerConn{
sessionID: t.SessionID,
stateless: t.Stateless,
eventStore: t.EventStore,
stateStore: t.StateStore,
jsonResponse: t.jsonResponse,
logger: ensureLogger(t.logger), // see #556: must be non-nil
incoming: make(chan jsonrpc.Message, 10),
done: make(chan struct{}),
streams: make(map[string]*stream),
requestStreams: make(map[jsonrpc.ID]string),
}
// Stream 0 corresponds to the standalone SSE stream.
//
// It is always text/event-stream, since it must carry arbitrarily many
// messages.
var err error
t.connection.streams[""], err = t.connection.newStream(ctx, nil, "")
if err != nil {
return nil, err
}
return t.connection, nil
}
type streamableServerConn struct {
sessionID string
stateless bool
jsonResponse bool
eventStore EventStore
stateStore ServerSessionStateStore
logger *slog.Logger
incoming chan jsonrpc.Message // messages from the client to the server
mu sync.Mutex // guards all fields below
// Sessions are closed exactly once.
isDone bool
done chan struct{}
// Sessions can have multiple logical connections (which we call streams),
// corresponding to HTTP requests. Additionally, streams may be resumed by
// subsequent HTTP requests, when the HTTP connection is terminated
// unexpectedly.
//
// Therefore, we use a logical stream ID to key the stream state, and
// perform the accounting described below when incoming HTTP requests are
// handled.
// streams holds the logical streams for this session, keyed by their ID.
//
// Lifecycle: streams persist until all of their responses are received from
// the server.
streams map[string]*stream
// requestStreams maps incoming requests to their logical stream ID.
//
// Lifecycle: requestStreams persist until their response is received.
requestStreams map[jsonrpc.ID]string
}
func (c *streamableServerConn) SessionID() string {
return c.sessionID
}
func (c *streamableServerConn) sessionUpdated(state ServerSessionState) {
if c.stateStore == nil || c.sessionID == "" || c.stateless {
return
}
stateCopy := state
if err := c.stateStore.Save(context.Background(), c.sessionID, &stateCopy); err != nil {
c.logger.Error("session state save failed", "session_id", c.sessionID, "error", err)
}
}
// A stream is a single logical stream of SSE events within a server session.
// A stream begins with a client request, or with a client GET that has
// no Last-Event-ID header.
//
// A stream ends only when its session ends; we cannot determine its end otherwise,
// since a client may send a GET with a Last-Event-ID that references the stream
// at any time.
type stream struct {
// id is the logical ID for the stream, unique within a session.
//
// The standalone SSE stream has id "".
id string
// mu guards the fields below, as well as storage of new messages in the
// connection's event store (if any).
mu sync.Mutex
// If non-nil, deliver writes data directly to the HTTP response.
//
// Only one HTTP response may receive messages at a given time. An active
// HTTP connection acquires ownership of the stream by setting this field.
deliver func(data []byte, final bool) error
// streamRequests is the set of unanswered incoming requests for the stream.
//
// Requests are removed when their response has been received.
requests map[jsonrpc.ID]struct{}
}
// doneLocked reports whether the stream is logically complete.
//
// s.mu must be held while calling this function.
func (s *stream) doneLocked() bool {
return len(s.requests) == 0 && s.id != ""
}
func (c *streamableServerConn) newStream(ctx context.Context, requests map[jsonrpc.ID]struct{}, id string) (*stream, error) {
if c.eventStore != nil {
if err := c.eventStore.Open(ctx, c.sessionID, id); err != nil {
return nil, err
}
}
return &stream{
id: id,
requests: requests,
}, nil
}
// We track the incoming request ID inside the handler context using
// idContextValue, so that notifications and server->client calls that occur in
// the course of handling incoming requests are correlated with the incoming
// request that caused them, and can be dispatched as server-sent events to the
// correct HTTP request.
//
// Currently, this is implemented in [ServerSession.handle]. This is not ideal,
// because it means that a user of the MCP package couldn't implement the
// streamable transport, as they'd lack this privileged access.
//
// If we ever wanted to expose this mechanism, we have a few options:
// 1. Make ServerSession an interface, and provide an implementation of
// ServerSession to handlers that closes over the incoming request ID.
// 2. Expose a 'HandlerTransport' interface that allows transports to provide
// a handler middleware, so that we don't hard-code this behavior in
// ServerSession.handle.
// 3. Add a `func ForRequest(context.Context) jsonrpc.ID` accessor that lets
// any transport access the incoming request ID.
//
// For now, by giving only the StreamableServerTransport access to the request
// ID, we avoid having to make this API decision.
type idContextKey struct{}
// ServeHTTP handles a single HTTP request for the session.
func (t *StreamableServerTransport) ServeHTTP(w http.ResponseWriter, req *http.Request) {
if t.connection == nil {
http.Error(w, "transport not connected", http.StatusInternalServerError)
return
}
switch req.Method {
case http.MethodGet:
t.connection.serveGET(w, req)
case http.MethodPost:
t.connection.servePOST(w, req)
default:
// Should not be reached, as this is checked in StreamableHTTPHandler.ServeHTTP.
w.Header().Set("Allow", "GET, POST")
http.Error(w, "unsupported method", http.StatusMethodNotAllowed)
return
}
}
// serveGET streams messages to a hanging http GET, with stream ID and last
// message parsed from the Last-Event-ID header.
//
// It returns an HTTP status code and error message.
func (c *streamableServerConn) serveGET(w http.ResponseWriter, req *http.Request) {
// streamID "" corresponds to the default GET request.
streamID := ""
// By default, we haven't seen a last index. Since indices start at 0, we represent
// that by -1. This is incremented just before each event is written.
lastIdx := -1
if len(req.Header.Values("Last-Event-ID")) > 0 {
eid := req.Header.Get("Last-Event-ID")
var ok bool
streamID, lastIdx, ok = parseEventID(eid)
if !ok {
http.Error(w, fmt.Sprintf("malformed Last-Event-ID %q", eid), http.StatusBadRequest)
return
}
if c.eventStore == nil {
http.Error(w, "stream replay unsupported", http.StatusBadRequest)
return
}
}
ctx, cancel := context.WithCancel(req.Context())
defer cancel()
stream, done := c.acquireStream(ctx, w, streamID, &lastIdx)
if stream == nil {
return
}
// Release the stream when we're done.
defer func() {
stream.mu.Lock()
stream.deliver = nil
stream.mu.Unlock()
}()
select {
case <-ctx.Done():
// request cancelled
case <-done:
// request complete
case <-c.done:
// session closed
}
}
// writeEvent writes an SSE event to w corresponding to the given stream, data, and index.
// lastIdx is incremented before writing, so that it continues to point to the index of the
// last event written to the stream.
func (c *streamableServerConn) writeEvent(w http.ResponseWriter, stream *stream, data []byte, lastIdx *int) error {
*lastIdx++
e := Event{
Name: "message",
Data: data,
}
if c.eventStore != nil {
e.ID = formatEventID(stream.id, *lastIdx)
}
if _, err := writeEvent(w, e); err != nil {
return err
}
return nil
}
// acquireStream acquires the stream and replays all events since lastIdx, if
// any, updating lastIdx accordingly. If non-nil, the resulting stream will be
// registered for receiving new messages, and the resulting done channel will
// be closed when all related messages have been delivered.
//
// If any errors occur, they will be written to w and the resulting stream will
// be nil. The resulting stream may also be nil if the stream is complete.
//
// Importantly, this function must hold the stream mutex until done replaying
// all messages, so that no delivery or storage of new messages occurs while
// the stream is still replaying.
func (c *streamableServerConn) acquireStream(ctx context.Context, w http.ResponseWriter, streamID string, lastIdx *int) (*stream, chan struct{}) {
// if tempStream is set, the stream is done and we're just replaying messages.
//
// We record a temporary stream to claim exclusive replay rights.
tempStream := false
c.mu.Lock()
s, ok := c.streams[streamID]
if !ok {
// The stream is logically done, but claim exclusive rights to replay it by
// adding a temporary entry in the streams map.
//
// We create this entry with a non-nil deliver function, to ensure it isn't
// claimed by another request before we lock it below.
tempStream = true
s = &stream{
id: streamID,
deliver: func([]byte, bool) error { return nil },
}
c.streams[streamID] = s
// Since this stream is transient, we must clean up after replaying.
defer func() {
c.mu.Lock()
delete(c.streams, streamID)
c.mu.Unlock()
}()
}
c.mu.Unlock()
s.mu.Lock()
defer s.mu.Unlock()
// Check that this stream wasn't claimed by another request.
if !tempStream && s.deliver != nil {
http.Error(w, "stream ID conflicts with ongoing stream", http.StatusConflict)
return nil, nil
}
// Collect events to replay. Collect them all before writing, so that we
// have an opportunity to set the HTTP status code on an error.
//
// As indicated above, we must do that while holding stream.mu, so that no
// new messages are added to the eventstore until we've replayed all previous
// messages, and registered our delivery function.
var toReplay [][]byte
if c.eventStore != nil {
for data, err := range c.eventStore.After(ctx, c.SessionID(), s.id, *lastIdx) {
if err != nil {
// We can't replay events, perhaps because the underlying event store
// has garbage collected its storage.
//
// We must be careful here: any 404 will signal to the client that the
// *session* is not found, rather than the stream.
//
// 400 is not really accurate, but should at least have no side effects.
// Other SDKs (typescript) do not have a mechanism for events to be purged.
http.Error(w, "failed to replay events", http.StatusBadRequest)
return nil, nil
}
toReplay = append(toReplay, data)
}
}
w.Header().Set("Cache-Control", "no-cache, no-transform")
w.Header().Set("Content-Type", "text/event-stream") // Accept checked in [StreamableHTTPHandler]
w.Header().Set("Connection", "keep-alive")
if s.id == "" {
// Issue #410: the standalone SSE stream is likely not to receive messages
// for a long time. Ensure that headers are flushed.
w.WriteHeader(http.StatusOK)
if f, ok := w.(http.Flusher); ok {
f.Flush()
}
}
for _, data := range toReplay {
if err := c.writeEvent(w, s, data, lastIdx); err != nil {
return nil, nil
}
}
if tempStream || s.doneLocked() {
// Nothing more to do.
return nil, nil
}
// The stream is not done: register a delivery function before the stream is
// unlocked, allowing the connection to write new events.
done := make(chan struct{})
s.deliver = func(data []byte, final bool) error {
if err := ctx.Err(); err != nil {
return err
}
err := c.writeEvent(w, s, data, lastIdx)
if final {
close(done)
}
return err
}
return s, done
}
// servePOST handles an incoming message, and replies with either an outgoing
// message stream or single response object, depending on whether the
// jsonResponse option is set.
//
// It returns an HTTP status code and error message.
func (c *streamableServerConn) servePOST(w http.ResponseWriter, req *http.Request) {
if len(req.Header.Values("Last-Event-ID")) > 0 {
http.Error(w, "can't send Last-Event-ID for POST request", http.StatusBadRequest)
return
}
// Read incoming messages.
body, err := io.ReadAll(req.Body)
if err != nil {
http.Error(w, "failed to read body", http.StatusBadRequest)
return
}
if len(body) == 0 {
http.Error(w, "POST requires a non-empty body", http.StatusBadRequest)
return
}
incoming, isBatch, err := readBatch(body)
if err != nil {
http.Error(w, fmt.Sprintf("malformed payload: %v", err), http.StatusBadRequest)
return
}
protocolVersion := req.Header.Get(protocolVersionHeader)
if protocolVersion == "" {
protocolVersion = protocolVersion20250326
}
if isBatch && protocolVersion >= protocolVersion20250618 {
http.Error(w, fmt.Sprintf("JSON-RPC batching is not supported in %s and later (request version: %s)", protocolVersion20250618, protocolVersion), http.StatusBadRequest)
return
}
// TODO(rfindley): no tests fail if we reject batch JSON requests entirely.
// We need to test this with older protocol versions.
// if isBatch && c.jsonResponse {
// http.Error(w, "server does not support batch requests", http.StatusBadRequest)
// return
// }
calls := make(map[jsonrpc.ID]struct{})
tokenInfo := auth.TokenInfoFromContext(req.Context())
isInitialize := false
for _, msg := range incoming {
if jreq, ok := msg.(*jsonrpc.Request); ok {
// Preemptively check that this is a valid request, so that we can fail
// the HTTP request. If we didn't do this, a request with a bad method or
// missing ID could be silently swallowed.
if _, err := checkRequest(jreq, serverMethodInfos); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
if jreq.Method == methodInitialize {
isInitialize = true
}
jreq.Extra = &RequestExtra{
TokenInfo: tokenInfo,
Header: req.Header,
}
if jreq.IsCall() {
calls[jreq.ID] = struct{}{}
}
}
}
// If we don't have any calls, we can just publish the incoming messages and return.
// No need to track a logical stream.
if len(calls) == 0 {
for _, msg := range incoming {
select {
case c.incoming <- msg:
case <-c.done:
// The session is closing. Since we haven't yet written any data to the
// response, we can signal to the client that the session is gone.
http.Error(w, "session is closing", http.StatusNotFound)
return
}
}
w.WriteHeader(http.StatusAccepted)
return
}
// Invariant: we have at least one call.
//
// Create a logical stream to track its responses.
// Important: don't publish the incoming messages until the stream is
// registered, as the server may attempt to respond to imcoming messages as