fix(mcp-server): support forwarding mcp-server request (#25190)

pkg/apigateway/handler/backendproxy.go

@@ -65,6 +65,10 @@ func (h *SBackendServiceProxyHandler) requestManipulator(ctx context.Context, r
 	if slashPos <= 0 {
 		return r, httperrors.NewBadRequestError("invalid request URL %s", r.URL.Path)
 	}
+	serviceName := path[:slashPos]
+	// Tell upstream (e.g. mcp-server SSE) the external path prefix so endpoint
+	// events point clients back through the gateway: /api/s/<service>/message
+	r.Header.Set("X-Forwarded-Prefix", "/api/s/"+serviceName)
 	path = path[slashPos:]
 	if strings.HasPrefix(path, "/r/") {
 		path = path[len("/r/"):]

pkg/appsrv/response.go

@@ -18,8 +18,10 @@ import (
 	"bufio"
 	"context"
 	"fmt"
+	"mime"
 	"net"
 	"net/http"
+	"strings"
 
 	"yunion.io/x/onecloud/pkg/httperrors"
 )
@@ -36,6 +38,8 @@ type responseWriterChannel struct {
 	bodyResp   chan responseWriterResponse
 	statusChan chan int
 	statusResp chan bool
+	flushChan  chan struct{}
+	flushResp  chan struct{}
 
 	isClosed bool
 }
@@ -47,6 +51,8 @@ func newResponseWriterChannel(backend http.ResponseWriter) responseWriterChannel
 		bodyResp:   make(chan responseWriterResponse),
 		statusChan: make(chan int),
 		statusResp: make(chan bool),
+		flushChan:  make(chan struct{}),
+		flushResp:  make(chan struct{}),
 		isClosed:   false,
 	}
 }
@@ -76,16 +82,34 @@ func (w *responseWriterChannel) WriteHeader(status int) {
 	<-w.statusResp
 }
 
-// implent http.Flusher
+// Flush implements http.Flusher. Flush must run on the same goroutine as Write
+// (wait loop); otherwise SSE via ReverseProxy never reaches the client.
 func (w *responseWriterChannel) Flush() {
 	if w.isClosed {
 		return
 	}
+	w.flushChan <- struct{}{}
+	<-w.flushResp
+}
+
+func (w *responseWriterChannel) flushBackend() {
 	if f, ok := w.backend.(http.Flusher); ok {
 		f.Flush()
 	}
 }
 
+func (w *responseWriterChannel) isEventStream() bool {
+	ct := w.backend.Header().Get("Content-Type")
+	if ct == "" {
+		return false
+	}
+	baseCT, _, err := mime.ParseMediaType(ct)
+	if err != nil {
+		return strings.HasPrefix(ct, "text/event-stream")
+	}
+	return baseCT == "text/event-stream"
+}
+
 // Hijack implements the Hijacker.Hijack method. Our response is both a ResponseWriter
 // and a Hijacker.
 func (w *responseWriterChannel) Hijack() (rwc net.Conn, buf *bufio.ReadWriter, err error) {
@@ -121,6 +145,11 @@ func (w *responseWriterChannel) wait(ctx context.Context, workerChan chan *SWork
 			// log.Infof("Recive body: %s, more: %v", len(bytes), more)
 			if more {
 				c, e := w.backend.Write(bytes)
+				// SSE / streaming: flush immediately after each chunk so clients
+				// see events without waiting for handler completion.
+				if e == nil && w.isEventStream() {
+					w.flushBackend()
+				}
 				w.bodyResp <- responseWriterResponse{count: c, err: e}
 			} else {
 				stop = true
@@ -133,6 +162,13 @@ func (w *responseWriterChannel) wait(ctx context.Context, workerChan chan *SWork
 			} else {
 				stop = true
 			}
+		case _, more := <-w.flushChan:
+			if more {
+				w.flushBackend()
+				w.flushResp <- struct{}{}
+			} else {
+				stop = true
+			}
 		}
 	}
 	return err
@@ -148,4 +184,6 @@ func (w *responseWriterChannel) closeChannels() {
 	close(w.bodyResp)
 	close(w.statusChan)
 	close(w.statusResp)
+	close(w.flushChan)
+	close(w.flushResp)
 }

pkg/mcp-server/server/access_log.go

@@ -0,0 +1,109 @@
+// Copyright 2019 Yunion
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package server
+
+import (
+	"bufio"
+	"crypto/sha256"
+	"encoding/base64"
+	"fmt"
+	"net"
+	"net/http"
+	"os"
+	"time"
+
+	"yunion.io/x/log"
+	"yunion.io/x/pkg/utils"
+)
+
+var accessLogHostId = genAccessLogHostId()
+
+func genAccessLogHostId() string {
+	hostname, _ := os.Hostname()
+	h := sha256.New()
+	fmt.Fprintf(h, "mcp-server:%s", hostname)
+	return base64.URLEncoding.EncodeToString(h.Sum(nil))
+}
+
+type accessLogResponseWriter struct {
+	http.ResponseWriter
+	status int
+}
+
+func (w *accessLogResponseWriter) WriteHeader(code int) {
+	if w.status == 0 {
+		w.status = code
+	}
+	w.ResponseWriter.WriteHeader(code)
+}
+
+func (w *accessLogResponseWriter) Write(b []byte) (int, error) {
+	if w.status == 0 {
+		w.status = http.StatusOK
+	}
+	return w.ResponseWriter.Write(b)
+}
+
+func (w *accessLogResponseWriter) Flush() {
+	if f, ok := w.ResponseWriter.(http.Flusher); ok {
+		f.Flush()
+	}
+}
+
+func (w *accessLogResponseWriter) Unwrap() http.ResponseWriter {
+	return w.ResponseWriter
+}
+
+func (w *accessLogResponseWriter) Hijack() (net.Conn, *bufio.ReadWriter, error) {
+	h, ok := w.ResponseWriter.(http.Hijacker)
+	if !ok {
+		return nil, nil, fmt.Errorf("not a hijacker")
+	}
+	return h.Hijack()
+}
+
+func genAccessRequestId(w http.ResponseWriter, r *http.Request) string {
+	rid := r.Header.Get("X-Request-Id")
+	if len(rid) == 0 {
+		rid = utils.GenRequestId(3)
+	} else {
+		rid = fmt.Sprintf("%s-%s", rid, utils.GenRequestId(3))
+	}
+	w.Header().Set("X-Request-Id", rid)
+	w.Header().Set("X-Request-Host-Id", accessLogHostId)
+	return rid
+}
+
+// withAccessLog wraps handler with appsrv-style access logs:
+//
+//	hostId status requestId METHOD /path (remote) durationMs
+func withAccessLog(next http.Handler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		lrw := &accessLogResponseWriter{ResponseWriter: w, status: 0}
+		rid := genAccessRequestId(lrw, r)
+		start := time.Now()
+		next.ServeHTTP(lrw, r)
+		if lrw.status == 0 {
+			lrw.status = http.StatusOK
+		}
+		durationMs := float64(time.Since(start).Nanoseconds()) / 1e6
+		remote := r.RemoteAddr
+		if peer := r.Header.Get("X-Yunion-Peer-Service-Name"); peer != "" {
+			remote = fmt.Sprintf("%s:%s", r.RemoteAddr, peer)
+		}
+		log.Infof("%s %d %s %s %s (%s) %.2fms",
+			accessLogHostId, lrw.status, rid, r.Method, r.URL, remote, durationMs)
+	})
+}

pkg/mcp-server/server/server.go

@@ -193,7 +193,13 @@ func (s *CloudpodsMCPServer) Start() error {
 	sseServer := server.NewSSEServer(
 		s.mcpServer,
 		server.WithSSEContextFunc(contextFunc),
-		server.WithHTTPServer(&http.Server{Handler: mux}),
+		// Gateway proxies /api/s/mcp-server/sse → /sse and sets X-Forwarded-Prefix.
+		// Clients must POST messages to /api/s/mcp-server/message, not /message.
+		server.WithDynamicBasePath(func(r *http.Request, _ string) string {
+			return strings.TrimSuffix(r.Header.Get("X-Forwarded-Prefix"), "/")
+		}),
+		server.WithUseFullURLForMessageEndpoint(false),
+		server.WithHTTPServer(&http.Server{Handler: withAccessLog(mux)}),
 	)
 	mux.HandleFunc("/version", func(w http.ResponseWriter, r *http.Request) {
 		appsrv.VersionHandler(context.Background(), w, r)

pkg/proxy/http.go

@@ -20,6 +20,7 @@ import (
 	"net/http"
 	"net/http/httputil"
 	"net/url"
+	"time"
 
 	"yunion.io/x/log"
 	"yunion.io/x/pkg/appctx"
@@ -68,15 +69,24 @@ func (p *SReverseProxy) ServeHTTP(ctx context.Context, w http.ResponseWriter, r
 	}
 	log.Debugf("Forwarding to servie: %q, url: %q", p.serviceName, remoteUrl.String())
 	proxy := httputil.NewSingleHostReverseProxy(remoteUrl)
+	// SSE / long-polling: flush immediately; never ask upstream for gzip (buffers whole stream).
+	proxy.FlushInterval = -1
 	proxy.Transport = &http.Transport{
 		Proxy:              http.ProxyFromEnvironment,
 		TLSClientConfig:    &tls.Config{InsecureSkipVerify: true},
 		DisableKeepAlives:  true,
+		DisableCompression: true,
+		IdleConnTimeout:    90 * time.Second,
+	}
+	proxy.ModifyResponse = func(resp *http.Response) error {
+		resp.Header.Set("X-Accel-Buffering", "no")
+		return nil
 	}
 	r, err = p.manipulator(ctx, r)
 	if err != nil {
 		httperrors.InternalServerError(ctx, w, "%v", err)
 		return
 	}
+	r.Header.Del("Accept-Encoding")
 	proxy.ServeHTTP(w, r)
 }