Repository: modelcontextprotocol/go-sdk

Issue:
Streamable HTTP support
This issue tracks work on the Streamable HTTP transport:

- [x] Add a `StreamableServerTransport` type.
- [x] Add a `StreamableClientTransport` type.
- [x] Support 'stateless' mode for server transport (not documented, but implicit: see https://github.com/modelcontextprotocol/modelcontextprotocol/issues/282#issuecomment-2998128187).
- [x] Add configurable event store for the streamable server transport (see TypeScript's 'EventStore').
- [x] Support retryability/replayability for the streamable client transport.
- [x] Support a single JSON response (application/json)

Discussion:
@samthanawalla: I can take a look at "Support retryability/replayability for the streamable client transport."@findleyr: @samthanawalla sure, that would be great. I started thinking about persistence in https://go.dev/cl/682555, but haven't done anything toward retryability. I think there is some good prior art in other SDKs. For example:
https://github.com/modelcontextprotocol/typescript-sdk/blob/main/src/client/streamableHttp.ts#L51

I will work on server persistence APIs.@samthanawalla: We also need to support application/json #129 @jpmcb: > Support 'stateless' mode for server transport

I'd be happy to take a look at this since it's important for alot of servers I've been building.

Adding an option to `ServerOptions` to not expect session management seems like an ideal solution. This is how alot of other SDKs implement this (like in the Python SDK):

```python
from mcp.server.fastmcp import FastMCP

# Stateless server (no session persistence)
mcp = FastMCP("StatelessServer", stateless_http=True)
```@findleyr: @jpmcb thank you, and sorry for the missing support. I've been away for a few weeks, but @samthanawalla, @jba and I have a meeting tomorrow to talk about outstanding items, of which streamable transport is the most important.  I think it's likely that we'll prioritize this above essentially everything else, and with Sam working on configurable session storage, it makes sense for us to do stateless mode as well.@whiletrue111: So does it support streamable HTTP?
@jba: > So does it support streamable HTTP?

Yes, at main.@findleyr: Stateless support is next after #257.@findleyr: This will be also be completed by #307.

Solution patch:
diff --git a/internal/jsonrpc2/conn.go b/internal/jsonrpc2/conn.go
index 3223945..fbe0688 100644
--- a/internal/jsonrpc2/conn.go
+++ b/internal/jsonrpc2/conn.go
@@ -125,7 +125,7 @@ func (c *Connection) updateInFlight(f func(*inFlightState)) {
 		// that and avoided making any updates that would cause the state to be
 		// non-idle.)
 		if !s.idle() {
-			panic("jsonrpc2_v2: updateInFlight transitioned to non-idle when already done")
+			panic("jsonrpc2: updateInFlight transitioned to non-idle when already done")
 		}
 		return
 	default:
@@ -718,7 +718,7 @@ func (c *Connection) processResult(from any, req *incomingRequest, result any, e
 	req.cancel()
 	c.updateInFlight(func(s *inFlightState) {
 		if s.incoming == 0 {
-			panic("jsonrpc2_v2: processResult called when incoming count is already zero")
+			panic("jsonrpc2: processResult called when incoming count is already zero")
 		}
 		s.incoming--
 	})
diff --git a/mcp/create-repo.sh b/mcp/create-repo.sh
deleted file mode 100755
index a68964b..0000000
--- a/mcp/create-repo.sh
+++ /dev/null
@@ -1,80 +0,0 @@
-#!/bin/bash
-
-# This script creates an MCP SDK repo from x/tools/internal/mcp (and friends).
-# It will be used as a one-off to create github.com/modelcontextprotocol/go-sdk.
-#
-# Requires https://github.com/newren/git-filter-repo.
-
-set -eu
-
-# Check if exactly one argument is provided
-if [ "$#" -ne 1 ]; then
-  echo "create-repo.sh: create a standalone mcp SDK repo from x/tools"
-  echo "Usage: $0 <mcp repo destination>"
-  exit 1
-fi >&2
-
-src=$(go list -m -f {{.Dir}} golang.org/x/tools)
-dest="$1"
-
-echo "Filtering MCP commits from ${src} to ${dest}..." >&2
-
-startdir=$(pwd)
-tempdir=$(mktemp -d)
-function cleanup {
-  echo "cleaning up ${tempdir}"
-  rm -rf "$tempdir"
-} >&2
-trap cleanup EXIT SIGINT
-
-echo "Checking out to ${tempdir}"
-
-git clone --bare "${src}" "${tempdir}"
-git -C "${tempdir}" --git-dir=. filter-repo \
-  --path internal/mcp/jsonschema --path-rename internal/mcp/jsonschema:jsonschema \
-  --path internal/mcp/design --path-rename internal/mcp/design:design \
-  --path internal/mcp/examples --path-rename internal/mcp/examples:examples \
-  --path internal/mcp/internal --path-rename internal/mcp/internal:internal \
-  --path internal/mcp/README.md --path-rename internal/mcp/README.md:README.md \
-  --path internal/mcp/CONTRIBUTING.md --path-rename internal/mcp/CONTRIBUTING.md:CONTRIBUTING.md \
-  --path internal/mcp --path-rename internal/mcp:mcp \
-  --path internal/jsonrpc2_v2 --path-rename internal/jsonrpc2_v2:internal/jsonrpc2 \
-  --path internal/xcontext \
-  --replace-text "${startdir}/mcp-repo-replace.txt" \
-  --force
-mkdir ${dest}
-cd "${dest}"
-git init
-
-cat << EOF > LICENSE
-MIT License
-
-Copyright (c) 2025 Go MCP SDK Authors
-
-Permission is hereby granted, free of charge, to any person obtaining a copy
-of this software and associated documentation files (the "Software"), to deal
-in the Software without restriction, including without limitation the rights
-to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
-copies of the Software, and to permit persons to whom the Software is
-furnished to do so, subject to the following conditions:
-
-The above copyright notice and this permission notice shall be included in all
-copies or substantial portions of the Software.
-
-THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
-IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
-FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
-AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
-LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
-OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
-SOFTWARE.
-EOF
-
-git add LICENSE && git commit -m "Initial commit: add LICENSE"
-git remote add filtered_source "${tempdir}"
-git pull filtered_source master --allow-unrelated-histories
-git remote remove filtered_source
-go mod init github.com/modelcontextprotocol/go-sdk && go get go@1.23.0
-go mod tidy
-git add go.mod go.sum
-git commit -m "all: add go.mod and go.sum file"
diff --git a/mcp/server.go b/mcp/server.go
index 960d35f..e88455a 100644
--- a/mcp/server.go
+++ b/mcp/server.go
@@ -29,7 +29,7 @@ const DefaultPageSize = 1000
 // A Server is an instance of an MCP server.
 //
 // Servers expose server-side MCP features, which can serve one or more MCP
-// sessions by using [Server.Start] or [Server.Run].
+// sessions by using [Server.Run].
 type Server struct {
 	// fixed at creation
 	impl *Implementation
@@ -69,13 +69,21 @@ type ServerOptions struct {
 	SubscribeHandler func(context.Context, *SubscribeParams) error
 	// Function called when a client session unsubscribes from a resource.
 	UnsubscribeHandler func(context.Context, *UnsubscribeParams) error
+	// If true, advertises the prompts capability during initialization,
+	// even if no prompts have been registered.
+	HasPrompts bool
+	// If true, advertises the resources capability during initialization,
+	// even if no resources have been registered.
+	HasResources bool
+	// If true, advertises the tools capability during initialization,
+	// even if no tools have been registered.
+	HasTools bool
 }
 
 // NewServer creates a new MCP server. The resulting server has no features:
 // add features using the various Server.AddXXX methods, and the [AddTool] function.
 //
-// The server can be connected to one or more MCP clients using [Server.Start]
-// or [Server.Run].
+// The server can be connected to one or more MCP clients using [Server.Run].
 //
 // The first argument must not be nil.
 //
@@ -230,16 +238,17 @@ func (s *Server) capabilities() *serverCapabilities {
 	defer s.mu.Unlock()
 
 	caps := &serverCapabilities{
+		// TODO(samthanawalla): check for completionHandler before advertising capability.
 		Completions: &completionCapabilities{},
 		Logging:     &loggingCapabilities{},
 	}
-	if s.tools.len() > 0 {
+	if s.opts.HasTools || s.tools.len() > 0 {
 		caps.Tools = &toolCapabilities{ListChanged: true}
 	}
-	if s.prompts.len() > 0 {
+	if s.opts.HasPrompts || s.prompts.len() > 0 {
 		caps.Prompts = &promptCapabilities{ListChanged: true}
 	}
-	if s.resources.len() > 0 || s.resourceTemplates.len() > 0 {
+	if s.opts.HasResources || s.resources.len() > 0 || s.resourceTemplates.len() > 0 {
 		caps.Resources = &resourceCapabilities{ListChanged: true}
 		if s.opts.SubscribeHandler != nil {
 			caps.Resources.Subscribe = true
@@ -733,7 +742,7 @@ func (ss *ServerSession) handle(ctx context.Context, req *jsonrpc.Request) (any,
 	}
 	// For the streamable transport, we need the request ID to correlate
 	// server->client calls and notifications to the incoming request from which
-	// they originated. See [idContext] for details.
+	// they originated. See [idContextKey] for details.
 	ctx = context.WithValue(ctx, idContextKey{}, req.ID)
 	return handleReceive(ctx, ss, req)
 }
diff --git a/mcp/shared.go b/mcp/shared.go
index 49dd2b1..cb5234a 100644
--- a/mcp/shared.go
+++ b/mcp/shared.go
@@ -44,7 +44,7 @@ type MethodHandler[S Session] func(ctx context.Context, _ S, method string, para
 // the compiler would complain.
 type methodHandler any // MethodHandler[*ClientSession] | MethodHandler[*ServerSession]
 
-// A Session is either a ClientSession or a ServerSession.
+// A Session is either a [ClientSession] or a [ServerSession].
 type Session interface {
 	*ClientSession | *ServerSession
 	// ID returns the session ID, or the empty string if there is none.
@@ -57,7 +57,7 @@ type Session interface {
 	getConn() *jsonrpc2.Connection
 }
 
-// Middleware is a function from MethodHandlers to MethodHandlers.
+// Middleware is a function from [MethodHandler] to [MethodHandler].
 type Middleware[S Session] func(MethodHandler[S]) MethodHandler[S]
 
 // addMiddleware wraps the handler in the middleware functions.
diff --git a/mcp/streamable.go b/mcp/streamable.go
index e5ffa64..80304a0 100644
--- a/mcp/streamable.go
+++ b/mcp/streamable.go
@@ -179,11 +179,10 @@ func NewStreamableServerTransport(sessionID string, opts *StreamableServerTransp
 		id:             sessionID,
 		incoming:       make(chan jsonrpc.Message, 10),
 		done:           make(chan struct{}),
-		outgoing:       make(map[StreamID][][]byte),
-		signals:        make(map[StreamID]chan struct{}),
+		streams:        make(map[StreamID]*stream),
 		requestStreams: make(map[jsonrpc.ID]StreamID),
-		streamRequests: make(map[StreamID]map[jsonrpc.ID]struct{}),
 	}
+	t.streams[0] = newStream(0)
 	if opts != nil {
 		t.opts = *opts
 	}
@@ -205,11 +204,11 @@ type StreamableServerTransport struct {
 	id       string
 	opts     StreamableServerTransportOptions
 	incoming chan jsonrpc.Message // messages from the client to the server
+	done     chan struct{}
 
 	mu sync.Mutex
 	// Sessions are closed exactly once.
 	isDone bool
-	done   chan struct{}
 
 	// Sessions can have multiple logical connections, corresponding to HTTP
 	// requests. Additionally, logical sessions may be resumed by subsequent HTTP
@@ -219,31 +218,10 @@ type StreamableServerTransport struct {
 	// perform the accounting described below when incoming HTTP requests are
 	// handled.
 	//
-	// The accounting is complicated. It is tempting to merge some of the maps
-	// below, but they each have different lifecycles, as indicated by Lifecycle:
-	// comments.
-	//
 	// TODO(rfindley): simplify.
 
-	// outgoing is the collection of outgoing messages, keyed by the logical
-	// stream ID where they should be delivered.
-	//
-	// streamID 0 is used for messages that don't correlate with an incoming
-	// request.
-	//
-	// Lifecycle: persists for the duration of the session.
-	outgoing map[StreamID][][]byte
-
-	// signals maps a logical stream ID to a 1-buffered channel, owned by an
-	// incoming HTTP request, that signals that there are messages available to
-	// write into the HTTP response. Signals guarantees that at most one HTTP
-	// response can receive messages for a logical stream. After claiming
-	// the stream, incoming requests should read from outgoing, to ensure
-	// that no new messages are missed.
-	//
-	// Lifecycle: signals persists for the duration of an HTTP POST or GET
-	// request for the given streamID.
-	signals map[StreamID]chan struct{}
+	// streams holds the logical streams for this session, keyed by their ID.
+	streams map[StreamID]*stream
 
 	// requestStreams maps incoming requests to their logical stream ID.
 	//
@@ -251,27 +229,63 @@ type StreamableServerTransport struct {
 	//
 	// TODO(rfindley): clean up once requests are handled.
 	requestStreams map[jsonrpc.ID]StreamID
+}
 
-	// streamRequests tracks the set of unanswered incoming RPCs for each logical
-	// stream.
+// 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.
+	// ID 0 is used for messages that don't correlate with an incoming request.
+	id StreamID
+
+	// signal is a 1-buffered channel, owned by an incoming HTTP request, that signals
+	// that there are messages available to write into the HTTP response.
+	// In addition, the presence of a channel guarantees that at most one HTTP response
+	// can receive messages for a logical stream. After claiming the stream, incoming
+	// requests should read from outgoing, to ensure that no new messages are missed.
 	//
-	// When the server has responded to each request, the stream should be
-	// closed.
+	// To simplify locking, signal is an atomic. We need an atomic.Pointer, because
+	// you can't set an atomic.Value to nil.
 	//
-	// Lifecycle: streamRequests values persist as until the requests have been
+	// Lifecycle: each channel value persists for the duration of an HTTP POST or
+	// GET request for the given streamID.
+	signal atomic.Pointer[chan struct{}]
+
+	// The following mutable fields are protected by the mutex of the containing
+	// StreamableServerTransport.
+
+	// outgoing is the list of outgoing messages, enqueued by server methods that
+	// write notifications and responses, and dequeued by streamResponse.
+	outgoing [][]byte
+
+	// streamRequests is the set of unanswered incoming RPCs for the stream.
+	//
+	// Lifecycle: requests values persist until the requests have been
 	// replied to by the server. Notably, NOT until they are sent to an HTTP
 	// response, as delivery is not guaranteed.
-	streamRequests map[StreamID]map[jsonrpc.ID]struct{}
+	requests map[jsonrpc.ID]struct{}
 }
 
-type StreamID int64
+func newStream(id StreamID) *stream {
+	return &stream{
+		id:       id,
+		requests: make(map[jsonrpc.ID]struct{}),
+	}
+}
 
-// a streamableMsg is an SSE event with an index into its logical stream.
-type streamableMsg struct {
-	idx   int
-	event Event
+func signalChanPtr() *chan struct{} {
+	c := make(chan struct{}, 1)
+	return &c
 }
 
+// A StreamID identifies a stream of SSE events. It is unique within the stream's
+// [ServerSession].
+type StreamID int64
+
 // Connect implements the [Transport] interface.
 //
 // TODO(rfindley): Connect should return a new object.
@@ -304,19 +318,25 @@ type idContextKey struct{}
 
 // ServeHTTP handles a single HTTP request for the session.
 func (t *StreamableServerTransport) ServeHTTP(w http.ResponseWriter, req *http.Request) {
+	status := 0
+	message := ""
 	switch req.Method {
 	case http.MethodGet:
-		t.serveGET(w, req)
+		status, message = t.serveGET(w, req)
 	case http.MethodPost:
-		t.servePOST(w, req)
+		status, message = t.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)
+		status = http.StatusMethodNotAllowed
+		message = "unsupported method"
+	}
+	if status != 0 && status != http.StatusOK {
+		http.Error(w, message, status)
 	}
 }
 
-func (t *StreamableServerTransport) serveGET(w http.ResponseWriter, req *http.Request) {
+func (t *StreamableServerTransport) serveGET(w http.ResponseWriter, req *http.Request) (int, string) {
 	// connID 0 corresponds to the default GET request.
 	id := StreamID(0)
 	// By default, we haven't seen a last index. Since indices start at 0, we represent
@@ -328,44 +348,39 @@ func (t *StreamableServerTransport) serveGET(w http.ResponseWriter, req *http.Re
 		var ok bool
 		id, lastIdx, ok = parseEventID(eid)
 		if !ok {
-			http.Error(w, fmt.Sprintf("malformed Last-Event-ID %q", eid), http.StatusBadRequest)
-			return
+			return http.StatusBadRequest, fmt.Sprintf("malformed Last-Event-ID %q", eid)
 		}
 	}
 
 	t.mu.Lock()
-	if _, ok := t.signals[id]; ok {
-		http.Error(w, "stream ID conflicts with ongoing stream", http.StatusBadRequest)
-		t.mu.Unlock()
-		return
-	}
-	signal := make(chan struct{}, 1)
-	t.signals[id] = signal
+	stream, ok := t.streams[id]
 	t.mu.Unlock()
-
-	t.streamResponse(w, req, id, lastIdx, signal)
+	if !ok {
+		return http.StatusBadRequest, "unknown stream"
+	}
+	if !stream.signal.CompareAndSwap(nil, signalChanPtr()) {
+		// The CAS returned false, meaning that the comparison failed: stream.signal is not nil.
+		return http.StatusBadRequest, "stream ID conflicts with ongoing stream"
+	}
+	return t.streamResponse(stream, w, req, lastIdx)
 }
 
-func (t *StreamableServerTransport) servePOST(w http.ResponseWriter, req *http.Request) {
+func (t *StreamableServerTransport) servePOST(w http.ResponseWriter, req *http.Request) (int, string) {
 	if len(req.Header.Values("Last-Event-ID")) > 0 {
-		http.Error(w, "can't send Last-Event-ID for POST request", http.StatusBadRequest)
-		return
+		return http.StatusBadRequest, "can't send Last-Event-ID for POST request"
 	}
 
 	// Read incoming messages.
 	body, err := io.ReadAll(req.Body)
 	if err != nil {
-		http.Error(w, "failed to read body", http.StatusBadRequest)
-		return
+		return http.StatusBadRequest, "failed to read body"
 	}
 	if len(body) == 0 {
-		http.Error(w, "POST requires a non-empty body", http.StatusBadRequest)
-		return
+		return http.StatusBadRequest, "POST requires a non-empty body"
 	}
 	incoming, _, err := readBatch(body)
 	if err != nil {
-		http.Error(w, fmt.Sprintf("malformed payload: %v", err), http.StatusBadRequest)
-		return
+		return http.StatusBadRequest, fmt.Sprintf("malformed payload: %v", err)
 	}
 	requests := make(map[jsonrpc.ID]struct{})
 	for _, msg := range incoming {
@@ -375,18 +390,18 @@ func (t *StreamableServerTransport) servePOST(w http.ResponseWriter, req *http.R
 	}
 
 	// Update accounting for this request.
-	id := StreamID(t.nextStreamID.Add(1))
-	signal := make(chan struct{}, 1)
+	stream := newStream(StreamID(t.nextStreamID.Add(1)))
 	t.mu.Lock()
+	t.streams[stream.id] = stream
 	if len(requests) > 0 {
-		t.streamRequests[id] = make(map[jsonrpc.ID]struct{})
+		stream.requests = make(map[jsonrpc.ID]struct{})
 	}
 	for reqID := range requests {
-		t.requestStreams[reqID] = id
-		t.streamRequests[id][reqID] = struct{}{}
+		t.requestStreams[reqID] = stream.id
+		stream.requests[reqID] = struct{}{}
 	}
-	t.signals[id] = signal
 	t.mu.Unlock()
+	stream.signal.Store(signalChanPtr())
 
 	// Publish incoming messages.
 	for _, msg := range incoming {
@@ -396,16 +411,12 @@ func (t *StreamableServerTransport) servePOST(w http.ResponseWriter, req *http.R
 	// TODO(rfindley): consider optimizing for a single incoming request, by
 	// responding with application/json when there is only a single message in
 	// the response.
-	t.streamResponse(w, req, id, -1, signal)
+	return t.streamResponse(stream, w, req, -1)
 }
 
 // lastIndex is the index of the last seen event if resuming, else -1.
-func (t *StreamableServerTransport) streamResponse(w http.ResponseWriter, req *http.Request, id StreamID, lastIndex int, signal chan struct{}) {
-	defer func() {
-		t.mu.Lock()
-		delete(t.signals, id)
-		t.mu.Unlock()
-	}()
+func (t *StreamableServerTransport) streamResponse(stream *stream, w http.ResponseWriter, req *http.Request, lastIndex int) (int, string) {
+	defer stream.signal.Store(nil)
 
 	writes := 0
 
@@ -414,11 +425,12 @@ func (t *StreamableServerTransport) streamResponse(w http.ResponseWriter, req *h
 		lastIndex++
 		e := Event{
 			Name: "message",
-			ID:   formatEventID(id, lastIndex),
+			ID:   formatEventID(stream.id, lastIndex),
 			Data: data,
 		}
 		if _, err := writeEvent(w, e); err != nil {
 			// Connection closed or broken.
+			// TODO(#170): log when we add server-side logging.
 			return false
 		}
 		writes++
@@ -432,7 +444,7 @@ func (t *StreamableServerTransport) streamResponse(w http.ResponseWriter, req *h
 
 	if lastIndex >= 0 {
 		// Resume.
-		for data, err := range t.opts.EventStore.After(req.Context(), t.SessionID(), id, lastIndex) {
+		for data, err := range t.opts.EventStore.After(req.Context(), t.SessionID(), stream.id, lastIndex) {
 			if err != nil {
 				// TODO: reevaluate these status codes.
 				// Maybe distinguish between storage errors, which are 500s, and missing
@@ -441,13 +453,12 @@ func (t *StreamableServerTransport) streamResponse(w http.ResponseWriter, req *h
 				if errors.Is(err, ErrEventsPurged) {
 					status = http.StatusInsufficientStorage
 				}
-				http.Error(w, err.Error(), status)
-				return
+				return status, err.Error()
 			}
 			// The iterator yields events beginning just after lastIndex, or it would have
 			// yielded an error.
 			if !write(data) {
-				return
+				return 0, ""
 			}
 		}
 	}
@@ -456,42 +467,41 @@ stream:
 	// Repeatedly collect pending outgoing events and send them.
 	for {
 		t.mu.Lock()
-		outgoing := t.outgoing[id]
-		t.outgoing[id] = nil
+		outgoing := stream.outgoing
+		stream.outgoing = nil
 		t.mu.Unlock()
 
 		for _, data := range outgoing {
-			if err := t.opts.EventStore.Append(req.Context(), t.id, id, data); err != nil {
-				http.Error(w, err.Error(), http.StatusInternalServerError)
-				return
+			if err := t.opts.EventStore.Append(req.Context(), t.SessionID(), stream.id, data); err != nil {
+				return http.StatusInternalServerError, err.Error()
 			}
 			if !write(data) {
-				return
+				return 0, ""
 			}
 		}
 
 		t.mu.Lock()
-		nOutstanding := len(t.streamRequests[id])
+		nOutstanding := len(stream.requests)
 		t.mu.Unlock()
 		// If all requests have been handled and replied to, we should terminate this connection.
 		// "After the JSON-RPC response has been sent, the server SHOULD close the SSE stream."
 		// §6.4, https://modelcontextprotocol.io/specification/2025-06-18/basic/transports#sending-messages-to-the-server
-		// TODO(jba): why not terminate regardless of http method?
+		// TODO(jba,findleyr): why not terminate regardless of http method?
 		if req.Method == http.MethodPost && nOutstanding == 0 {
 			if writes == 0 {
 				// Spec: If the server accepts the input, the server MUST return HTTP
 				// status code 202 Accepted with no body.
 				w.WriteHeader(http.StatusAccepted)
 			}
-			return
+			return 0, ""
 		}
 
 		select {
-		case <-signal: // there are new outgoing messages
+		case <-*stream.signal.Load(): // there are new outgoing messages
 			// return to top of loop
 		case <-t.done: // session is closed
 			if writes == 0 {
-				http.Error(w, "session terminated", http.StatusGone)
+				return http.StatusGone, "session terminated"
 			}
 			break stream
 		case <-req.Context().Done():
@@ -501,6 +511,7 @@ stream:
 			break stream
 		}
 	}
+	return 0, ""
 }
 
 // Event IDs: encode both the logical connection ID and the index, as
@@ -585,30 +596,32 @@ func (t *StreamableServerTransport) Write(ctx context.Context, msg jsonrpc.Messa
 	t.mu.Lock()
 	defer t.mu.Unlock()
 	if t.isDone {
-		return fmt.Errorf("session is closed") // TODO: should this be EOF?
+		return errors.New("session is closed")
 	}
 
-	if _, ok := t.streamRequests[forConn]; !ok && forConn != 0 {
+	stream := t.streams[forConn]
+	if stream == nil {
+		return fmt.Errorf("no stream with ID %d", forConn)
+	}
+	if len(stream.requests) == 0 && forConn != 0 {
 		// No outstanding requests for this connection, which means it is logically
 		// done. This is a sequencing violation from the server, so we should report
 		// a side-channel error here. Put the message on the general queue to avoid
 		// dropping messages.
-		forConn = 0
+		stream = t.streams[0]
 	}
 
-	t.outgoing[forConn] = append(t.outgoing[forConn], data)
+	stream.outgoing = append(stream.outgoing, data)
 	if replyTo.IsValid() {
 		// Once we've put the reply on the queue, it's no longer outstanding.
-		delete(t.streamRequests[forConn], replyTo)
-		if len(t.streamRequests[forConn]) == 0 {
-			delete(t.streamRequests, forConn)
-		}
+		delete(stream.requests, replyTo)
 	}
 
-	// Signal work.
-	if c, ok := t.signals[forConn]; ok {
+	// Signal streamResponse that new work is available.
+	signalp := stream.signal.Load()
+	if signalp != nil {
 		select {
-		case c <- struct{}{}:
+		case *signalp <- struct{}{}:
 		default:
 		}
 	}
@@ -800,6 +813,8 @@ func (s *streamableClientConn) Write(ctx context.Context, msg jsonrpc.Message) e
 	return nil
 }
 
+// postMessage POSTs msg to the server and reads the response.
+// It returns the session ID from the response.
 func (s *streamableClientConn) postMessage(ctx context.Context, sessionID string, msg jsonrpc.Message) (string, error) {
 	data, err := jsonrpc2.EncodeMessage(msg)
 	if err != nil {
@@ -836,9 +851,17 @@ func (s *streamableClientConn) postMessage(ctx context.Context, sessionID string
 		// Section 2.1: The SSE stream is initiated after a POST.
 		go s.handleSSE(resp)
 	case "application/json":
-		// TODO: read the body and send to s.incoming (in a select that also receives from s.done).
+		body, err := io.ReadAll(resp.Body)
 		resp.Body.Close()
-		return "", fmt.Errorf("streamable HTTP client does not yet support raw JSON responses")
+		if err != nil {
+			return "", err
+		}
+		select {
+		case s.incoming <- body:
+		case <-s.done:
+			// The connection was closed by the client; exit gracefully.
+		}
+		return sessionID, nil
 	default:
 		resp.Body.Close()
 		return "", fmt.Errorf("unsupported content type %q", ct)
diff --git a/mcp/transport.go b/mcp/transport.go
index a7de506..5175f6f 100644
--- a/mcp/transport.go
+++ b/mcp/transport.go
@@ -10,6 +10,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"log"
 	"net"
 	"os"
 	"sync"
@@ -120,6 +121,7 @@ func connect[H handler](ctx context.Context, t Transport, b binder[H]) (H, error
 		OnDone: func() {
 			b.disconnect(h)
 		},
+		OnInternalError: func(err error) { log.Printf("jsonrpc2 error: %v", err) },
 	})
 	assert(preempter.conn != nil, "unbound preempter")
 	h.setConn(conn)
@@ -132,7 +134,7 @@ type canceller struct {
 	conn *jsonrpc2.Connection
 }
 
-// Preempt implements jsonrpc2.Preempter.
+// Preempt implements [jsonrpc2.Preempter].
 func (c *canceller) Preempt(ctx context.Context, req *jsonrpc.Request) (result any, err error) {
 	if req.Method == "notifications/cancelled" {
 		var params CancelledParams
@@ -201,7 +203,7 @@ type loggingConn struct {
 
 func (c *loggingConn) SessionID() string { return c.delegate.SessionID() }
 
-// loggingReader is a stream middleware that logs incoming messages.
+// Read is a stream middleware that logs incoming messages.
 func (s *loggingConn) Read(ctx context.Context) (jsonrpc.Message, error) {
 	msg, err := s.delegate.Read(ctx)
 	if err != nil {
@@ -216,7 +218,7 @@ func (s *loggingConn) Read(ctx context.Context) (jsonrpc.Message, error) {
 	return msg, err
 }
 
-// loggingWriter is a stream middleware that logs outgoing messages.
+// Write is a stream middleware that logs outgoing messages.
 func (s *loggingConn) Write(ctx context.Context, msg jsonrpc.Message) error {
 	err := s.delegate.Write(ctx, msg)
 	if err != nil {



Test patch:
diff --git a/mcp/server_test.go b/mcp/server_test.go
index 9a36c96..79c607e 100644
--- a/mcp/server_test.go
+++ b/mcp/server_test.go
@@ -333,6 +333,22 @@ func TestServerCapabilities(t *testing.T) {
 				Tools:       &toolCapabilities{ListChanged: true},
 			},
 		},
+		{
+			name:            "With initial capabilities",
+			configureServer: func(s *Server) {},
+			serverOpts: ServerOptions{
+				HasPrompts:   true,
+				HasResources: true,
+				HasTools:     true,
+			},
+			wantCapabilities: &serverCapabilities{
+				Completions: &completionCapabilities{},
+				Logging:     &loggingCapabilities{},
+				Prompts:     &promptCapabilities{ListChanged: true},
+				Resources:   &resourceCapabilities{ListChanged: true},
+				Tools:       &toolCapabilities{ListChanged: true},
+			},
+		},
 	}
 
 	for _, tc := range testCases {
diff --git a/mcp/streamable_test.go b/mcp/streamable_test.go
index 864265e..185bc63 100644
--- a/mcp/streamable_test.go
+++ b/mcp/streamable_test.go
@@ -158,7 +158,8 @@ func TestClientReplay(t *testing.T) {
 	client := NewClient(testImpl, &ClientOptions{
 		ProgressNotificationHandler: func(ctx context.Context, cc *ClientSession, params *ProgressNotificationParams) {
 			notifications <- params.Message
-		}})
+		},
+	})
 	clientSession, err := client.Connect(ctx, NewStreamableClientTransport(proxy.URL, nil))
 	if err != nil {
 		t.Fatalf("client.Connect() failed: %v", err)
@@ -678,6 +679,51 @@ func mustMarshal(t *testing.T, v any) json.RawMessage {
 	return data
 }
 
+func TestStreamableClientTransportApplicationJSON(t *testing.T) {
+	// Test handling of application/json responses.
+	ctx := context.Background()
+	resp := func(id int64, result any, err error) *jsonrpc.Response {
+		return &jsonrpc.Response{
+			ID:     jsonrpc2.Int64ID(id),
+			Result: mustMarshal(t, result),
+			Error:  err,
+		}
+	}
+	initResult := &InitializeResult{
+		Capabilities: &serverCapabilities{
+			Completions: &completionCapabilities{},
+			Logging:     &loggingCapabilities{},
+			Tools:       &toolCapabilities{ListChanged: true},
+		},
+		ProtocolVersion: latestProtocolVersion,
+		ServerInfo:      &Implementation{Name: "testServer", Version: "v1.0.0"},
+	}
+	initResp := resp(1, initResult, nil)
+
+	serverHandler := func(w http.ResponseWriter, r *http.Request) {
+		data, err := jsonrpc2.EncodeMessage(initResp)
+		if err != nil {
+			t.Fatal(err)
+		}
+		w.Header().Set("Content-Type", "application/json")
+		w.Write(data)
+	}
+
+	httpServer := httptest.NewServer(http.HandlerFunc(serverHandler))
+	defer httpServer.Close()
+
+	transport := NewStreamableClientTransport(httpServer.URL, nil)
+	client := NewClient(testImpl, nil)
+	session, err := client.Connect(ctx, transport)
+	if err != nil {
+		t.Fatalf("client.Connect() failed: %v", err)
+	}
+	defer session.Close()
+	if diff := cmp.Diff(initResult, session.initializeResult); diff != "" {
+		t.Errorf("mismatch (-want, +got):\n%s", diff)
+	}
+}
+
 func TestEventID(t *testing.T) {
 	tests := []struct {
 		sid StreamID