Merge pull request #73 from dgageot/deprecate-stdio-over-tcp

Deprecate stdio over TCP

Author: dgageot

cmd/docker-mcp/commands/gateway.go

@@ -1,6 +1,7 @@
 package commands
 
 import (
+	"errors"
 	"os"
 
 	"github.com/spf13/cobra"
@@ -56,7 +57,11 @@ func gatewayCommand(docker docker.Client) *cobra.Command {
 		Short: "Run the gateway",
 		Args:  cobra.NoArgs,
 		RunE: func(cmd *cobra.Command, _ []string) error {
-			if options.Port == 0 && options.Transport != "stdio" {
+			if options.Transport == "stdio" {
+				if options.Port != 0 {
+					return errors.New("cannot use --port with --transport=stdio")
+				}
+			} else if options.Port == 0 {
 				options.Port = 8811
 			}
 

cmd/docker-mcp/internal/gateway/run.go

@@ -6,7 +6,6 @@ import (
 	"net"
 	"os"
 	"strings"
-	"sync"
 	"time"
 
 	"github.com/mark3labs/mcp-go/server"
@@ -90,34 +89,14 @@ func (g *Gateway) Run(ctx context.Context) error {
 	}
 	toolCallbacks := interceptors.Callbacks(g.LogCalls, g.BlockSecrets, customInterceptors)
 
-	// TODO: cleanup stopped servers. That happens in stdio over TCP mode.
-	var (
-		lock            sync.Mutex
-		changeListeners []func(*Capabilities)
+	mcpServer := server.NewMCPServer(
+		"Docker AI MCP Gateway",
+		"2.0.1",
+		server.WithToolHandlerMiddleware(toolCallbacks),
 	)
 
-	capabilities, err := g.listServersAndCapabilities(ctx, configuration)
-	if err != nil {
-		return fmt.Errorf("listing capabilities: %w", err)
-	}
-
-	newMCPServer := func() *server.MCPServer {
-		mcpServer := server.NewMCPServer(
-			"Docker AI MCP Gateway",
-			"2.0.1",
-			server.WithToolHandlerMiddleware(toolCallbacks),
-		)
-
-		// TODO: This will create a new server instance with an outdated vision of the capabilities.
-		refreshCapabilities(mcpServer, capabilities)
-
-		lock.Lock()
-		changeListeners = append(changeListeners, func(newCapabilities *Capabilities) {
-			refreshCapabilities(mcpServer, newCapabilities)
-		})
-		lock.Unlock()
-
-		return mcpServer
+	if err := g.reloadConfiguration(ctx, mcpServer, configuration); err != nil {
+		return fmt.Errorf("loading configuration: %w", err)
 	}
 
 	// Optionally watch for configuration updates.
@@ -137,19 +116,10 @@ func (g *Gateway) Run(ctx context.Context) error {
 						continue
 					}
 
-					capabilities, err := g.listServersAndCapabilities(ctx, configuration)
-					if err != nil {
+					if err := g.reloadConfiguration(ctx, mcpServer, configuration); err != nil {
 						logf("> Unable to list capabilities: %s", err)
 						continue
 					}
-
-					g.health.SetUnhealthy()
-					lock.Lock()
-					for _, listener := range changeListeners {
-						listener(capabilities)
-					}
-					lock.Unlock()
-					g.health.SetHealthy()
 				}
 			}
 		}()
@@ -162,31 +132,25 @@ func (g *Gateway) Run(ctx context.Context) error {
 	}
 
 	// Start the server
-	g.health.SetHealthy()
 	switch strings.ToLower(g.Transport) {
 	case "stdio":
-		if g.Port == 0 {
-			log("> Start stdio server")
-			return g.startStdioServer(ctx, newMCPServer, os.Stdin, os.Stdout)
-		}
-
-		log("> Start stdio over TCP server on port", g.Port)
-		return g.startStdioOverTCPServer(ctx, newMCPServer, ln)
+		log("> Start stdio server")
+		return g.startStdioServer(ctx, mcpServer, os.Stdin, os.Stdout)
 
 	case "sse":
 		log("> Start sse server on port", g.Port)
-		return g.startSseServer(ctx, newMCPServer, ln)
+		return g.startSseServer(ctx, mcpServer, ln)
 
 	case "streaming":
 		log("> Start streaming server on port", g.Port)
-		return g.startStreamingServer(ctx, newMCPServer, ln)
+		return g.startStreamingServer(ctx, mcpServer, ln)
 
 	default:
 		return fmt.Errorf("unknown transport %q, expected 'stdio', 'sse' or 'streaming", g.Transport)
 	}
 }
 
-func (g *Gateway) listServersAndCapabilities(ctx context.Context, configuration Configuration) (*Capabilities, error) {
+func (g *Gateway) reloadConfiguration(ctx context.Context, mcpServer *server.MCPServer, configuration Configuration) error {
 	// Which servers are enabled in the registry.yaml?
 	serverNames := configuration.ServerNames()
 	if len(serverNames) == 0 {
@@ -200,19 +164,20 @@ func (g *Gateway) listServersAndCapabilities(ctx context.Context, configuration
 	log("- Listing MCP tools...")
 	capabilities, err := g.listCapabilities(ctx, configuration, serverNames)
 	if err != nil {
-		return nil, fmt.Errorf("listing resources: %w", err)
+		return fmt.Errorf("listing resources: %w", err)
 	}
 	log(">", len(capabilities.Tools), "tools listed in", time.Since(startList))
 
-	return capabilities, nil
-}
-
-func refreshCapabilities(s *server.MCPServer, c *Capabilities) {
-	s.SetTools(c.Tools...)
-	s.SetPrompts(c.Prompts...)
-	s.SetResources(c.Resources...)
-	s.RemoveAllResourceTemplates()
-	for _, v := range c.ResourceTemplates {
-		s.AddResourceTemplate(v.ResourceTemplate, v.Handler)
+	// Update the server's capabilities.
+	g.health.SetUnhealthy()
+	mcpServer.SetTools(capabilities.Tools...)
+	mcpServer.SetPrompts(capabilities.Prompts...)
+	mcpServer.SetResources(capabilities.Resources...)
+	mcpServer.RemoveAllResourceTemplates()
+	for _, v := range capabilities.ResourceTemplates {
+		mcpServer.AddResourceTemplate(v.ResourceTemplate, v.Handler)
 	}
+	g.health.SetHealthy()
+
+	return nil
 }

cmd/docker-mcp/internal/gateway/transport.go

@@ -7,17 +7,15 @@ import (
 	"net/http"
 
 	"github.com/mark3labs/mcp-go/server"
-
-	"github.com/docker/mcp-gateway/cmd/docker-mcp/internal/sockets"
 )
 
-func (g *Gateway) startStdioServer(ctx context.Context, newMCPServer func() *server.MCPServer, stdin io.Reader, stdout io.Writer) error {
-	return server.NewStdioServer(newMCPServer()).Listen(ctx, stdin, stdout)
+func (g *Gateway) startStdioServer(ctx context.Context, mcpServer *server.MCPServer, stdin io.Reader, stdout io.Writer) error {
+	return server.NewStdioServer(mcpServer).Listen(ctx, stdin, stdout)
 }
 
-func (g *Gateway) startSseServer(ctx context.Context, newMCPServer func() *server.MCPServer, ln net.Listener) error {
+func (g *Gateway) startSseServer(ctx context.Context, mcpServer *server.MCPServer, ln net.Listener) error {
 	mux := http.NewServeMux()
-	sseServer := server.NewSSEServer(newMCPServer())
+	sseServer := server.NewSSEServer(mcpServer)
 	mux.Handle("/sse", sseServer.SSEHandler())
 	mux.Handle("/message", sseServer.MessageHandler())
 	mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
@@ -40,9 +38,9 @@ func (g *Gateway) startSseServer(ctx context.Context, newMCPServer func() *serve
 	return httpServer.Serve(ln)
 }
 
-func (g *Gateway) startStreamingServer(ctx context.Context, newMCPServer func() *server.MCPServer, ln net.Listener) error {
+func (g *Gateway) startStreamingServer(ctx context.Context, mcpServer *server.MCPServer, ln net.Listener) error {
 	mux := http.NewServeMux()
-	mux.Handle("/mcp", server.NewStreamableHTTPServer(newMCPServer()))
+	mux.Handle("/mcp", server.NewStreamableHTTPServer(mcpServer))
 	mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
 		http.Redirect(w, r, "/mcp", http.StatusTemporaryRedirect)
 	})
@@ -63,30 +61,3 @@ func (g *Gateway) startStreamingServer(ctx context.Context, newMCPServer func()
 	}()
 	return httpServer.Serve(ln)
 }
-
-func (g *Gateway) startStdioOverTCPServer(ctx context.Context, newMCPServer func() *server.MCPServer, ln net.Listener) error {
-	for {
-		select {
-		case <-ctx.Done():
-			return ctx.Err()
-		default:
-			conn, err := sockets.AcceptWithContext(ctx, ln)
-			if err != nil {
-				if ctx.Err() != nil {
-					return ctx.Err()
-				}
-				logf("Error accepting the connection: %v", err)
-				continue
-			}
-
-			newServer := server.NewStdioServer(newMCPServer())
-			go func() {
-				defer conn.Close()
-
-				if err := newServer.Listen(ctx, conn, conn); err != nil {
-					logf("Error listening: %v", err)
-				}
-			}()
-		}
-	}
-}