From 1a388e3e9ed93b10b7c3e12f9a2b30214251873b Mon Sep 17 00:00:00 2001 From: RuoJi6 <79234113+RuoJi6@users.noreply.github.com> Date: Tue, 15 Sep 2026 00:56:16 +0800 Subject: [PATCH] feat: support legacy SSE MCP servers (#117) --- README.md | 5 + db/schema.sql | 7 +- mcphttp/client.go | 248 ++++++++++++++++++++++++- mcphttp/client_sse_test.go | 107 +++++++++++ server/mcpdiscover.go | 12 +- server/platform_tools.go | 2 +- web/src/app/(main)/system/mcp/page.tsx | 24 ++- web/src/lib/types.ts | 2 +- 8 files changed, 386 insertions(+), 21 deletions(-) create mode 100644 mcphttp/client_sse_test.go diff --git a/README.md b/README.md index c83c86c..01a62a4 100644 --- a/README.md +++ b/README.md @@ -101,6 +101,11 @@ docker compose up -d # 拉取 autumn27/artex 镜像 + postgres 镜像已含常用工具（ripgrep/curl/vim/npm/nmap…）；`./skills` 与 `./data` 以绑定挂载持久化。 +远程 MCP 可在系统设置中选择 `http`（Streamable HTTP）或 `sse`（旧版 SSE）。 +旧版 SSE 服务通常使用 `GET /sse` 建立事件流，再通过服务返回的 +`/message?sessionId=...` 接收 JSON-RPC 请求；配置时将 URL 填为 `/sse`，请求头按 +`Authorization=Bearer ` 填写。 + ### 方式三：下载预编译二进制（Releases） 到 [Releases](https://github.com/Autumn-27/ARTEX/releases) 下载对应平台的 zip，解压后得到 `artex` + `start.sh`（Windows 为 `start.bat`）+ `skills/` + `config.example.json`： diff --git a/db/schema.sql b/db/schema.sql index 3b4dc2e..c34b778 100644 --- a/db/schema.sql +++ b/db/schema.sql @@ -819,7 +819,7 @@ CREATE TABLE IF NOT EXISTS agent_prompt_vars ( CREATE TABLE IF NOT EXISTS mcp_servers ( id BIGSERIAL PRIMARY KEY, name TEXT NOT NULL UNIQUE, - transport TEXT NOT NULL CHECK (transport IN ('stdio','http')), + transport TEXT NOT NULL CHECK (transport IN ('stdio','http','sse')), command TEXT, args JSONB NOT NULL DEFAULT '[]', env JSONB NOT NULL DEFAULT '{}', @@ -829,6 +829,11 @@ CREATE TABLE IF NOT EXISTS mcp_servers ( created_at TIMESTAMPTZ NOT NULL DEFAULT now(), updated_at TIMESTAMPTZ NOT NULL DEFAULT now() ); + +-- Allow legacy MCP SSE servers on databases created before SSE support. +ALTER TABLE mcp_servers DROP CONSTRAINT IF EXISTS mcp_servers_transport_check; +ALTER TABLE mcp_servers ADD CONSTRAINT mcp_servers_transport_check + CHECK (transport IN ('stdio','http','sse')); -- 旧库补列(schema.sql 每次启动都会 Exec)。 ALTER TABLE mcp_servers ADD COLUMN IF NOT EXISTS insecure BOOLEAN NOT NULL DEFAULT false; DROP TRIGGER IF EXISTS trg_mcp_upd ON mcp_servers; diff --git a/mcphttp/client.go b/mcphttp/client.go index 345f346..ee31fd3 100644 --- a/mcphttp/client.go +++ b/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) { diff --git a/mcphttp/client_sse_test.go b/mcphttp/client_sse_test.go new file mode 100644 index 0000000..b470f9c --- /dev/null +++ b/mcphttp/client_sse_test.go @@ -0,0 +1,107 @@ +package mcphttp + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" +) + +func TestLegacySSEClientDiscoverAndCall(t *testing.T) { + var mu sync.Mutex + var stream http.ResponseWriter + var flusher http.Flusher + sessions := make(map[string]bool) + messages := make(chan []byte, 8) + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/sse": + id := "test-session" + mu.Lock() + sessions[id] = true + stream, flusher = w, w.(http.Flusher) + mu.Unlock() + w.Header().Set("Content-Type", "text/event-stream") + w.Header().Set("Cache-Control", "no-cache") + fmt.Fprintf(w, "event: endpoint\ndata: %s/message?sessionId=%s\n\n", "http://"+r.Host, id) + flusher.Flush() + for { + select { + case payload := <-messages: + fmt.Fprintf(w, "event: message\ndata: %s\n\n", payload) + flusher.Flush() + case <-r.Context().Done(): + return + } + } + case "/message": + mu.Lock() + knownSession := sessions["test-session"] + mu.Unlock() + if r.URL.Query().Get("sessionId") != "test-session" || !knownSession { + http.Error(w, "unknown session", http.StatusBadRequest) + return + } + var req rpcRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + w.WriteHeader(http.StatusAccepted) + if req.ID == 0 { + return + } + var result any + switch req.Method { + case "initialize": + result = map[string]any{"protocolVersion": "2024-11-05", "capabilities": map[string]any{"tools": map[string]any{}}, "serverInfo": map[string]any{"name": "test", "version": "1"}} + case "tools/list": + result = map[string]any{"tools": []any{map[string]any{"name": "echo", "description": "echo input", "inputSchema": map[string]any{"type": "object"}}}} + case "tools/call": + result = map[string]any{"content": []any{map[string]any{"type": "text", "text": "ok"}}} + default: + result = map[string]any{} + } + resultJSON, err := json.Marshal(result) + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + payload, _ := json.Marshal(rpcResponse{JSONRPC: "2.0", ID: req.ID, Result: json.RawMessage(resultJSON)}) + mu.Lock() + streamReady := stream != nil + mu.Unlock() + if streamReady { + messages <- payload + } + default: + http.NotFound(w, r) + } + })) + defer srv.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + c, err := NewSSE(ctx, "test", srv.URL+"/sse", map[string]string{"Authorization ": "Bearer test"}, false) + if err != nil { + t.Fatal(err) + } + defer c.Close() + tools, err := c.Tools(ctx) + if err != nil { + t.Fatal(err) + } + if len(tools) != 1 || !strings.Contains(tools[0].Name(), "echo") { + t.Fatalf("unexpected tools: %#v", tools) + } + got, err := c.Call(ctx, "echo", map[string]any{"value": "x"}) + if err != nil || got != "ok" { + t.Fatalf("Call() = %q, %v", got, err) + } +} diff --git a/server/mcpdiscover.go b/server/mcpdiscover.go index edf199b..857cc29 100644 --- a/server/mcpdiscover.go +++ b/server/mcpdiscover.go @@ -6,13 +6,14 @@ import ( "log" "time" - "github.com/Autumn-27/norma/mcp" - actool "github.com/Autumn-27/norma/tool" "github.com/Autumn-27/artex/db" "github.com/Autumn-27/artex/mcphttp" + "github.com/Autumn-27/norma/mcp" + actool "github.com/Autumn-27/norma/tool" ) -// mcpClient is the shared surface of a connected MCP server (stdio or remote http), +// mcpClient is the shared surface of a connected MCP server (stdio, Streamable HTTP, +// or legacy SSE), // so tools/list and cleanup are handled uniformly regardless of transport. type mcpClient interface { Tools(context.Context) ([]actool.CoreTool, error) @@ -33,6 +34,11 @@ func connectMCP(ctx context.Context, m *db.MCPServer) (mcpClient, error) { } // env map doubles as HTTP headers (e.g. Authorization). return mcphttp.New(ctx, m.Name, m.URL, jsonStrMap(m.Env), m.Insecure) + case "sse": + if m.URL == "" { + return nil, fmt.Errorf("sse 传输缺少 URL") + } + return mcphttp.NewSSE(ctx, m.Name, m.URL, jsonStrMap(m.Env), m.Insecure) default: return nil, fmt.Errorf("未知传输方式 %q", m.Transport) } diff --git a/server/platform_tools.go b/server/platform_tools.go index aa7ab2d..871ad8b 100644 --- a/server/platform_tools.go +++ b/server/platform_tools.go @@ -292,7 +292,7 @@ func (a mcpToolInput) toDB() *db.MCPServer { } func (s *Server) toolCreateMCP() actool.CoreTool { - return wrTool("create_mcp", "创建一个 MCP 服务器(stdio/http)。创建后其工具需按 agent 可见性授权。", + return wrTool("create_mcp", "创建一个 MCP 服务器(stdio/http/sse)。创建后其工具需按 agent 可见性授权。", mcpSchema(false), func(_ context.Context, in json.RawMessage) (actool.Result, error) { var a mcpToolInput diff --git a/web/src/app/(main)/system/mcp/page.tsx b/web/src/app/(main)/system/mcp/page.tsx index 267f6b8..152401c 100644 --- a/web/src/app/(main)/system/mcp/page.tsx +++ b/web/src/app/(main)/system/mcp/page.tsx @@ -23,7 +23,7 @@ import { Tabs, TabsList, TabsTrigger, TabsContent } from "@/components/ui/tabs"; import { api } from "@/lib/api"; import type { MCPServer, MCPTool, Agent } from "@/lib/types"; -type Transport = "stdio" | "http"; +type Transport = "stdio" | "http" | "sse"; type FormState = { name: string; transport: Transport; @@ -86,7 +86,10 @@ export default function MCPPage() { const trimmed = line.trim(); if (!trimmed) continue; const idx = trimmed.indexOf("="); - if (idx > 0) out[trimmed.slice(0, idx)] = trimmed.slice(idx + 1); + if (idx > 0) { + const key = trimmed.slice(0, idx).trim(); + if (key) out[key] = trimmed.slice(idx + 1).trim(); + } } return out; } @@ -140,16 +143,16 @@ export default function MCPPage() { toast.error("请填写命令"); return; } - if (form.transport === "http" && !form.url.trim()) { + if (form.transport !== "stdio" && !form.url.trim()) { toast.error("请填写远程 URL"); return; } setSaving(true); try { const base = - form.transport === "http" + form.transport !== "stdio" ? { - transport: "http" as const, + transport: form.transport, url: form.url.trim(), command: "", args: [] as string[], @@ -245,6 +248,13 @@ export default function MCPPage() { > http（远程） +
@@ -300,7 +310,7 @@ export default function MCPPage() { )}
@@ -308,7 +318,7 @@ export default function MCPPage() { id="m-env" className="font-mono" placeholder={ - form.transport === "http" ? "Authorization=Bearer xxxx" : "API_KEY=xxxx\nFOO=bar" + form.transport !== "stdio" ? "Authorization=Bearer xxxx" : "API_KEY=xxxx\nFOO=bar" } value={form.env} onChange={(e) => setF({ env: e.target.value })} diff --git a/web/src/lib/types.ts b/web/src/lib/types.ts index bd9df9b..b9c04f8 100644 --- a/web/src/lib/types.ts +++ b/web/src/lib/types.ts @@ -1064,7 +1064,7 @@ export interface AgentDetail { export interface MCPServer { id: number; name: string; - transport: "stdio" | "http"; + transport: "stdio" | "http" | "sse"; command?: string; args: string[]; env: Record;