streamable.go

// 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.

// NOTE: see streamable_server.go and streamable_client.go for detailed
// documentation of the streamable server design.
// TODO: move the client and server logic into those files.

package mcp

import (
	"bytes"
	"context"
	crand "crypto/rand"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"maps"
	"math"
	"math/rand/v2"
	"mime"
	"net"
	"net/http"
	"slices"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/modelcontextprotocol/go-sdk/auth"
	internaljson "github.com/modelcontextprotocol/go-sdk/internal/json"
	"github.com/modelcontextprotocol/go-sdk/internal/jsonrpc2"
	"github.com/modelcontextprotocol/go-sdk/internal/mcpgodebug"
	"github.com/modelcontextprotocol/go-sdk/internal/util"
	"github.com/modelcontextprotocol/go-sdk/internal/xcontext"
	"github.com/modelcontextprotocol/go-sdk/jsonrpc"
	"golang.org/x/oauth2"
)

// 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
	// userID is the user ID from the TokenInfo when the session was created.
	// If non-empty, subsequent requests must have the same user ID to prevent
	// session hijacking.
	userID string

	// 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
}

// StreamableHTTPRequestSummary contains redacted metadata about a single
// JSON-RPC message decoded from a streamable HTTP POST body.
type StreamableHTTPRequestSummary struct {
	// Method is the method of a decoded JSON-RPC request. It is empty for a
	// response. The method has not yet been validated and may contain an
	// arbitrary attacker-controlled string.
	Method string

	// RequestID is valid only for a JSON-RPC call. IDs use the same coercion
	// rules as [jsonrpc.DecodeMessage].
	RequestID jsonrpc.ID

	// IsNotification reports whether the message is a JSON-RPC notification.
	IsNotification bool

	// IsResponse reports whether the message is a JSON-RPC response.
	IsResponse bool
}

// 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 signals 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 read or set the Mcp-Session-Id header, and
	// uses a temporary session with default initialization parameters for each
	// request. [ServerOptions.GetSessionID] is not consulted. 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].
	// In Stateless mode, GET and DELETE requests return 405 Method Not Allowed.
	//
	// This mode aligns with the sessionless direction of the MCP spec; see
	// [SEP-2567]. The previous behavior, in which stateless servers still
	// honored Mcp-Session-Id, can be restored temporarily via the
	// MCPGODEBUG compatibility parameter "allowsessionsinstateless=1".
	//
	// [SEP-2567]: https://github.com/modelcontextprotocol/modelcontextprotocol/pull/2567
	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

	// DisableLocalhostProtection disables automatic DNS rebinding protection.
	// By default, requests arriving via a localhost address (127.0.0.1, [::1])
	// that have a non-localhost Host header are rejected with 403 Forbidden.
	// This protects against DNS rebinding attacks regardless of whether the
	// server is listening on localhost specifically or on 0.0.0.0.
	//
	// Only disable this if you understand the security implications.
	// See: https://modelcontextprotocol.io/specification/2025-11-25/basic/security_best_practices#local-mcp-server-compromise
	DisableLocalhostProtection bool

	// CrossOriginProtection allows to customize cross-origin protection.
	// The deny handler set in the CrossOriginProtection through SetDenyHandler
	// is ignored.
	// If nil, no cross-origin protection is applied.
	//
	// Deprecated: wrap the handler with cross-origin protection middleware
	// instead. For example:
	//
	//   handler := mcp.NewStreamableHTTPHandler(...)
	//   protection := http.NewCrossOriginProtection()
	//   protectedHandler := protection.Handler(handler)
	CrossOriginProtection *http.CrossOriginProtection

	// MaxRequestBodyBytes limits the number of bytes read from any incoming
	// HTTP request body. Requests that exceed this limit are rejected with
	// 413 Request Entity Too Large.
	//
	// The limit is enforced during the read, so it applies uniformly to
	// requests using Content-Length, Transfer-Encoding: chunked, or HTTP/2
	// (which has no Content-Length).
	//
	// If zero, [DefaultMaxRequestBodyBytes] is used.
	// A negative value disables the limit entirely; do not use this on
	// servers exposed to untrusted clients.
	MaxRequestBodyBytes int64

	// PropagateRequestCancellation, when true, ties the in-flight handler
	// context to the originating HTTP request's context. Only applies to
	// requests using the >= 2026-07-28 protocol, where the POST is the whole
	// request lifecycle.
	// The handler context cancels whenever the HTTP request context does as the
	// response can no longer be delivered, so cancelling the handler is safe.
	//
	// Requests using older protocol versions (including those routed through
	// the allowsessionsinstateless compatibility path) are unaffected.
	PropagateRequestCancellation bool

	// OnRequestSummary, when non-nil, observes redacted metadata for a single
	// JSON-RPC message decoded from a streamable HTTP POST body. It is not called
	// for JSON-RPC batches. The callback receives the HTTP request's context and
	// runs synchronously before validation and dispatch of the decoded message.
	// It may be called concurrently for different requests and should return
	// promptly; in particular, it must not wait for processing of the same
	// request. Panics are not recovered. Only the summary is redacted; the context
	// may contain values added by authentication or other middleware.
	//
	// The callback is not invoked for requests rejected before this point,
	// including HTTP, authorization, session-routing, and connection failures,
	// even if connection setup previously inspected the body. Use HTTP middleware
	// to observe those failures. Use [Server.AddReceivingMiddleware] to observe
	// dispatched messages; this callback additionally observes decoded messages
	// that are rejected before dispatch without exposing their parameters.
	OnRequestSummary func(context.Context, StreamableHTTPRequestSummary)

	// StreamKeepAlive is how long the response stream of a subscriptions/listen
	// request may carry no bytes before an SSE comment line is written, so
	// that idle-timeout intermediaries do not sever the long-lived stream, as
	// the 2026-07-28 Streamable HTTP specification encourages. SSE clients
	// ignore comment lines. Other SSE responses are not kept alive.
	//
	// The keep-alive starts only after the listen acknowledgment has
	// committed the response headers, since until then the HTTP status may
	// still have to change (see #1229). A write failure ends the stream as a
	// disconnect.
	//
	// If zero, [DefaultStreamKeepAlive] is used. A negative value disables
	// the keep-alive.
	StreamKeepAlive time.Duration
}

// DefaultMaxRequestBodyBytes is the default value used for
// [StreamableHTTPOptions.MaxRequestBodyBytes] when it is left at zero.
const DefaultMaxRequestBodyBytes = 4 << 20 // 4 MiB

// DefaultStreamKeepAlive is the default value used for
// [StreamableHTTPOptions.StreamKeepAlive] when it is left at zero.
const DefaultStreamKeepAlive = 30 * time.Second

// 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
	}

	h.opts.Logger = ensureLogger(h.opts.Logger)

	if h.opts.MaxRequestBodyBytes == 0 {
		h.opts.MaxRequestBodyBytes = DefaultMaxRequestBodyBytes
	}
	if h.opts.StreamKeepAlive == 0 {
		h.opts.StreamKeepAlive = DefaultStreamKeepAlive
	}

	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()
	}
}

// allowsessionsinstateless is a compatibility parameter that restores the old
// behavior of reading and using Mcp-Session-Id headers in stateless mode. When
// set to "1", stateless servers will read the session ID from the request
// header (or generate one via GetSessionID), set it on response headers, and
// accept DELETE requests. When unset (the default), stateless servers ignore
// session IDs entirely and reject DELETE with 405.
// See the documentation for the mcpgodebug package for instructions how to enable it.
// The option will be removed in the 1.9.0 version of the SDK.
var allowsessionsinstateless = mcpgodebug.Value("allowsessionsinstateless")

// noprotocolerrorbody is a compatibility parameter that restores the previous
// behavior of [streamableClientConn.checkResponse]. When unset (the default),
// the client attempts to decode the response body of a non-2xx response as a
// JSON-RPC error. If successful, the underlying JSON-RPC error is surfaced to
// the caller and wrapped with [jsonrpc2.ErrRejected] so that per-call
// rejections do not tear down the session. When set to "1", the client
// ignores response bodies on non-2xx responses and any non-transient error
// permanently fails the connection.
var noprotocolerrorbody = mcpgodebug.Value("noprotocolerrorbody")

// plaintextstatefulrejection is a compatibility parameter that restores the
// previous behavior of a stateful [StreamableHTTPHandler] when it receives a
// request carrying SEP-2575 per-request metadata (i.e. an
// `io.modelcontextprotocol/protocolVersion` `_meta` field, or an
// `MCP-Protocol-Version` header >= 2026-07-28). When unset (the default), the
// server responds with a JSON-RPC error of code
// [CodeUnsupportedProtocolVersion] and an [UnsupportedProtocolVersionData]
// payload listing its supported legacy versions, as required by SEP-2575. When
// set to "1", the server instead responds with a plain-text `http.Error` body
// mentioning the `StreamableHTTPOptions.Stateless` field.
var plaintextstatefulrejection = mcpgodebug.Value("plaintextstatefulrejection")

// writeJSONRPCError writes a JSON-RPC error response with the given HTTP
// status code, request ID (may be a zero ID for errors that occur before the
// request body has been parsed), and JSON-RPC error.
func writeJSONRPCError(w http.ResponseWriter, status int, id jsonrpc.ID, jerr *jsonrpc.Error) {
	resp := &jsonrpc.Response{ID: id, Error: jerr}
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(status)
	if data, err := jsonrpc2.EncodeMessage(resp); err == nil {
		w.Write(data)
	}
}

func (h *StreamableHTTPHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
	// DNS rebinding protection: auto-enabled for localhost servers.
	// See: https://modelcontextprotocol.io/specification/2025-11-25/basic/security_best_practices#local-mcp-server-compromise
	if !h.opts.DisableLocalhostProtection {
		if localAddr, ok := req.Context().Value(http.LocalAddrContextKey).(net.Addr); ok && localAddr != nil {
			if util.IsLoopback(localAddr.String()) && !util.IsLoopback(req.Host) {
				http.Error(w, fmt.Sprintf("Forbidden: invalid Host header %q", req.Host), http.StatusForbidden)
				return
			}
		}
	}

	if h.opts.CrossOriginProtection != nil {
		if err := h.opts.CrossOriginProtection.Check(req); err != nil {
			http.Error(w, err.Error(), http.StatusForbidden)
			return
		}
	}

	// Bound the request body to protect against OOM attacks.
	if req.Body != nil && h.opts.MaxRequestBodyBytes > 0 {
		req.Body = http.MaxBytesReader(w, req.Body, h.opts.MaxRequestBodyBytes)
	}

	// [§2.7] of the spec (2025-06-18): validate the MCP-Protocol-Version
	// header. If provided, it must be a version the server negotiates, which
	// [ServerOptions.SupportedProtocolVersions] may have narrowed. If absent,
	// the version is unknown (the request may be an initialize for any
	// version).
	//
	// [§2.7]: https://modelcontextprotocol.io/specification/2025-06-18/basic/transports#protocol-version-header
	protocolVersion := req.Header.Get(protocolVersionHeader)
	var supportedVersions []string
	if server := h.getServer(req); server != nil {
		supportedVersions = server.protocolVersions
	} else {
		supportedVersions = supportedProtocolVersions
	}
	if protocolVersion != "" && !slices.Contains(supportedVersions, protocolVersion) && protocolVersion < protocolVersion20260728 {
		http.Error(w, fmt.Sprintf("Bad Request: Unsupported protocol version (supported versions: %s)", strings.Join(supportedVersions, ",")), http.StatusBadRequest)
		return
	}
	req = req.WithContext(context.WithValue(req.Context(), protocolVersionContextKey{}, protocolVersion))

	if h.opts.Stateless {
		h.serveStateless(w, req)
	} else {
		h.serveStateful(w, req)
	}
}

// serveStateless handles requests for stateless servers.
// Stateless servers only support POST. Each request creates a temporary
// session that is closed when the request completes.
//
// When the allowsessionsinstateless compatibility flag is set, DELETE is also
// accepted (as a no-op) and session IDs are read from the request header.
func (h *StreamableHTTPHandler) serveStateless(w http.ResponseWriter, req *http.Request) {
	legacySessions := allowsessionsinstateless == "1"

	if req.Method == http.MethodDelete && legacySessions {
		h.serveStatelessLegacyDELETE(w, req)
		return
	}

	if req.Method != http.MethodPost {
		// RFC 9110 §15.5.6: 405 responses MUST include Allow header.
		w.Header().Set("Allow", "POST")
		http.Error(w, "Method Not Allowed", http.StatusMethodNotAllowed)
		return
	}

	if baseMediaType(req.Header.Get("Content-Type")) != "application/json" {
		http.Error(w, "Content-Type must be 'application/json'", http.StatusUnsupportedMediaType)
		return
	}

	// Accept must contain both 'application/json' and 'text/event-stream'.
	jsonOK, streamOK := streamableAccepts(req.Header.Values("Accept"))
	if !jsonOK || !streamOK {
		http.Error(w, "Accept must contain both 'application/json' and 'text/event-stream'", http.StatusBadRequest)
		return
	}

	server := h.getServer(req)
	if server == nil {
		http.Error(w, "no server available", http.StatusBadRequest)
		return
	}

	info, err := h.ephemeralConnectOpts(req)
	if err != nil {
		var mbe *http.MaxBytesError
		if errors.As(err, &mbe) {
			http.Error(w, fmt.Sprintf("request body exceeds %d bytes", mbe.Limit), http.StatusRequestEntityTooLarge)
			return
		}
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}

	var sessionID string
	if legacySessions && !info.usesNewProtocol {
		sessionID = req.Header.Get(sessionIDHeader)
		if sessionID == "" {
			sessionID = server.opts.GetSessionID()
		}
	}

	transport := h.newStreamableServerTransport(sessionID, true)
	transport.shouldPropagateCancellation = info.usesNewProtocol && (info.isSubscriptionsListen || h.opts.PropagateRequestCancellation)

	session, err := connectStreamable(req.Context(), server, transport, info.opts)
	if err != nil {
		h.opts.Logger.Error(fmt.Sprintf("failed to connect: %v", err))
		http.Error(w, "failed connection", http.StatusInternalServerError)
		return
	}
	defer session.Close()

	transport.ServeHTTP(w, req)
}

// serveStatelessLegacyDELETE handles DELETE requests in stateless mode when the
// allowsessionsinstateless compatibility flag is set. DELETE requires a
// Mcp-Session-Id header but is otherwise a no-op since stateless servers don't
// persist sessions.
func (h *StreamableHTTPHandler) serveStatelessLegacyDELETE(w http.ResponseWriter, req *http.Request) {
	sessionID := req.Header.Get(sessionIDHeader)
	if sessionID == "" {
		http.Error(w, "Bad Request: DELETE requires an Mcp-Session-Id header", http.StatusBadRequest)
		return
	}
	w.WriteHeader(http.StatusNoContent)
}

type ephemeralConnectInfo struct {
	opts                  *ServerSessionOptions
	usesNewProtocol       bool
	isSubscriptionsListen bool
}

// ephemeralConnectOpts peeks at the request body to determine connection
// parameters and whether protocol version >= 2026-06-30 (SEP-2575).
//
// For old-protocol requests, default session state is synthesized so that
// the session's init gate doesn't reject the request.
//
// It is used for both stateless servers and stateful servers with no session ID.
func (h *StreamableHTTPHandler) ephemeralConnectOpts(req *http.Request) (*ephemeralConnectInfo, error) {
	protocolVersion := protocolVersionFromContext(req.Context())
	if protocolVersion == "" {
		protocolVersion = protocolVersion20250326
	}

	var hasInitialize, hasInitialized, usesNewProtocol, isSubscriptionsListen bool
	body, err := io.ReadAll(req.Body)
	if err != nil {
		// Preserve *http.MaxBytesError so serveStateless can respond with 413.
		return nil, fmt.Errorf("failed to read body: %w", err)
	}
	req.Body.Close()
	req.Body = io.NopCloser(bytes.NewBuffer(body))
	msgs, _, err := readBatch(body)
	if err == nil {
		for _, msg := range msgs {
			if r, ok := msg.(*jsonrpc.Request); ok {
				switch r.Method {
				case methodInitialize:
					hasInitialize = true
				case notificationInitialized:
					hasInitialized = true
				case methodSubscriptionsListen:
					isSubscriptionsListen = true
				}
				if protocolVersion >= protocolVersion20260728 {
					usesNewProtocol = true
				}
			}
		}
	}

	state := new(ServerSessionState)
	// Only synthesize fake InitializeParams/InitializedParams for old-protocol
	// requests.
	if !hasInitialize && !usesNewProtocol {
		state.InitializeParams = &InitializeParams{
			ProtocolVersion: protocolVersion,
		}
		// The header carries the version an earlier handshake settled on, or the
		// 2025-03-26 the transports spec has a server assume without one; either
		// way it is the version this request is served under, not only declared.
		state.NegotiatedProtocolVersion = protocolVersion
	}
	if !hasInitialized && !usesNewProtocol {
		state.InitializedParams = new(InitializedParams)
	}
	if !usesNewProtocol {
		state.LogLevel = "info"
	}
	return &ephemeralConnectInfo{
		opts: &ServerSessionOptions{
			State: state,
		},
		usesNewProtocol:       usesNewProtocol,
		isSubscriptionsListen: isSubscriptionsListen,
	}, nil
}

func connectStreamable(ctx context.Context, server *Server, transport *StreamableServerTransport, opts *ServerSessionOptions) (*ServerSession, error) {
	s, err := server.Connect(ctx, transport, opts)
	if err != nil {
		return nil, err
	}
	transport.connection.server = server
	transport.connection.toolLookup = server.getServerTool
	return s, nil
}

func (h *StreamableHTTPHandler) newStreamableServerTransport(sessionID string, stateless bool) *StreamableServerTransport {
	return &StreamableServerTransport{
		SessionID:        sessionID,
		Stateless:        stateless,
		EventStore:       h.opts.EventStore,
		jsonResponse:     h.opts.JSONResponse,
		streamKeepAlive:  h.opts.StreamKeepAlive,
		logger:           h.opts.Logger,
		onRequestSummary: h.opts.OnRequestSummary,
	}
}

// serveStateful handles requests for stateful servers.
// Stateful servers support GET, POST, and DELETE, and maintain persistent
// sessions keyed by session ID.
func (h *StreamableHTTPHandler) serveStateful(w http.ResponseWriter, req *http.Request) {
	switch req.Method {
	case http.MethodGet:
		h.serveStatefulGET(w, req)
	case http.MethodPost:
		h.serveStatefulPOST(w, req)
	case http.MethodDelete:
		h.serveStatefulDELETE(w, req)
	default:
		// RFC 9110 §15.5.6: 405 responses MUST include Allow header.
		w.Header().Set("Allow", "GET, POST, DELETE")
		http.Error(w, "Method Not Allowed", http.StatusMethodNotAllowed)
	}
}

// lookupSession looks up a session by the Mcp-Session-Id header value.
// It returns the session info and whether the caller should proceed. If ok is
// false, an error response has been written. The sessionID must be non-empty;
// callers are responsible for checking this before calling lookupSession.
func (h *StreamableHTTPHandler) lookupSession(w http.ResponseWriter, req *http.Request, sessionID string) (info *sessionInfo, ok bool) {
	h.mu.Lock()
	info = h.sessions[sessionID]
	h.mu.Unlock()
	if info == nil {
		http.Error(w, "session not found", http.StatusNotFound)
		return nil, false
	}
	if info.userID != "" {
		tokenInfo := auth.TokenInfoFromContext(req.Context())
		if tokenInfo == nil || tokenInfo.UserID != info.userID {
			http.Error(w, "session user mismatch", http.StatusForbidden)
			return nil, false
		}
	}
	return info, true
}

// serveStatefulGET handles GET requests for standalone SSE streams.
// GET requires a valid Mcp-Session-Id header.
func (h *StreamableHTTPHandler) serveStatefulGET(w http.ResponseWriter, req *http.Request) {
	if _, streamOK := streamableAccepts(req.Header.Values("Accept")); !streamOK {
		http.Error(w, "Accept must contain 'text/event-stream' for GET requests", http.StatusBadRequest)
		return
	}

	sessionID := req.Header.Get(sessionIDHeader)
	if sessionID == "" {
		http.Error(w, "Bad Request: GET requires an Mcp-Session-Id header", http.StatusBadRequest)
		return
	}

	sessInfo, ok := h.lookupSession(w, req, sessionID)
	if !ok {
		return
	}

	sessInfo.transport.ServeHTTP(w, req)
}

// serveStatefulDELETE handles DELETE requests for session termination.
// DELETE requires a valid Mcp-Session-Id header.
func (h *StreamableHTTPHandler) serveStatefulDELETE(w http.ResponseWriter, req *http.Request) {
	sessionID := req.Header.Get(sessionIDHeader)
	if sessionID == "" {
		http.Error(w, "Bad Request: DELETE requires an Mcp-Session-Id header", http.StatusBadRequest)
		return
	}

	sessInfo, ok := h.lookupSession(w, req, sessionID)
	if !ok {
		return
	}

	sessInfo.session.Close()
	w.WriteHeader(http.StatusNoContent)
}

// serveStatefulPOST handles POST requests for stateful servers.
// POST may arrive with or without a Mcp-Session-Id header. Without a session
// ID, a new session is created (this is the normal path for the first
// initialize request).
func (h *StreamableHTTPHandler) serveStatefulPOST(w http.ResponseWriter, req *http.Request) {
	if baseMediaType(req.Header.Get("Content-Type")) != "application/json" {
		http.Error(w, "Content-Type must be 'application/json'", http.StatusUnsupportedMediaType)
		return
	}

	jsonOK, streamOK := streamableAccepts(req.Header.Values("Accept"))
	if !jsonOK || !streamOK {
		http.Error(w, "Accept must contain both 'application/json' and 'text/event-stream'", http.StatusBadRequest)
		return
	}

	sessionID := req.Header.Get(sessionIDHeader)

	// Look up existing session if a session ID was provided.
	if sessionID != "" {
		sessInfo, ok := h.lookupSession(w, req, sessionID)
		if !ok {
			return
		}
		sessInfo.startPOST()
		defer sessInfo.endPOST()
		sessInfo.transport.ServeHTTP(w, req)
		return
	}

	// No session ID: create a new session.
	server := h.getServer(req)
	if server == nil {
		http.Error(w, "no server available", http.StatusBadRequest)
		return
	}
	sessionID = server.opts.GetSessionID()

	transport := h.newStreamableServerTransport(sessionID, false)

	// Sessions without a session ID (GetSessionID returned "") are ephemeral:
	// there's no way to address them, so they are closed after the request.
	// This can happen when ServerOptions.GetSessionID is explicitly set to
	// return "" to suppress session IDs entirely. It also covers any request
	// that arrives before a session exists (e.g. initialize or ping) on a
	// server configured this way.
	if sessionID == "" {
		info, err := h.ephemeralConnectOpts(req)
		if err != nil {
			http.Error(w, err.Error(), http.StatusBadRequest)
			return
		}
		session, err := connectStreamable(req.Context(), server, transport, info.opts)
		if err != nil {
			h.opts.Logger.Error(fmt.Sprintf("failed to connect: %v", err))
			http.Error(w, "failed connection", http.StatusInternalServerError)
			return
		}
		defer session.Close()
		transport.ServeHTTP(w, req)
		return
	}

	connectOpts := &ServerSessionOptions{
		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)
				}
			}
		},
	}

	// 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 := connectStreamable(req.Context(), server, transport, connectOpts)
	if err != nil {
		h.opts.Logger.Error(fmt.Sprintf("failed to connect: %v", err))
		http.Error(w, "failed connection", http.StatusInternalServerError)
		return
	}
	// Capture the user ID from the token info to enable session hijacking
	// prevention on subsequent requests.
	var userID string
	if tokenInfo := auth.TokenInfoFromContext(req.Context()); tokenInfo != nil {
		userID = tokenInfo.UserID
	}
	sessInfo := &sessionInfo{
		session:   session,
		transport: transport,
		userID:    userID,
	}

	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 {
			session.Close()
		}
	}()

	sessInfo.startPOST()
	defer sessInfo.endPOST()
	sessInfo.transport.ServeHTTP(w, req)
}

func streamableAccepts(values []string) (jsonOK, streamOK bool) {
	for _, value := range values {
		for _, raw := range strings.Split(value, ",") {
			token := strings.TrimSpace(raw)
			// Ignore Accept parameters like ";charset=utf-8"; match the base media type.
			base, _, _ := strings.Cut(token, ";")
			switch strings.ToLower(strings.TrimSpace(base)) {
			case "application/json", "application/*":
				jsonOK = true
			case "text/event-stream", "text/*":
				streamOK = true
			case "*/*":
				jsonOK = true
				streamOK = true
			}
		}
	}
	return jsonOK, streamOK
}

func baseMediaType(value string) string {
	mediaType, _, err := mime.ParseMediaType(value)
	if err != nil {
		return ""
	}
	return mediaType
}

// 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

	// 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

	// streamKeepAlive is the idle interval after which an SSE comment is
	// written to a stream; see [StreamableHTTPOptions.StreamKeepAlive].
	//
	// TODO: streamKeepAlive should be exported, like jsonResponse and logger,
	// once users can write their own streamable HTTP handler.
	streamKeepAlive time.Duration

	// 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

	// shouldPropagateCancellation is forwarded to the underlying
	// [streamableServerConn]. See its docstring.
	shouldPropagateCancellation bool

	// onRequestSummary is forwarded from StreamableHTTPOptions for transports
	// created by StreamableHTTPHandler.
	onRequestSummary func(context.Context, StreamableHTTPRequestSummary)

	// 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,
		jsonResponse:                t.jsonResponse,
		streamKeepAlive:             t.streamKeepAlive,
		logger:                      ensureLogger(t.logger), // see #556: must be non-nil
		shouldPropagateCancellation: t.shouldPropagateCancellation,
		onRequestSummary:            t.onRequestSummary,
		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
}

// The streamable HTTP transport supports every legacy SDK protocol version,
// but the SEP-2575 >= 2026-07-28 protocol is only supported when the
// transport is configured as stateless.
func (t *StreamableServerTransport) SupportsProtocolVersion(version string) bool {
	if version >= protocolVersion20260728 {
		return t.Stateless && slices.Contains(supportedProtocolVersions, version)
	}
	return slices.Contains(supportedProtocolVersions, version)
}

type streamableServerConn struct {
	sessionID        string
	stateless        bool
	jsonResponse     bool
	eventStore       EventStore
	onRequestSummary func(context.Context, StreamableHTTPRequestSummary)

	// streamKeepAlive is the idle interval for SSE keep-alive comments, or
	// non-positive to disable them; see [StreamableHTTPOptions.StreamKeepAlive].
	streamKeepAlive time.Duration

	// shouldPropagateCancellation is true when the underlying HTTP request's
	// lifetime IS the connection's cancellation signal (e.g., a stateless
	// POST that owns a long-lived subscriptions/listen stream). It is read
	// by the [cancellationPropagator] interface so the jsonrpc2 layer wires
	// handler contexts to observe the carrier's cancellation.
	shouldPropagateCancellation bool

	logger *slog.Logger

	server     *Server
	toolLookup func(name string) (*serverTool, bool)

	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
}

// propagateCancellation implements [cancellationPropagator]. It returns true
// when this connection is bound to a single HTTP request whose lifetime
// should drive request-handler cancellation — for example, a stateless POST
// carrying a long-lived subscriptions/listen stream that must unwind when
// the client TCP-disconnects.
func (c *streamableServerConn) propagateCancellation() bool {
	return c.shouldPropagateCancellation
}

// 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

	// logger is used for logging errors during stream operations.
	logger *slog.Logger

	// mu guards the fields below, as well as storage of new messages in the
	// connection's event store (if any).
	mu sync.Mutex

	// If pendingJSONMessages is non-nil, this is a JSON stream and messages are
	// collected here until the stream is complete, at which point they are
	// flushed as a single JSON response. Note that the non-nilness of this field
	// is significant, as it signals the expected content type.
	//
	// Note: if we remove support for batching, this could just be a bool.
	pendingJSONMessages []json.RawMessage

	// isBatch reports whether the request that opened the stream was a JSON-RPC
	// batch. Batch responses must remain arrays even when they contain one item.
	isBatch bool

	// w is the HTTP response writer for this stream. A non-nil w indicates
	// that the stream is claimed by an HTTP request (the hanging POST or GET);
	// it is set to nil when the request completes.
	w http.ResponseWriter

	// done is closed to release the hanging HTTP request.
	//
	// Invariant: a non-nil done implies w is also non-nil, though the converse
	// is not necessarily true: done is set to nil when it is closed, to avoid
	// duplicate closure.
	done chan struct{}

	// lastIdx is the index of the last written SSE event, for event ID generation.
	// It starts at -1 since indices start at 0.
	lastIdx int

	// lastWrite is when bytes were last written to w. The zero value means
	// nothing has been written to the current w, so its headers are still
	// uncommitted and the HTTP status can still be changed. Reset by release.
	lastWrite time.Time

	// committed, if non-nil, is closed by the first write to w. The keep-alive
	// goroutine parks on it instead of polling.
	committed chan struct{}

	// protocolVersion is the protocol version for this stream.
	protocolVersion string

	// requests is the set of unanswered incoming requests for the stream.
	//
	// Requests are removed when their response has been received.
	// In practice, there is only one request, but in the 2025-03-26 version of
	// the spec and earlier there was a concept of batching, in which POST
	// payloads could hold multiple requests or responses.
	requests map[jsonrpc.ID]struct{}

	// isListen reports whether this stream was opened by a
	// subscriptions/listen request. Listen streams are always SSE, live for
	// the duration of the subscription, and act as the target for
	// out-of-band notifications routed through this connection.
	isListen bool
}

// close sends a 'close' event to the client (if protocolVersion >= 2025-11-25
// and reconnectAfter > 0) and closes the done channel.
//
// The done channel is set to nil after closing, so that done != nil implies
// the stream is active and done is open. This simplifies checks elsewhere.
func (s *stream) close(reconnectAfter time.Duration) {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.done == nil {
		return // stream not connected or already closed
	}
	if s.protocolVersion >= protocolVersion20251125 && reconnectAfter > 0 {
		reconnectStr := strconv.FormatInt(reconnectAfter.Milliseconds(), 10)
		if _, err := writeEvent(s.w, Event{
			Name:  "close",
			Retry: reconnectStr,
		}); err != nil {
			s.logger.Warn(fmt.Sprintf("Writing close event: %v", err))
		}
	}
	close(s.done)
	s.done = nil
}

// release releases the stream from its HTTP request, allowing it to be
// claimed by another request (e.g., for resumption).
func (s *stream) release() {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.w = nil
	s.done = nil // may already be nil, if the stream is done or closed
	s.lastWrite = time.Time{}
	s.committed = nil
}

// markWrittenLocked records a write to s.w, for the keep-alive.
//
// s.mu must be held.
func (s *stream) markWrittenLocked() {
	s.lastWrite = time.Now()
	if s.committed != nil {
		close(s.committed)
		s.committed = nil
	}
}

// startKeepAliveLocked starts the keep-alive goroutine for the HTTP request
// currently claiming the stream. ctx is that request's context.
//
// s.mu must be held.
func (s *stream) startKeepAliveLocked(ctx context.Context, interval time.Duration) {
	// Nothing has been written yet, so a SEP-2575 status override may still
	// be needed (see deliverLocked): wait for the first event.
	committed := make(chan struct{})
	s.committed = committed
	go s.keepAlive(ctx, interval, committed)
}

// keepAlive writes an SSE comment to the stream whenever it has been idle for
// interval, until ctx is done or the stream is released or closed. A failed
// write closes the stream, releasing the hanging request so that a dead peer
// is noticed within one interval.
func (s *stream) keepAlive(ctx context.Context, interval time.Duration, committed chan struct{}) {
	if committed != nil {
		select {
		case <-ctx.Done():
			return
		case <-committed:
		}
	}
	timer := time.NewTimer(interval)
	defer timer.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-timer.C:
		}
		if ctx.Err() != nil {
			// The request ended; don't touch a stream that a later request may
			// have re-acquired.
			return
		}
		s.mu.Lock()
		if s.done == nil {
			s.mu.Unlock()
			return
		}
		if wait := interval - time.Since(s.lastWrite); !s.lastWrite.IsZero() && wait > 0 {
			s.mu.Unlock()
			timer.Reset(wait)
			continue
		}
		_, err := fmt.Fprint(s.w, ":\n\n")
		if err == nil {
			// Ignore returned error as flushing is best-effort.
			_ = http.NewResponseController(s.w).Flush()
			s.markWrittenLocked()
		} else {
			close(s.done)
			s.done = nil
		}
		s.mu.Unlock()
		if err != nil {
			// A client that closes its connection cancels ctx before any write
			// fails, so reaching this means the peer vanished without closing.
			s.logger.Warn(fmt.Sprintf("Writing keep-alive: %v", err))
			return
		}
		timer.Reset(interval)
	}
}

// extractErrorStatus reports the HTTP status to send when the given
// outgoing message is a JSON-RPC error response under the SEP-2575 protocol
// (>= 2026-07-28).
//
// Per SEP-2575:
//   - MethodNotFound (-32601) MUST return HTTP 404.
//   - InvalidParams (-32602), UnsupportedProtocolVersion (-32022) and
//     CodeMissingRequiredClientCapabilities (-32021) MUST
//     return HTTP 400.
func extractErrorStatus(ctx context.Context, msg jsonrpc.Message) int {
	if protocolVersionFromContext(ctx) < protocolVersion20260728 {
		return 0
	}
	resp, ok := msg.(*jsonrpc.Response)
	if !ok || resp.Error == nil {
		return 0
	}
	var jerr *jsonrpc.Error
	if !errors.As(resp.Error, &jerr) {
		return 0
	}
	switch jerr.Code {
	case jsonrpc.CodeMethodNotFound:
		return http.StatusNotFound
	case jsonrpc.CodeInvalidParams, CodeUnsupportedProtocolVersion, CodeMissingRequiredClientCapabilities:
		return http.StatusBadRequest
	}
	return 0
}

// deliverLocked writes data to the stream (for SSE) or stores it in
// pendingJSONMessages (for JSON mode). The eventID is used for SSE event ID;
// pass "" to omit.
//
// If responseTo is valid, it is removed from the requests map. When all
// requests have been responded to, the done channel is closed and set to nil.
//
// If overrideStatus is non-zero, data is treated as a SEP-2575 protocol-level
// error response (>= 2026-07-28): it is written as a single raw JSON-RPC
// response body with Content-Type: application/json and HTTP status
// overrideStatus.
//
// Returns true if the stream is now done (all requests have been responded to).
// The done value is always accurate, even if an error is returned.
//
// s.mu must be held when calling this method.
func (s *stream) deliverLocked(data []byte, eventID string, responseTo jsonrpc.ID, overrideStatus int) (done bool, err error) {
	// First, record the response. We must do this *before* returning an error
	// below, as even if the stream is disconnected we want to update our
	// accounting.
	if responseTo.IsValid() {
		delete(s.requests, responseTo)
	}
	// Now, try to deliver the message to the client.
	done = len(s.requests) == 0 && s.id != ""
	if s.done == nil {
		return done, fmt.Errorf("stream not connected or already closed")
	}
	if done {
		defer func() { close(s.done); s.done = nil }()
	}
	// SEP-2575 protocol-level error override: write the error as a raw
	// JSON-RPC response with the spec-mandated HTTP status, bypassing any
	// SSE framing.
	if overrideStatus != 0 {
		s.w.Header().Set("Content-Type", "application/json")
		s.w.Header().Del("X-Accel-Buffering")
		s.w.WriteHeader(overrideStatus)
		if _, err := s.w.Write(data); err != nil {
			return done, err
		}
		return done, nil
	}
	// Try to write to the response.
	//
	// If we get here, the request is still hanging (because s.done != nil
	// implies s.w != nil), but may have been cancelled by the client/http layer:
	// there's a brief race between request cancellation and releasing the
	// stream.
	if s.pendingJSONMessages != nil {
		s.pendingJSONMessages = append(s.pendingJSONMessages, data)
		if done {
			// Flush all pending messages as JSON response.
			var toWrite []byte
			if len(s.pendingJSONMessages) == 1 && !s.isBatch {
				toWrite = s.pendingJSONMessages[0]
			} else {
				toWrite, err = json.Marshal(s.pendingJSONMessages)
				if err != nil {
					return done, err
				}
			}
			if _, err := s.w.Write(toWrite); err != nil {
				return done, err
			}
		}
	} else {
		// SSE mode: write event to response writer.
		s.lastIdx++
		if _, err := writeEvent(s.w, Event{Name: "message", Data: data, ID: eventID}); err != nil {
			return done, err
		}
		s.markWrittenLocked()
	}
	return done, nil
}

// doneLocked reports whether the stream is logically complete.
//
// s.requests was populated when reading the POST body, requests are deleted as
// they are responded to. Once all requests have been responded to, the stream
// is done.
//
// 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 && protocolVersionFromContext(ctx) < protocolVersion20260728 {
		if err := c.eventStore.Open(ctx, c.sessionID, id); err != nil {
			return nil, err
		}
	}
	return &stream{
		id:       id,
		requests: requests,
		lastIdx:  -1, // indices start at 0, incremented before each write
		logger:   c.logger,
	}, 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{}

// protocolVersionContextKey stores the protocol version extracted from the
// MCP-Protocol-Version HTTP header for use by lower layers.
type protocolVersionContextKey struct{}

// protocolVersionFromContext returns the protocol version from the context, or
// the empty string if not set. An empty string means the version is unknown
// (e.g. the header was absent).
func protocolVersionFromContext(ctx context.Context) string {
	v, _ := ctx.Value(protocolVersionContextKey{}).(string)
	return v
}

// 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(lastEventIDHeader)) > 0 {
		eid := req.Header.Get(lastEventIDHeader)
		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 := req.Context()

	protocolVersion := protocolVersionFromContext(ctx)
	if protocolVersion == "" {
		protocolVersion = protocolVersion20250326
	}

	stream, done := c.acquireStream(ctx, w, streamID, lastIdx, protocolVersion)
	if stream == nil {
		return
	}
	defer stream.release()
	c.hangResponse(ctx, done)
}

// hangResponse blocks the HTTP response until one of three conditions is met:
//   - ctx is cancelled (the client disconnected or the request timed out)
//   - done is closed (all responses have been sent, or the stream was explicitly closed)
//   - the session is closed
//
// This keeps the HTTP connection open so that server-sent events can be
// written to the response.
func (c *streamableServerConn) hangResponse(ctx context.Context, done <-chan struct{}) {
	select {
	case <-ctx.Done():
	case <-done:
	case <-c.done:
	}
}

// acquireStream replays all events since lastIdx, and acquires the ongoing
// stream, if any. If non-nil, the resulting stream will be registered for
// receiving new messages, and the stream's 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.
//
// protocolVersion is the protocol version for this stream, used to determine
// feature support (e.g. prime and close events were added in 2025-11-25).
func (c *streamableServerConn) acquireStream(ctx context.Context, w http.ResponseWriter, streamID string, lastIdx int, protocolVersion string) (*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. The spec
	// (https://modelcontextprotocol.io/specification/2025-11-25/basic/transports#resumability-and-redelivery)
	// does not explicitly require exclusive replay, but we enforce it defensively.
	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 w, to ensure it isn't claimed by
		// another request before we lock it below.
		tempStream = true
		s = &stream{
			id: streamID,
			w:  w,
		}
		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.w != 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
			}
			if len(data) > 0 {
				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.
		//
		// On HTTP/2, headers and body travel as separate frames (HEADERS and
		// DATA). Reverse proxies (e.g. Envoy, Caddy, net/http/httputil)
		// commonly buffer the HEADERS frame until they have a DATA frame to
		// coalesce it with — there is no HTTP/2 equivalent of HTTP/1.1's
		// Transfer-Encoding: chunked signal that says "this is streaming, send
		// headers now". Calling Flush() alone is not sufficient: it pushes
		// the kernel buffer to the proxy, but the proxy still holds the
		// HEADERS frame.
		//
		// Write an SSE comment (lines starting with ":" are ignored by
		// clients per RFC) so a DATA frame is produced, which forces the
		// proxy to forward both frames. See:
		//   https://github.com/golang/go/issues/31125
		//   https://github.com/caddyserver/caddy/issues/4247
		w.WriteHeader(http.StatusOK)
		fmt.Fprint(w, ": ok\n\n")
		rc := http.NewResponseController(w)
		// Ignore returned error as flushing is best-effort.
		_ = rc.Flush()
	}

	for _, data := range toReplay {
		lastIdx++
		e := Event{Name: "message", Data: data}
		if c.eventStore != nil {
			e.ID = formatEventID(s.id, lastIdx)
		}
		if _, err := writeEvent(w, e); err != nil {
			return nil, nil
		}
	}

	if tempStream || s.doneLocked() {
		// Nothing more to do.
		return nil, nil
	}

	// The stream is not done: set up delivery state before the stream is
	// unlocked, allowing the connection to write new events.
	s.w = w
	s.done = make(chan struct{})
	s.lastIdx = lastIdx
	s.protocolVersion = protocolVersion
	return s, 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(lastEventIDHeader)) > 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 {
		var mbe *http.MaxBytesError
		if errors.As(err, &mbe) {
			http.Error(w, fmt.Sprintf("request body exceeds %d bytes", mbe.Limit), http.StatusRequestEntityTooLarge)
			return
		}
		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
	}
	// TODO(#674): once we've documented the support matrix for 2025-03-26 and
	// earlier, drop support for matching entirely; that will simplify this
	// logic.
	incoming, isBatch, err := readBatch(body)
	if err != nil {
		http.Error(w, fmt.Sprintf("malformed payload: %v", err), http.StatusBadRequest)
		return
	}
	if c.onRequestSummary != nil && !isBatch && len(incoming) == 1 {
		c.onRequestSummary(req.Context(), summarizeStreamableHTTPRequest(incoming[0]))
	}

	protocolVersion := protocolVersionFromContext(req.Context())
	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
	isSubscriptionsListen := false
	var initializeProtocolVersion string
	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.
			// Use the server's receiving method infos (which include any custom
			// methods registered via AddReceivingCustomMethod) when available;
			// fall back to the standard methods otherwise, e.g. in tests that
			// exercise streamableServerConn directly without a server.
			methodInfos := serverMethodInfos
			if c.server != nil {
				methodInfos = c.server.receivingMethodInfos()
			}
			if _, err := checkRequest(jreq, methodInfos); err != nil {
				if protocolVersion >= protocolVersion20260728 && errors.Is(err, jsonrpc2.ErrNotHandled) && jreq.IsCall() {
					writeJSONRPCError(w, http.StatusNotFound, jreq.ID, &jsonrpc.Error{
						Code:    jsonrpc.CodeMethodNotFound,
						Message: err.Error(),
					})
					return
				}
				http.Error(w, err.Error(), http.StatusBadRequest)
				return
			}
			if jreq.Method == methodInitialize {
				isInitialize = true
				// Extract the protocol version from InitializeParams.
				var params InitializeParams
				if err := internaljson.Unmarshal(jreq.Params, &params); err == nil {
					initializeProtocolVersion = params.ProtocolVersion
				}
			}
			if jreq.Method == methodSubscriptionsListen {
				isSubscriptionsListen = true
			}
			// SEP-2575: requests carrying `_meta.protocolVersion` require the
			// Mcp-Protocol-Version HTTP header to be present and to match the
			// per-request `_meta.protocolVersion` value.
			// The new (>= 2026-07-28) protocol is supported on the HTTP transport
			// only when [StreamableHTTPOptions.Stateless] is true.
			//
			// The `_meta` triple is defined for calls only, notifications are
			// excluded.
			//
			// TODO: this validation can be moved within validateMcpHeaders.
			var metaVersion string
			if meta := extractRequestMeta(jreq.Params); meta != nil {
				metaVersion, _ = meta[MetaKeyProtocolVersion].(string)
			}
			if (jreq.IsCall() && protocolVersion >= protocolVersion20260728) || metaVersion != "" {
				// Extract again the protocol version from the context to see what the client
				// is advertising in the Mcp-Protocol-Version HTTP header.
				headerVersion := protocolVersionFromContext(req.Context())
				// server/discover is exempt from the stateful
				// rejection as it should learn about the supported protocols from the
				// DiscoverResult response.
				if !c.stateless && jreq.Method != methodDiscover {
					if plaintextstatefulrejection == "1" {
						http.Error(w, fmt.Sprintf(
							"Bad Request: protocol version %q is only supported on stateless HTTP servers (set StreamableHTTPOptions.Stateless = true)",
							protocolVersion),
							http.StatusBadRequest)
						return
					}
					// Advertise only the legacy versions this stateful server accepts
					// (i.e. exclude 2026-07-28, since that's what the request asked for
					// and what we're rejecting).
					advertised := supportedProtocolVersions
					if c.server != nil {
						advertised = c.server.protocolVersions
					}
					legacyVersions := slices.DeleteFunc(slices.Clone(advertised), func(v string) bool {
						return v >= protocolVersion20260728
					})
					data, _ := json.Marshal(UnsupportedProtocolVersionData{
						Supported: legacyVersions,
						Requested: protocolVersion,
					})
					c.logger.Warn(fmt.Sprintf(
						"rejecting request with protocol version %q: this server is stateful; set StreamableHTTPOptions.Stateless = true to accept it",
						protocolVersion))
					writeJSONRPCError(w, http.StatusBadRequest, jreq.ID, &jsonrpc.Error{
						Code:    CodeUnsupportedProtocolVersion,
						Message: fmt.Sprintf("protocol version %q is not supported by this server", protocolVersion),
						Data:    data,
					})
					return
				}
				if headerVersion == "" {
					writeJSONRPCError(w, http.StatusBadRequest, jreq.ID, &jsonrpc.Error{
						Code: CodeHeaderMismatch,
						Message: fmt.Sprintf(
							"%s header is required for requests carrying %q",
							protocolVersionHeader, MetaKeyProtocolVersion),
					})
					return
				}
				if metaVersion == "" {
					writeJSONRPCError(w, http.StatusBadRequest, jreq.ID, &jsonrpc.Error{
						Code: jsonrpc.CodeInvalidParams,
						Message: fmt.Sprintf(
							"missing or invalid _meta field %q",
							MetaKeyProtocolVersion),
					})
					return
				}
				if headerVersion != metaVersion {
					writeJSONRPCError(w, http.StatusBadRequest, jreq.ID, &jsonrpc.Error{
						Code: CodeHeaderMismatch,
						Message: fmt.Sprintf(
							"%s header %q does not match request %s %q",
							protocolVersionHeader, headerVersion,
							MetaKeyProtocolVersion, metaVersion),
					})
					return
				}
			}
			// Include metadata for all requests (including notifications).
			jreq.Extra = &RequestExtra{
				TokenInfo: tokenInfo,
				Header:    req.Header,
			}
			if jreq.IsCall() {
				calls[jreq.ID] = struct{}{}
				// See the doc for CloseSSEStream: allow the request handler to
				// explicitly close the ongoing stream.
				jreq.Extra.(*RequestExtra).CloseSSEStream = func(args CloseSSEStreamArgs) {
					// This mechanism was designed to trigger client reconnection with
					// Last-Event-ID for server-initiated disconnect scenarios. It is
					// deprecated in protocol version 2026-07-28.
					if protocolVersion >= protocolVersion20260728 {
						return
					}
					c.mu.Lock()
					streamID, ok := c.requestStreams[jreq.ID]
					var stream *stream
					if ok {
						stream = c.streams[streamID]
					}
					c.mu.Unlock()

					if stream != nil {
						stream.close(args.RetryAfter)
					}
				}
			}
		}
	}

	// Validate MCP standard headers (Mcp-Method, Mcp-Name, Mcp-Param-*)
	if !isBatch && len(incoming) == 1 {
		if err := validateMcpHeaders(req.Header, incoming[0], c.toolLookup); err != nil {
			resp := &jsonrpc.Response{
				Error: jsonrpc2.NewError(CodeHeaderMismatch, err.Error()),
			}
			if jreq, ok := incoming[0].(*jsonrpc.Request); ok {
				resp.ID = jreq.ID
			}
			w.Header().Set("Content-Type", "application/json")
			w.WriteHeader(http.StatusBadRequest)
			if data, err := jsonrpc2.EncodeMessage(resp); err == nil {
				w.Write(data)
			}
			return
		}
	}

	// The prime and close events were added in protocol version 2025-11-25 (SEP-1699).
	// Use the version from InitializeParams if this is an initialize request,
	// otherwise use the protocol version header.
	effectiveVersion := protocolVersion
	if isInitialize && initializeProtocolVersion != "" {
		effectiveVersion = initializeProtocolVersion
	}

	// If we don't have any calls, we can just publish the incoming messages and return.
	// No need to track a logical stream.
	//
	// See section [§2.1.4] of the spec: "If the server accepts the input, the
	// server MUST return HTTP status code 202 Accepted with no body."
	//
	// [§2.1.4]: https://modelcontextprotocol.io/specification/2025-11-25/basic/transports#sending-messages-to-the-server
	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 incoming messages as
	// soon as they're published.
	stream, err := c.newStream(req.Context(), calls, crand.Text())
	if err != nil {
		http.Error(w, fmt.Sprintf("storing stream: %v", err), http.StatusInternalServerError)
		return
	}
	stream.isListen = isSubscriptionsListen
	stream.isBatch = isBatch

	// subscriptions/listen is inherently a long-lived SSE endpoint (SEP-2575):
	// it has no synchronous result, the response stream stays open until the
	// client cancels, and the server pushes notifications on it as they occur.
	// Force SSE mode (bypassing JSONResponse) so the buffered application/json
	// path doesn't deadlock waiting for a completion that won't come.
	useSSE := !c.jsonResponse || isSubscriptionsListen

	// Set response headers. Accept was checked in [StreamableHTTPHandler].
	w.Header().Set("Cache-Control", "no-cache, no-transform")
	if useSSE {
		w.Header().Set("Content-Type", "text/event-stream")
		w.Header().Set("Connection", "keep-alive")
		// The spec recommends this for SSE: ask reverse proxies not to buffer.
		w.Header().Set("X-Accel-Buffering", "no")
	} else {
		w.Header().Set("Content-Type", "application/json")
	}
	if c.sessionID != "" && isInitialize {
		w.Header().Set(sessionIDHeader, c.sessionID)
	}

	// Set up stream delivery state.
	stream.w = w
	done := make(chan struct{})
	stream.done = done
	stream.protocolVersion = effectiveVersion

	// Reject any call whose ID is already in flight on this session,
	// atomically and without partial registration.
	c.mu.Lock()
	for reqID := range calls {
		if _, ok := c.requestStreams[reqID]; ok {
			c.mu.Unlock()
			writeJSONRPCError(w, http.StatusBadRequest, reqID, &jsonrpc.Error{
				Code:    jsonrpc.CodeInvalidRequest,
				Message: fmt.Sprintf("duplicate in-flight request ID %v", reqID.Raw()),
			})
			return
		}
	}
	c.streams[stream.id] = stream
	for reqID := range calls {
		c.requestStreams[reqID] = stream.id
	}
	c.mu.Unlock()

	// TODO(rfindley): if we have no event store, we should really cancel all
	// remaining requests here, since the client will never get the results.
	defer stream.release()

	if !useSSE {
		// JSON mode: collect messages in pendingJSONMessages until done.
		// Set pendingJSONMessages to a non-nil value to signal that this is an
		// application/json stream.
		stream.pendingJSONMessages = []json.RawMessage{}
	} else {
		// SSE mode: write a priming event if supported.
		//
		// SEP-2575 removes Last-Event-ID-based resumable streams for protocol
		// version >= 2026-07-28.
		if c.eventStore != nil && effectiveVersion >= protocolVersion20251125 && effectiveVersion < protocolVersion20260728 {
			// Write a priming event, as defined by [§2.1.6] of the spec.
			//
			// [§2.1.6]: https://modelcontextprotocol.io/specification/2025-11-25/basic/transports#sending-messages-to-the-server
			//
			// We must also write it to the event store in order for indexes to
			// align.
			if err := c.eventStore.Append(req.Context(), c.sessionID, stream.id, nil); err != nil {
				c.logger.Warn(fmt.Sprintf("Storing priming event: %v", err))
			}
			stream.lastIdx++
			e := Event{Name: "prime", ID: formatEventID(stream.id, stream.lastIdx)}
			if _, err := writeEvent(w, e); err != nil {
				c.logger.Warn(fmt.Sprintf("Writing priming event: %v", err))
			}
		}
		if c.streamKeepAlive > 0 && stream.isListen {
			stream.mu.Lock()
			stream.startKeepAliveLocked(req.Context(), c.streamKeepAlive)
			stream.mu.Unlock()
		}
	}

	// Publish incoming messages.
	for _, msg := range incoming {
		select {
		case c.incoming <- msg:
		// Note: don't select on req.Context().Done() here, since we've already
		// received the requests and may have already published a response message
		// or notification. The client could resume the stream.
		//
		// In fact, this send could be in a separate goroutine.
		case <-c.done:
			// Session closed: we don't know if any data has been written, so it's
			// too late to write a status code here.
			return
		}
	}

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

func summarizeStreamableHTTPRequest(msg jsonrpc.Message) StreamableHTTPRequestSummary {
	var summary StreamableHTTPRequestSummary
	switch msg := msg.(type) {
	case *jsonrpc.Request:
		summary.Method = msg.Method
		if msg.IsCall() {
			summary.RequestID = msg.ID
		} else {
			summary.IsNotification = true
		}
	case *jsonrpc.Response:
		summary.IsResponse = true
	}
	return summary
}

// Event IDs: encode both the logical connection ID and the index, as
// <streamID>_<idx>, to be consistent with the typescript implementation.

// formatEventID returns the event ID to use for the logical connection ID
// streamID and message index idx.
//
// See also [parseEventID].
func formatEventID(sid string, idx int) string {
	return fmt.Sprintf("%s_%d", sid, idx)
}

// parseEventID parses a Last-Event-ID value into a logical stream id and
// index.
//
// See also [formatEventID].
func parseEventID(eventID string) (streamID string, idx int, ok bool) {
	parts := strings.Split(eventID, "_")
	if len(parts) != 2 {
		return "", 0, false
	}
	streamID = parts[0]
	idx, err := strconv.Atoi(parts[1])
	if err != nil || idx < 0 {
		return "", 0, false
	}
	return streamID, idx, true
}

// Read implements the [Connection] interface.
func (c *streamableServerConn) Read(ctx context.Context) (jsonrpc.Message, error) {
	select {
	case <-ctx.Done():
		return nil, ctx.Err()
	case msg, ok := <-c.incoming:
		if !ok {
			return nil, io.EOF
		}
		return msg, nil
	case <-c.done:
		return nil, io.EOF
	}
}

// Write implements the [Connection] interface.
func (c *streamableServerConn) Write(ctx context.Context, msg jsonrpc.Message) error {
	// Throughout this function, note that any error that wraps ErrRejected
	// indicates a does not cause the connection to break.
	//
	// Most errors don't break the connection: unlike a true bidirectional
	// stream, a failure to deliver to a stream is not an indication that the
	// logical session is broken.
	data, err := jsonrpc2.EncodeMessage(msg)
	if err != nil {
		return err
	}

	if req, ok := msg.(*jsonrpc.Request); ok && req.IsCall() && (c.stateless || c.sessionID == "") {
		// Requests aren't possible with stateless servers, or when there's no session ID.
		return fmt.Errorf("%w: stateless servers cannot make requests", jsonrpc2.ErrRejected)
	}

	// Find the incoming request that this write relates to, if any.
	var (
		relatedRequest jsonrpc.ID
		responseTo     jsonrpc.ID // if valid, the message is a response to this request
	)
	if resp, ok := msg.(*jsonrpc.Response); ok {
		// If the message is a response, it relates to its request (of course).
		relatedRequest = resp.ID
		responseTo = resp.ID
	} else {
		// Otherwise, we check to see if it request was made in the context of an
		// ongoing request. This may not be the case if the request was made with
		// an unrelated context.
		if v := ctx.Value(idContextKey{}); v != nil {
			relatedRequest = v.(jsonrpc.ID)
		}
	}

	// If the stream is application/json, but the message is not a response, we
	// must send it out of band to the standalone SSE stream.
	if c.jsonResponse && !responseTo.IsValid() {
		relatedRequest = jsonrpc.ID{}
	}

	// Write the message to the stream.
	var s *stream
	c.mu.Lock()
	if relatedRequest.IsValid() {
		if streamID, ok := c.requestStreams[relatedRequest]; ok {
			s = c.streams[streamID]
		}
	} else {
		// In stateless mode there will always be only one stream per connection.
		// If that stream was open to listen for subscription notifications,
		// automatically select as the one to write the notification to.
		for _, stream := range c.streams {
			if stream.isListen {
				s = stream
				break
			}
		}
		if s == nil {
			s = c.streams[""] // standalone SSE stream
		}
	}
	if responseTo.IsValid() {
		// Once we've responded to a request, disallow related messages by removing
		// the stream association. This also releases memory.
		delete(c.requestStreams, responseTo)
	}
	sessionClosed := c.isDone
	c.mu.Unlock()

	if s == nil {
		// The request was made in the context of an ongoing request, but that
		// request is complete.
		//
		// In the future, we could be less strict and allow the request to land on
		// the standalone SSE stream.
		return fmt.Errorf("%w: write to closed stream", jsonrpc2.ErrRejected)
	}
	if sessionClosed {
		return errors.New("session is closed")
	}

	s.mu.Lock()
	defer s.mu.Unlock()

	// Store in eventStore before delivering.
	// TODO(rfindley): we should only append if the response is SSE, not JSON, by
	// pushing down into the delivery layer.
	delivered := false
	var errs []error
	protocolVersion := protocolVersionFromContext(ctx)
	if c.eventStore != nil && protocolVersion < protocolVersion20260728 {
		if err := c.eventStore.Append(ctx, c.sessionID, s.id, data); err != nil {
			errs = append(errs, err)
		} else {
			delivered = true
		}
	}

	// Compute eventID for SSE streams with event store.
	// Use s.lastIdx + 1 because deliverLocked increments before writing.
	var eventID string
	if c.eventStore != nil && protocolVersion < protocolVersion20260728 {
		eventID = formatEventID(s.id, s.lastIdx+1)
	}

	// SEP-2575: map protocol-level JSON-RPC error codes to HTTP status codes
	// on the new protocol (>= 2026-07-28). When non-zero, deliverLocked will
	// write the body as raw application/json with the override status.
	overrideStatus := extractErrorStatus(ctx, msg)

	done, err := s.deliverLocked(data, eventID, responseTo, overrideStatus)
	if err != nil {
		errs = append(errs, err)
	} else {
		delivered = true
	}

	if done {
		c.mu.Lock()
		delete(c.streams, s.id)
		c.mu.Unlock()
	}

	if !delivered {
		return fmt.Errorf("%w: undelivered message: %v", jsonrpc2.ErrRejected, errors.Join(errs...))
	}
	return nil
}

// Close implements the [Connection] interface.
func (c *streamableServerConn) Close() error {
	c.mu.Lock()
	defer c.mu.Unlock()
	if !c.isDone {
		c.isDone = true
		close(c.done)
		if c.eventStore != nil {
			// TODO: find a way to plumb a context here, or an event store with a long-running
			// close operation can take arbitrary time. Alternative: impose a fixed timeout here.
			return c.eventStore.SessionClosed(context.TODO(), c.sessionID)
		}
	}
	return nil
}

// A StreamableClientTransport is a [Transport] that can communicate with an MCP
// endpoint serving the streamable HTTP transport defined by the 2025-03-26
// version of the spec.
type StreamableClientTransport struct {
	Endpoint   string
	HTTPClient *http.Client
	// MaxRetries is the maximum number of times to attempt a reconnect before giving up.
	// It defaults to 5. To disable retries, use a negative number.
	MaxRetries int

	// DisableStandaloneSSE controls whether the client establishes a standalone SSE stream
	// for receiving server-initiated messages.
	//
	// When false (the default), after initialization the client sends an HTTP GET request
	// to establish a persistent server-sent events (SSE) connection. This allows the server
	// to send messages to the client at any time, such as ToolListChangedNotification or
	// other server-initiated requests and notifications. The connection persists for the
	// lifetime of the session and automatically reconnects if interrupted.
	//
	// When true, the client does not establish the standalone SSE stream. The client will
	// only receive responses to its own POST requests. Server-initiated messages will not
	// be received.
	//
	// According to the MCP specification, the standalone SSE stream is optional.
	// Setting DisableStandaloneSSE to true is useful when:
	//   - You only need request-response communication and don't need server-initiated notifications
	//   - The server doesn't properly handle GET requests for SSE streams
	//   - You want to avoid maintaining a persistent connection
	DisableStandaloneSSE bool

	// OAuthHandler is an optional field that, if provided, will be used to authorize the requests.
	OAuthHandler auth.OAuthHandler

	// MaxEventSize bounds the number of bytes buffered while reading a single
	// server-sent event. A value of 0 selects [DefaultMaxEventSize], a negative
	// value disables the cap.
	MaxEventSize int

	// TODO(rfindley): propose exporting these.
	// If strict is set, the transport is in 'strict mode', where any violation
	// of the MCP spec causes a failure.
	strict bool
	// If logger is set, it is used to log aspects of the transport, such as spec
	// violations that were ignored.
	logger *slog.Logger
}

// These settings are not (yet) exposed to the user in
// StreamableClientTransport.
const (
	// reconnectGrowFactor is the multiplicative factor by which the delay increases after each attempt.
	// A value of 1.0 results in a constant delay, while a value of 2.0 would double it each time.
	// It must be 1.0 or greater if MaxRetries is greater than 0.
	reconnectGrowFactor = 1.5
	// reconnectMaxDelay caps the backoff delay, preventing it from growing indefinitely.
	reconnectMaxDelay = 30 * time.Second
	// closeDeleteTimeout bounds the session-termination DELETE sent by Close.
	// That request is best-effort: a peer that accepts the connection but never
	// answers must not be able to hold teardown open until the TCP timeout.
	// See issue #1183.
	closeDeleteTimeout = 5 * time.Second
)

var (
	// reconnectInitialDelay is the base delay for the first reconnect attempt.
	//
	// Mutable for testing.
	reconnectInitialDelay atomic.Int64
)

func init() {
	reconnectInitialDelay.Store(int64(1 * time.Second))
}

// Connect implements the [Transport] interface.
//
// The resulting [Connection] writes messages via POST requests to the
// transport URL, and reads messages from hanging requests. If the server
// provides a session ID via the Mcp-Session-Id response header, subsequent
// requests include it; sessionless servers that omit the header are fully
// supported.
//
// When closed, the connection issues a DELETE request to terminate the
// session, unless no session was established.
func (t *StreamableClientTransport) Connect(ctx context.Context) (Connection, error) {
	client := t.HTTPClient
	if client == nil {
		client = http.DefaultClient
	}
	maxRetries := t.MaxRetries
	if maxRetries == 0 {
		maxRetries = 5
	} else if maxRetries < 0 {
		maxRetries = 0
	}
	// Create a new cancellable context that will manage the connection's lifecycle.
	// This is crucial for cleanly shutting down the background SSE listener by
	// cancelling its blocking network operations, which prevents hangs on exit.
	//
	// This context should be detached from the incoming context: the standalone
	// SSE request should not break when the connection context is done.
	//
	// For example, consider that the user may want to wait at most 5s to connect
	// to the server, and therefore uses a context with a 5s timeout when calling
	// client.Connect. Let's suppose that Connect returns after 1s, and the user
	// starts using the resulting session. If we didn't detach here, the session
	// would break after 4s, when the background SSE stream is terminated.
	//
	// Instead, creating a cancellable context detached from the incoming context
	// allows us to preserve context values (which may be necessary for auth
	// middleware), yet only cancel the standalone stream when the connection is closed.
	connCtx, cancel := context.WithCancel(xcontext.Detach(ctx))
	conn := &streamableClientConn{
		url:                  t.Endpoint,
		client:               client,
		incoming:             make(chan jsonrpc.Message, 10),
		done:                 make(chan struct{}),
		maxRetries:           maxRetries,
		strict:               t.strict,
		logger:               ensureLogger(t.logger), // must be non-nil for safe logging
		ctx:                  connCtx,
		cancel:               cancel,
		failed:               make(chan struct{}),
		disableStandaloneSSE: t.DisableStandaloneSSE,
		oauthHandler:         t.OAuthHandler,
		maxEventSize:         t.MaxEventSize,
	}
	return conn, nil
}

type streamableClientConn struct {
	url        string
	client     *http.Client
	ctx        context.Context    // connection context, detached from Connect
	cancel     context.CancelFunc // cancels ctx
	incoming   chan jsonrpc.Message
	maxRetries int
	strict     bool         // from [StreamableClientTransport.strict]
	logger     *slog.Logger // from [StreamableClientTransport.logger]

	// disableStandaloneSSE controls whether to disable the standalone SSE stream
	// for receiving server-to-client notifications when no request is in flight.
	disableStandaloneSSE bool // from [StreamableClientTransport.DisableStandaloneSSE]

	// oauthHandler is the OAuth handler for the connection.
	oauthHandler auth.OAuthHandler // from [StreamableClientTransport.OAuthHandler]

	// maxEventSize bounds the number of bytes buffered while reading a single
	// server-sent event. Resolved from [StreamableClientTransport.MaxEventSize].
	maxEventSize int

	// Guard calls to Close, as it may be called multiple times.
	closeOnce sync.Once
	closeErr  error
	done      chan struct{} // signal graceful termination

	// Logical reads are distributed across multiple http requests. Whenever any
	// of them fails to process their response, we must break the connection, by
	// failing the pending Read.
	//
	// Achieve this by storing the failure message, and signalling when reads are
	// broken. See also [streamableClientConn.fail] and
	// [streamableClientConn.failure].
	failOnce sync.Once
	_failure error
	failed   chan struct{} // signal failure

	// Guard the initialization state.
	mu                sync.Mutex
	initializedResult *InitializeResult
	sessionID         string
}

var _ clientConnection = (*streamableClientConn)(nil)

func (c *streamableClientConn) sessionUpdated(state clientSessionState) {
	c.mu.Lock()
	c.initializedResult = state.InitializeResult
	c.mu.Unlock()

	// Under SEP-2575 (protocol version >= 2026-07-28) the standalone HTTP GET
	// SSE stream is removed.
	if state.InitializeResult == nil ||
		state.InitializeResult.ProtocolVersion >= protocolVersion20260728 {
		return
	}

	// Start the standalone SSE stream as soon as we have the initialized
	// result, if continuous listening is enabled.
	//
	// § 2.2: The client MAY issue an HTTP GET to the MCP endpoint. This can be
	// used to open an SSE stream, allowing the server to communicate to the
	// client, without the client first sending data via HTTP POST.
	//
	// We have to wait for initialized, because until we've received
	// initialized, we don't know whether the server requires a sessionID.
	//
	// § 2.5: A server using the Streamable HTTP transport MAY assign a session
	// ID at initialization time, by including it in a Mcp-Session-Id header
	// on the HTTP response containing the InitializeResult.
	if !c.disableStandaloneSSE {
		c.connectStandaloneSSE()
	}
}

func (c *streamableClientConn) connectStandaloneSSE() {
	resp, err := c.connectSSE(c.ctx, "", 0, true)
	if err != nil {
		// If the client didn't cancel the request, and failure breaks the logical
		// session.
		if c.ctx.Err() == nil {
			c.fail(fmt.Errorf("standalone SSE request failed (session ID: %v): %v", c.sessionID, err))
		}
		return
	}

	// [§2.2.3]: "The server MUST either return Content-Type:
	// text/event-stream in response to this HTTP GET, or else return HTTP
	// 405 Method Not Allowed, indicating that the server does not offer an
	// SSE stream at this endpoint."
	//
	// [§2.2.3]: https://modelcontextprotocol.io/specification/2025-06-18/basic/transports#listening-for-messages-from-the-server
	if resp.StatusCode == http.StatusMethodNotAllowed {
		// The server doesn't support the standalone SSE stream.
		resp.Body.Close()
		return
	}
	if baseMediaType(resp.Header.Get("Content-Type")) != "text/event-stream" {
		// modelcontextprotocol/go-sdk#736: some servers return 200 OK or redirect with
		// non-SSE content type instead of text/event-stream for the standalone
		// SSE stream.
		c.logger.Warn(fmt.Sprintf("got Content-Type %s instead of text/event-stream for standalone SSE stream", resp.Header.Get("Content-Type")))
		resp.Body.Close()
		return
	}
	if resp.StatusCode >= 400 && resp.StatusCode < 500 && !c.strict {
		// modelcontextprotocol/go-sdk#393,#610: some servers return NotFound or
		// other status codes instead of MethodNotAllowed for the standalone SSE
		// stream.
		//
		// Treat this like MethodNotAllowed in non-strict mode.
		c.logger.Warn(fmt.Sprintf("got %d instead of 405 for standalone SSE stream", resp.StatusCode))
		resp.Body.Close()
		return
	}
	summary := "standalone SSE stream"
	if err := c.checkResponse(c.ctx, summary, resp); err != nil {
		c.fail(err)
		return
	}
	go c.handleSSE(c.ctx, summary, resp, nil)
}

// fail handles an asynchronous error while reading.
//
// If err is non-nil, it is terminal, and subsequent (or pending) Reads will
// fail.
//
// If err wraps ErrSessionMissing, the failure indicates that the session is no
// longer present on the server, and no final DELETE will be performed when
// closing the connection.
func (c *streamableClientConn) fail(err error) {
	if err != nil {
		c.failOnce.Do(func() {
			c._failure = err
			close(c.failed)
		})
	}
}

func (c *streamableClientConn) failure() error {
	select {
	case <-c.failed:
		return c._failure
	default:
		return nil
	}
}

func (c *streamableClientConn) SessionID() string {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.sessionID
}

// Read implements the [Connection] interface.
func (c *streamableClientConn) Read(ctx context.Context) (jsonrpc.Message, error) {
	if err := c.failure(); err != nil {
		return nil, err
	}
	select {
	case <-ctx.Done():
		return nil, ctx.Err()
	case <-c.failed:
		return nil, c.failure()
	case <-c.done:
		return nil, io.EOF
	case msg := <-c.incoming:
		return msg, nil
	}
}

// Write implements the [Connection] interface.
func (c *streamableClientConn) Write(ctx context.Context, msg jsonrpc.Message) error {
	if err := c.failure(); err != nil {
		return err
	}

	var requestSummary string
	var requestMethod string
	var forCall *jsonrpc.Request
	switch msg := msg.(type) {
	case *jsonrpc.Request:
		requestSummary = fmt.Sprintf("sending %q", msg.Method)
		if msg.IsCall() {
			forCall = msg
			requestMethod = msg.Method
		}
	case *jsonrpc.Response:
		requestSummary = fmt.Sprintf("sending jsonrpc response #%d", msg.ID)
	default:
		panic("unreachable")
	}

	data, err := jsonrpc.EncodeMessage(msg)
	if err != nil {
		return fmt.Errorf("%s: %v", requestSummary, err)
	}

	doRequest := func() (*http.Request, *http.Response, error) {
		req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.url, bytes.NewReader(data))
		if err != nil {
			return nil, nil, err
		}
		req.Header.Set("Content-Type", "application/json")
		req.Header.Set("Accept", "application/json, text/event-stream")

		if err := c.setMCPHeaders(req, msg); err != nil {
			// Failure to set headers means that the request was not sent.
			// Wrap with ErrRejected so the jsonrpc2 connection doesn't set writeErr
			// and permanently break the connection.
			return nil, nil, fmt.Errorf("%s: %w: %w", requestSummary, jsonrpc2.ErrRejected, err)
		}
		// Keep this after the setMCPHeaders call to ensure that the
		// protocol version header is set.
		setStandardHeaders(ctx, req.Header, msg)
		resp, err := c.client.Do(req)
		if err != nil {
			// Any error from client.Do means the request didn't reach the server.
			// Wrap with ErrRejected so the jsonrpc2 connection doesn't set writeErr
			// and permanently break the connection.
			err = fmt.Errorf("%s: %w: %w", requestSummary, jsonrpc2.ErrRejected, err)
		}
		return req, resp, err
	}

	req, resp, err := doRequest()
	if err != nil {
		return err
	}

	if (resp.StatusCode == http.StatusUnauthorized || resp.StatusCode == http.StatusForbidden) && c.oauthHandler != nil {
		if err := c.oauthHandler.Authorize(ctx, req, resp); err != nil {
			// If the caller's context was cancelled while we were running the
			// authorization flow, treat the connection as failed so subsequent
			// operations on it (e.g. the cancellation notify the call layer
			// sends in response to ctx cancellation) short-circuit instead of
			// re-invoking the OAuth handler. Otherwise the user gets prompted
			// to authorize a request they have already abandoned. See #882.
			//
			// We check ctx.Err() rather than the error returned by Authorize,
			// because the handler is user-implemented and may return an error
			// that does not wrap context.Canceled (e.g. a custom sentinel or
			// a fmt.Errorf with %v). The context itself is the authoritative
			// source for whether the caller abandoned the request.
			ctxErr := ctx.Err()
			if errors.Is(ctxErr, context.Canceled) || errors.Is(ctxErr, context.DeadlineExceeded) {
				c.fail(fmt.Errorf("%s: authorization cancelled: %w", requestSummary, err))
			}
			// Wrap with ErrRejected so the jsonrpc2 connection doesn't set writeErr
			// and permanently break the connection.
			// Wrap the authorization error as well for client inspection.
			return fmt.Errorf("%s: %w: %w", requestSummary, jsonrpc2.ErrRejected, err)
		}
		// Retry the request after successful authorization.
		_, resp, err = doRequest()
		if err != nil {
			return err
		}
	}

	if err := c.checkResponse(ctx, requestSummary, resp); err != nil {
		if (requestMethod == methodDiscover || requestMethod == methodSubscriptionsListen) && !errors.Is(err, jsonrpc2.ErrRejected) {
			// Wrap the discover or subscriptions/listen failure with ErrRejected so
			// the jsonrpc2 layer doesn't set writeErr, which would break the connection
			// (preventing the legacy initialize fallback from succeeding, or breaking
			// subsequent RPCs if subscriptions/listen fails). This covers the case
			// where a server rejects these requests with a non-JSON-RPC body (e.g.
			// plain text 404 or 400), which checkResponse cannot classify as a
			// per-call rejection on its own.
			err = fmt.Errorf("%w: %w", err, jsonrpc2.ErrRejected)
		} else if !errors.Is(err, jsonrpc2.ErrRejected) {
			// Only fail the connection for non-transient errors.
			// Transient errors (wrapped with ErrRejected) should not break the connection.
			c.fail(err)
		}
		return err
	}

	if sessionID := resp.Header.Get(sessionIDHeader); sessionID != "" {
		c.mu.Lock()
		hadSessionID := c.sessionID
		if hadSessionID == "" {
			c.sessionID = sessionID
		}
		c.mu.Unlock()
		if hadSessionID != "" && hadSessionID != sessionID {
			resp.Body.Close()
			return fmt.Errorf("mismatching session IDs %q and %q", hadSessionID, sessionID)
		}
	}

	if forCall == nil {
		resp.Body.Close()

		// [§2.1.4]: "If the input is a JSON-RPC response or notification:
		// If the server accepts the input, the server MUST return HTTP status code 202 Accepted with no body."
		//
		// [§2.1.4]: https://modelcontextprotocol.io/specification/2025-06-18/basic/transports#listening-for-messages-from-the-server
		if resp.StatusCode != http.StatusNoContent && resp.StatusCode != http.StatusAccepted {
			errMsg := fmt.Sprintf("unexpected status code %d from non-call", resp.StatusCode)
			// Some servers return 200, even with an empty json body.
			//
			// In strict mode, return an error to the caller.
			c.logger.Warn(errMsg)
			if c.strict {
				return errors.New(errMsg)
			}
		}
		return nil
	}

	contentType := baseMediaType(resp.Header.Get("Content-Type"))
	switch contentType {
	case "application/json":
		go c.handleJSON(requestSummary, resp)

	case "text/event-stream":
		var forCall *jsonrpc.Request
		if jsonReq, ok := msg.(*jsonrpc.Request); ok && jsonReq.IsCall() {
			forCall = jsonReq
		}
		// Handle the resulting stream. Note that ctx comes from the call, and
		// therefore is already cancelled when the JSON-RPC request is cancelled
		// (or rather, context cancellation is what *triggers* JSON-RPC
		// cancellation)
		go c.handleSSE(ctx, requestSummary, resp, forCall)

	default:
		resp.Body.Close()
		return fmt.Errorf("%s: unsupported content type %q", requestSummary, contentType)
	}
	return nil
}

func (c *streamableClientConn) setMCPHeaders(req *http.Request, msg jsonrpc.Message) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	if c.oauthHandler != nil {
		ts, err := c.oauthHandler.TokenSource(c.ctx)
		if err != nil {
			return err
		}
		if ts != nil {
			token, err := ts.Token()
			if err != nil {
				// If the error is an invalid_grant oauth2.RetrieveError it indicates
				// that the token source doesn't have valid authorization for the token
				// endpoint, per RFC 6749 section 5.2. For example, the refresh token
				// may be expired or invalid.
				//
				// In that case, ignore the error, skip setting the Authorization
				// header, and proceed with the request. Callers that support
				// authorization flows get a 401/403 response and trigger the
				// Authorize() flow to refresh their token.
				var retrieveErr *oauth2.RetrieveError
				if !errors.As(err, &retrieveErr) || retrieveErr.ErrorCode != "invalid_grant" {
					return err
				}
			} else if token != nil {
				req.Header.Set("Authorization", "Bearer "+token.AccessToken)
			}
		}
	}
	if pv := protocolVersionFromMessage(msg); pv != "" {
		req.Header.Set(protocolVersionHeader, pv)
	} else if c.initializedResult != nil {
		req.Header.Set(protocolVersionHeader, c.initializedResult.ProtocolVersion)
	} else if pv := protocolVersionFromContext(req.Context()); pv != "" {
		req.Header.Set(protocolVersionHeader, pv)
	}
	if c.sessionID != "" {
		req.Header.Set(sessionIDHeader, c.sessionID)
	}

	return nil
}

// protocolVersionFromMessage recovers the SEP-2575 `_meta.protocolVersion`
// value from an outgoing JSON-RPC request, if present. It returns "" for
// notifications, responses, requests without a `_meta.protocolVersion`, or a
// nil msg.
func protocolVersionFromMessage(msg jsonrpc.Message) string {
	req, ok := msg.(*jsonrpc.Request)
	if !ok || req == nil {
		return ""
	}
	meta := extractRequestMeta(req.Params)
	if meta == nil {
		return ""
	}
	v, _ := meta[MetaKeyProtocolVersion].(string)
	return v
}

func (c *streamableClientConn) handleJSON(requestSummary string, resp *http.Response) {
	body, err := io.ReadAll(resp.Body)
	resp.Body.Close()
	if err != nil {
		c.fail(fmt.Errorf("%s: failed to read body: %v", requestSummary, err))
		return
	}
	msg, err := jsonrpc.DecodeMessage(body)
	if err != nil {
		c.fail(fmt.Errorf("%s: failed to decode response: %v", requestSummary, err))
		return
	}
	select {
	case c.incoming <- msg:
	case <-c.done:
		// The connection was closed by the client; exit gracefully.
	}
}

// handleSSE manages the lifecycle of an SSE connection. It can be either
// persistent (for the main GET listener) or temporary (for a POST response).
//
// If forCall is set, it is the call that initiated the stream, and the
// stream is complete when we receive its response. Otherwise, this is the
// standalone stream.
func (c *streamableClientConn) handleSSE(ctx context.Context, requestSummary string, resp *http.Response, forCall *jsonrpc2.Request) {
	// Track the last event ID to detect progress.
	// The retry counter is only reset when progress is made (lastEventID advances).
	// This prevents infinite retry loops when a server repeatedly terminates
	// connections without making progress (#679).
	var prevLastEventID string
	retriesWithoutProgress := 0

	for {
		lastEventID, reconnectDelay, clientClosed := c.processStream(ctx, requestSummary, resp, forCall)

		// If the connection was closed by the client, we're done.
		if clientClosed {
			return
		}
		// If we don't have a last event ID, we can never get the call response, so
		// there's nothing to resume. For the standalone stream, we can reconnect,
		// but we may just miss messages.
		if lastEventID == "" && forCall != nil {
			return
		}

		// Check if we made progress (lastEventID advanced).
		// Only reset the retry counter when actual progress is made.
		if lastEventID != "" && lastEventID != prevLastEventID {
			// Progress was made: reset the retry counter.
			retriesWithoutProgress = 0
			prevLastEventID = lastEventID
		} else {
			// No progress: increment the retry counter.
			retriesWithoutProgress++
			if retriesWithoutProgress > c.maxRetries {
				if ctx.Err() == nil {
					c.fail(fmt.Errorf("%s: exceeded %d retries without progress (session ID: %v)", requestSummary, c.maxRetries, c.sessionID))
				}
				return
			}
		}

		// The stream was interrupted or ended by the server. Attempt to reconnect.
		newResp, err := c.connectSSE(ctx, lastEventID, reconnectDelay, false)
		if err != nil {
			// If the client didn't cancel this request, any failure to execute it
			// breaks the logical MCP session.
			if ctx.Err() == nil {
				// All reconnection attempts failed: fail the connection.
				c.fail(fmt.Errorf("%s: failed to reconnect (session ID: %v): %v", requestSummary, c.sessionID, err))
			}
			return
		}

		resp = newResp
		if err := c.checkResponse(ctx, requestSummary, resp); err != nil {
			c.fail(err)
			return
		}
	}
}

// checkResponse checks the status code of the provided response, and
// translates it into an error if the request was unsuccessful.
//
// The response body is close if a non-nil error is returned.
func (c *streamableClientConn) checkResponse(ctx context.Context, requestSummary string, resp *http.Response) (err error) {
	defer func() {
		if err != nil {
			resp.Body.Close()
		}
	}()
	// Transient server errors (502, 503, 504, 429) should not break the connection.
	// Wrap them with ErrRejected so the jsonrpc2 layer doesn't set writeErr.
	if isTransientHTTPStatus(resp.StatusCode) {
		return fmt.Errorf("%w: %s: %v", jsonrpc2.ErrRejected, requestSummary, http.StatusText(resp.StatusCode))
	}
	// By default, always try to decode the body and surface the underlying
	// JSON-RPC error, wrapping it with ErrRejected to prevent the connection from closing.
	// Setting MCPGODEBUG=noprotocolerrorbody=1 restores the previous behavior.
	if noprotocolerrorbody != "1" && (resp.StatusCode < 200 || resp.StatusCode >= 300) {
		body, _ := io.ReadAll(resp.Body)
		msg, _ := jsonrpc.DecodeMessage(body)
		if response, ok := msg.(*jsonrpc.Response); ok && response.Error != nil {
			return fmt.Errorf("%s: %w: %w: %v", requestSummary, response.Error, jsonrpc2.ErrRejected, http.StatusText(resp.StatusCode))
		}
	}
	// §2.5.3: "The server MAY terminate the session at any time, after
	// which it MUST respond to requests containing that session ID with HTTP
	// 404 Not Found."
	// Only a stateful connection with a known session ID can experience ErrSessionMissing.
	if sessionID := c.SessionID(); resp.StatusCode == http.StatusNotFound && sessionID != "" {
		// Return an ErrSessionMissing to avoid sending a redundant DELETE when the
		// session is already gone.
		return fmt.Errorf("%s: failed to connect (session ID: %v): %w", requestSummary, sessionID, ErrSessionMissing)
	}
	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
		return fmt.Errorf("%s: %v", requestSummary, http.StatusText(resp.StatusCode))
	}
	return nil
}

// processStream reads from a single response body, sending events to the
// incoming channel. It returns the ID of the last processed event and a flag
// indicating if the connection was closed by the client. If resp is nil, it
// returns "", false.
func (c *streamableClientConn) processStream(ctx context.Context, requestSummary string, resp *http.Response, forCall *jsonrpc.Request) (lastEventID string, reconnectDelay time.Duration, clientClosed bool) {
	defer func() {
		// Drain any remaining unprocessed body. This allows the connection to be re-used after closing.
		io.Copy(io.Discard, resp.Body)
		resp.Body.Close()
	}()
	for evt, err := range scanEventsLimited(resp.Body, c.maxEventSize) {
		if err != nil {
			if ctx.Err() != nil {
				return "", 0, true // don't reconnect: client cancelled
			}

			// Malformed events are hard errors that indicate corrupted data or protocol
			// violations. These should fail the connection permanently.
			if errors.Is(err, errMalformedEvent) {
				c.fail(fmt.Errorf("%s: %v", requestSummary, err))
				return "", 0, true
			}

			break
		}

		if evt.ID != "" {
			lastEventID = evt.ID
		}

		if evt.Retry != "" {
			if n, err := strconv.ParseInt(evt.Retry, 10, 64); err == nil {
				reconnectDelay = time.Duration(n) * time.Millisecond
			}
		}

		// According to SSE specification
		// (https://html.spec.whatwg.org/multipage/server-sent-events.html#event-stream-interpretation)
		// events with an empty data buffer are allowed.
		// In MCP these can be priming events (SEP-1699) that carry only a Last-Event-ID for stream resumption.
		if len(evt.Data) == 0 {
			continue
		}

		// According to SSE spec, events with no name default to "message"
		if evt.Name != "" && evt.Name != "message" {
			continue
		}

		msg, err := jsonrpc.DecodeMessage(evt.Data)
		if err != nil {
			c.fail(fmt.Errorf("%s: failed to decode event: %v", requestSummary, err))
			return "", 0, true
		}

		select {
		case c.incoming <- msg:
			// Check if this is the response to our call, which terminates the request.
			// (it could also be a server->client request or notification).
			if jsonResp, ok := msg.(*jsonrpc.Response); ok && forCall != nil {
				// TODO: we should never get a response when forReq is nil (the standalone SSE request).
				// We should detect this case.
				if jsonResp.ID == forCall.ID {
					return "", 0, true
				}
			}

		case <-c.done:
			// The connection was closed by the client; exit gracefully.
			return "", 0, true
		}
	}
	// The loop finished without an error, indicating the server closed the stream.
	//
	// If the lastEventID is "", the stream is not retryable and we should
	// report a synthetic error for the call.
	//
	// Note that this is different from the cancellation case above, since the
	// caller is still waiting for a response that will never come.
	if lastEventID == "" && forCall != nil {
		errmsg := &jsonrpc2.Response{
			ID:    forCall.ID,
			Error: fmt.Errorf("request terminated without response"),
		}
		select {
		case c.incoming <- errmsg:
		case <-c.done:
		}
	}
	return lastEventID, reconnectDelay, false
}

// connectSSE handles the logic of connecting a text/event-stream connection.
//
// If lastEventID is set, it is the last-event ID of a stream being resumed.
//
// If connection fails, connectSSE retries with an exponential backoff
// strategy. It returns a new, valid HTTP response if successful, or an error
// if all retries are exhausted.
//
// reconnectDelay is the delay set by the server using the SSE retry field, or
// 0.
//
// If initial is set, this is the initial attempt.
//
// If connectSSE exits due to context cancellation, the result is (nil, ctx.Err()).
func (c *streamableClientConn) connectSSE(ctx context.Context, lastEventID string, reconnectDelay time.Duration, initial bool) (*http.Response, error) {
	var finalErr error
	attempt := 0
	if !initial {
		// We've already connected successfully once, so delay subsequent
		// reconnections. Otherwise, if the server returns 200 but terminates the
		// connection, we'll reconnect as fast as we can, ad infinitum.
		//
		// TODO: we should consider also setting a limit on total attempts for one
		// logical request.
		attempt = 1
	}
	delay := calculateReconnectDelay(attempt)
	if reconnectDelay > 0 {
		delay = reconnectDelay // honor the server's requested initial delay
	}
	for ; attempt <= c.maxRetries; attempt++ {
		select {
		case <-c.done:
			return nil, fmt.Errorf("connection closed by client during reconnect")

		case <-ctx.Done():
			// If the connection context is canceled, the request below will not
			// succeed anyway.
			return nil, ctx.Err()

		case <-time.After(delay):
			req, err := http.NewRequestWithContext(ctx, http.MethodGet, c.url, nil)
			if err != nil {
				return nil, err
			}
			if err := c.setMCPHeaders(req, nil); err != nil {
				return nil, err
			}
			if lastEventID != "" {
				req.Header.Set(lastEventIDHeader, lastEventID)
			}
			req.Header.Set("Accept", "text/event-stream")
			resp, err := c.client.Do(req)
			if err != nil {
				finalErr = err // Store the error and try again.
				delay = calculateReconnectDelay(attempt + 1)
				continue
			}
			return resp, nil
		}
	}
	// If the loop completes, all retries have failed, or the client is closing.
	if finalErr != nil {
		return nil, fmt.Errorf("connection failed after %d attempts: %w", c.maxRetries, finalErr)
	}
	return nil, fmt.Errorf("connection aborted after %d attempts", c.maxRetries)
}

// Close implements the [Connection] interface.
func (c *streamableClientConn) Close() error {
	c.closeOnce.Do(func() {
		if errors.Is(c.failure(), ErrSessionMissing) {
			// If the session is missing, no need to delete it.
		} else if c.SessionID() == "" {
			// No session was established (e.g. the server is stateless),
			// so there is nothing to delete.
		} else {
			reqCtx, stop := context.WithTimeout(c.ctx, closeDeleteTimeout)
			defer stop()
			req, err := http.NewRequestWithContext(reqCtx, http.MethodDelete, c.url, nil)
			if err != nil {
				c.closeErr = err
			} else {
				if err := c.setMCPHeaders(req, nil); err != nil {
					c.closeErr = err
				} else if resp, err := c.client.Do(req); err != nil {
					c.closeErr = err
				} else {
					resp.Body.Close()
				}
			}
		}

		// Cancel any hanging network requests after cleanup.
		c.cancel()
		close(c.done)
	})
	return c.closeErr
}

// calculateReconnectDelay calculates a delay using exponential backoff with full jitter.
func calculateReconnectDelay(attempt int) time.Duration {
	if attempt == 0 {
		return 0
	}
	// Calculate the exponential backoff using the grow factor.
	backoffDuration := time.Duration(float64(reconnectInitialDelay.Load()) * math.Pow(reconnectGrowFactor, float64(attempt-1)))
	// Cap the backoffDuration at maxDelay.
	backoffDuration = min(backoffDuration, reconnectMaxDelay)

	// Use a full jitter using backoffDuration
	jitter := rand.N(backoffDuration)

	return backoffDuration + jitter
}

// isTransientHTTPStatus reports whether the HTTP status code indicates a
// transient server error that should not permanently break the connection.
func isTransientHTTPStatus(statusCode int) bool {
	switch statusCode {
	case http.StatusInternalServerError, // 500
		http.StatusBadGateway,         // 502
		http.StatusServiceUnavailable, // 503
		http.StatusGatewayTimeout,     // 504
		http.StatusTooManyRequests:    // 429
		return true
	}
	return false
}