feat: support legacy SSE MCP servers (#117)

mcphttp/client.go

@@ -1,4 +1,4 @@
-// Package mcphttp is a remote (Streamable HTTP) MCP client for ARTEX.
+// Package mcphttp contains remote Streamable HTTP and legacy SSE MCP clients.
 //
 // The core mcp package only speaks stdio; this adapts an HTTP/JSON-RPC MCP server
 // to the same tool.CoreTool shape (mcp__server__tool) WITHOUT touching the SDK.
@@ -14,9 +14,11 @@ import (
 	"context"
 	"crypto/tls"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"net/http"
+	"net/url"
 	"strings"
 	"sync"
 	"time"
@@ -27,6 +29,7 @@ import (
 )
 
 const protocolVersion = "2025-06-18"
+const legacySSEProtocolVersion = "2024-11-05"
 
 // ToolName mirrors mcp.ToolName so remote tools share the mcp__server__tool scheme.
 func ToolName(server, name string) string { return "mcp__" + server + "__" + name }
@@ -52,7 +55,8 @@ type rpcError struct {
 
 func (e *rpcError) Error() string { return fmt.Sprintf("mcp rpc error %d: %s", e.Code, e.Message) }
 
-// Client is a Streamable HTTP connection to one remote MCP server.
+// Client is a connection to one remote MCP server. New uses Streamable HTTP;
+// NewSSE uses the legacy GET /sse + POST /message transport.
 type Client struct {
 	server  string
 	url     string
@@ -62,6 +66,30 @@ type Client struct {
 	mu        sync.Mutex
 	nextID    int
 	sessionID string
+
+	// legacySSE is the pre-2025 MCP SSE transport used by servers such as GSL5.
+	legacySSE    bool
+	messageURL   string
+	streamBody   io.ReadCloser
+	streamCancel context.CancelFunc
+	streamDone   chan struct{}
+	legacyMu     sync.Mutex
+	pendingMu    sync.Mutex
+	pending      map[int]chan *rpcResponse
+	streamErr    chan error
+	protocol     string
+}
+
+func normalizeHeaders(headers map[string]string) (map[string]string, error) {
+	out := make(map[string]string, len(headers))
+	for key, value := range headers {
+		clean := strings.TrimSpace(key)
+		if clean == "" || strings.ContainsAny(clean, "\r\n") {
+			return nil, fmt.Errorf("invalid HTTP header name %q", key)
+		}
+		out[clean] = value
+	}
+	return out, nil
 }
 
 // New connects to a remote MCP endpoint and performs the initialize handshake.
@@ -69,15 +97,20 @@ type Client struct {
 // When insecure is true, TLS certificate verification is skipped so servers that
 // present a self-signed certificate can still be reached (issue #108).
 func New(ctx context.Context, server, url string, headers map[string]string, insecure bool) (*Client, error) {
+	cleanHeaders, err := normalizeHeaders(headers)
+	if err != nil {
+		return nil, err
+	}
 	hc := &http.Client{Timeout: 120 * time.Second}
 	if insecure {
 		hc.Transport = &http.Transport{TLSClientConfig: &tls.Config{InsecureSkipVerify: true}}
 	}
 	c := &Client{
-		server:  server,
-		url:     url,
-		headers: headers,
-		http:    hc,
+		server:   server,
+		url:      url,
+		headers:  cleanHeaders,
+		http:     hc,
+		protocol: protocolVersion,
 	}
 	if err := c.initialize(ctx); err != nil {
 		return nil, err
@@ -85,9 +118,134 @@ func New(ctx context.Context, server, url string, headers map[string]string, ins
 	return c, nil
 }
 
+// NewSSE connects to the legacy MCP SSE transport: a long-lived GET /sse
+// announces a per-session POST /message endpoint, while JSON-RPC responses are
+// delivered asynchronously as SSE message events.
+func NewSSE(ctx context.Context, server, sseURL string, headers map[string]string, insecure bool) (*Client, error) {
+	cleanHeaders, err := normalizeHeaders(headers)
+	if err != nil {
+		return nil, err
+	}
+	hc := &http.Client{} // the SSE stream is intentionally long-lived.
+	if insecure {
+		hc.Transport = &http.Transport{TLSClientConfig: &tls.Config{InsecureSkipVerify: true}}
+	}
+	streamCtx, cancel := context.WithCancel(context.Background())
+	req, err := http.NewRequestWithContext(streamCtx, http.MethodGet, sseURL, nil)
+	if err != nil {
+		cancel()
+		return nil, err
+	}
+	req.Header.Set("Accept", "text/event-stream")
+	for k, v := range cleanHeaders {
+		req.Header.Set(k, v)
+	}
+	resp, err := hc.Do(req)
+	if err != nil {
+		cancel()
+		return nil, err
+	}
+	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
+		defer resp.Body.Close()
+		cancel()
+		return nil, fmt.Errorf("mcp sse http %d: %s", resp.StatusCode, readSnippet(resp.Body))
+	}
+	reader := bufio.NewReader(resp.Body)
+	endpoint, err := readSSEEndpoint(reader, sseURL)
+	if err != nil {
+		resp.Body.Close()
+		cancel()
+		return nil, err
+	}
+	c := &Client{
+		server:       server,
+		url:          sseURL,
+		headers:      cleanHeaders,
+		http:         hc,
+		legacySSE:    true,
+		messageURL:   endpoint,
+		streamBody:   resp.Body,
+		streamCancel: cancel,
+		streamDone:   make(chan struct{}),
+		pending:      make(map[int]chan *rpcResponse),
+		streamErr:    make(chan error, 1),
+		protocol:     legacySSEProtocolVersion,
+	}
+	go c.readLegacySSE(reader)
+	if err := c.initialize(ctx); err != nil {
+		_ = c.Close()
+		return nil, err
+	}
+	return c, nil
+}
+
+func readSSEEndpoint(r *bufio.Reader, base string) (string, error) {
+	for {
+		line, err := r.ReadString('\n')
+		if err != nil {
+			return "", fmt.Errorf("mcp sse endpoint: %w", err)
+		}
+		line = strings.TrimRight(line, "\r\n")
+		if !strings.HasPrefix(line, "data:") {
+			continue
+		}
+		candidate := strings.TrimSpace(strings.TrimPrefix(line, "data:"))
+		if candidate == "" {
+			continue
+		}
+		u, err := url.Parse(candidate)
+		if err != nil {
+			return "", fmt.Errorf("mcp sse endpoint URL: %w", err)
+		}
+		if !u.IsAbs() {
+			b, err := url.Parse(base)
+			if err != nil {
+				return "", err
+			}
+			candidate = b.ResolveReference(u).String()
+		}
+		return candidate, nil
+	}
+}
+
+func (c *Client) readLegacySSE(r *bufio.Reader) {
+	defer close(c.streamDone)
+	for {
+		line, err := r.ReadString('\n')
+		if err != nil {
+			if !errors.Is(err, context.Canceled) {
+				select {
+				case c.streamErr <- err:
+				default:
+				}
+			}
+			return
+		}
+		line = strings.TrimRight(line, "\r\n")
+		if !strings.HasPrefix(line, "data:") {
+			continue
+		}
+		var response rpcResponse
+		if err := json.Unmarshal([]byte(strings.TrimSpace(strings.TrimPrefix(line, "data:"))), &response); err != nil || response.ID == 0 {
+			continue
+		}
+		c.pendingMu.Lock()
+		ch := c.pending[response.ID]
+		delete(c.pending, response.ID)
+		c.pendingMu.Unlock()
+		if ch != nil {
+			ch <- &response
+		}
+	}
+}
+
 func (c *Client) initialize(ctx context.Context) error {
+	version := c.protocol
+	if version == "" {
+		version = protocolVersion
+	}
 	if _, err := c.call(ctx, "initialize", map[string]any{
-		"protocolVersion": protocolVersion,
+		"protocolVersion": version,
 		"capabilities":    map[string]any{},
 		"clientInfo":      map[string]any{"name": "artex", "version": "0.2"},
 	}); err != nil {
@@ -194,6 +352,19 @@ func (c *Client) Call(ctx context.Context, tool string, args any) (string, error
 // Close terminates the MCP session best-effort (DELETE with the session id, per the
 // Streamable HTTP spec). Servers that don't track sessions simply ignore it.
 func (c *Client) Close() error {
+	if c.legacySSE {
+		if c.streamCancel != nil {
+			c.streamCancel()
+		}
+		if c.streamBody != nil {
+			_ = c.streamBody.Close()
+		}
+		select {
+		case <-c.streamDone:
+		case <-time.After(time.Second):
+		}
+		return nil
+	}
 	c.mu.Lock()
 	sid := c.sessionID
 	c.mu.Unlock()
@@ -243,6 +414,9 @@ func (c *Client) notify(ctx context.Context, method string, params any) error {
 // server replies 202 Accepted with no body. Otherwise the reply is parsed from JSON
 // or from an SSE stream, whichever the server chose.
 func (c *Client) roundTrip(ctx context.Context, body rpcRequest, expectResp bool) (*rpcResponse, error) {
+	if c.legacySSE {
+		return c.legacyRoundTrip(ctx, body, expectResp)
+	}
 	data, err := json.Marshal(body)
 	if err != nil {
 		return nil, err
@@ -253,7 +427,11 @@ func (c *Client) roundTrip(ctx context.Context, body rpcRequest, expectResp bool
 	}
 	req.Header.Set("Content-Type", "application/json")
 	req.Header.Set("Accept", "application/json, text/event-stream")
-	req.Header.Set("MCP-Protocol-Version", protocolVersion)
+	version := c.protocol
+	if version == "" {
+		version = protocolVersion
+	}
+	req.Header.Set("MCP-Protocol-Version", version)
 	for k, v := range c.headers {
 		req.Header.Set(k, v)
 	}
@@ -295,6 +473,60 @@ func (c *Client) roundTrip(ctx context.Context, body rpcRequest, expectResp bool
 	return &out, nil
 }
 
+// legacyRoundTrip posts to the endpoint announced by GET /sse. The HTTP POST
+// only acknowledges receipt; the JSON-RPC response arrives on the SSE stream.
+func (c *Client) legacyRoundTrip(ctx context.Context, body rpcRequest, expectResp bool) (*rpcResponse, error) {
+	// Serializing calls keeps the first implementation deterministic while the
+	// shared SSE reader still handles asynchronous delivery.
+	c.legacyMu.Lock()
+	defer c.legacyMu.Unlock()
+
+	data, err := json.Marshal(body)
+	if err != nil {
+		return nil, err
+	}
+	var responseCh chan *rpcResponse
+	if expectResp {
+		responseCh = make(chan *rpcResponse, 1)
+		c.pendingMu.Lock()
+		c.pending[body.ID] = responseCh
+		c.pendingMu.Unlock()
+		defer func() {
+			c.pendingMu.Lock()
+			delete(c.pending, body.ID)
+			c.pendingMu.Unlock()
+		}()
+	}
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.messageURL, bytes.NewReader(data))
+	if err != nil {
+		return nil, err
+	}
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("Accept", "application/json")
+	for k, v := range c.headers {
+		req.Header.Set(k, v)
+	}
+	resp, err := c.http.Do(req)
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
+		return nil, fmt.Errorf("mcp sse http %d: %s", resp.StatusCode, readSnippet(resp.Body))
+	}
+	if !expectResp {
+		return nil, nil
+	}
+	select {
+	case response := <-responseCh:
+		return response, nil
+	case err := <-c.streamErr:
+		return nil, fmt.Errorf("mcp sse stream: %w", err)
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	}
+}
+
 // parseSSE reads an SSE stream and returns the first data frame that is a JSON-RPC
 // response matching wantID (server-to-client requests/notifications are skipped).
 func parseSSE(r io.Reader, wantID int) (*rpcResponse, error) {