Repository: mark3labs/mcp-go

Issue:
bug: bufio.Reader is not thread-safe
# Problem Description

When using the `github.com/mark3labs/mcp-go` library for stdio-based MCP servers, a race condition occurs after exactly 2-4 concurrent tool calls, causing:

- `Error reading input: read /dev/stdin: bad file descriptor`
- `write |1: broken pipe` errors
- Complete server failure requiring restart
- 0% success rate for subsequent tool calls

# Root Cause Analysis

The MCP-Go library has a **hardcoded race condition** in `/server/stdio.go`:

```go
// Problematic code in MCP-Go library
if baseMessage.Method == "tools/call" {
    go func() {  // ❌ RACE CONDITION: Multiple goroutines compete for same stdin
        response := s.server.HandleMessage(ctx, rawMessage)
        if response != nil {
            s.writeResponse(response, stdout)
        }
    }()
}
```

**The Issue:**
- Multiple goroutines created for concurrent tool calls
- All goroutines access the same `bufio.Reader` on stdin
- `bufio.Reader` is **NOT thread-safe**
- Concurrent access corrupts the file descriptor
- Results in permanent stdin corruption

# After trying many workarounds for hours, Claude said:
The only reliable fix is to **completely replace** the MCP-Go stdio server with a custom synchronous implementation that eliminates all goroutines.

# The fix worked but I would like to use the mcp-go stdio

Discussion:

Solution patch:
diff --git a/server/stdio.go b/server/stdio.go
index 4d567d8..8c270e1 100644
--- a/server/stdio.go
+++ b/server/stdio.go
@@ -29,6 +29,20 @@ type StdioServer struct {
 	server      *MCPServer
 	errLogger   *log.Logger
 	contextFunc StdioContextFunc
+
+	// Thread-safe tool call processing
+	toolCallQueue  chan *toolCallWork
+	workerWg       sync.WaitGroup
+	workerPoolSize int
+	queueSize      int
+	writeMu        sync.Mutex // Protects concurrent writes
+}
+
+// toolCallWork represents a queued tool call request
+type toolCallWork struct {
+	ctx     context.Context
+	message json.RawMessage
+	writer  io.Writer
 }
 
 // StdioOption defines a function type for configuring StdioServer
@@ -50,6 +64,32 @@ func WithStdioContextFunc(fn StdioContextFunc) StdioOption {
 	}
 }
 
+// WithWorkerPoolSize sets the number of workers for processing tool calls
+func WithWorkerPoolSize(size int) StdioOption {
+	return func(s *StdioServer) {
+		const maxWorkerPoolSize = 100
+		if size > 0 && size <= maxWorkerPoolSize {
+			s.workerPoolSize = size
+		} else if size > maxWorkerPoolSize {
+			s.errLogger.Printf("Worker pool size %d exceeds maximum (%d), using maximum", size, maxWorkerPoolSize)
+			s.workerPoolSize = maxWorkerPoolSize
+		}
+	}
+}
+
+// WithQueueSize sets the size of the tool call queue
+func WithQueueSize(size int) StdioOption {
+	return func(s *StdioServer) {
+		const maxQueueSize = 10000
+		if size > 0 && size <= maxQueueSize {
+			s.queueSize = size
+		} else if size > maxQueueSize {
+			s.errLogger.Printf("Queue size %d exceeds maximum (%d), using maximum", size, maxQueueSize)
+			s.queueSize = maxQueueSize
+		}
+	}
+}
+
 // stdioSession is a static client session, since stdio has only one client.
 type stdioSession struct {
 	notifications      chan mcp.JSONRPCNotification
@@ -218,6 +258,8 @@ func NewStdioServer(server *MCPServer) *StdioServer {
 			"",
 			log.LstdFlags,
 		), // Default to discarding logs
+		workerPoolSize: 5,   // Default worker pool size
+		queueSize:      100, // Default queue size
 	}
 }
 
@@ -281,6 +323,30 @@ func (s *StdioServer) processInputStream(ctx context.Context, reader *bufio.Read
 	}
 }
 
+// toolCallWorker processes tool calls from the queue
+func (s *StdioServer) toolCallWorker(ctx context.Context) {
+	defer s.workerWg.Done()
+
+	for {
+		select {
+		case work, ok := <-s.toolCallQueue:
+			if !ok {
+				// Channel closed, exit worker
+				return
+			}
+			// Process the tool call
+			response := s.server.HandleMessage(work.ctx, work.message)
+			if response != nil {
+				if err := s.writeResponse(response, work.writer); err != nil {
+					s.errLogger.Printf("Error writing tool response: %v", err)
+				}
+			}
+		case <-ctx.Done():
+			return
+		}
+	}
+}
+
 // readNextLine reads a single line from the input reader in a context-aware manner.
 // It uses channels to make the read operation cancellable via context.
 // Returns the read line and any error encountered. If the context is cancelled,
@@ -315,6 +381,9 @@ func (s *StdioServer) Listen(
 	stdin io.Reader,
 	stdout io.Writer,
 ) error {
+	// Initialize the tool call queue
+	s.toolCallQueue = make(chan *toolCallWork, s.queueSize)
+
 	// Set a static client context since stdio only has one client
 	if err := s.server.RegisterSession(ctx, &stdioSessionInstance); err != nil {
 		return fmt.Errorf("register session: %w", err)
@@ -332,9 +401,23 @@ func (s *StdioServer) Listen(
 
 	reader := bufio.NewReader(stdin)
 
+	// Start worker pool for tool calls
+	for i := 0; i < s.workerPoolSize; i++ {
+		s.workerWg.Add(1)
+		go s.toolCallWorker(ctx)
+	}
+
 	// Start notification handler
 	go s.handleNotifications(ctx, stdout)
-	return s.processInputStream(ctx, reader, stdout)
+
+	// Process input stream
+	err := s.processInputStream(ctx, reader, stdout)
+
+	// Shutdown workers gracefully
+	close(s.toolCallQueue)
+	s.workerWg.Wait()
+
+	return err
 }
 
 // processMessage handles a single JSON-RPC message and writes the response.
@@ -367,16 +450,25 @@ func (s *StdioServer) processMessage(
 		Method string `json:"method"`
 	}
 	if json.Unmarshal(rawMessage, &baseMessage) == nil && baseMessage.Method == "tools/call" {
-		// Process tool calls concurrently to avoid blocking on sampling requests
-		go func() {
+		// Queue tool calls for processing by workers
+		select {
+		case s.toolCallQueue <- &toolCallWork{
+			ctx:     ctx,
+			message: rawMessage,
+			writer:  writer,
+		}:
+			return nil
+		case <-ctx.Done():
+			return ctx.Err()
+		default:
+			// Queue is full, process synchronously as fallback
+			s.errLogger.Printf("Tool call queue full, processing synchronously")
 			response := s.server.HandleMessage(ctx, rawMessage)
 			if response != nil {
-				if err := s.writeResponse(response, writer); err != nil {
-					s.errLogger.Printf("Error writing tool response: %v", err)
-				}
+				return s.writeResponse(response, writer)
 			}
-		}()
-		return nil
+			return nil
+		}
 	}
 
 	// Handle other messages synchronously
@@ -462,6 +554,10 @@ func (s *StdioServer) writeResponse(
 		return err
 	}
 
+	// Protect concurrent writes
+	s.writeMu.Lock()
+	defer s.writeMu.Unlock()
+
 	// Write response followed by newline
 	if _, err := fmt.Fprintf(writer, "%s\n", responseBytes); err != nil {
 		return err



Test patch:
diff --git a/server/stdio_test.go b/server/stdio_test.go
index 4ec7259..8fb542a 100644
--- a/server/stdio_test.go
+++ b/server/stdio_test.go
@@ -4,10 +4,13 @@ import (
 	"bufio"
 	"context"
 	"encoding/json"
+	"fmt"
 	"io"
 	"log"
 	"os"
+	"sync"
 	"testing"
+	"time"
 
 	"github.com/mark3labs/mcp-go/mcp"
 )
@@ -267,4 +270,217 @@ func TestStdioServer(t *testing.T) {
 			t.Errorf("unexpected server error: %v", err)
 		}
 	})
+
+	t.Run("Can handle concurrent tool calls", func(t *testing.T) {
+		// Create pipes for stdin and stdout
+		stdinReader, stdinWriter := io.Pipe()
+		stdoutReader, stdoutWriter := io.Pipe()
+
+		// Track tool call executions (sync.Map is already thread-safe)
+		var callCount sync.Map
+
+		// Create server with test tools
+		mcpServer := NewMCPServer("test", "1.0.0")
+
+		// Add multiple tools that simulate work and track concurrent execution
+		for i := 0; i < 5; i++ {
+			toolName := fmt.Sprintf("test_tool_%d", i)
+			mcpServer.AddTool(
+				mcp.NewTool(toolName),
+				func(name string) func(ctx context.Context, request mcp.CallToolRequest) (*mcp.CallToolResult, error) {
+					return func(ctx context.Context, request mcp.CallToolRequest) (*mcp.CallToolResult, error) {
+						// Track concurrent executions
+						count, _ := callCount.LoadOrStore(name, 0)
+						callCount.Store(name, count.(int)+1)
+
+						// Simulate some work
+						time.Sleep(10 * time.Millisecond)
+
+						return mcp.NewToolResultText(fmt.Sprintf("Result from %s", name)), nil
+					}
+				}(toolName),
+			)
+		}
+
+		stdioServer := NewStdioServer(mcpServer)
+		stdioServer.SetErrorLogger(log.New(io.Discard, "", 0))
+
+		// Create context with cancel
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+
+		// Start server
+		serverErrCh := make(chan error, 1)
+		go func() {
+			err := stdioServer.Listen(ctx, stdinReader, stdoutWriter)
+			if err != nil && err != io.EOF && err != context.Canceled {
+				serverErrCh <- err
+			}
+			stdoutWriter.Close()
+			close(serverErrCh)
+		}()
+
+		// Initialize the session
+		initRequest := map[string]any{
+			"jsonrpc": "2.0",
+			"id":      1,
+			"method":  "initialize",
+			"params": map[string]any{
+				"protocolVersion": "2024-11-05",
+				"clientInfo": map[string]any{
+					"name":    "test-client",
+					"version": "1.0.0",
+				},
+			},
+		}
+
+		requestBytes, _ := json.Marshal(initRequest)
+		if _, err := stdinWriter.Write(append(requestBytes, '\n')); err != nil {
+			t.Fatalf("Failed to write init request: %v", err)
+		}
+
+		// Read init response
+		scanner := bufio.NewScanner(stdoutReader)
+		scanner.Scan()
+
+		// Send multiple concurrent tool calls
+		var wg sync.WaitGroup
+		responseChan := make(chan string, 10)
+
+		// Send 10 concurrent tool calls
+		for i := 0; i < 10; i++ {
+			wg.Add(1)
+			go func(id int) {
+				defer wg.Done()
+
+				toolRequest := map[string]any{
+					"jsonrpc": "2.0",
+					"id":      id + 2,
+					"method":  "tools/call",
+					"params": map[string]any{
+						"name": fmt.Sprintf("test_tool_%d", id%5),
+					},
+				}
+
+				requestBytes, _ := json.Marshal(toolRequest)
+				if _, err := stdinWriter.Write(append(requestBytes, '\n')); err != nil {
+					t.Errorf("Failed to write tool request %d: %v", id, err)
+				}
+			}(i)
+		}
+
+		// Read all responses
+		go func() {
+			for i := 0; i < 10; i++ {
+				if scanner.Scan() {
+					responseChan <- scanner.Text()
+				}
+			}
+			close(responseChan)
+		}()
+
+		// Wait for all requests to be sent
+		wg.Wait()
+
+		// Collect responses
+		responses := make([]string, 0, 10)
+		timeout := time.After(2 * time.Second)
+
+	collectLoop:
+		for len(responses) < 10 {
+			select {
+			case resp, ok := <-responseChan:
+				if !ok {
+					break collectLoop
+				}
+				responses = append(responses, resp)
+			case <-timeout:
+				t.Fatal("Timeout waiting for responses")
+			}
+		}
+		// Verify we got all responses
+		if len(responses) != 10 {
+			t.Errorf("Expected 10 responses, got %d", len(responses))
+		}
+
+		// Verify no errors in responses
+		for _, resp := range responses {
+			var response map[string]any
+			if err := json.Unmarshal([]byte(resp), &response); err != nil {
+				t.Errorf("Failed to unmarshal response: %v", err)
+				continue
+			}
+
+			if response["error"] != nil {
+				t.Errorf("Unexpected error in response: %v", response["error"])
+			}
+
+			// Verify response has expected structure
+			if response["result"] == nil {
+				t.Error("Expected result in response")
+			}
+		}
+
+		// Verify tools were called
+		callCount.Range(func(key, value interface{}) bool {
+			toolName := key.(string)
+			count := value.(int)
+			if count == 0 {
+				t.Errorf("Tool %s was not called", toolName)
+			}
+			return true
+		})
+
+		// Clean up
+		cancel()
+		stdinWriter.Close()
+
+		// Check for server errors
+		if err := <-serverErrCh; err != nil {
+			t.Errorf("Server error: %v", err)
+		}
+	})
+
+	t.Run("Configuration options respect bounds", func(t *testing.T) {
+		mcpServer := NewMCPServer("test", "1.0.0")
+
+		// Test worker pool size bounds
+		stdioServer := NewStdioServer(mcpServer)
+		WithWorkerPoolSize(150)(stdioServer)
+		if stdioServer.workerPoolSize != 100 { // Should use maximum
+			t.Errorf("Expected maximum worker pool size 100, got %d", stdioServer.workerPoolSize)
+		}
+
+		// Test valid worker pool size
+		stdioServer = NewStdioServer(mcpServer)
+		WithWorkerPoolSize(50)(stdioServer)
+		if stdioServer.workerPoolSize != 50 {
+			t.Errorf("Expected worker pool size 50, got %d", stdioServer.workerPoolSize)
+		}
+
+		// Test queue size bounds
+		stdioServer = NewStdioServer(mcpServer)
+		WithQueueSize(20000)(stdioServer)
+		if stdioServer.queueSize != 10000 { // Should use maximum
+			t.Errorf("Expected maximum queue size 10000, got %d", stdioServer.queueSize)
+		}
+
+		// Test valid queue size
+		stdioServer = NewStdioServer(mcpServer)
+		WithQueueSize(500)(stdioServer)
+		if stdioServer.queueSize != 500 {
+			t.Errorf("Expected queue size 500, got %d", stdioServer.queueSize)
+		}
+
+		// Test zero and negative values
+		stdioServer = NewStdioServer(mcpServer)
+		WithWorkerPoolSize(0)(stdioServer)
+		WithQueueSize(-10)(stdioServer)
+		if stdioServer.workerPoolSize != 5 {
+			t.Errorf("Expected default worker pool size 5 for zero input, got %d", stdioServer.workerPoolSize)
+		}
+		if stdioServer.queueSize != 100 {
+			t.Errorf("Expected default queue size 100 for negative input, got %d", stdioServer.queueSize)
+		}
+	})
 }