Merge remote-tracking branch 'origin/main' into merge-main-into-release-0.7.0

Author: Patryk-Stefanski

Dockerfile

@@ -11,7 +11,7 @@ COPY internal/ internal/
 COPY api/ api/
 
 ARG LDFLAGS=""
-RUN CGO_ENABLED=0 GOOS=linux go build -ldflags "${LDFLAGS}" -o mcp_gateway cmd/mcp-broker-router/main.go
+RUN CGO_ENABLED=0 GOOS=linux go build -ldflags "${LDFLAGS}" -o mcp_gateway ./cmd/mcp-broker-router/
 
 FROM alpine:3.22.1
 

cmd/mcp-broker-router/broker.go

@@ -0,0 +1,110 @@
+package main
+
+import (
+	"fmt"
+	"net/http"
+	"os"
+	"time"
+
+	mcpv1alpha1 "github.com/Kuadrant/mcp-gateway/api/v1alpha1"
+	"github.com/Kuadrant/mcp-gateway/internal/broker"
+	"github.com/mark3labs/mcp-go/server"
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/propagation"
+)
+
+func (a *app) createBroker() {
+	invalidToolPolicy := mcpv1alpha1.InvalidToolPolicy(a.brokerCfg.invalidToolPolicy)
+	if invalidToolPolicy != mcpv1alpha1.InvalidToolPolicyFilterOut && invalidToolPolicy != mcpv1alpha1.InvalidToolPolicyRejectServer {
+		panic("--invalid-tool-policy must be FilterOut or RejectServer")
+	}
+
+	managerTickerInterval := time.Duration(a.brokerCfg.managerTickerIntervalSecs) * time.Second
+	if managerTickerInterval <= 0 {
+		panic("flag mcp-check-interval cannot be 0 or less seconds")
+	}
+
+	a.mcpBroker = broker.NewBroker(a.logger.With("component", "broker"),
+		broker.WithEnforceCapabilityFilter(a.brokerCfg.enforceCapabilityFiltering),
+		broker.WithTrustedHeadersPublicKey(os.Getenv("TRUSTED_HEADER_PUBLIC_KEY")),
+		broker.WithManagerTickerInterval(managerTickerInterval),
+		broker.WithInvalidToolPolicy(invalidToolPolicy),
+		broker.WithElicitationEnabled(a.brokerCfg.enableURLElicitation),
+		broker.WithDiscoveryToolsEnabled(a.brokerCfg.discoveryToolsEnabled),
+		broker.WithDiscoveryToolThreshold(a.brokerCfg.discoveryToolThreshold),
+		broker.WithSessionCache(a.sessionCache),
+	)
+	a.tokenHandler = broker.NewTokenHandler(a.sessionCache, a.tokenElicitMap, *a.logger)
+	a.elicitHandler = &broker.ElicitationHandler{
+		ElicitationMap: a.tokenElicitMap,
+		Config:         a.mcpConfig,
+	}
+	a.setUpHTTPServer()
+}
+
+func (a *app) setUpHTTPServer() {
+	cfg := &a.brokerCfg
+	mux := http.NewServeMux()
+
+	mux.HandleFunc("/", func(w http.ResponseWriter, _ *http.Request) {
+		_, _ = fmt.Fprint(w, "Hello, World!  BTW, the MCP server is on /mcp")
+	})
+
+	mux.HandleFunc("/healthz", func(w http.ResponseWriter, _ *http.Request) {
+		w.WriteHeader(http.StatusOK)
+	})
+
+	mux.HandleFunc("/readyz", func(w http.ResponseWriter, _ *http.Request) {
+		if !a.mcpBroker.IsReady() {
+			w.WriteHeader(http.StatusServiceUnavailable)
+			return
+		}
+		w.WriteHeader(http.StatusOK)
+	})
+
+	oauthHandler := broker.ProtectedResourceHandler{Logger: a.logger}
+	mux.HandleFunc("/.well-known/oauth-protected-resource", oauthHandler.Handle)
+	mux.HandleFunc("/.well-known/oauth-protected-resource/", oauthHandler.Handle)
+
+	// WriteTimeout of 0 (disabled) is important for SSE connections (GET /mcp).
+	// SSE streams notifications indefinitely - any write timeout would kill the connection.
+	writeTimeout := time.Duration(cfg.writeTimeoutSecs) * time.Second
+
+	httpSrv := &http.Server{
+		Addr:         cfg.addr,
+		Handler:      mux,
+		ReadTimeout:  5 * time.Second,
+		WriteTimeout: writeTimeout,
+	}
+
+	streamableHTTPOpts := []server.StreamableHTTPOption{
+		server.WithStreamableHTTPServer(httpSrv),
+	}
+	if a.jwtMgr != nil {
+		streamableHTTPOpts = append(streamableHTTPOpts, server.WithSessionIdManager(a.jwtMgr))
+	}
+	streamableHTTPServer := server.NewStreamableHTTPServer(a.mcpBroker.MCPServer(), streamableHTTPOpts...)
+
+	mux.HandleFunc("OPTIONS /mcp", func(w http.ResponseWriter, r *http.Request) {
+		a.logger.Debug("Handling OPTIONS", "Mcp-Session-Id", r.Header.Get("Mcp-Session-Id"))
+		w.WriteHeader(http.StatusOK)
+	})
+
+	mux.HandleFunc("/status", a.mcpBroker.HandleStatusRequest)
+	mux.HandleFunc("/status/", a.mcpBroker.HandleStatusRequest)
+	if cfg.enableURLElicitation {
+		mux.Handle("/tokens", a.tokenHandler)
+		mux.Handle("/mcp/elicitation", a.elicitHandler)
+	}
+	mux.Handle("/mcp", traceContextMiddleware(streamableHTTPServer))
+
+	a.brokerServer = httpSrv
+	a.mcpServer = streamableHTTPServer
+}
+
+func traceContextMiddleware(next http.Handler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		ctx := otel.GetTextMapPropagator().Extract(r.Context(), propagation.HeaderCarrier(r.Header))
+		next.ServeHTTP(w, r.WithContext(ctx))
+	})
+}