@@ -17,10 +17,24 @@ import (
1717 "time"
1818
1919 "github.com/mark3labs/mcp-go/mcp"
20+ "github.com/mark3labs/mcp-go/util"
2021)
2122
2223type StreamableHTTPCOption func(*StreamableHTTP)
2324
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.
28+//
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
35+ }
36+}
37+
2438// WithHTTPClient sets a custom HTTP client on the StreamableHTTP transport.
2539func WithHTTPBasicClient(client *http.Client) StreamableHTTPCOption {
2640 return func(sc *StreamableHTTP) {
@@ -54,6 +68,19 @@ func WithHTTPOAuth(config OAuthConfig) StreamableHTTPCOption {
5468 }
5569}
5670
71+func WithLogger(logger util.Logger) StreamableHTTPCOption {
72+ return func(sc *StreamableHTTP) {
73+ sc.logger = logger
74+ }
75+}
76+
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)
81+ }
82+}
83+
5784// StreamableHTTP implements Streamable HTTP transport.
5885//
5986// It transmits JSON-RPC messages over individual HTTP requests. One message per request.
@@ -64,19 +91,22 @@ func WithHTTPOAuth(config OAuthConfig) StreamableHTTPCOption {
6491//
6592// The current implementation does not support the following features:
6693// - 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)
6994// - resuming stream
7095// (https://modelcontextprotocol.io/specification/2025-03-26/basic/transports#resumability-and-redelivery)
7196// - server -> client request
7297type StreamableHTTP struct {
73− serverURL *url.URL
74− httpClient *http.Client
75− headers map[string]string
76− headerFunc HTTPHeaderFunc
98+ serverURL *url.URL
99+ httpClient *http.Client
100+ headers map[string]string
101+ headerFunc HTTPHeaderFunc
102+ logger util.Logger
103+ getListeningEnabled bool
77104
78105 sessionID atomic.Value // string
79106
107+ initialized chan struct{}
108+ initializedOnce sync.Once
109+
80110 notificationHandler func(mcp.JSONRPCNotification)
81111 notifyMu sync.RWMutex
82112
@@ -95,15 +125,19 @@ func NewStreamableHTTP(serverURL string, options ...StreamableHTTPCOption) (*Str
95125 }
96126
97127 smc := &StreamableHTTP{
98− serverURL: parsedURL,
99− httpClient: &http.Client{},
100− headers: make(map[string]string),
101− closed: make(chan struct{}),
128+ serverURL: parsedURL,
129+ httpClient: &http.Client{},
130+ headers: make(map[string]string),
131+ closed: make(chan struct{}),
132+ logger: util.DefaultLogger(),
133+ initialized: make(chan struct{}),
102134 }
103135 smc.sessionID.Store("") // set initial value to simplify later usage
104136
105137 for _, opt := range options {
106− opt(smc)
138+ if opt != nil {
139+ opt(smc)
140+ }
107141 }
108142
109143 // If OAuth is configured, set the base URL for metadata discovery
@@ -118,7 +152,20 @@ func NewStreamableHTTP(serverURL string, options ...StreamableHTTPCOption) (*Str
118152
119153// Start initiates the HTTP connection to the server.
120154func (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 {
157+ go func() {
158+ select {
159+ case <-c.initialized:
160+ ctx, cancel := c.contextAwareOfClientClose(ctx)
161+ defer cancel()
162+ c.listenForever(ctx)
163+ case <-c.closed:
164+ return
165+ }
166+ }()
167+ }
168+
122169 return nil
123170}
124171
@@ -142,13 +189,13 @@ func (c *StreamableHTTP) Close() error {
142189 defer cancel()
143190 req, err := http.NewRequestWithContext(ctx, http.MethodDelete, c.serverURL.String(), nil)
144191 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)
146193 return
147194 }
148195 req.Header.Set(headerKeySessionID, sessionId)
149196 res, err := c.httpClient.Do(req)
150197 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)
152199 return
153200 }
154201 res.Body.Close()
@@ -185,77 +232,29 @@ func (c *StreamableHTTP) SendRequest(
185232 request JSONRPCRequest,
186233) (*JSONRPCResponse, error) {
187234
188− // Create a combined context that could be canceled when the client is closed
189− newCtx, cancel := context.WithCancel(ctx)
190− defer cancel()
191− go func() {
192− select {
193− case <-c.closed:
194− cancel()
195− case <-newCtx.Done():
196− // The original context was canceled, no need to do anything
197− }
198− }()
199− ctx = newCtx
200−
201235 // Marshal request
202236 requestBody, err := json.Marshal(request)
203237 if err != nil {
204238 return nil, fmt.Errorf("failed to marshal request: %w", err)
205239 }
206240
207− // Create HTTP request
208− req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.serverURL.String(), bytes.NewReader(requestBody))
209− if err != nil {
210− return nil, fmt.Errorf("failed to create request: %w", err)
211− }
212−
213− // Set headers
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))
219− }
220− for k, v := range c.headers {
221− req.Header.Set(k, v)
222− }
223−
224− // Add OAuth authorization if configured
225− if c.oauthHandler != nil {
226− authHeader, err := c.oauthHandler.GetAuthorizationHeader(ctx)
227− if err != nil {
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,
232− }
233− }
234− return nil, fmt.Errorf("failed to get authorization header: %w", err)
235− }
236− req.Header.Set("Authorization", authHeader)
237− }
238−
239− if c.headerFunc != nil {
240− for k, v := range c.headerFunc(ctx) {
241− req.Header.Set(k, v)
242− }
243− }
241+ ctx, cancel := c.contextAwareOfClientClose(ctx)
242+ defer cancel()
244243
245− // Send request
246− resp, err := c.httpClient.Do(req)
244+ resp, err := c.sendHTTP(ctx, http.MethodPost, bytes.NewReader(requestBody), "application/json, text/event-stream")
247245 if err != nil {
248− return nil, fmt.Errorf("failed to send request: %w", err)
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. )
250+ } else {
251+ return nil, fmt.Errorf("failed to send request: %w", err)
252+ }
249253 }
250254 defer resp.Body.Close()
251255
252256 // Check if we got an error response
253257 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")
258− }
259258
260259 // Handle OAuth unauthorized error
261260 if resp.StatusCode == http.StatusUnauthorized && c.oauthHandler != nil {
@@ -279,6 +278,10 @@ func (c *StreamableHTTP) SendRequest(
279278 if sessionID := resp.Header.Get(headerKeySessionID); sessionID != "" {
280279 c.sessionID.Store(sessionID)
281280 }
281+
282+ c.initializedOnce.Do(func() {
283+ close(c.initialized)
284+ })
282285 }
283286
284287 // Handle different response types
@@ -300,16 +303,77 @@ func (c *StreamableHTTP) SendRequest(
300303
301304 case "text/event-stream":
302305 // Server is using SSE for streaming responses
303− return c.handleSSEResponse(ctx, resp.Body)
306+ return c.handleSSEResponse(ctx, resp.Body, false)
304307
305308 default:
306309 return nil, fmt.Errorf("unexpected content type: %s", resp.Header.Get("Content-Type"))
307310 }
308311}
309312
313+func (c *StreamableHTTP) sendHTTP(
314+ ctx context.Context,
315+ method string,
316+ body io.Reader,
317+ acceptType string,
318+) (resp *http.Response, err error) {
319+
320+ // Create HTTP request
321+ req, err := http.NewRequestWithContext(ctx, method, c.serverURL.String(), body)
322+ if err != nil {
323+ return nil, fmt.Errorf("failed to create request: %w", err)
324+ }
325+
326+ // Set headers
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)
332+ }
333+ for k, v := range c.headers {
334+ req.Header.Set(k, v)
335+ }
336+
337+ // Add OAuth authorization if configured
338+ if c.oauthHandler != nil {
339+ authHeader, err := c.oauthHandler.GetAuthorizationHeader(ctx)
340+ if err != nil {
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,
345+ }
346+ }
347+ return nil, fmt.Errorf("failed to get authorization header: %w", err)
348+ }
349+ req.Header.Set("Authorization", authHeader)
350+ }
351+
352+ if c.headerFunc != nil {
353+ for k, v := range c.headerFunc(ctx) {
354+ req.Header.Set(k, v)
355+ }
356+ }
357+
358+ // Send request
359+ resp, err = c.httpClient.Do(req)
360+ if err != nil {
361+ return nil, fmt.Errorf("failed to send request: %w", err)
362+ }
363+
364+ // universal handling for session terminated
365+ if resp.StatusCode == http.StatusNotFound {
366+ c.sessionID.CompareAndSwap(sessionID, "")
367+ return nil, ErrSessionTerminated
368+ }
369+
370+ return resp, nil
371+}
372+
310373// handleSSEResponse processes an SSE stream for a specific request.
311374// It returns the final result for the request once received, or an error.
312−func (c *StreamableHTTP) handleSSEResponse(ctx context.Context, reader io.ReadCloser) (*JSONRPCResponse, error) {
375+// If ignoreResponse is true, it won't return when a response messge is received. This is for continuous listening.
376+func (c *StreamableHTTP) handleSSEResponse(ctx context.Context, reader io.ReadCloser, ignoreResponse bool) (*JSONRPCResponse, error) {
313377
314378 // Create a channel for this specific request
315379 responseChan := make(chan *JSONRPCResponse, 1)
@@ -328,7 +392,7 @@ func (c *StreamableHTTP) handleSSEResponse(ctx context.Context, reader io.ReadCl
328392
329393 var message JSONRPCResponse
330394 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)
332396 return
333397 }
334398
@@ -336,7 +400,7 @@ func (c *StreamableHTTP) handleSSEResponse(ctx context.Context, reader io.ReadCl
336400 if message.ID.IsNil() {
337401 var notification mcp.JSONRPCNotification
338402 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)
340404 return
341405 }
342406 c.notifyMu.RLock()
@@ -347,7 +411,9 @@ func (c *StreamableHTTP) handleSSEResponse(ctx context.Context, reader io.ReadCl
347411 return
348412 }
349413
350− responseChan <- &message
414+ if !ignoreResponse {
415+ responseChan <- &message
416+ }
351417 })
352418 }()
353419
@@ -393,7 +459,7 @@ func (c *StreamableHTTP) readSSE(ctx context.Context, reader io.ReadCloser, hand
393459 case <-ctx.Done():
394460 return
395461 default:
396− fmt.Printf("SSE stream error: %v\n", err)
462+ c.logger.Errorf("SSE stream error: %v", err)
397463 return
398464 }
399465 }
@@ -432,44 +498,10 @@ func (c *StreamableHTTP) SendNotification(ctx context.Context, notification mcp.
432498 }
433499
434500 // Create HTTP request
435− req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.serverURL.String(), bytes.NewReader(requestBody))
436− if err != nil {
437− return fmt.Errorf("failed to create request: %w", err)
438− }
439−
440− // Set headers
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))
445− }
446− for k, v := range c.headers {
447− req.Header.Set(k, v)
448− }
449−
450− // Add OAuth authorization if configured
451− if c.oauthHandler != nil {
452− authHeader, err := c.oauthHandler.GetAuthorizationHeader(ctx)
453− if err != nil {
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,
458− }
459− }
460− return fmt.Errorf("failed to get authorization header: %w", err)
461− }
462− req.Header.Set("Authorization", authHeader)
463− }
464−
465− if c.headerFunc != nil {
466− for k, v := range c.headerFunc(ctx) {
467− req.Header.Set(k, v)
468− }
469− }
501+ ctx, cancel := c.contextAwareOfClientClose(ctx)
502+ defer cancel()
470503
471− // Send request
472− resp, err := c.httpClient.Do(req)
504+ resp, err := c.sendHTTP(ctx, http.MethodPost, bytes.NewReader(requestBody), "application/json, text/event-stream")
473505 if err != nil {
474506 return fmt.Errorf("failed to send request: %w", err)
475507 }
@@ -513,3 +545,84 @@ func (c *StreamableHTTP) GetOAuthHandler() *OAuthHandler {
513545func (c *StreamableHTTP) IsOAuthEnabled() bool {
514546 return c.oauthHandler != nil
515547}
548+
549+func (c *StreamableHTTP) listenForever(ctx context.Context) {
550+ c.logger.Infof("listening to server forever")
551+ for {
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")
556+ return
557+ }
558+
559+ select {
560+ case <-ctx.Done():
561+ return
562+ default:
563+ }
564+
565+ if err != nil {
566+ c.logger.Errorf("failed to listen to server. retry in 1 second: %v", err)
567+ }
568+ time.Sleep(retryInterval)
569+ }
570+}
571+
572+var (
573+ ErrSessionTerminated = fmt.Errorf("session terminated (404). need to re-initialize")
574+ ErrGetMethodNotAllowed = fmt.Errorf("GET method not allowed")
575+
576+ retryInterval = 1 * time.Second // a variable is convenient for testing
577+)
578+
579+func (c *StreamableHTTP) createGETConnectionToServer(ctx context.Context) error {
580+
581+ resp, err := c.sendHTTP(ctx, http.MethodGet, nil, "text/event-stream")
582+ if err != nil {
583+ return fmt.Errorf("failed to send request: %w", err)
584+ }
585+ defer resp.Body.Close()
586+
587+ // Check if we got an error response
588+ if resp.StatusCode == http.StatusMethodNotAllowed {
589+ return ErrGetMethodNotAllowed
590+ }
591+
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)
595+ }
596+
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)
601+ }
602+
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)
609+ if err != nil {
610+ return fmt.Errorf("failed to handle SSE response: %w", err)
611+ }
612+
613+ return nil
614+}
615+
616+func (c *StreamableHTTP) contextAwareOfClientClose(ctx context.Context) (context.Context, context.CancelFunc) {
617+ newCtx, cancel := context.WithCancel(ctx)
618+ go func() {
619+ select {
620+ case <-c.closed:
621+ cancel()
622+ case <-newCtx.Done():
623+ // The original context was canceled
624+ cancel()
625+ }
626+ }()
627+ return newCtx, cancel
628+}