Merge pull request #1462 from VirajSalaka/sync-fix-mcp-2

Make MCP Proxies sync based on Eventing Mechanism

Author: Krishanx92

common/eventhub/types.go

@@ -36,6 +36,8 @@ const (
 	EventTypeLLMProxy EventType = "LLM_PROXY"
 	// EventTypeLLMTemplate represents an LLM template change event
 	EventTypeLLMTemplate EventType = "LLM_TEMPLATE"
+	// EventTypeMCPProxy represents an MCP proxy change event
+	EventTypeMCPProxy EventType = "MCP_PROXY"
 
 	// EmptyEventData is the canonical JSON payload for events that do not
 	// require additional data beyond the top-level event fields.

gateway/gateway-controller/cmd/controller/main.go

@@ -189,6 +189,18 @@ func main() {
 	}
 	log.Info("Loaded API keys", slog.Int("count", apiKeyXDSManager.GetAPIKeyCount()))
 
+	// MCP proxies are stored in their source form and need to be rehydrated into
+	// the derived RestAPI representation before startup snapshot and policy work.
+	for _, storedCfg := range configStore.GetAllByKind(string(api.Mcp)) {
+		if err := utils.HydrateStoredMCPConfig(storedCfg); err != nil {
+			log.Error("Failed to hydrate stored MCP proxy configuration",
+				slog.String("id", storedCfg.UUID),
+				slog.String("handle", storedCfg.Handle),
+				slog.Any("error", err))
+			os.Exit(1)
+		}
+	}
+
 	// Initialize xDS snapshot manager with router config
 	snapshotManager := xds.NewSnapshotManager(configStore, log, &cfg.Router, db, cfg)
 
@@ -292,8 +304,9 @@ func main() {
 	loadedAPIs := configStore.GetAll()
 	derivedCount := 0
 	for _, apiConfig := range loadedAPIs {
-		// Derive policy configuration from API
-		if apiConfig.Kind == "RestApi" || apiConfig.Kind == "WebSubApi" {
+		// MCP proxies hydrate into RestAPI form, so they can participate in the
+		// same policy bootstrapping path as the other routed artifacts.
+		if apiConfig.Kind == "RestApi" || apiConfig.Kind == "WebSubApi" || apiConfig.Kind == "Mcp" {
 			storedPolicy := policybuilder.DerivePolicyFromAPIConfig(apiConfig, &cfg.Router, cfg, policyDefinitions)
 			if storedPolicy != nil {
 				if err := policyStore.Set(storedPolicy); err != nil {

gateway/gateway-controller/pkg/api/handlers/handlers.go

@@ -113,6 +113,7 @@ func NewAPIServer(
 	parser := config.NewParser()
 	httpClient := &http.Client{Timeout: 10 * time.Second}
 	routerConfig := &systemConfig.Router
+	mcpDeploymentService := utils.NewMCPDeploymentService(store, db, snapshotManager, policyManager, policyValidator)
 
 	server := &APIServer{
 		store:                store,
@@ -124,7 +125,7 @@ func NewAPIServer(
 		validator:            validator,
 		logger:               logger,
 		deploymentService:    deploymentService,
-		mcpDeploymentService: utils.NewMCPDeploymentService(store, db, snapshotManager, policyManager, policyValidator),
+		mcpDeploymentService: mcpDeploymentService,
 		llmDeploymentService: utils.NewLLMDeploymentService(store, db, snapshotManager, lazyResourceManager, templateDefinitions,
 			deploymentService, routerConfig, policyVersionResolver, policyValidator),
 		apiKeyService:      apiKeyService,
@@ -138,6 +139,7 @@ func NewAPIServer(
 	}
 	if eventHub != nil {
 		server.llmDeploymentService.SetEventHub(eventHub, systemConfig.Controller.Server.GatewayID)
+		server.mcpDeploymentService.SetEventHub(eventHub, systemConfig.Controller.Server.GatewayID)
 	}
 
 	// Create RestAPI service and handler
@@ -238,6 +240,9 @@ func (s *APIServer) SearchDeployments(c *gin.Context, kind string) {
 	}
 
 	configs := s.store.GetAllByKind(kind)
+	if kind == string(api.Mcp) && s.mcpDeploymentService != nil {
+		configs = s.mcpDeploymentService.ListMCPProxies()
+	}
 
 	// Filter based on kind to return appropriate response format
 	if kind == string(api.Mcp) {
@@ -1294,7 +1299,7 @@ func (s *APIServer) CreateMCPProxy(c *gin.Context) {
 	}
 
 	// Build and add policy config derived from API configuration if policies are present
-	if s.policyManager != nil {
+	if s.policyManager != nil && s.eventHub == nil {
 		storedPolicy := s.buildStoredPolicyFromAPI(cfg)
 		if storedPolicy != nil {
 			if err := s.policyManager.AddPolicy(storedPolicy); err != nil {
@@ -1315,7 +1320,7 @@ func (s *APIServer) ListMCPProxies(c *gin.Context, params api.ListMCPProxiesPara
 		s.SearchDeployments(c, string(api.Mcp))
 		return
 	}
-	configs := s.store.GetAllByKind(string(api.Mcp))
+	configs := s.mcpDeploymentService.ListMCPProxies()
 
 	items := make([]api.MCPProxyListItem, len(configs))
 	for i, cfg := range configs {
@@ -1472,7 +1477,7 @@ func (s *APIServer) UpdateMCPProxy(c *gin.Context, id string) {
 		slog.String("handle", handle))
 
 	// Rebuild and update derived policy configuration
-	if s.policyManager != nil {
+	if s.policyManager != nil && s.eventHub == nil {
 		storedPolicy := s.buildStoredPolicyFromAPI(updated)
 		if storedPolicy != nil {
 			if err := s.policyManager.AddPolicy(storedPolicy); err != nil {

gateway/gateway-controller/pkg/api/handlers/handlers_test.go

@@ -756,6 +756,7 @@ func createTestAPIServerWithDB(db storage.Storage) *APIServer {
 
 	deploymentService := utils.NewAPIDeploymentService(store, db, nil, validator, routerCfg)
 	server.deploymentService = deploymentService
+	server.mcpDeploymentService = utils.NewMCPDeploymentService(store, db, nil, nil, nil)
 	server.llmDeploymentService = utils.NewLLMDeploymentService(
 		store,
 		db,
@@ -905,6 +906,66 @@ func createTestRestAPIRequestBody(t *testing.T, handle, displayName, version, co
 	return body
 }
 
+func createTestMCPRequestBody(t *testing.T, handle, displayName, version, contextPath string) []byte {
+	t.Helper()
+	upstreamURL := "http://backend.example.com"
+
+	mcpConfig := api.MCPProxyConfiguration{
+		ApiVersion: api.MCPProxyConfigurationApiVersionGatewayApiPlatformWso2Comv1alpha1,
+		Kind:       api.Mcp,
+		Metadata: api.Metadata{
+			Name: handle,
+		},
+		Spec: api.MCPProxyConfigData{
+			DisplayName: displayName,
+			Version:     version,
+			Context:     stringPtr(contextPath),
+			Upstream: api.MCPProxyConfigData_Upstream{
+				Url: &upstreamURL,
+			},
+		},
+	}
+
+	body, err := json.Marshal(mcpConfig)
+	require.NoError(t, err)
+	return body
+}
+
+func createTestMCPStoredConfig(t *testing.T, id, handle, displayName, version, contextPath string, desiredState models.DesiredState) *models.StoredConfig {
+	t.Helper()
+	upstreamURL := "http://backend.example.com"
+
+	cfg := &models.StoredConfig{
+		UUID:        id,
+		Kind:        string(api.Mcp),
+		Handle:      handle,
+		DisplayName: displayName,
+		Version:     version,
+		SourceConfiguration: api.MCPProxyConfiguration{
+			ApiVersion: api.MCPProxyConfigurationApiVersionGatewayApiPlatformWso2Comv1alpha1,
+			Kind:       api.Mcp,
+			Metadata: api.Metadata{
+				Name: handle,
+			},
+			Spec: api.MCPProxyConfigData{
+				DisplayName: displayName,
+				Version:     version,
+				Context:     stringPtr(contextPath),
+				Upstream: api.MCPProxyConfigData_Upstream{
+					Url: &upstreamURL,
+				},
+			},
+		},
+		DesiredState: desiredState,
+		Origin:       models.OriginGatewayAPI,
+		CreatedAt:    time.Now(),
+		UpdatedAt:    time.Now(),
+	}
+
+	require.NoError(t, utils.HydrateStoredMCPConfig(cfg))
+	return cfg
+}
+
 func attachTestEventHub(server *APIServer, hub eventhub.EventHub, gatewayID string) {
 	server.eventHub = hub
 	server.gatewayID = gatewayID
@@ -930,6 +991,9 @@ func attachTestEventHub(server *APIServer, hub eventhub.EventHub, gatewayID stri
 	if server.llmDeploymentService != nil {
 		server.llmDeploymentService.SetEventHub(hub, gatewayID)
 	}
+	if server.mcpDeploymentService != nil {
+		server.mcpDeploymentService.SetEventHub(hub, gatewayID)
+	}
 }
 
 func seedAPIForAPIKeyHandlerTests(t *testing.T, server *APIServer, handle string) *models.StoredConfig {
@@ -1634,10 +1698,22 @@ func TestDeleteLLMProxyNotFound(t *testing.T) {
 	t.Skip("Skipping test that requires full deployment service setup")
 }
 
-// TestListMCPProxies tests listing MCP proxies
-// Note: This test requires full deployment service setup
 func TestListMCPProxies(t *testing.T) {
-	t.Skip("Skipping test that requires full deployment service setup")
+	server := createTestAPIServer()
+	mockDB := server.db.(*MockStorage)
+
+	cfg := createTestMCPStoredConfig(t, "0000-mcp-id-0000-000000000000", "test-mcp", "Test MCP", "v1.0.0", "/mcp", models.StateDeployed)
+	require.NoError(t, mockDB.SaveConfig(cfg))
+
+	c, w := createTestContext("GET", "/mcp-proxies", nil)
+	server.ListMCPProxies(c, api.ListMCPProxiesParams{})
+
+	assert.Equal(t, http.StatusOK, w.Code)
+
+	var response map[string]any
+	require.NoError(t, json.Unmarshal(w.Body.Bytes(), &response))
+	assert.Equal(t, "success", response["status"])
+	assert.Equal(t, float64(1), response["count"])
 }
 
 // TestListMCPProxiesWithFilters tests listing MCP proxies with filters
@@ -1670,6 +1746,67 @@ func TestDeleteMCPProxyNotFound(t *testing.T) {
 	t.Skip("Skipping test that requires full deployment service setup")
 }
 
+func TestCreateMCPProxyWithDBAndEventHub(t *testing.T) {
+	server := createTestAPIServer()
+	mockDB := server.db.(*MockStorage)
+	mockHub := &mockEventHub{}
+	attachTestEventHub(server, mockHub, "test-gateway")
+
+	body := createTestMCPRequestBody(t, "test-mcp", "Test MCP", "v1.0.0", "/mcp")
+	c, w := createTestContextWithHeader("POST", "/mcp-proxies", body, map[string]string{
+		"Content-Type": "application/json",
+	})
+	c.Set(middleware.CorrelationIDKey, "corr-id-create-mcp")
+
+	server.CreateMCPProxy(c)
+
+	assert.Equal(t, http.StatusCreated, w.Code)
+
+	cfg, err := mockDB.GetConfigByKindAndHandle(string(api.Mcp), "test-mcp")
+	require.NoError(t, err)
+	require.Len(t, mockHub.publishedEvents, 1)
+	assert.Equal(t, "test-gateway", mockHub.publishedEvents[0].gatewayID)
+	assert.Equal(t, eventhub.EventTypeMCPProxy, mockHub.publishedEvents[0].event.EventType)
+	assert.Equal(t, "CREATE", mockHub.publishedEvents[0].event.Action)
+	assert.Equal(t, cfg.UUID, mockHub.publishedEvents[0].event.EntityID)
+	assert.Equal(t, "corr-id-create-mcp", mockHub.publishedEvents[0].event.EventID)
+	assert.Equal(t, eventhub.EmptyEventData, mockHub.publishedEvents[0].event.EventData)
+
+	_, err = server.store.Get(cfg.UUID)
+	require.Error(t, err)
+}
+
+func TestDeleteMCPProxyWithDBAndEventHub(t *testing.T) {
+	server := createTestAPIServer()
+	mockDB := server.db.(*MockStorage)
+	mockHub := &mockEventHub{}
+	attachTestEventHub(server, mockHub, "test-gateway")
+
+	cfg := createTestMCPStoredConfig(t, "0000-mcp-delete-id-0000-000000000000", "test-mcp", "Test MCP", "v1.0.0", "/mcp", models.StateDeployed)
+	require.NoError(t, mockDB.SaveConfig(cfg))
+	require.NoError(t, server.store.Add(cfg))
+
+	c, w := createTestContext("DELETE", "/mcp-proxies/test-mcp", nil)
+	c.Set(middleware.CorrelationIDKey, "corr-id-delete-mcp")
+
+	server.DeleteMCPProxy(c, "test-mcp")
+
+	assert.Equal(t, http.StatusOK, w.Code)
+	require.Len(t, mockHub.publishedEvents, 1)
+	assert.Equal(t, "test-gateway", mockHub.publishedEvents[0].gatewayID)
+	assert.Equal(t, eventhub.EventTypeMCPProxy, mockHub.publishedEvents[0].event.EventType)
+	assert.Equal(t, "DELETE", mockHub.publishedEvents[0].event.Action)
+	assert.Equal(t, cfg.UUID, mockHub.publishedEvents[0].event.EntityID)
+	assert.Equal(t, "corr-id-delete-mcp", mockHub.publishedEvents[0].event.EventID)
+	assert.Equal(t, eventhub.EmptyEventData, mockHub.publishedEvents[0].event.EventData)
+
+	_, err := mockDB.GetConfig(cfg.UUID)
+	require.Error(t, err)
+
+	_, err = server.store.Get(cfg.UUID)
+	require.NoError(t, err)
+}
+
 // TestGenerateAPIKeyNoAuth tests CreateAPIKey without authentication
 func TestGenerateAPIKeyNoAuth(t *testing.T) {
 	server := createTestAPIServer()
@@ -3027,6 +3164,7 @@ func TestBuildStoredPolicyFromAPIWebSubApiWithPolicies(t *testing.T) {
 // Test ListMCPProxies with stored configs that have unmarshal issues
 func TestListMCPProxiesUnmarshalError(t *testing.T) {
 	server := createTestAPIServer()
+	mockDB := server.db.(*MockStorage)
 
 	// Add MCP config, then replace SourceConfiguration with something that can't be marshaled to JSON
 	cfg := &models.StoredConfig{
@@ -3048,8 +3186,8 @@ func TestListMCPProxiesUnmarshalError(t *testing.T) {
 		CreatedAt: time.Now(),
 		UpdatedAt: time.Now(),
 	}
-	_ = server.store.Add(cfg)
-	// Mutate SourceConfiguration to something that can't be JSON marshaled
+	require.NoError(t, mockDB.SaveConfig(cfg))
+	// Mutate the DB-backed object to something that can't be JSON marshaled.
 	cfg.SourceConfiguration = make(chan int)
 
 	c, w := createTestContext("GET", "/mcp-proxies", nil)