Skip to content

Commit 28dce5c

Browse files
wucm667wucm667
committed
mcp: remove support for batching on streamable connections
Batching was already deprecated and rejected for protocol versions >= 2025-06-18. Remove it entirely to simplify the streamable transport code. Replace readBatch calls with single message decoding, remove batch-related checks and comments, and delete batch-specific tests. Closes #911 Signed-off-by: wucm667 <stevenwucongmin@gmail.com>
1 parent 438cd43 commit 28dce5c

2 files changed

Lines changed: 71 additions & 144 deletions

File tree

‎mcp/streamable.go‎

Lines changed: 71 additions & 101 deletions
Original file line numberDiff line numberDiff line change
@@ -435,16 +435,14 @@ func (h *StreamableHTTPHandler) ServeHTTP(w http.ResponseWriter, req *http.Reque
435435
// Reset the body so that it can be read later.
436436
req.Body = io.NopCloser(bytes.NewBuffer(body))
437437

438-
msgs, _, err := readBatch(body)
438+
msg, err := jsonrpc2.DecodeMessage(body)
439439
if err == nil {
440-
for _, msg := range msgs {
441-
if req, ok := msg.(*jsonrpc.Request); ok {
442-
switch req.Method {
443-
case methodInitialize:
444-
hasInitialize = true
445-
case notificationInitialized:
446-
hasInitialized = true
447-
}
440+
if req, ok := msg.(*jsonrpc.Request); ok {
441+
switch req.Method {
442+
case methodInitialize:
443+
hasInitialize = true
444+
case notificationInitialized:
445+
hasInitialized = true
448446
}
449447
}
450448
}
@@ -726,8 +724,6 @@ type stream struct {
726724
// collected here until the stream is complete, at which point they are
727725
// flushed as a single JSON response. Note that the non-nilness of this field
728726
// is significant, as it signals the expected content type.
729-
//
730-
// Note: if we remove support for batching, this could just be a bool.
731727
pendingJSONMessages []json.RawMessage
732728

733729
// w is the HTTP response writer for this stream. A non-nil w indicates
@@ -752,9 +748,6 @@ type stream struct {
752748
// requests is the set of unanswered incoming requests for the stream.
753749
//
754750
// Requests are removed when their response has been received.
755-
// In practice, there is only one request, but in the 2025-03-26 version of
756-
// the spec and earlier there was a concept of batching, in which POST
757-
// payloads could hold multiple requests or responses.
758751
requests map[jsonrpc.ID]struct{}
759752
}
760753

@@ -1113,10 +1106,7 @@ func (c *streamableServerConn) servePOST(w http.ResponseWriter, req *http.Reques
11131106
http.Error(w, "POST requires a non-empty body", http.StatusBadRequest)
11141107
return
11151108
}
1116-
// TODO(#674): once we've documented the support matrix for 2025-03-26 and
1117-
// earlier, drop support for matching entirely; that will simplify this
1118-
// logic.
1119-
incoming, isBatch, err := readBatch(body)
1109+
incoming, err := jsonrpc2.DecodeMessage(body)
11201110
if err != nil {
11211111
http.Error(w, fmt.Sprintf("malformed payload: %v", err), http.StatusBadRequest)
11221112
return
@@ -1127,81 +1117,65 @@ func (c *streamableServerConn) servePOST(w http.ResponseWriter, req *http.Reques
11271117
protocolVersion = protocolVersion20250326
11281118
}
11291119

1130-
if isBatch && protocolVersion >= protocolVersion20250618 {
1131-
http.Error(w, fmt.Sprintf("JSON-RPC batching is not supported in %s and later (request version: %s)", protocolVersion20250618, protocolVersion), http.StatusBadRequest)
1132-
return
1133-
}
1134-
1135-
// TODO(rfindley): no tests fail if we reject batch JSON requests entirely.
1136-
// We need to test this with older protocol versions.
1137-
// if isBatch && c.jsonResponse {
1138-
// http.Error(w, "server does not support batch requests", http.StatusBadRequest)
1139-
// return
1140-
// }
1141-
11421120
calls := make(map[jsonrpc.ID]struct{})
11431121
tokenInfo := auth.TokenInfoFromContext(req.Context())
11441122
isInitialize := false
11451123
var initializeProtocolVersion string
1146-
for _, msg := range incoming {
1147-
if jreq, ok := msg.(*jsonrpc.Request); ok {
1148-
// Preemptively check that this is a valid request, so that we can fail
1149-
// the HTTP request. If we didn't do this, a request with a bad method or
1150-
// missing ID could be silently swallowed.
1151-
if _, err := checkRequest(jreq, serverMethodInfos); err != nil {
1152-
http.Error(w, err.Error(), http.StatusBadRequest)
1153-
return
1124+
if jreq, ok := incoming.(*jsonrpc.Request); ok {
1125+
// Preemptively check that this is a valid request, so that we can fail
1126+
// the HTTP request. If we didn't do this, a request with a bad method or
1127+
// missing ID could be silently swallowed.
1128+
if _, err := checkRequest(jreq, serverMethodInfos); err != nil {
1129+
http.Error(w, err.Error(), http.StatusBadRequest)
1130+
return
1131+
}
1132+
if jreq.Method == methodInitialize {
1133+
isInitialize = true
1134+
// Extract the protocol version from InitializeParams.
1135+
var params InitializeParams
1136+
if err := internaljson.Unmarshal(jreq.Params, &params); err == nil {
1137+
initializeProtocolVersion = params.ProtocolVersion
11541138
}
1155-
if jreq.Method == methodInitialize {
1156-
isInitialize = true
1157-
// Extract the protocol version from InitializeParams.
1158-
var params InitializeParams
1159-
if err := internaljson.Unmarshal(jreq.Params, &params); err == nil {
1160-
initializeProtocolVersion = params.ProtocolVersion
1139+
}
1140+
// Include metadata for all requests (including notifications).
1141+
jreq.Extra = &RequestExtra{
1142+
TokenInfo: tokenInfo,
1143+
Header: req.Header,
1144+
}
1145+
if jreq.IsCall() {
1146+
calls[jreq.ID] = struct{}{}
1147+
// See the doc for CloseSSEStream: allow the request handler to
1148+
// explicitly close the ongoing stream.
1149+
jreq.Extra.(*RequestExtra).CloseSSEStream = func(args CloseSSEStreamArgs) {
1150+
c.mu.Lock()
1151+
streamID, ok := c.requestStreams[jreq.ID]
1152+
var stream *stream
1153+
if ok {
1154+
stream = c.streams[streamID]
11611155
}
1162-
}
1163-
// Include metadata for all requests (including notifications).
1164-
jreq.Extra = &RequestExtra{
1165-
TokenInfo: tokenInfo,
1166-
Header: req.Header,
1167-
}
1168-
if jreq.IsCall() {
1169-
calls[jreq.ID] = struct{}{}
1170-
// See the doc for CloseSSEStream: allow the request handler to
1171-
// explicitly close the ongoing stream.
1172-
jreq.Extra.(*RequestExtra).CloseSSEStream = func(args CloseSSEStreamArgs) {
1173-
c.mu.Lock()
1174-
streamID, ok := c.requestStreams[jreq.ID]
1175-
var stream *stream
1176-
if ok {
1177-
stream = c.streams[streamID]
1178-
}
1179-
c.mu.Unlock()
1156+
c.mu.Unlock()
11801157

1181-
if stream != nil {
1182-
stream.close(args.RetryAfter)
1183-
}
1158+
if stream != nil {
1159+
stream.close(args.RetryAfter)
11841160
}
11851161
}
11861162
}
11871163
}
11881164

11891165
// Validate MCP standard headers (Mcp-Method, Mcp-Name)
1190-
if !isBatch && len(incoming) == 1 {
1191-
if err := validateMcpHeaders(req.Header, incoming[0]); err != nil {
1192-
resp := &jsonrpc.Response{
1193-
Error: jsonrpc2.NewError(CodeHeaderMismatch, err.Error()),
1194-
}
1195-
if jreq, ok := incoming[0].(*jsonrpc.Request); ok {
1196-
resp.ID = jreq.ID
1197-
}
1198-
w.Header().Set("Content-Type", "application/json")
1199-
w.WriteHeader(http.StatusBadRequest)
1200-
if data, err := jsonrpc2.EncodeMessage(resp); err == nil {
1201-
w.Write(data)
1202-
}
1203-
return
1166+
if err := validateMcpHeaders(req.Header, incoming); err != nil {
1167+
resp := &jsonrpc.Response{
1168+
Error: jsonrpc2.NewError(CodeHeaderMismatch, err.Error()),
1169+
}
1170+
if jreq, ok := incoming.(*jsonrpc.Request); ok {
1171+
resp.ID = jreq.ID
12041172
}
1173+
w.Header().Set("Content-Type", "application/json")
1174+
w.WriteHeader(http.StatusBadRequest)
1175+
if data, err := jsonrpc2.EncodeMessage(resp); err == nil {
1176+
w.Write(data)
1177+
}
1178+
return
12051179
}
12061180

12071181
// The prime and close events were added in protocol version 2025-11-25 (SEP-1699).
@@ -1220,15 +1194,13 @@ func (c *streamableServerConn) servePOST(w http.ResponseWriter, req *http.Reques
12201194
//
12211195
// [§2.1.4]: https://modelcontextprotocol.io/specification/2025-11-25/basic/transports#sending-messages-to-the-server
12221196
if len(calls) == 0 {
1223-
for _, msg := range incoming {
1224-
select {
1225-
case c.incoming <- msg:
1226-
case <-c.done:
1227-
// The session is closing. Since we haven't yet written any data to the
1228-
// response, we can signal to the client that the session is gone.
1229-
http.Error(w, "session is closing", http.StatusNotFound)
1230-
return
1231-
}
1197+
select {
1198+
case c.incoming <- incoming:
1199+
case <-c.done:
1200+
// The session is closing. Since we haven't yet written any data to the
1201+
// response, we can signal to the client that the session is gone.
1202+
http.Error(w, "session is closing", http.StatusNotFound)
1203+
return
12321204
}
12331205
w.WriteHeader(http.StatusAccepted)
12341206
return
@@ -1303,19 +1275,17 @@ func (c *streamableServerConn) servePOST(w http.ResponseWriter, req *http.Reques
13031275
c.mu.Unlock()
13041276

13051277
// Publish incoming messages.
1306-
for _, msg := range incoming {
1307-
select {
1308-
case c.incoming <- msg:
1309-
// Note: don't select on req.Context().Done() here, since we've already
1310-
// received the requests and may have already published a response message
1311-
// or notification. The client could resume the stream.
1312-
//
1313-
// In fact, this send could be in a separate goroutine.
1314-
case <-c.done:
1315-
// Session closed: we don't know if any data has been written, so it's
1316-
// too late to write a status code here.
1317-
return
1318-
}
1278+
select {
1279+
case c.incoming <- incoming:
1280+
// Note: don't select on req.Context().Done() here, since we've already
1281+
// received the requests and may have already published a response message
1282+
// or notification. The client could resume the stream.
1283+
//
1284+
// In fact, this send could be in a separate goroutine.
1285+
case <-c.done:
1286+
// Session closed: we don't know if any data has been written, so it's
1287+
// too late to write a status code here.
1288+
return
13191289
}
13201290

13211291
c.hangResponse(req.Context(), done)

‎mcp/streamable_test.go‎

Lines changed: 0 additions & 43 deletions
Original file line numberDiff line numberDiff line change
@@ -966,49 +966,6 @@ func TestStreamableServerTransport(t *testing.T) {
966966
},
967967
wantSessions: 1,
968968
},
969-
{
970-
name: "batch rejected on 2025-06-18",
971-
requests: []streamableRequest{
972-
initialize,
973-
initialized,
974-
{
975-
method: "POST",
976-
// Explicitly set the protocol version header
977-
headers: http.Header{"MCP-Protocol-Version": {"2025-06-18"}},
978-
// Two messages => batch. Expect reject.
979-
messages: []jsonrpc.Message{
980-
req(101, "tools/call", &CallToolParams{Name: "tool"}),
981-
req(102, "tools/call", &CallToolParams{Name: "tool"}),
982-
},
983-
wantStatusCode: http.StatusBadRequest,
984-
wantBodyContaining: "batch",
985-
},
986-
},
987-
wantSessions: 1,
988-
},
989-
{
990-
name: "batch accepted on 2025-03-26",
991-
requests: []streamableRequest{
992-
initialize,
993-
initialized,
994-
{
995-
method: "POST",
996-
headers: http.Header{"MCP-Protocol-Version": {"2025-03-26"}},
997-
// Two messages => batch. Expect OK with two responses in order.
998-
messages: []jsonrpc.Message{
999-
// Note: only include one request here, because responses are not
1000-
// necessarily sorted.
1001-
req(201, "tools/call", &CallToolParams{Name: "tool"}),
1002-
req(0, "notifications/roots/list_changed", &RootsListChangedParams{}),
1003-
},
1004-
wantStatusCode: http.StatusOK,
1005-
wantMessages: []jsonrpc.Message{
1006-
resp(201, &CallToolResult{Content: []Content{}}, nil),
1007-
},
1008-
},
1009-
},
1010-
wantSessions: 1,
1011-
},
1012969
{
1013970
name: "tool notification",
1014971
tool: func(t *testing.T, ctx context.Context, req *CallToolRequest) {

0 commit comments

Comments
 (0)