From dd5a8ff77b125a60ea3e7658a52258639eb5b146 Mon Sep 17 00:00:00 2001 From: Markus Kusano Date: Fri, 5 Dec 2025 17:49:54 +0000 Subject: [PATCH] mcp: clarify documentation for the streamable transports After a deep dive with the team, document areas of the codebase identified as unclear. For now, this is a documentation-only commit (apart from one switch of jsonrpc2.Request.ID.IsValid->Request.IsCall). Special TODO(cleanup) comments are added for additional code changes in follow-up commits. Also, include high level doc comments for server and client transports, in new streamable_server.go and streamable_client.go. Eventually, the server and client transport logic should be migrated to these files, as streamable.go has gotten too large. However, we can do that at a future date, to minimize diff. --- internal/jsonrpc2/conn.go | 13 ++- mcp/shared.go | 20 +++- mcp/streamable.go | 73 +++++++++--- mcp/streamable_client.go | 226 ++++++++++++++++++++++++++++++++++++++ mcp/streamable_server.go | 153 ++++++++++++++++++++++++++ 5 files changed, 462 insertions(+), 23 deletions(-) create mode 100644 mcp/streamable_client.go create mode 100644 mcp/streamable_server.go diff --git a/internal/jsonrpc2/conn.go b/internal/jsonrpc2/conn.go index f4ac86f..9ed924a 100644 --- a/internal/jsonrpc2/conn.go +++ b/internal/jsonrpc2/conn.go @@ -55,9 +55,16 @@ type ConnectionOptions struct { } // Connection manages the jsonrpc2 protocol, connecting responses back to their -// calls. -// Connection is bidirectional; it does not have a designated server or client -// end. +// calls. Connection is bidirectional; it does not have a designated server or +// client end. +// +// Note that the word 'Connection' is overloaded: the mcp.Connection represents +// the bidirectional stream of messages between client an server. The +// jsonrpc2.Connection layers RPC logic on top of that stream, dispatching RPC +// handlers, and correlating requests with responses from the peer. +// +// Some of the complexity of the Connection type is grown out of its usage in +// gopls: it could probably be simplified based on our usage in MCP. type Connection struct { seq int64 // must only be accessed using atomic operations diff --git a/mcp/shared.go b/mcp/shared.go index 43a0d6f..f19236f 100644 --- a/mcp/shared.go +++ b/mcp/shared.go @@ -394,7 +394,7 @@ func notifySessions[S Session, P Params](sessions []S, method string, params P) for _, s := range sessions { req := newRequest(s, params) if err := handleNotify(ctx, method, req); err != nil { - // TODO(jba): surface this error better + // TODO(#218): surface this error better log.Printf("calling %s: %v", method, err) } } @@ -469,15 +469,23 @@ type RequestExtra struct { TokenInfo *auth.TokenInfo // bearer token info (e.g. from OAuth) if any Header http.Header // header from HTTP request, if any - // CloseStream closes the current request stream, if the current transport - // supports replaying requests. + // If set, CloseStream explicitly closes the current request stream. // - // If reconnectAfter is nonzero, it signals to the client to reconnect after - // the given duration. Otherwise, clients may determine their own - // reconnection policy. + // [SEP-1699] introduced server-side SSE stream disconnection: for + // long-running requests, servers may opt to close the SSE stream and + // ask the client to retry at a later time. CloseStream implements this + // feature; if reconnectAfter is set, an event is sent with a `retry:` field + // to configure the reconnection delay. + // + // [SEP-1699]: https://github.com/modelcontextprotocol/modelcontextprotocol/issues/1699 CloseStream func(reconnectAfter time.Duration) } +// TODO(cleanup): switch to an args struct here for forwards compatibility. +// type CloseStreamArgs struct { +// After time.Duration +// } + func (*ClientRequest[P]) isRequest() {} func (*ServerRequest[P]) isRequest() {} diff --git a/mcp/streamable.go b/mcp/streamable.go index 2fcd997..059a1a7 100644 --- a/mcp/streamable.go +++ b/mcp/streamable.go @@ -2,6 +2,10 @@ // 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(cleanup): move the client and server logic into those files. + package mcp import ( @@ -632,6 +636,9 @@ type stream struct { // streamRequests 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{} } @@ -645,6 +652,10 @@ func (s *stream) close(reconnectAfter time.Duration) { // 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 != "" @@ -751,6 +762,7 @@ func (c *streamableServerConn) serveGET(w http.ResponseWriter, req *http.Request stream.mu.Unlock() }() + // TODO(cleanup): factor out, to clarify closing patterns. select { case <-ctx.Done(): // request cancelled @@ -790,10 +802,10 @@ func (c *streamableServerConn) writeCloseEvent(w http.ResponseWriter, reconnectA } } -// acquireStream acquires the stream and replays all events since lastIdx, if -// any, updating lastIdx accordingly. If non-nil, the resulting stream will be -// registered for receiving new messages, and the resulting done channel will -// be closed when all related messages have been delivered. +// acquireStream replays all events since lastIdx, and acquires the ongoing +// stream, if any, updating lastIdx accordingly. If non-nil, the resulting +// stream will be registered for receiving new messages, and the resulting done +// channel will be closed when all related messages have been delivered. // // If any errors occur, they will be written to w and the resulting stream will // be nil. The resulting stream may also be nil if the stream is complete. @@ -803,11 +815,13 @@ func (c *streamableServerConn) writeCloseEvent(w http.ResponseWriter, reconnectA // the stream is still replaying. // // supportsPrimeClose indicates whether the client supports the prime and close -// events (protocol version 2025-11-25 or later). +// events (SEP-1699, added in protocol version 2025-11-25 or later). func (c *streamableServerConn) acquireStream(ctx context.Context, w http.ResponseWriter, streamID string, lastIdx *int, supportsPrimeClose bool) (*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. + // 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] @@ -896,20 +910,23 @@ func (c *streamableServerConn) acquireStream(ctx context.Context, w http.Respons // The stream is not done: register a delivery function before the stream is // unlocked, allowing the connection to write new events. var done = make(chan struct{}) + // TODO(cleanup): consolidate deliver functions, and be consistent about + // checking ctx.Err(). s.deliver = func(data []byte, final bool) error { select { case <-done: return fmt.Errorf("stream closed") + case <-ctx.Done(): + return ctx.Err() default: } - if err := ctx.Err(); err != nil { - return err - } if final { defer close(done) } return c.writeEvent(w, s.id, Event{Name: "message", Data: data}, lastIdx) } + // TODO(cleanup): consolidate closeLocked functions, and be consistent about + // checking ctx.Err(). if supportsPrimeClose { s.closeLocked = func(reconnectAfter time.Duration) { select { @@ -947,6 +964,9 @@ func (c *streamableServerConn) servePOST(w http.ResponseWriter, req *http.Reques 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) @@ -991,12 +1011,15 @@ func (c *streamableServerConn) servePOST(w http.ResponseWriter, req *http.Reques initializeProtocolVersion = params.ProtocolVersion } } + // 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 CloseStream: allow the request handler to explicitly + // close the ongoing stream. jreq.Extra.(*RequestExtra).CloseStream = func(reconnectAfter time.Duration) { c.mu.Lock() streamID, ok := c.requestStreams[jreq.ID] @@ -1025,6 +1048,11 @@ func (c *streamableServerConn) servePOST(w http.ResponseWriter, req *http.Reques // 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 { @@ -1074,6 +1102,10 @@ func (c *streamableServerConn) servePOST(w http.ResponseWriter, req *http.Reques // // In recent protocol versions, there should only be one message as // batching is disabled, as checked above. + // + // Note that in [streamableClientConn.Write], any non-response message is + // sent to the standalone stream (if any), when the connection is + // configured to respond with application/json. msgs = append(msgs, data) if !final { return nil @@ -1099,7 +1131,10 @@ func (c *streamableServerConn) servePOST(w http.ResponseWriter, req *http.Reques // Write events in the order we receive them. lastIndex := -1 if c.eventStore != nil && supportsPrimeClose { - // Write a priming event. + // 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 { @@ -1110,6 +1145,7 @@ func (c *streamableServerConn) servePOST(w http.ResponseWriter, req *http.Reques } } + // TODO(cleanup): consolidate per other TODO, check ctx.Err() stream.deliver = func(data []byte, final bool) error { select { case <-done: @@ -1121,6 +1157,7 @@ func (c *streamableServerConn) servePOST(w http.ResponseWriter, req *http.Reques } return c.writeEvent(w, stream.id, Event{Name: "message", Data: data}, &lastIndex) } + // TODO(cleanup): consolidate per other TODO. if supportsPrimeClose { stream.closeLocked = func(reconnectAfter time.Duration) { select { @@ -1163,6 +1200,8 @@ func (c *streamableServerConn) servePOST(w http.ResponseWriter, req *http.Reques // 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. @@ -1170,6 +1209,7 @@ func (c *streamableServerConn) servePOST(w http.ResponseWriter, req *http.Reques } } + // TODO(cleanup): factor this out, per other TODO. select { case <-req.Context().Done(): // request cancelled @@ -1236,7 +1276,7 @@ func (c *streamableServerConn) Write(ctx context.Context, msg jsonrpc.Message) e return err } - if req, ok := msg.(*jsonrpc.Request); ok && req.ID.IsValid() && (c.stateless || c.sessionID == "") { + 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) } @@ -1689,7 +1729,8 @@ func (c *streamableClientConn) Write(ctx context.Context, msg jsonrpc.Message) e if jsonReq, ok := msg.(*jsonrpc.Request); ok && jsonReq.IsCall() { forCall = jsonReq } - // TODO: should we cancel this logical SSE request if/when jsonReq is canceled? + // TODO(cleanup): decide if we should cancel this logical SSE request + // if/when jsonReq is canceled. go c.handleSSE(ctx, requestSummary, resp, false, forCall) default: @@ -1742,7 +1783,9 @@ func (c *streamableClientConn) handleJSON(requestSummary string, resp *http.Resp // // If forCall is set, it is the call that initiated the stream, and the // stream is complete when we receive its response. -func (c *streamableClientConn) handleSSE(ctx context.Context, requestSummary string, resp *http.Response, persistent bool, forCall *jsonrpc2.Request) { +func (c *streamableClientConn) handleSSE(ctx context.Context, requestSummary string, resp *http.Response, standalone bool, forCall *jsonrpc2.Request) { + // TODO(cleanup): standalone should be completely equivalent to forCall == nil. + // Enforce that at the callsite, and remove the standalone flag argument. for { // Connection was successful. Continue the loop with the new response. // @@ -1760,7 +1803,7 @@ func (c *streamableClientConn) handleSSE(ctx context.Context, requestSummary str } // If the stream has ended, then do not reconnect if the stream is // temporary (POST initiated SSE). - if lastEventID == "" && !persistent { + if lastEventID == "" && !standalone { return } @@ -1844,6 +1887,8 @@ func (c *streamableClientConn) processStream(ctx context.Context, requestSummary 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. diff --git a/mcp/streamable_client.go b/mcp/streamable_client.go new file mode 100644 index 0000000..41a1004 --- /dev/null +++ b/mcp/streamable_client.go @@ -0,0 +1,226 @@ +// 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. + +// TODO: move client-side streamable HTTP logic from streamable.go to this file. + +package mcp + +/* +Streamable HTTP Client Design + +This document describes the client-side implementation of the MCP streamable +HTTP transport, as defined by the MCP spec: +https://modelcontextprotocol.io/specification/2025-11-25/basic/transports#streamable-http + +# Overview + +The client-side streamable transport allows an MCP client to communicate with a +server over HTTP, sending messages via POST and receiving responses via either +JSON or server-sent events (SSE). The implementation consists of two main +components: + + ┌─────────────────────────────────────────────────────────────────┐ + │ [StreamableClientTransport] │ + │ Transport configuration; creates connections via Connect() │ + └─────────────────────────────────────────────────────────────────┘ + │ + ▼ + ┌─────────────────────────────────────────────────────────────────┐ + │ [streamableClientConn] │ + │ Connection implementation; handles HTTP request/response │ + └─────────────────────────────────────────────────────────────────┘ + │ + ├──────────────────────────────────────┐ + ▼ ▼ + ┌─────────────────────────────────────────┐ ┌────────────────────────────────────┐ + │ POST request handlers │ │ Standalone SSE stream │ + │ (one per outgoing message/call) │ │ (server-initiated messages) │ + └─────────────────────────────────────────┘ └────────────────────────────────────┘ + +# Sessions + +The client maintains a session with the server, identified by a session ID +(Mcp-Session-Id header): + + - Session ID is received from the server after initialization + - Client includes the session ID in all subsequent requests + - Session ends when the client calls Close() (sends DELETE) or server returns 404 + +[streamableClientConn] stores the session state: + - [streamableClientConn.sessionID]: Server-assigned session identifier + - [streamableClientConn.initializedResult]: Protocol version and server capabilities + +# Connection Lifecycle + +1. Connect: [StreamableClientTransport.Connect] creates a [streamableClientConn] + with a detached context for the connection's lifetime. The context is detached + to prevent the standalone SSE stream from being cancelled when the original + Connect context times out. + +2. Initialize: The MCP client sends initialize/initialized messages. Upon + receiving [InitializeResult], the connection: + - Stores the negotiated protocol version for the Mcp-Protocol-Version header + - Captures the session ID from the Mcp-Session-Id response header + - Starts the standalone SSE stream via [streamableClientConn.connectStandaloneSSE] + +3. Operation: Messages are sent via POST, responses received via JSON or SSE. + +4. Close: [streamableClientConn.Close] sends a DELETE request to terminate + the session (unless the session is already gone), then cancels the connection + context to clean up the standalone SSE stream. + +# Sending Messages (Write) + +[streamableClientConn.Write] sends all outgoing messages via HTTP POST: + + POST /endpoint + Content-Type: application/json + Accept: application/json, text/event-stream + Mcp-Protocol-Version: + Mcp-Session-Id: + + + +The server may respond with: + - 202 Accepted: Message received, no response body (notifications/responses) + - 200 OK with application/json: Single JSON-RPC response + - 200 OK with text/event-stream: SSE stream of responses + +# Receiving Messages (Read) + +[streamableClientConn.Read] returns messages from the [streamableClientConn.incoming] +channel, which is populated by multiple concurrent goroutines: + +1. POST response handlers ([streamableClientConn.handleJSON] and + [streamableClientConn.handleSSE]): Process responses from POST requests + +2. Standalone SSE stream: Receives server-initiated requests and notifications + +The client handles both response formats: + - JSON: [streamableClientConn.handleJSON] reads body, decodes message + - SSE: [streamableClientConn.handleSSE] scans events, decodes each message + +# Standalone SSE Stream + +After initialization, [streamableClientConn.sessionUpdated] triggers +[streamableClientConn.connectStandaloneSSE] to open a GET request for +server-initiated messages: + + GET /endpoint + Accept: text/event-stream + Mcp-Session-Id: + +Stream behavior: + - Optional: Server may return 405 Method Not Allowed (spec-compliant) or + other 4xx errors (tolerated in non-strict mode for compatibility) + - Persistent: Runs for the connection lifetime in a background goroutine + - Resumable: Uses Last-Event-ID header on reconnection if server provides event IDs + - Reconnects: Automatic reconnection with exponential backoff on interruption + +# Stream Resumption + +When an SSE stream (standalone or POST response) is interrupted, the client +attempts to reconnect using [streamableClientConn.connectSSE]: + +Event ID tracking: + - [streamableClientConn.processStream] tracks the last received event ID + - On reconnection, the Last-Event-ID header is set to resume from that point + - Server replays missed events if it has an [EventStore] configured + +See [calculateReconnectDelay] for the reconnect delay details. + +Server-initiated reconnection (SEP-1699) + - SSE retry field: Sets the delay for the next reconnect attempt + - If server doesn't provide event IDs, non-standalone streams don't reconnect + +# Response Formats + +The client must handle two response formats from POST requests: + +1. application/json: Single JSON-RPC response + - Body contains one JSON-RPC message + - Handled by [streamableClientConn.handleJSON] + - Simpler but doesn't support streaming or server-initiated messages + +2. text/event-stream: SSE stream of messages + - Body contains SSE events with JSON-RPC messages + - Handled by [streamableClientConn.handleSSE] + - Supports multiple messages and server-initiated communication + - Stream completes when the response to the originating call is received + +# HTTP Methods + + - POST: Send JSON-RPC messages (requests, responses, notifications) + - Used by [streamableClientConn.Write] + - Response may be JSON or SSE + + - GET: Open or resume SSE stream for server-initiated messages + - Used by [streamableClientConn.connectSSE] + - Always expects text/event-stream response (or 405) + + - DELETE: Terminate the session + - Used by [streamableClientConn.Close] + - Skipped if session is already known to be gone ([errSessionMissing]) + +# Error Handling + +Errors are categorized and handled differently: + +1. Transient (recoverable via reconnection): + - Network interruption during SSE streaming + - Connection reset or timeout + - Triggers reconnection in [streamableClientConn.handleSSE] + +2. Terminal (breaks the connection): + - 404 Not Found: Session terminated by server ([errSessionMissing]) + - Message decode errors: Protocol violation + - Context cancellation: Client closed connection + - Mismatched session IDs: Protocol error + - See issue #683: our terminal errors are too strict. + +Terminal errors are stored via [streamableClientConn.fail] and returned by +subsequent [streamableClientConn.Read] calls. The [streamableClientConn.failed] +channel signals that the connection is broken. + +Special case: [errSessionMissing] indicates the server has terminated the session, +so [streamableClientConn.Close] skips the DELETE request. + +# Protocol Version Header + +After initialization, all requests include: + + Mcp-Protocol-Version: + +This header (set by [streamableClientConn.setMCPHeaders]): + - Allows the server to handle requests per the negotiated protocol + - Is omitted before initialization completes + - Uses the version from [streamableClientConn.initializedResult] + +# Key Implementation Details + +[StreamableClientTransport] configuration: + - [StreamableClientTransport.Endpoint]: URL of the MCP server + - [StreamableClientTransport.HTTPClient]: Custom HTTP client (optional) + - [StreamableClientTransport.MaxRetries]: Reconnection attempts (default 5) + +[streamableClientConn] handles the [Connection] interface: + - [streamableClientConn.Read]: Returns messages from incoming channel + - [streamableClientConn.Write]: Sends messages via POST, starts response handlers + - [streamableClientConn.Close]: Sends DELETE, cancels context, closes done channel + +State management: + - [streamableClientConn.incoming]: Buffered channel for received messages + - [streamableClientConn.sessionID]: Server-assigned session identifier + - [streamableClientConn.initializedResult]: Cached for protocol version header + - [streamableClientConn.failed]: Channel closed on terminal error + - [streamableClientConn.done]: Channel closed on graceful shutdown + - [streamableClientConn.ctx]: Detached context for connection lifetime + - [streamableClientConn.cancel]: Cancels ctx to terminate SSE streams + +Context handling: + - Connection context is detached from [StreamableClientTransport.Connect] context + using [xcontext.Detach] to preserve context values (for auth middleware) while + preventing premature cancellation of the standalone SSE stream + - Individual POST requests use caller-provided contexts for cancellation +*/ diff --git a/mcp/streamable_server.go b/mcp/streamable_server.go new file mode 100644 index 0000000..120931c --- /dev/null +++ b/mcp/streamable_server.go @@ -0,0 +1,153 @@ +// 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. + +// TODO: move server-side streamable HTTP logic from streamable.go to this file. + +package mcp + +/* +Streamable HTTP Server Design + +This document describes the server-side implementation of the MCP streamable +HTTP transport, as defined by the MCP spec: +https://modelcontextprotocol.io/specification/2025-11-25/basic/transports#streamable-http + +# Overview + +The streamable HTTP transport enables MCP communication over HTTP, with +server-sent events (SSE) for server-to-client messages. The implementation +consists of several layered components: + + ┌─────────────────────────────────────────────────────────────────┐ + │ [StreamableHTTPHandler] │ + │ http.Handler that manages sessions and routes HTTP requests │ + └─────────────────────────────────────────────────────────────────┘ + │ + ▼ + ┌─────────────────────────────────────────────────────────────────┐ + │ [StreamableServerTransport] │ + │ transport implementation, one per session; exposes ServeHTTP │ + └─────────────────────────────────────────────────────────────────┘ + │ + ▼ + ┌─────────────────────────────────────────────────────────────────┐ + │ [streamableServerConn] │ + │ Connection implementation, handles message routing │ + └─────────────────────────────────────────────────────────────────┘ + │ + ▼ + ┌─────────────────────────────────────────────────────────────────┐ + │ [stream] │ + │ Logical message channel within a session, may be resumed │ + └─────────────────────────────────────────────────────────────────┘ + +# Sessions + +As with other transports, a session represents a logical MCP connection between +a client and server. In the streamable transport, sessions are identified by a +unique session ID (Mcp-Session-Id header) and persist across multiple HTTP +requests. + +[StreamableHTTPHandler] maintains a map of active sessions ([sessionInfo]), +each containing: + - The [ServerSession] (MCP-level session state) + - The [StreamableServerTransport] (for message I/O) + - Optional timeout management for idle session cleanup + +Sessions are created on the first POST request (typically containing the +initialize request) and destroyed either by: + - Client sending a DELETE request + - Session timeout due to inactivity + - Server explicitly closing the session + +# Streams + +Within a session, there can be multiple concurrent "streams" - logical channels +for message delivery. This is distinct from HTTP streams; a single [stream] may +span multiple HTTP request/response cycles (via resumption). + +There are two types of streams: + +1. Optional standalone SSE stream (id = ""): + - Created when client sends a GET request to the endpoint + - Used for server-initiated messages (requests/notifications to client) + - Persists for the lifetime of the session + - Only one standalone stream per session + +2. Request streams (id = random string): + - Created for each POST request containing JSON-RPC calls + - Used to route responses back to the originating HTTP request + - Completed when all responses have been sent + - Can be resumed via GET with Last-Event-ID if interrupted + +# Message Routing + +When the server writes a message, it must be routed to the correct [stream]: + + - Responses: Routed to the stream that originated the request + - Requests/Notifications made during request handling: Routed to the same + stream as the triggering request (via context) + - Requests/Notifications made outside request handling: Routed to the + standalone SSE stream + +This routing is implemented using: + - [streamableServerConn.requestStreams] maps request IDs to stream IDs + - [idContextKey] is used to store the originating request ID in Context + - [streamableServerConn.streams] maps stream IDs to [stream] objects + +# Stream Resumption + +If an HTTP connection is interrupted (network issues, etc.), clients can +resume a stream by sending a GET request with the Last-Event-ID header. +This requires an [EventStore] to be configured on the server. + + - [EventStore.Open] is called when a new stream is created + - [EventStore.Append] is called for each message written to the stream + - [EventStore.After] is called to replay messages after a given index + - [EventStore.SessionClosed] is called when the session ends + +Event IDs are formatted as "_" to identify both the +stream and position within that stream (see [formatEventID] and [parseEventID]). + +# Stateless Mode + +For simpler deployments, the handler supports "stateless" mode +([StreamableHTTPOptions.Stateless]) where: + - No session ID validation is performed + - Each request creates a temporary session that's closed after the request + - Server-to-client requests are not supported (no way to receive response) + +This mode is useful for simple tool servers that don't need bidirectional +communication. + +# Response Formats + +The server can respond to POST requests in two formats: + +1. text/event-stream (default): Messages sent as SSE events, supports + streaming multiple messages and server-initiated communication during + request handling. + +2. application/json ([StreamableHTTPOptions.JSONResponse]): Single JSON + response, simpler but doesn't support streaming. Server-initiated messages + during request handling go to the standalone SSE stream instead. + +# HTTP Methods + + - POST: Send JSON-RPC messages (requests, responses, notifications) + - GET: Open standalone SSE stream or resume an interrupted stream + - DELETE: Terminate the session + +# Key Implementation Details + +The [stream] struct manages delivery of messages to HTTP responses: + - [stream.deliver], if set, writes data to the current HTTP response + - [stream.closeLocked] sends a close event with retry delay + - [stream.requests] tracks pending request IDs (stream completes when empty) + +[streamableServerConn] handles the [Connection] interface: + - [streamableServerConn.Read] receives messages from the incoming channel (fed by POST handlers) + - [streamableServerConn.Write] routes messages to appropriate streams + - [streamableServerConn.Close] terminates the session and notifies the [EventStore] +*/