@cryptotaxi247 / netdata-1 / commits / 215875e5c

deps(mcp/bridge/stdio-golang): switch to github.com/coder/websocket (#21006)

Ilya Mashchenko committed Sep 18, 2025 at 20:12 UTC 215875e5cfbddc88e22ec24384a81ffe2782a900
6 files changed +62 -62
src/web/mcp/bridges/stdio-golang/README.md
+3 -3
@@ -5,7 +5,7 @@ This Go bridge converts MCP stdio communication to Netdata's MCP over WebSocket.
5 ## Requirements
6
7 - Go 1.16+
8 -- nhooyr.io/websocket package
8 +- github.com/coder/websocket package
9
10 ## Installation
11
@@ -32,7 +32,7 @@ Alternatively, you can build it manually:
32 go mod init netdata/nd-mcp-bridge
33
34 # Add dependencies
35 -go get nhooyr.io/websocket
35 +go get github.com/coder/websocket
36 go mod tidy
37
38 # Build the binary
@@ -89,7 +89,7 @@ The reconnection algorithm starts with a 1-second delay and doubles the wait tim
89 ## Implementation Notes
90
91 This implementation:
92 -- Uses the nhooyr.io/websocket library for WebSocket communication
92 +- Uses the github.com/coder/websocket library for WebSocket communication
93 - Explicitly sets WebSocket headers including Sec-WebSocket-Key and Sec-WebSocket-Version
94 - Implements proper WebSocket handshake to ensure compatibility with Netdata's WebSocket server
95 - Sends and receives raw text messages directly (without JSON serialization/deserialization)
src/web/mcp/bridges/stdio-golang/build.bat
+1 -1
@@ -34,7 +34,7 @@ if not exist "go.mod" (
34
35 REM Add required dependencies
36 echo Adding dependencies...
37 -go get nhooyr.io/websocket
37 +go get github.com/coder/websocket
38 go mod tidy
39
40 REM Build the binary
src/web/mcp/bridges/stdio-golang/build.sh
+1 -1
@@ -27,7 +27,7 @@ fi
27
28 # Add required dependencies
29 echo "Adding dependencies..."
30 -go get nhooyr.io/websocket
30 +go get github.com/coder/websocket
31 go mod tidy
32
33 # Build the binary
src/web/mcp/bridges/stdio-golang/go.mod
+1 -1
@@ -2,4 +2,4 @@ module netdata/nd-mcp-bridge
2
3 go 1.24.0
4
5 -require nhooyr.io/websocket v1.8.17
5 +require github.com/coder/websocket v1.8.14
src/web/mcp/bridges/stdio-golang/go.sum
+2 -2
@@ -1,2 +1,2 @@
1 -nhooyr.io/websocket v1.8.17 h1:KEVeLJkUywCKVsnLIDlD/5gtayKp8VoCkksHCGGfT9Y=
2 -nhooyr.io/websocket v1.8.17/go.mod h1:rN9OFWIUwuxg4fR5tELlYC04bXYowCP9GX47ivo2l+c=
1 +github.com/coder/websocket v1.8.14 h1:9L0p0iKiNOibykf283eHkKUHHrpG7f65OE3BhhO7v9g=
2 +github.com/coder/websocket v1.8.14/go.mod h1:NX3SzP+inril6yawo5CQXx8+fk145lPDC6pumgx0mVg=
src/web/mcp/bridges/stdio-golang/nd-mcp.go
+54 -54
@@ -17,7 +17,7 @@ import (
17 "syscall"
18 "time"
19
20 - "nhooyr.io/websocket"
20 + "github.com/coder/websocket"
21 )
22
23 // Connection states
@@ -35,10 +35,10 @@ const (
35
36 // JSON-RPC 2.0 related structures
37 type JsonRpcMessage struct {
38 - JsonRpc string `json:"jsonrpc"`
39 - Id interface{} `json:"id,omitempty"`
40 - Method string `json:"method,omitempty"`
41 - Result interface{} `json:"result,omitempty"`
38 + JsonRpc string `json:"jsonrpc"`
39 + Id interface{} `json:"id,omitempty"`
40 + Method string `json:"method,omitempty"`
41 + Result interface{} `json:"result,omitempty"`
42 Error *JsonRpcError `json:"error,omitempty"`
43 }
44
@@ -80,10 +80,10 @@ func main() {
80 }
81
82 // Set up channels for communication
83 - stdinCh := make(chan string, 100) // Buffer stdin messages
84 - reconnectCh := make(chan struct{}, 1) // Signal for immediate reconnection
85 - doneCh := make(chan struct{}) // Signal for program termination
86 - stdinClosedCh := make(chan struct{}) // Signal that stdin is closed
83 + stdinCh := make(chan string, 100) // Buffer stdin messages
84 + reconnectCh := make(chan struct{}, 1) // Signal for immediate reconnection
85 + doneCh := make(chan struct{}) // Signal for program termination
86 + stdinClosedCh := make(chan struct{}) // Signal that stdin is closed
87
88 // Global state
89 var state int // Connection state
@@ -91,7 +91,7 @@ func main() {
91 var messageQueueMu sync.Mutex
92 messageQueue := []string{}
93 stdinActive := true
94 -
94 +
95 // Pending requests that are waiting for connection to be established
96 var pendingMu sync.Mutex
97 pendingRequests := make(map[interface{}]*PendingRequest)
@@ -136,7 +136,7 @@ func main() {
136 pendingMu.Lock()
137 if req, exists := pendingRequests[msgId]; exists {
138 fmt.Fprintf(os.Stderr, "%s: Connection timeout for request ID %v, sending error response\n", programName, msgId)
139 -
139 +
140 // Create and send error response
141 errorResponse := createJsonRpcError(
142 msgId,
@@ -144,16 +144,16 @@ func main() {
144 "MCP server connection failed",
145 map[string]string{"details": "Could not establish connection to Netdata within timeout period"},
146 )
147 -
147 +
148 fmt.Println(errorResponse)
149 -
149 +
150 // Get the original message
151 originalMessage := req.Message
152 -
152 +
153 // Remove request from pending
154 delete(pendingRequests, msgId)
155 pendingMu.Unlock()
156 -
156 +
157 // Also remove from messageQueue if it exists there
158 messageQueueMu.Lock()
159 for i, msg := range messageQueue {
@@ -191,23 +191,23 @@ func main() {
191 go func() {
192 scanner := bufio.NewScanner(os.Stdin)
193 scannerRunning := true
194 -
194 +
195 // Make scanner buffer larger to handle large messages
196 const maxScanBufferSize = 1024 * 1024 // 1MB
197 buf := make([]byte, maxScanBufferSize)
198 scanner.Buffer(buf, maxScanBufferSize)
199 -
199 +
200 for scannerRunning && scanner.Scan() {
201 text := scanner.Text()
202 -
202 +
203 // Parse for JSON-RPC ID
204 msgId, _ := parseJsonRpcMessage(text)
205 -
205 +
206 // Check connection state
207 stateMu.Lock()
208 currentState := state
209 stateMu.Unlock()
210 -
210 +
211 if currentState != stateConnected && msgId != nil {
212 // Store as pending request with timeout
213 pendingMu.Lock()
@@ -218,13 +218,13 @@ func main() {
218 }
219 pendingRequests[msgId] = req
220 pendingMu.Unlock()
221 -
221 +
222 fmt.Fprintf(os.Stderr, "%s: Received request with ID %v, setting response timeout\n", programName, msgId)
223 }
224 -
224 +
225 // Queue the message
226 stdinCh <- text
227 -
227 +
228 // If we're not connected, trigger immediate reconnection
229 if currentState != stateConnected {
230 select {
@@ -235,14 +235,14 @@ func main() {
235 }
236 }
237 }
238 -
238 +
239 // Check for scanner error
240 if err := scanner.Err(); err != nil {
241 fmt.Fprintf(os.Stderr, "%s: ERROR: stdin read error: %v\n", programName, err)
242 } else {
243 fmt.Fprintf(os.Stderr, "%s: End of stdin\n", programName)
244 }
245 -
245 +
246 // Signal that stdin is closed
247 stdinActive = false
248 close(stdinClosedCh)
@@ -252,10 +252,10 @@ func main() {
252 baseDelay := 1 * time.Second
253 maxDelay := 60 * time.Second
254 attempt := 0
255 -
255 +
256 // Timer for reconnection backoff
257 var timer *time.Timer
258 -
258 +
259 // Main connection loop
260 for {
261 select {
@@ -271,12 +271,12 @@ func main() {
271 if timer != nil {
272 timer.Stop()
273 }
274 -
274 +
275 // Only proceed with immediate reconnection if we're not already connecting/connected
276 stateMu.Lock()
277 currentState := state
278 stateMu.Unlock()
279 -
279 +
280 if currentState == stateDisconnected {
281 // Reset the reconnection timer
282 attempt = 0
@@ -293,17 +293,17 @@ func main() {
293 fmt.Fprintf(os.Stderr, "%s: Stdin closed and disconnected, exiting\n", programName)
294 return
295 }
296 -
296 +
297 delaySeconds := math.Min(float64(maxDelay.Seconds()),
298 float64(baseDelay.Seconds())*math.Pow(2, float64(attempt-1))*(0.5+jitter()))
299 delay := time.Duration(delaySeconds * float64(time.Second))
300 -
300 +
301 fmt.Fprintf(os.Stderr, "%s: Reconnecting in %.1f seconds (attempt %d)...\n",
302 programName, delaySeconds, attempt)
303 -
303 +
304 // Create timer and wait for it to expire, or for signals
305 timer = time.NewTimer(delay)
306 -
306 +
307 select {
308 case <-timer.C:
309 // Timer expired, continue to connection attempt
@@ -330,7 +330,7 @@ func main() {
330
331 // Set up connection context with cancellation
332 ctx, cancel := context.WithCancel(context.Background())
333 -
333 +
334 // Set up connection timeout
335 connectionCtx, connectionCancel := context.WithTimeout(ctx, 15*time.Second)
336 defer connectionCancel()
@@ -347,15 +347,15 @@ func main() {
347 CompressionMode: websocket.CompressionContextTakeover,
348 HTTPHeader: header,
349 })
350 -
350 +
351 // Connection failed
352 if err != nil {
353 fmt.Fprintf(os.Stderr, "%s: ERROR: websocket connection failed: %v\n", programName, err)
354 cancel()
355 -
355 +
356 // Increment attempt counter and try again
357 attempt++
358 -
358 +
359 // Update state to disconnected
360 stateMu.Lock()
361 state = stateDisconnected
@@ -400,7 +400,7 @@ func main() {
400 messageQueue = nil // Clear the queue
401 }
402 messageQueueMu.Unlock()
403 -
403 +
404 // Memory management: Limit size of pendingRequests to prevent memory leaks
405 pendingMu.Lock()
406 if len(pendingRequests) > 1000 {
@@ -413,10 +413,10 @@ func main() {
413
414 // Signal for connection closure
415 connClosed := make(chan struct{})
416 -
416 +
417 // Create wait group for goroutines
418 var wg sync.WaitGroup
419 -
419 +
420 // Goroutine for handling termination signals during connection
421 wg.Add(1)
422 go func() {
@@ -435,7 +435,7 @@ func main() {
435 return
436 }
437 }()
438 -
438 +
439 // Goroutine for sending messages to websocket
440 wg.Add(1)
441 go func() {
@@ -456,14 +456,14 @@ func main() {
456 if err != nil {
457 // Check if this was a message with ID
458 msgId, _ := parseJsonRpcMessage(msg)
459 -
459 +
460 select {
461 case <-connClosed:
462 // Connection already known to be closed, queue the message
463 messageQueueMu.Lock()
464 messageQueue = append(messageQueue, msg)
465 messageQueueMu.Unlock()
466 -
466 +
467 // If the message had an ID, keep the pending status
468 if msgId != nil {
469 pendingMu.Lock()
@@ -483,7 +483,7 @@ func main() {
483 messageQueueMu.Lock()
484 messageQueue = append(messageQueue, msg)
485 messageQueueMu.Unlock()
486 -
486 +
487 // Same ID handling as above
488 if msgId != nil {
489 pendingMu.Lock()
@@ -523,7 +523,7 @@ func main() {
523 go func() {
524 defer wg.Done()
525 defer close(connClosed)
526 -
526 +
527 for {
528 messageType, message, err := conn.Read(ctx)
529 if err != nil {
@@ -533,10 +533,10 @@ func main() {
533 }
534 return
535 }
536 -
536 +
537 if messageType == websocket.MessageText {
538 fmt.Println(string(message))
539 -
539 +
540 // Check if this is a response to a request with an ID
541 var response JsonRpcMessage
542 if err := json.Unmarshal(message, &response); err == nil && response.JsonRpc == "2.0" && response.Id != nil {
@@ -561,7 +561,7 @@ func main() {
561 defer wg.Done()
562 ticker := time.NewTicker(pingInterval)
563 defer ticker.Stop()
564 -
564 +
565 for {
566 select {
567 case <-ticker.C:
@@ -581,26 +581,26 @@ func main() {
581
582 // Wait for connection closed
583 <-connClosed
584 -
584 +
585 // Update state to disconnected
586 stateMu.Lock()
587 state = stateDisconnected
588 stateMu.Unlock()
589 -
589 +
590 // Clean up
591 cancel()
592 wg.Wait()
593 -
593 +
594 // Don't automatically reconnect if stdin closed and no messages in queue
595 messageQueueMu.Lock()
596 hasMessages := len(messageQueue) > 0
597 messageQueueMu.Unlock()
598 -
598 +
599 if !stdinActive && !hasMessages {
600 fmt.Fprintf(os.Stderr, "%s: Stdin closed and no pending messages, exiting\n", programName)
601 return
602 }
603 -
603 +
604 // Increment attempt counter for next reconnection
605 attempt++
606 }
@@ -613,4 +613,4 @@ func jitter() float64 {
613 return 0.5 // Fall back to 0.5 if there's an error
614 }
615 return float64(n.Int64()) / 1000.0
616 -}
\ No newline at end of file
616 +}