feat: update Go filter mcp-server (#1950)

Co-authored-by: johnlanni <zty98751@alibaba-inc.com>

plugins/golang-filter/mcp-server/config.go

@@ -7,12 +7,14 @@ import (
 	"google.golang.org/protobuf/types/known/anypb"
 
 	"github.com/alibaba/higress/plugins/golang-filter/mcp-server/internal"
-	_ "github.com/alibaba/higress/plugins/golang-filter/mcp-server/servers/gorm" // 导入gorm包以执行其init函数
+	_ "github.com/alibaba/higress/plugins/golang-filter/mcp-server/servers/gorm"
 	"github.com/envoyproxy/envoy/contrib/golang/common/go/api"
 	envoyHttp "github.com/envoyproxy/envoy/contrib/golang/filters/http/source/go/pkg/http"
 )
 
 const Name = "mcp-server"
+const Version = "1.0.0"
+const DefaultServerName = "default"
 
 func init() {
 	envoyHttp.RegisterHttpFilterFactoryAndConfigParser(Name, filterFactory, &parser{})
@@ -23,6 +25,7 @@ type config struct {
 	redisClient   *internal.RedisClient
 	stopChan      chan struct{}
 	servers       []*internal.SSEServer
+	defaultServer *internal.SSEServer
 }
 
 type parser struct {
@@ -113,6 +116,9 @@ func (p *parser) Merge(parent interface{}, child interface{}) interface{} {
 	if childConfig.servers != nil {
 		newConfig.servers = append(newConfig.servers, childConfig.servers...)
 	}
+	if childConfig.defaultServer != nil {
+		newConfig.defaultServer = childConfig.defaultServer
+	}
 	return &newConfig
 }
 

plugins/golang-filter/mcp-server/filter.go

@@ -1,10 +1,13 @@
 package main
 
 import (
+	"fmt"
 	"net/http"
 	"net/http/httptest"
 	"net/url"
+	"strings"
 
+	"github.com/alibaba/higress/plugins/golang-filter/mcp-server/internal"
 	"github.com/envoyproxy/envoy/contrib/golang/common/go/api"
 )
 
@@ -18,40 +21,60 @@ type filter struct {
 	config    *config
 
 	req        *http.Request
-	sse        bool
+	serverName string
 	message    bool
 	bodyBuffer []byte
 }
 
+type RequestURL struct {
+	method    string
+	scheme    string
+	host      string
+	path      string
+	baseURL   string
+	parsedURL *url.URL
+}
+
+func NewRequestURL(header api.RequestHeaderMap) *RequestURL {
+	method, _ := header.Get(":method")
+	scheme, _ := header.Get(":scheme")
+	host, _ := header.Get(":authority")
+	path, _ := header.Get(":path")
+	baseURL := fmt.Sprintf("%s://%s", scheme, host)
+	parsedURL, _ := url.Parse(path)
+	api.LogInfof("RequestURL: method=%s, scheme=%s, host=%s, path=%s", method, scheme, host, path)
+	return &RequestURL{method: method, scheme: scheme, host: host, path: path, baseURL: baseURL, parsedURL: parsedURL}
+}
+
 // Callbacks which are called in request path
 // The endStream is true if the request doesn't have body
 func (f *filter) DecodeHeaders(header api.RequestHeaderMap, endStream bool) api.StatusType {
-	fullPath, _ := header.Get(":path")
-	parsedURL, _ := url.Parse(fullPath)
-	f.path = parsedURL.Path
-	method, _ := header.Get(":method")
+	url := NewRequestURL(header)
+	f.path = url.parsedURL.Path
+
 	for _, server := range f.config.servers {
 		if f.path == server.GetSSEEndpoint() {
-			if method != http.MethodGet {
+			if url.method != http.MethodGet {
 				f.callbacks.DecoderFilterCallbacks().SendLocalReply(http.StatusMethodNotAllowed, "Method not allowed", nil, 0, "")
 			} else {
-				f.sse = true
+				f.serverName = server.GetServerName()
 				body := "SSE connection create"
 				f.callbacks.DecoderFilterCallbacks().SendLocalReply(http.StatusOK, body, nil, 0, "")
 			}
 			api.LogInfof("%s SSE connection started", server.GetServerName())
+			server.SetBaseURL(url.baseURL)
 			return api.LocalReply
 		} else if f.path == server.GetMessageEndpoint() {
-			if method != http.MethodPost {
+			if url.method != http.MethodPost {
 				f.callbacks.DecoderFilterCallbacks().SendLocalReply(http.StatusMethodNotAllowed, "Method not allowed", nil, 0, "")
 			}
 			// Create a new http.Request object
 			f.req = &http.Request{
-				Method: method,
-				URL:    parsedURL,
+				Method: url.method,
+				URL:    url.parsedURL,
 				Header: make(http.Header),
 			}
-			api.LogDebugf("Message request: %v", parsedURL)
+			api.LogDebugf("Message request: %v", url.parsedURL)
 			// Copy headers from api.RequestHeaderMap to http.Header
 			header.Range(func(key, value string) bool {
 				f.req.Header.Add(key, value)
@@ -65,11 +88,27 @@ func (f *filter) DecodeHeaders(header api.RequestHeaderMap, endStream bool) api.
 			}
 		}
 	}
-	if endStream {
-		return api.Continue
-	} else {
-		return api.StopAndBuffer
+	if !strings.HasSuffix(url.parsedURL.Path, f.config.ssePathSuffix) {
+		if endStream {
+			return api.Continue
+		} else {
+			return api.StopAndBuffer
+		}
 	}
+
+	if url.method != http.MethodGet {
+		f.callbacks.DecoderFilterCallbacks().SendLocalReply(http.StatusMethodNotAllowed, "Method not allowed", nil, 0, "")
+	} else {
+		f.config.defaultServer = internal.NewSSEServer(internal.NewMCPServer(DefaultServerName, Version),
+			internal.WithSSEEndpoint(f.config.ssePathSuffix),
+			internal.WithMessageEndpoint(strings.TrimSuffix(url.parsedURL.Path, f.config.ssePathSuffix)),
+			internal.WithRedisClient(f.config.redisClient))
+		f.serverName = f.config.defaultServer.GetServerName()
+		body := "SSE connection create"
+		f.callbacks.DecoderFilterCallbacks().SendLocalReply(http.StatusOK, body, nil, 0, "")
+		f.config.defaultServer.SetBaseURL(url.baseURL)
+	}
+	return api.LocalReply
 }
 
 // DecodeData might be called multiple times during handling the request body.
@@ -101,7 +140,7 @@ func (f *filter) DecodeData(buffer api.BufferInstance, endStream bool) api.Statu
 // Callbacks which are called in response path
 // The endStream is true if the response doesn't have body
 func (f *filter) EncodeHeaders(header api.ResponseHeaderMap, endStream bool) api.StatusType {
-	if f.sse {
+	if f.serverName != "" {
 		header.Set("Content-Type", "text/event-stream")
 		header.Set("Cache-Control", "no-cache")
 		header.Set("Connection", "keep-alive")
@@ -115,21 +154,35 @@ func (f *filter) EncodeHeaders(header api.ResponseHeaderMap, endStream bool) api
 // EncodeData might be called multiple times during handling the response body.
 // The endStream is true when handling the last piece of the body.
 func (f *filter) EncodeData(buffer api.BufferInstance, endStream bool) api.StatusType {
-	for _, server := range f.config.servers {
-		if f.sse {
+	if f.serverName != "" {
+		// handle specific server
+		for _, server := range f.config.servers {
+			if f.serverName == server.GetServerName() {
+				buffer.Reset()
+				server.HandleSSE(f.callbacks)
+				return api.Running
+			}
+		}
+		// handle default server
+		if f.serverName == f.config.defaultServer.GetServerName() {
 			buffer.Reset()
-			server.HandleSSE(f.callbacks)
-			f.sse = false
+			f.config.defaultServer.HandleSSE(f.callbacks)
 			return api.Running
 		}
+		return api.Continue
 	}
 	return api.Continue
 }
 
-// OnDestroy 或 OnStreamComplete 中停止 goroutine
+// OnDestroy stops the goroutine
 func (f *filter) OnDestroy(reason api.DestroyReason) {
-	if f.sse && f.config.stopChan != nil {
-		api.LogInfo("Stopping SSE connection")
-		close(f.config.stopChan)
+	if f.serverName != "" && f.config.stopChan != nil {
+		select {
+		case <-f.config.stopChan:
+			return
+		default:
+			api.LogInfo("Stopping SSE connection")
+			close(f.config.stopChan)
+		}
 	}
 }

plugins/golang-filter/mcp-server/internal/server.go

@@ -1,4 +1,3 @@
-// Package server provides MCP (Model Control Protocol) server implementations.
 package internal
 
 import (

plugins/golang-filter/mcp-server/internal/sse.go

@@ -4,6 +4,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"net/http"
+	"sync"
 
 	"github.com/envoyproxy/envoy/contrib/golang/common/go/api"
 	"github.com/google/uuid"
@@ -17,10 +18,14 @@ type SSEServer struct {
 	baseURL         string
 	messageEndpoint string
 	sseEndpoint     string
-	sessions        map[string]bool
+	sessions        sync.Map
 	redisClient     *RedisClient // Redis client for pub/sub
 }
 
+func (s *SSEServer) SetBaseURL(baseURL string) {
+	s.baseURL = baseURL
+}
+
 func (s *SSEServer) GetMessageEndpoint() string {
 	return s.messageEndpoint
 }
@@ -69,7 +74,6 @@ func NewSSEServer(server *MCPServer, opts ...Option) *SSEServer {
 		server:          server,
 		sseEndpoint:     "/sse",
 		messageEndpoint: "/message",
-		sessions:        make(map[string]bool),
 	}
 
 	// Apply all options
@@ -84,12 +88,8 @@ func NewSSEServer(server *MCPServer, opts ...Option) *SSEServer {
 func (s *SSEServer) HandleSSE(cb api.FilterCallbackHandler) {
 	sessionID := uuid.New().String()
 
-	s.sessions[sessionID] = true
-
-	// sessionStore, _ := json.Marshal(s.sessions)
-	// TODO: sse:sessions?
-	// s.redisClient.Set("sse:sessions", string(sessionStore), 0)
-	defer delete(s.sessions, sessionID)
+	s.sessions.Store(sessionID, true)
+	defer s.sessions.Delete(sessionID)
 
 	channel := fmt.Sprintf("sse:%s", sessionID)
 

plugins/golang-filter/mcp-server/servers/gorm/server.go

@@ -52,7 +52,7 @@ func (c *DBConfig) NewServer() (*internal.MCPServer, error) {
 
 	// Add query tool
 	mcpServer.AddTool(
-		mcp.NewToolWithRawSchema("query", "Run a read-only SQL query in clickhouse database with repository git data", GetQueryToolSchema()),
+		mcp.NewToolWithRawSchema("query", "Run a read-only SQL query in database", GetQueryToolSchema()),
 		HandleQueryTool(dbClient),
 	)