SANDBOX-1567: Add stateless mode configuration (#39)

* Add stateless mode configuration and e2e

Signed-off-by: Feny Mehta <fbm3307@gmail.com>

Author: fbm3307

Containerfile

@@ -32,6 +32,7 @@ ENV ARGOCD_URL=https://argocd-server
 ENV ARGOCD_TOKEN=secure-token
 ENV ARGOCD_MCP_SERVER_INSECURE=false
 ENV ARGOCD_MCP_SERVER_DEBUG=false
+ENV ARGOCD_MCP_SERVER_STATELESS=false
 ENV ARGOCD_MCP_SERVER_LISTEN_HOST=127.0.0.1
 ENV ARGOCD_MCP_SERVER_LISTEN_PORT=8080
 
@@ -40,4 +41,4 @@ USER 1001
 
 EXPOSE ${ARGOCD_MCP_SERVER_LISTEN_PORT}
 
-CMD /usr/local/bin/argocd-mcp-server --transport http --argocd-url $ARGOCD_URL --argocd-token $ARGOCD_TOKEN --insecure $ARGOCD_MCP_SERVER_INSECURE --debug $ARGOCD_MCP_SERVER_DEBUG --listen $ARGOCD_MCP_SERVER_LISTEN_HOST:$ARGOCD_MCP_SERVER_LISTEN_PORT
+CMD /usr/local/bin/argocd-mcp-server --transport http --argocd-url $ARGOCD_URL --argocd-token $ARGOCD_TOKEN --insecure=$ARGOCD_MCP_SERVER_INSECURE --debug=$ARGOCD_MCP_SERVER_DEBUG --stateless=$ARGOCD_MCP_SERVER_STATELESS --listen $ARGOCD_MCP_SERVER_LISTEN_HOST:$ARGOCD_MCP_SERVER_LISTEN_PORT

cmd/start_server.go

@@ -17,7 +17,7 @@ import (
 )
 
 var transport, listen, argocdURL, argocdToken string
-var argocdInsecure, debug bool
+var argocdInsecure, debug, stateless bool
 
 func init() {
 	startServerCmd.Flags().StringVar(&argocdURL, "argocd-url", "", "Specify the URL of the Argo CD server to query (required)")
@@ -30,6 +30,7 @@ func init() {
 	}
 	startServerCmd.Flags().BoolVar(&argocdInsecure, "insecure", false, "Allow insecure TLS connections to the Argo CD server")
 	startServerCmd.Flags().BoolVar(&debug, "debug", false, "Enable debug mode")
+	startServerCmd.Flags().BoolVar(&stateless, "stateless", false, "Enable stateless mode where the server does not send change notifications (required for multiple replicas)")
 	startServerCmd.Flags().StringVar(&transport, "transport", "http", "Choose between 'stdio' or 'http' transport")
 	startServerCmd.Flags().StringVar(&listen, "listen", "127.0.0.1:8080", "Specify the host and port to listen on when using the 'http' transport")
 }
@@ -59,15 +60,18 @@ var startServerCmd = &cobra.Command{
 		logger := slog.New(slog.NewTextHandler(cmd.ErrOrStderr(), &slog.HandlerOptions{
 			Level: lvl,
 		}))
-		logger.Info("starting the Argo CD MCP server", "transport", transport, "listen", listen, "argocd-url", argocdURL, "insecure", argocdInsecure, "debug", debug)
+		logger.Info("starting the Argo CD MCP server", "transport", transport, "listen", listen, "argocd-url", argocdURL, "insecure", argocdInsecure, "debug", debug, "stateless", stateless)
 		if debug {
 			lvl.Set(slog.LevelDebug)
 			logger.Debug("debug mode enabled")
 		}
 		cl := argocd.NewClient(argocdURL, argocdToken, argocdInsecure)
-		srv := server.New(logger, cl)
+		srv := server.New(logger, cl, stateless)
 		switch transport {
 		case "stdio":
+			if stateless {
+				return fmt.Errorf("stateless mode is not supported for stdio transport")
+			}
 			t := &mcp.LoggingTransport{
 				Transport: &mcp.StdioTransport{},
 				Writer:    cmd.ErrOrStderr(),
@@ -79,9 +83,17 @@ var startServerCmd = &cobra.Command{
 			mux := http.NewServeMux()
 
 			// MCP endpoint
+			// Stateless mode configuration from server settings.
+			// When Stateless is true, the server will not send notifications to clients
+			// (e.g., tools/list_changed, prompts/list_changed). This disables dynamic
+			// tool and prompt updates but is useful for container deployments, load
+			// balancing, and serverless environments where maintaining client state
+			// is not desired or possible.
 			mux.Handle("/mcp", mcp.NewStreamableHTTPHandler(func(*http.Request) *mcp.Server {
 				return srv
-			}, nil))
+			}, &mcp.StreamableHTTPOptions{
+				Stateless: stateless,
+			}))
 
 			// Metrics endpoint
 			mux.Handle("/metrics", promhttp.Handler())

compose.yaml

@@ -40,3 +40,35 @@ services:
       - ARGOCD_MCP_SERVER_DEBUG=true
       - ARGOCD_MCP_SERVER_LISTEN_HOST=0.0.0.0 # here we need to bind to 0.0.0.0 to be able to connect to the server from the host machine
       - ARGOCD_MCP_SERVER_LISTEN_PORT=8080
+  argocd-mcp-server-stateless-replica1:
+    # instance of the Argo CD MCP server with a valid URL for the Argo CD server and stateless mode enabled
+    image: argocd-mcp-server
+    environment:
+      - ARGOCD_URL=http://argocd-server-mock:50084
+      - ARGOCD_TOKEN=secure-token
+      - ARGOCD_MCP_SERVER_INSECURE=true
+      - ARGOCD_MCP_SERVER_DEBUG=true
+      - ARGOCD_MCP_SERVER_STATELESS=true
+      - ARGOCD_MCP_SERVER_LISTEN_HOST=0.0.0.0
+      - ARGOCD_MCP_SERVER_LISTEN_PORT=8080
+  argocd-mcp-server-stateless-replica2:
+    # instance of the Argo CD MCP server with a valid URL for the Argo CD server and stateless mode enabled
+    image: argocd-mcp-server
+    environment:
+      - ARGOCD_URL=http://argocd-server-mock:50084
+      - ARGOCD_TOKEN=secure-token
+      - ARGOCD_MCP_SERVER_INSECURE=true
+      - ARGOCD_MCP_SERVER_DEBUG=true
+      - ARGOCD_MCP_SERVER_STATELESS=true
+      - ARGOCD_MCP_SERVER_LISTEN_HOST=0.0.0.0
+      - ARGOCD_MCP_SERVER_LISTEN_PORT=8080
+  load-balancer:
+    # load balancer to balance the traffic between the Argo CD MCP server instances
+    image: nginx:alpine
+    ports:
+      - "50090:80"
+    volumes:
+      - ./test/e2e/nginx.conf:/etc/nginx/nginx.conf:ro,z
+    depends_on:
+      - argocd-mcp-server-stateless-replica1
+      - argocd-mcp-server-stateless-replica2

internal/server/server.go

@@ -9,13 +9,21 @@ import (
 	"github.com/modelcontextprotocol/go-sdk/mcp"
 )
 
-func New(logger *slog.Logger, cl *argocd.Client) *mcp.Server {
+func New(logger *slog.Logger, cl *argocd.Client, stateless bool) *mcp.Server {
+	// Configure server capabilities based on stateless mode
+	// When stateless is true, disable ListChanged notifications for tools and prompts
+	// This prevents the server from attempting to send notifications that may not
+	// reach the client in multi-replica deployments
 	s := mcp.NewServer(
 		&mcp.Implementation{
 			Name:    "argocd-mcp-server",
 			Version: "0.1",
 		},
 		&mcp.ServerOptions{
+			Capabilities: &mcp.ServerCapabilities{
+				Tools:   &mcp.ToolCapabilities{ListChanged: !stateless},
+				Prompts: &mcp.PromptCapabilities{ListChanged: !stateless},
+			},
 			InitializedHandler: func(_ context.Context, ir *mcp.InitializedRequest) {
 				logger.Debug("initialized", "session_id", ir.Session.ID())
 			},

test/e2e/nginx.conf

@@ -0,0 +1,34 @@
+events {
+    worker_connections 1024;
+}
+
+http {
+    # Define upstream servers for round-robin load balancing
+    upstream mcp_backends {
+        # Round-robin by default
+        server argocd-mcp-server-stateless-replica1:8080;
+        server argocd-mcp-server-stateless-replica2:8080;
+    }
+
+    # Access log for debugging
+    access_log /var/log/nginx/access.log;
+    error_log /var/log/nginx/error.log debug;
+
+    server {
+        listen 80;
+        server_name localhost;
+
+        # Proxy all endpoints to backend servers
+        location / {
+            proxy_pass http://mcp_backends;
+            proxy_http_version 1.1;
+            proxy_set_header Host $host;
+            proxy_set_header Connection '';
+            
+            # SSE support for MCP streaming
+            proxy_buffering off;
+            proxy_cache off;
+            chunked_transfer_encoding on;
+        }
+    }
+}

test/e2e/server_test.go

@@ -23,7 +23,13 @@ import (
 // Note: make sure you ran `task install` before running this test
 // ------------------------------------------------------------------------------------------------
 
-func TestServer(t *testing.T) {
+// TestServer verifies basic MCP functionality with both stdio and http transports.
+// Both transports run in stateful mode (ListChanged enabled) and test:
+// - Tool calls (unhealthyApplications, unhealthyApplicationResources)
+// - Error handling (argocd-error, unreachable scenarios)
+// - Metrics collection (for http transport)
+// - Session reuse across multiple tool calls
+func TestStatefulServer(t *testing.T) {
 
 	testdata := []struct {
 		name string
@@ -35,11 +41,16 @@ func TestServer(t *testing.T) {
 		},
 		{
 			name: "http",
-			init: newHTTPSession("http://localhost:50081/mcp"),
+			init: func(t *testing.T) *mcp.ClientSession {
+				ctx := context.Background()
+				session, err := newHTTPSession(ctx, "http://localhost:50081/mcp", "e2e-test-client")
+				require.NoError(t, err)
+				return session
+			},
 		},
 	}
 
-	// test stdio and http transports with a valid Argo CD client
+	// Test stdio and http transports with a valid Argo CD client (stateful mode)
 	for _, td := range testdata {
 		t.Run(td.name, func(t *testing.T) {
 			// given
@@ -212,6 +223,11 @@ func TestServer(t *testing.T) {
 					assert.Equal(t, mcpCallsDurationSecondsInfBucketBefore+1, mcpCallsDurationSecondsInfBucketAfter)
 				}
 			})
+
+			t.Run("verify/capabilities/listChanged", func(t *testing.T) {
+				// Both stdio and http transports use stateful mode by default
+				assertListChanged(t, session, true)
+			})
 		})
 	}
 
@@ -225,7 +241,12 @@ func TestServer(t *testing.T) {
 		},
 		{
 			name: "http-unreachable",
-			init: newHTTPSession("http://localhost:50082/mcp"), // invalid URL and token for the Argo CD server
+			init: func(t *testing.T) *mcp.ClientSession {
+				ctx := context.Background()
+				session, err := newHTTPSession(ctx, "http://localhost:50082/mcp", "e2e-test-client")
+				require.NoError(t, err)
+				return session
+			}, // invalid URL and token for the Argo CD server
 		},
 	}
 
@@ -250,6 +271,55 @@ func TestServer(t *testing.T) {
 	}
 }
 
+// TestStateless verifies that multiple stateless server instances work correctly
+// with load balancing across replicas. This comprehensive test validates:
+// - Initialize response and capabilities with no ListChanged notifications
+// - Session reuse across multiple requests (list tools and call tools)
+// - Tools functionality with content validation
+func TestStatelessServer(t *testing.T) {
+	ctx := context.Background()
+	serverURL := "http://localhost:50090/mcp"
+
+	// Initialize a single session for the entire test
+	session, err := newHTTPSession(ctx, serverURL, "e2e-test-stateless")
+	require.NoError(t, err)
+	defer session.Close()
+
+	// Step 1: Validate initialize response for stateless mode
+	assertInitializeResponse(t, session, true)
+
+	// Step 2: Verify ListChanged is false (no notifications available in stateless mode)
+	assertListChanged(t, session, false)
+
+	// Step 3: Verify session can be reused by listing tools multiple times
+	for i := 0; i < 5; i++ {
+		tools, listErr := session.ListTools(ctx, &mcp.ListToolsParams{})
+		require.NoError(t, listErr, "should list tools on request %d", i)
+		assert.NotEmpty(t, tools.Tools, "should have tools on request %d", i)
+	}
+
+	// Step 4: Verify tools work correctly with content validation
+	result, callErr := session.CallTool(ctx, &mcp.CallToolParams{
+		Name: "unhealthyApplications",
+	})
+	require.NoError(t, callErr)
+	require.False(t, result.IsError, "tool call should succeed")
+	assert.NotEmpty(t, result.Content, "tool should return content")
+
+	// Verify the content is correct
+	expectedContent := map[string]any{
+		"degraded":    []any{"a-degraded-application", "another-degraded-application"},
+		"progressing": []any{"a-progressing-application", "another-progressing-application"},
+		"outOfSync":   []any{"an-out-of-sync-application", "another-out-of-sync-application"},
+	}
+	expectedContentText, marshalErr := json.Marshal(expectedContent)
+	require.NoError(t, marshalErr)
+
+	resultContent, ok := result.Content[0].(*mcp.TextContent)
+	require.True(t, ok)
+	assert.JSONEq(t, string(expectedContentText), resultContent.Text)
+}
+
 func getMetrics(t *testing.T, mcpServerURL string, labels map[string]string) (int64, int64) { //nolint:unparam
 	labelStrings := make([]string, 0, 2*len(labels))
 	for k, v := range labels {
@@ -287,19 +357,6 @@ func newStdioSession(mcpServerDebug bool, argocdURL string, argocdToken string,
 	}
 }
 
-func newHTTPSession(mcpServerURL string) func(*testing.T) *mcp.ClientSession {
-	return func(t *testing.T) *mcp.ClientSession {
-		ctx := context.Background()
-		cl := mcp.NewClient(&mcp.Implementation{Name: "e2e-test-client", Version: "v1.0.0"}, nil)
-		session, err := cl.Connect(ctx, &mcp.StreamableClientTransport{
-			MaxRetries: 5,
-			Endpoint:   mcpServerURL,
-		}, nil)
-		require.NoError(t, err)
-		return session
-	}
-}
-
 func newStdioServerCmd(ctx context.Context, mcpServerDebug bool, argocdURL string, argocdToken string, argocdInsecureURL bool) *exec.Cmd {
 	return exec.CommandContext(ctx, //nolint:gosec
 		"argocd-mcp-server",
@@ -310,3 +367,68 @@ func newStdioServerCmd(ctx context.Context, mcpServerDebug bool, argocdURL strin
 		"--insecure", strconv.FormatBool(argocdInsecureURL),
 	)
 }
+
+func newHTTPSession(ctx context.Context, endpoint, clientName string) (*mcp.ClientSession, error) {
+	client := mcp.NewClient(&mcp.Implementation{
+		Name:    clientName,
+		Version: "1.0.0",
+	}, nil)
+	return client.Connect(ctx, &mcp.StreamableClientTransport{
+		MaxRetries: 5,
+		Endpoint:   endpoint,
+	}, nil)
+}
+
+func assertListChanged(t *testing.T, session *mcp.ClientSession, expected bool) {
+	t.Helper()
+	initResult := session.InitializeResult()
+	require.NotNil(t, initResult, "should have initialize result")
+	require.NotNil(t, initResult.Capabilities, "should have capabilities")
+
+	if initResult.Capabilities.Tools != nil {
+		assert.Equal(t, expected, initResult.Capabilities.Tools.ListChanged,
+			"Tools.ListChanged should be %t", expected)
+	}
+	if initResult.Capabilities.Prompts != nil {
+		assert.Equal(t, expected, initResult.Capabilities.Prompts.ListChanged,
+			"Prompts.ListChanged should be %t", expected)
+	}
+}
+
+// assertInitializeResponse performs comprehensive validation of the initialize response
+func assertInitializeResponse(t *testing.T, session *mcp.ClientSession, stateless bool) {
+	t.Helper()
+
+	initResult := session.InitializeResult()
+	require.NotNil(t, initResult, "should have initialize result")
+
+	// Verify server info exists
+	require.NotNil(t, initResult.ServerInfo, "should have server info")
+	assert.NotEmpty(t, initResult.ServerInfo.Name, "server name should not be empty")
+	assert.NotEmpty(t, initResult.ServerInfo.Version, "server version should not be empty")
+
+	// Verify protocol version exists
+	assert.NotEmpty(t, initResult.ProtocolVersion, "protocol version should not be empty")
+
+	// Verify capabilities
+	require.NotNil(t, initResult.Capabilities, "should have capabilities")
+
+	// In stateless mode: ListChanged should be false (no notifications)
+	// In stateful mode: ListChanged should be true (notifications enabled)
+
+	// Tools capability
+	require.NotNil(t, initResult.Capabilities.Tools, "should have tools capability")
+	if stateless {
+		assert.False(t, initResult.Capabilities.Tools.ListChanged, "stateless mode should have Tools.ListChanged=false")
+	} else {
+		assert.True(t, initResult.Capabilities.Tools.ListChanged, "stateful mode should have Tools.ListChanged=true")
+	}
+
+	// Prompts capability
+	require.NotNil(t, initResult.Capabilities.Prompts, "should have prompts capability")
+	if stateless {
+		assert.False(t, initResult.Capabilities.Prompts.ListChanged, "stateless mode should have Prompts.ListChanged=false")
+	} else {
+		assert.True(t, initResult.Capabilities.Prompts.ListChanged, "stateful mode should have Prompts.ListChanged=true")
+	}
+}