@@ -17,10 +17,24 @@ import (
19 "github.com/mark3labs/mcp-go/mcp"
19 "github.com/mark3labs/mcp-go/mcp"
20+ "github.com/mark3labs/mcp-go/util"
22type StreamableHTTPCOption func(*StreamableHTTP)
23type StreamableHTTPCOption func(*StreamableHTTP)
25+// WithContinuousListening enables receiving server-to-client notifications when no request is in flight.
26+// In particular, if you want to receive global notifications from the server (like ToolListChangedNotification),
27+// you should enable this option.
29+// It will establish a standalone long-live GET HTTP connection to the server.
30+// https://modelcontextprotocol.io/specification/2025-03-26/basic/transports#listening-for-messages-from-the-server
31+// NOTICE: Even enabled, the server may not support this feature.
32+func WithContinuousListening() StreamableHTTPCOption {
33+ return func(sc *StreamableHTTP) {
34+ sc.getListeningEnabled = true
24// WithHTTPClient sets a custom HTTP client on the StreamableHTTP transport.
38// WithHTTPClient sets a custom HTTP client on the StreamableHTTP transport.
25func WithHTTPBasicClient(client *http.Client) StreamableHTTPCOption {
39func WithHTTPBasicClient(client *http.Client) StreamableHTTPCOption {
26 return func(sc *StreamableHTTP) {
40 return func(sc *StreamableHTTP) {
@@ -54,6 +68,19 @@ func WithHTTPOAuth(config OAuthConfig) StreamableHTTPCOption {
71+func WithLogger(logger util.Logger) StreamableHTTPCOption {
72+ return func(sc *StreamableHTTP) {
77+// WithSession creates a client with a pre-configured session
78+func WithSession(sessionID string) StreamableHTTPCOption {
79+ return func(sc *StreamableHTTP) {
80+ sc.sessionID.Store(sessionID)
57// StreamableHTTP implements Streamable HTTP transport.
84// StreamableHTTP implements Streamable HTTP transport.
59// It transmits JSON-RPC messages over individual HTTP requests. One message per request.
86// It transmits JSON-RPC messages over individual HTTP requests. One message per request.
@@ -64,19 +91,22 @@ func WithHTTPOAuth(config OAuthConfig) StreamableHTTPCOption {
65// The current implementation does not support the following features:
92// The current implementation does not support the following features:
66// - batching
93// - batching
67−// - continuously listening for server notifications when no request is in flight
68−// (https://modelcontextprotocol.io/specification/2025-03-26/basic/transports#listening-for-messages-from-the-server)
69// - resuming stream
94// - resuming stream
70// (https://modelcontextprotocol.io/specification/2025-03-26/basic/transports#resumability-and-redelivery)
95// (https://modelcontextprotocol.io/specification/2025-03-26/basic/transports#resumability-and-redelivery)
71// - server -> client request
96// - server -> client request
72type StreamableHTTP struct {
97type StreamableHTTP struct {
73− serverURL *url.URL
98+ serverURL *url.URL
74− httpClient *http.Client
99+ httpClient *http.Client
75− headers map[string]string
100+ headers map[string]string
76− headerFunc HTTPHeaderFunc
101+ headerFunc HTTPHeaderFunc
103+ getListeningEnabled bool
78 sessionID atomic.Value // string
105 sessionID atomic.Value // string
107+ initialized chan struct{}
108+ initializedOnce sync.Once
80 notificationHandler func(mcp.JSONRPCNotification)
110 notificationHandler func(mcp.JSONRPCNotification)
81 notifyMu sync.RWMutex
111 notifyMu sync.RWMutex
@@ -95,15 +125,19 @@ func NewStreamableHTTP(serverURL string, options ...StreamableHTTPCOption) (*Str
97 smc := &StreamableHTTP{
127 smc := &StreamableHTTP{
98− serverURL: parsedURL,
128+ serverURL: parsedURL,
99− httpClient: &http.Client{},
129+ httpClient: &http.Client{},
100− headers: make(map[string]string),
130+ headers: make(map[string]string),
131+ closed: make(chan struct{}),
132+ logger: util.DefaultLogger(),
101− closed: make(chan struct{}),
133+ initialized: make(chan struct{}),
103 smc.sessionID.Store("") // set initial value to simplify later usage
135 smc.sessionID.Store("") // set initial value to simplify later usage
105 for _, opt := range options {
137 for _, opt := range options {
106− opt(smc)
139+ opt(smc)
109 // If OAuth is configured, set the base URL for metadata discovery
143 // If OAuth is configured, set the base URL for metadata discovery
@@ -118,7 +152,20 @@ func NewStreamableHTTP(serverURL string, options ...StreamableHTTPCOption) (*Str
119// Start initiates the HTTP connection to the server.
153// Start initiates the HTTP connection to the server.
120func (c *StreamableHTTP) Start(ctx context.Context) error {
154func (c *StreamableHTTP) Start(ctx context.Context) error {
121− // For Streamable HTTP, we don't need to establish a persistent connection
155+ // For Streamable HTTP, we don't need to establish a persistent connection by default
156+ if c.getListeningEnabled {
159+ case <-c.initialized:
160+ ctx, cancel := c.contextAwareOfClientClose(ctx)
162+ c.listenForever(ctx)
122 return nil
169 return nil
@@ -142,13 +189,13 @@ func (c *StreamableHTTP) Close() error {
142 defer cancel()
189 defer cancel()
143 req, err := http.NewRequestWithContext(ctx, http.MethodDelete, c.serverURL.String(), nil)
190 req, err := http.NewRequestWithContext(ctx, http.MethodDelete, c.serverURL.String(), nil)
144 if err != nil {
191 if err != nil {
145− fmt.Printf("failed to create close request\n: %v", err)
192+ c.logger.Errorf("failed to create close request: %v", err)
148 req.Header.Set(headerKeySessionID, sessionId)
195 req.Header.Set(headerKeySessionID, sessionId)
149 res, err := c.httpClient.Do(req)
196 res, err := c.httpClient.Do(req)
150 if err != nil {
197 if err != nil {
151− fmt.Printf("failed to send close request\n: %v", err)
198+ c.logger.Errorf("failed to send close request: %v", err)
154 res.Body.Close()
201 res.Body.Close()
@@ -185,77 +232,29 @@ func (c *StreamableHTTP) SendRequest(
185 request JSONRPCRequest,
232 request JSONRPCRequest,
186) (*JSONRPCResponse, error) {
233) (*JSONRPCResponse, error) {
188− // Create a combined context that could be canceled when the client is closed
189− newCtx, cancel := context.WithCancel(ctx)
195− case <-newCtx.Done():
196− // The original context was canceled, no need to do anything
201 // Marshal request
235 // Marshal request
202 requestBody, err := json.Marshal(request)
236 requestBody, err := json.Marshal(request)
203 if err != nil {
237 if err != nil {
204 return nil, fmt.Errorf("failed to marshal request: %w", err)
238 return nil, fmt.Errorf("failed to marshal request: %w", err)
207− // Create HTTP request
241+ ctx, cancel := c.contextAwareOfClientClose(ctx)
208− req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.serverURL.String(), bytes.NewReader(requestBody))
242+ defer cancel()
210− return nil, fmt.Errorf("failed to create request: %w", err)
214− req.Header.Set("Content-Type", "application/json")
215− req.Header.Set("Accept", "application/json, text/event-stream")
216− sessionID := c.sessionID.Load()
217− if sessionID != "" {
218− req.Header.Set(headerKeySessionID, sessionID.(string))
220− for k, v := range c.headers {
221− req.Header.Set(k, v)
224− // Add OAuth authorization if configured
225− if c.oauthHandler != nil {
226− authHeader, err := c.oauthHandler.GetAuthorizationHeader(ctx)
228− // If we get an authorization error, return a specific error that can be handled by the client
229− if err.Error() == "no valid token available, authorization required" {
230− return nil, &OAuthAuthorizationRequiredError{
231− Handler: c.oauthHandler,
234− return nil, fmt.Errorf("failed to get authorization header: %w", err)
236− req.Header.Set("Authorization", authHeader)
239− if c.headerFunc != nil {
240− for k, v := range c.headerFunc(ctx) {
241− req.Header.Set(k, v)
245− // Send request
244+ resp, err := c.sendHTTP(ctx, http.MethodPost, bytes.NewReader(requestBody), "application/json, text/event-stream")
246− resp, err := c.httpClient.Do(req)
247 if err != nil {
245 if err != nil {
246+ if errors.Is(err, ErrSessionTerminated) && request.Method == string(mcp.MethodInitialize) {
247+ // If the request is initialize, should not return a SessionTerminated error
248+ // It should be a genuine endpoint-routing issue.
249+ // ( Fall through to return StatusCode checking. )
248− return nil, fmt.Errorf("failed to send request: %w", err)
251+ return nil, fmt.Errorf("failed to send request: %w", err)
250 defer resp.Body.Close()
254 defer resp.Body.Close()
252 // Check if we got an error response
256 // Check if we got an error response
253 if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusAccepted {
257 if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusAccepted {
254− // handle session closed
255− if resp.StatusCode == http.StatusNotFound {
256− c.sessionID.CompareAndSwap(sessionID, "")
257− return nil, fmt.Errorf("session terminated (404). need to re-initialize")
260 // Handle OAuth unauthorized error
259 // Handle OAuth unauthorized error
261 if resp.StatusCode == http.StatusUnauthorized && c.oauthHandler != nil {
260 if resp.StatusCode == http.StatusUnauthorized && c.oauthHandler != nil {
@@ -279,6 +278,10 @@ func (c *StreamableHTTP) SendRequest(
279 if sessionID := resp.Header.Get(headerKeySessionID); sessionID != "" {
278 if sessionID := resp.Header.Get(headerKeySessionID); sessionID != "" {
280 c.sessionID.Store(sessionID)
279 c.sessionID.Store(sessionID)
282+ c.initializedOnce.Do(func() {
283+ close(c.initialized)
284 // Handle different response types
287 // Handle different response types
@@ -300,16 +303,77 @@ func (c *StreamableHTTP) SendRequest(
301 case "text/event-stream":
304 case "text/event-stream":
302 // Server is using SSE for streaming responses
305 // Server is using SSE for streaming responses
303− return c.handleSSEResponse(ctx, resp.Body)
306+ return c.handleSSEResponse(ctx, resp.Body, false)
306 return nil, fmt.Errorf("unexpected content type: %s", resp.Header.Get("Content-Type"))
309 return nil, fmt.Errorf("unexpected content type: %s", resp.Header.Get("Content-Type"))
313+func (c *StreamableHTTP) sendHTTP(
314+ ctx context.Context,
318+) (resp *http.Response, err error) {
320+ // Create HTTP request
321+ req, err := http.NewRequestWithContext(ctx, method, c.serverURL.String(), body)
323+ return nil, fmt.Errorf("failed to create request: %w", err)
327+ req.Header.Set("Content-Type", "application/json")
328+ req.Header.Set("Accept", acceptType)
329+ sessionID := c.sessionID.Load().(string)
330+ if sessionID != "" {
331+ req.Header.Set(headerKeySessionID, sessionID)
333+ for k, v := range c.headers {
334+ req.Header.Set(k, v)
337+ // Add OAuth authorization if configured
338+ if c.oauthHandler != nil {
339+ authHeader, err := c.oauthHandler.GetAuthorizationHeader(ctx)
341+ // If we get an authorization error, return a specific error that can be handled by the client
342+ if err.Error() == "no valid token available, authorization required" {
343+ return nil, &OAuthAuthorizationRequiredError{
344+ Handler: c.oauthHandler,
347+ return nil, fmt.Errorf("failed to get authorization header: %w", err)
349+ req.Header.Set("Authorization", authHeader)
352+ if c.headerFunc != nil {
353+ for k, v := range c.headerFunc(ctx) {
354+ req.Header.Set(k, v)
359+ resp, err = c.httpClient.Do(req)
361+ return nil, fmt.Errorf("failed to send request: %w", err)
364+ // universal handling for session terminated
365+ if resp.StatusCode == http.StatusNotFound {
366+ c.sessionID.CompareAndSwap(sessionID, "")
367+ return nil, ErrSessionTerminated
310// handleSSEResponse processes an SSE stream for a specific request.
373// handleSSEResponse processes an SSE stream for a specific request.
311// It returns the final result for the request once received, or an error.
374// It returns the final result for the request once received, or an error.
375+// If ignoreResponse is true, it won't return when a response messge is received. This is for continuous listening.
312−func (c *StreamableHTTP) handleSSEResponse(ctx context.Context, reader io.ReadCloser) (*JSONRPCResponse, error) {
376+func (c *StreamableHTTP) handleSSEResponse(ctx context.Context, reader io.ReadCloser, ignoreResponse bool) (*JSONRPCResponse, error) {
314 // Create a channel for this specific request
378 // Create a channel for this specific request
315 responseChan := make(chan *JSONRPCResponse, 1)
379 responseChan := make(chan *JSONRPCResponse, 1)
@@ -328,7 +392,7 @@ func (c *StreamableHTTP) handleSSEResponse(ctx context.Context, reader io.ReadCl
329 var message JSONRPCResponse
393 var message JSONRPCResponse
330 if err := json.Unmarshal([]byte(data), &message); err != nil {
394 if err := json.Unmarshal([]byte(data), &message); err != nil {
331− fmt.Printf("failed to unmarshal message: %v\n", err)
395+ c.logger.Errorf("failed to unmarshal message: %v", err)
@@ -336,7 +400,7 @@ func (c *StreamableHTTP) handleSSEResponse(ctx context.Context, reader io.ReadCl
336 if message.ID.IsNil() {
400 if message.ID.IsNil() {
337 var notification mcp.JSONRPCNotification
401 var notification mcp.JSONRPCNotification
338 if err := json.Unmarshal([]byte(data), ¬ification); err != nil {
402 if err := json.Unmarshal([]byte(data), ¬ification); err != nil {
339− fmt.Printf("failed to unmarshal notification: %v\n", err)
403+ c.logger.Errorf("failed to unmarshal notification: %v", err)
342 c.notifyMu.RLock()
406 c.notifyMu.RLock()
@@ -347,7 +411,9 @@ func (c *StreamableHTTP) handleSSEResponse(ctx context.Context, reader io.ReadCl
414+ if !ignoreResponse {
350− responseChan <- &message
415+ responseChan <- &message
@@ -393,7 +459,7 @@ func (c *StreamableHTTP) readSSE(ctx context.Context, reader io.ReadCloser, hand
393 case <-ctx.Done():
459 case <-ctx.Done():
396− fmt.Printf("SSE stream error: %v\n", err)
462+ c.logger.Errorf("SSE stream error: %v", err)
@@ -432,44 +498,10 @@ func (c *StreamableHTTP) SendNotification(ctx context.Context, notification mcp.
434 // Create HTTP request
500 // Create HTTP request
435− req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.serverURL.String(), bytes.NewReader(requestBody))
501+ ctx, cancel := c.contextAwareOfClientClose(ctx)
436− if err != nil {
502+ defer cancel()
437− return fmt.Errorf("failed to create request: %w", err)
441− req.Header.Set("Content-Type", "application/json")
442− req.Header.Set("Accept", "application/json, text/event-stream")
443− if sessionID := c.sessionID.Load(); sessionID != "" {
444− req.Header.Set(headerKeySessionID, sessionID.(string))
446− for k, v := range c.headers {
447− req.Header.Set(k, v)
450− // Add OAuth authorization if configured
451− if c.oauthHandler != nil {
452− authHeader, err := c.oauthHandler.GetAuthorizationHeader(ctx)
454− // If we get an authorization error, return a specific error that can be handled by the client
455− if errors.Is(err, ErrOAuthAuthorizationRequired) {
456− return &OAuthAuthorizationRequiredError{
457− Handler: c.oauthHandler,
460− return fmt.Errorf("failed to get authorization header: %w", err)
462− req.Header.Set("Authorization", authHeader)
465− if c.headerFunc != nil {
466− for k, v := range c.headerFunc(ctx) {
467− req.Header.Set(k, v)
471− // Send request
504+ resp, err := c.sendHTTP(ctx, http.MethodPost, bytes.NewReader(requestBody), "application/json, text/event-stream")
472− resp, err := c.httpClient.Do(req)
473 if err != nil {
505 if err != nil {
474 return fmt.Errorf("failed to send request: %w", err)
506 return fmt.Errorf("failed to send request: %w", err)
@@ -513,3 +545,84 @@ func (c *StreamableHTTP) GetOAuthHandler() *OAuthHandler {
513func (c *StreamableHTTP) IsOAuthEnabled() bool {
545func (c *StreamableHTTP) IsOAuthEnabled() bool {
514 return c.oauthHandler != nil
546 return c.oauthHandler != nil
549+func (c *StreamableHTTP) listenForever(ctx context.Context) {
550+ c.logger.Infof("listening to server forever")
552+ err := c.createGETConnectionToServer(ctx)
553+ if errors.Is(err, ErrGetMethodNotAllowed) {
554+ // server does not support listening
555+ c.logger.Errorf("server does not support listening")
566+ c.logger.Errorf("failed to listen to server. retry in 1 second: %v", err)
568+ time.Sleep(retryInterval)
573+ ErrSessionTerminated = fmt.Errorf("session terminated (404). need to re-initialize")
574+ ErrGetMethodNotAllowed = fmt.Errorf("GET method not allowed")
576+ retryInterval = 1 * time.Second // a variable is convenient for testing
579+func (c *StreamableHTTP) createGETConnectionToServer(ctx context.Context) error {
581+ resp, err := c.sendHTTP(ctx, http.MethodGet, nil, "text/event-stream")
583+ return fmt.Errorf("failed to send request: %w", err)
585+ defer resp.Body.Close()
587+ // Check if we got an error response
588+ if resp.StatusCode == http.StatusMethodNotAllowed {
589+ return ErrGetMethodNotAllowed
592+ if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusAccepted {
593+ body, _ := io.ReadAll(resp.Body)
594+ return fmt.Errorf("request failed with status %d: %s", resp.StatusCode, body)
597+ // handle SSE response
598+ contentType := resp.Header.Get("Content-Type")
599+ if contentType != "text/event-stream" {
600+ return fmt.Errorf("unexpected content type: %s", contentType)
603+ // When ignoreResponse is true, the function will never return expect context is done.
604+ // NOTICE: Due to the ambiguity of the specification, other SDKs may use the GET connection to transfer the response
605+ // messages. To be more compatible, we should handle this response, however, as the transport layer is message-based,
606+ // currently, there is no convenient way to handle this response.
607+ // So we ignore the response here. It's not a bug, but may be not compatible with other SDKs.
608+ _, err = c.handleSSEResponse(ctx, resp.Body, true)
610+ return fmt.Errorf("failed to handle SSE response: %w", err)
616+func (c *StreamableHTTP) contextAwareOfClientClose(ctx context.Context) (context.Context, context.CancelFunc) {
617+ newCtx, cancel := context.WithCancel(ctx)
622+ case <-newCtx.Done():
623+ // The original context was canceled
627+ return newCtx, cancel