diff --git a/pkg/agent/agent.go b/pkg/agent/agent.go index 5749149c1..4f3a7e54a 100644 --- a/pkg/agent/agent.go +++ b/pkg/agent/agent.go @@ -139,7 +139,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 4d2886a80..37ad8dc2b 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 7bddbfc31..aa7e3ec65 100644 --- a/pkg/agent/steering.go +++ b/pkg/agent/steering.go @@ -390,8 +390,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 958927767..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" @@ -12,6 +13,7 @@ import ( "strings" "sync" "sync/atomic" + "time" "github.com/modelcontextprotocol/go-sdk/mcp" @@ -381,14 +383,25 @@ func connectServer( DisableStandaloneSSE: disableStandaloneSSE, } + // 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{ + Transport: baseTransport, + } + // 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: baseTransport, + headers: cfg.Headers, } logger.DebugCF("mcp", "Added custom HTTP headers", map[string]any{ @@ -397,6 +410,8 @@ func connectServer( }) } + sseTransport.HTTPClient = mcpHTTPClient + transport = sseTransport case "stdio": if cfg.Command == "" { @@ -457,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) } @@ -473,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) + } +}