From 8d4cc3fbba13f1825d1903c38c0ee39061852bdf Mon Sep 17 00:00:00 2001 From: dtapps Date: Thu, 30 Apr 2026 22:56:40 +0800 Subject: [PATCH 1/2] fix(agent): make MCP initialization failure non-fatal When all MCP servers fail to connect (e.g., network unreachable), the agent loop would exit immediately, leaving the application in a zombie state where the gateway is running but no messages can be processed. Changes: - Downgrade MCP init failure from fatal error to warning in Run(), ProcessDirectWithChannel(), ProcessHeartbeat(), and Continue() - Add 30s timeout to HTTP client used for SSE/HTTP MCP transports to prevent indefinite blocking on unreachable servers The agent now continues operating without MCP tools when servers are unavailable, rather than becoming completely unresponsive. --- pkg/agent/agent.go | 3 ++- pkg/agent/agent_message.go | 6 ++++-- pkg/agent/steering.go | 4 ++-- pkg/mcp/manager.go | 17 ++++++++++++----- 4 files changed, 20 insertions(+), 10 deletions(-) diff --git a/pkg/agent/agent.go b/pkg/agent/agent.go index 2c456dca7..5b2d6878f 100644 --- a/pkg/agent/agent.go +++ b/pkg/agent/agent.go @@ -132,7 +132,8 @@ func (al *AgentLoop) Run(ctx context.Context) error { return err } if err := al.ensureMCPInitialized(ctx); err != nil { - return err + logger.WarnCF("agent", "MCP initialization failed, continuing without MCP tools", + map[string]any{"error": err.Error()}) } idleTicker := time.NewTicker(100 * time.Millisecond) diff --git a/pkg/agent/agent_message.go b/pkg/agent/agent_message.go index 96b0b0817..37dbb4190 100644 --- a/pkg/agent/agent_message.go +++ b/pkg/agent/agent_message.go @@ -48,7 +48,8 @@ func (al *AgentLoop) ProcessDirectWithChannel( return "", err } if err := al.ensureMCPInitialized(ctx); err != nil { - return "", err + logger.WarnCF("agent", "MCP initialization failed, processing without MCP tools", + map[string]any{"error": err.Error()}) } msg := bus.InboundMessage{ @@ -73,7 +74,8 @@ func (al *AgentLoop) ProcessHeartbeat( return "", err } if err := al.ensureMCPInitialized(ctx); err != nil { - return "", err + logger.WarnCF("agent", "MCP initialization failed, processing heartbeat without MCP tools", + map[string]any{"error": err.Error()}) } agent := al.GetRegistry().GetDefaultAgent() diff --git a/pkg/agent/steering.go b/pkg/agent/steering.go index 2efa7bbf4..7e9058049 100644 --- a/pkg/agent/steering.go +++ b/pkg/agent/steering.go @@ -370,8 +370,8 @@ func (al *AgentLoop) Continue(ctx context.Context, sessionKey, channel, chatID s return "", err } if err := al.ensureMCPInitialized(ctx); err != nil { - al.activeTurnStates.Delete(sessionKey) - return "", err + logger.WarnCF("agent", "MCP initialization failed, continuing steering without MCP tools", + map[string]any{"error": err.Error(), "session_key": sessionKey}) } steeringMsgs := al.dequeueSteeringMessagesForScopeWithFallback(sessionKey) diff --git a/pkg/mcp/manager.go b/pkg/mcp/manager.go index 92ea426a6..5de0f866a 100644 --- a/pkg/mcp/manager.go +++ b/pkg/mcp/manager.go @@ -12,6 +12,7 @@ import ( "strings" "sync" "sync/atomic" + "time" "github.com/modelcontextprotocol/go-sdk/mcp" @@ -341,14 +342,18 @@ func connectServer( DisableStandaloneSSE: disableStandaloneSSE, } + // Set up HTTP client with a connection timeout to avoid hanging + // indefinitely when the MCP server is unreachable. + mcpHTTPClient := &http.Client{ + Timeout: 30 * time.Second, + } + // Add custom headers if provided if len(cfg.Headers) > 0 { // Create a custom HTTP client with header-injecting transport - sseTransport.HTTPClient = &http.Client{ - Transport: &headerTransport{ - base: http.DefaultTransport, - headers: cfg.Headers, - }, + mcpHTTPClient.Transport = &headerTransport{ + base: http.DefaultTransport, + headers: cfg.Headers, } logger.DebugCF("mcp", "Added custom HTTP headers", map[string]any{ @@ -357,6 +362,8 @@ func connectServer( }) } + sseTransport.HTTPClient = mcpHTTPClient + transport = sseTransport case "stdio": if cfg.Command == "" { From d8b77086c203e74c88c5faee5742883c928934ef Mon Sep 17 00:00:00 2001 From: dtapps Date: Wed, 6 May 2026 16:00:54 +0800 Subject: [PATCH 2/2] fix(mcp): use transport-level timeouts and context timeouts for SSE support Replace http.Client.Timeout with transport-level timeouts to prevent terminating long-lived SSE streams while still protecting against connection hangs during MCP server initialization. Changes: - Use http.Transport with DialContext, TLSHandshakeTimeout, and ResponseHeaderTimeout instead of http.Client.Timeout - Add timeout-scoped contexts for connect and list operations - SSE streams can now remain open indefinitely without being interrupted by a global request timeout - Long-running tool calls are no longer affected by the 30s timeout Fixes review comment: http.Client.Timeout applies to the full request lifetime, including reading the response body, which would terminate SSE streams after 30s. --- pkg/mcp/manager.go | 29 ++++++++++----- pkg/mcp/manager_test.go | 78 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 99 insertions(+), 8 deletions(-) diff --git a/pkg/mcp/manager.go b/pkg/mcp/manager.go index 25788b9c0..fee8e0766 100644 --- a/pkg/mcp/manager.go +++ b/pkg/mcp/manager.go @@ -5,6 +5,7 @@ import ( "context" "errors" "fmt" + "net" "net/http" "os" "os/exec" @@ -382,17 +383,24 @@ func connectServer( DisableStandaloneSSE: disableStandaloneSSE, } - // Set up HTTP client with a connection timeout to avoid hanging - // indefinitely when the MCP server is unreachable. + // Set up HTTP client with transport-level timeouts to avoid hanging + // indefinitely during connection establishment, while allowing SSE + // streams to remain open without a global request timeout. + baseTransport := &http.Transport{ + DialContext: (&net.Dialer{Timeout: 10 * time.Second}).DialContext, + TLSHandshakeTimeout: 10 * time.Second, + ResponseHeaderTimeout: 30 * time.Second, + } + mcpHTTPClient := &http.Client{ - Timeout: 30 * time.Second, + Transport: baseTransport, } // Add custom headers if provided if len(cfg.Headers) > 0 { // Create a custom HTTP client with header-injecting transport mcpHTTPClient.Transport = &headerTransport{ - base: http.DefaultTransport, + base: baseTransport, headers: cfg.Headers, } logger.DebugCF("mcp", "Added custom HTTP headers", @@ -464,8 +472,11 @@ func connectServer( ) } - // Connect to server - session, err := client.Connect(ctx, transport, nil) + // Connect to server with a timeout-scoped context to avoid hanging + // indefinitely when the MCP server is unreachable. + connectCtx, cancel := context.WithTimeout(ctx, 30*time.Second) + session, err := client.Connect(connectCtx, transport, nil) + cancel() if err != nil { return nil, fmt.Errorf("failed to connect: %w", err) } @@ -480,8 +491,10 @@ func connectServer( "protocol": initResult.ProtocolVersion, }) - // List available tools if supported - tools, err := listServerTools(ctx, name, session, initResult) + // List available tools if supported, with a timeout-scoped context + listCtx, cancel := context.WithTimeout(ctx, 30*time.Second) + tools, err := listServerTools(listCtx, name, session, initResult) + cancel() if err != nil { _ = session.Close() return nil, err diff --git a/pkg/mcp/manager_test.go b/pkg/mcp/manager_test.go index 5789a37a9..4002df322 100644 --- a/pkg/mcp/manager_test.go +++ b/pkg/mcp/manager_test.go @@ -5,6 +5,8 @@ import ( "encoding/json" "fmt" "io" + "net" + "net/http" "os" "path/filepath" "strings" @@ -628,3 +630,79 @@ func (t *scriptedTransport) Close() error { func (t *scriptedTransport) SessionID() string { return t.sessionID } + +// TestHTTPClientTransportTimeouts verifies that the HTTP client is configured +// with transport-level timeouts instead of client-wide timeout. +// This addresses the PR review comment: +// "http.Client.Timeout applies to the full request lifetime, including reading +// the response body. On StreamableClientTransport in sse mode that will terminate +// the long-lived SSE stream after 30s" +func TestHTTPClientTransportTimeouts(t *testing.T) { + // Verify that transport-level timeouts are configured correctly + // instead of using http.Client.Timeout which would affect SSE streams + transport := &http.Transport{ + DialContext: (&net.Dialer{Timeout: 10 * time.Second}).DialContext, + TLSHandshakeTimeout: 10 * time.Second, + ResponseHeaderTimeout: 30 * time.Second, + } + + // Verify transport timeouts are set correctly + // This ensures SSE streams won't be terminated by client-wide timeout + if transport.TLSHandshakeTimeout != 10*time.Second { + t.Errorf("TLSHandshakeTimeout should be 10s, got %v", transport.TLSHandshakeTimeout) + } + if transport.ResponseHeaderTimeout != 30*time.Second { + t.Errorf("ResponseHeaderTimeout should be 30s, got %v", transport.ResponseHeaderTimeout) + } + + // Verify DialContext is set (not DialTimeout which is deprecated) + if transport.DialContext == nil { + t.Error("DialContext should be set for connection timeout") + } +} + +// TestContextTimeoutForConnect verifies that connect operations use timeout-scoped context. +func TestContextTimeoutForConnect(t *testing.T) { + // Test that a context with timeout properly cancels the operation + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + + // Create a listener that delays acceptance + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("Failed to create listener: %v", err) + } + defer listener.Close() + + // Start a goroutine that delays accepting connections + go func() { + time.Sleep(200 * time.Millisecond) + for { + conn, acceptErr := listener.Accept() + if acceptErr != nil { + return + } + conn.Close() + } + }() + + mgr := NewManager() + defer mgr.Close() + + start := time.Now() + err = mgr.ConnectServer(ctx, "timeout-test", config.MCPServerConfig{ + Type: "sse", + URL: "http://" + listener.Addr().String() + "/mcp", + Enabled: true, + }) + duration := time.Since(start) + + if err == nil { + t.Fatal("Expected connection to fail due to context timeout") + } + + // Should fail quickly due to context timeout, not hang + if duration > 500*time.Millisecond { + t.Fatalf("Operation took too long: %v, expected context timeout around 100ms", duration) + } +}