From c481cad3254fbcc52a2ed2cbedef0d75759728a5 Mon Sep 17 00:00:00 2001 From: suryakanta subudhi Date: Sat, 3 Oct 2026 02:35:37 +0530 Subject: [PATCH 1/2] mcp: bound server/discover probe on stdio transports Fixes #1332 --- docs/protocol.md | 4 +- internal/docs/protocol.src.md | 4 +- mcp/client.go | 32 +++++++- mcp/client_test.go | 143 ++++++++++++++++++++++++++++++++++ 4 files changed, 180 insertions(+), 3 deletions(-) diff --git a/docs/protocol.md b/docs/protocol.md index 06c1b77b3..a512d6fcf 100644 --- a/docs/protocol.md +++ b/docs/protocol.md @@ -126,7 +126,9 @@ request. Servers implementing `2026-07-28` MUST implement it. - **Client**: `Client.Connect` calls `server/discover` first and uses the result to negotiate a mutually supported version. If discovery fails or the server does not support the latest version, the client falls back to the - legacy `initialize` handshake. + legacy `initialize` handshake. On stdio transports, where an older server may + silently ignore requests before `initialize`, the client also falls back if + discovery goes unanswered for a few seconds. A server advertises and negotiates every protocol version the SDK supports; set [`ServerOptions.SupportedProtocolVersions`](https://pkg.go.dev/github.com/modelcontextprotocol/go-sdk/mcp#ServerOptions) diff --git a/internal/docs/protocol.src.md b/internal/docs/protocol.src.md index 7d6b01b5c..b4952fda8 100644 --- a/internal/docs/protocol.src.md +++ b/internal/docs/protocol.src.md @@ -70,7 +70,9 @@ request. Servers implementing `2026-07-28` MUST implement it. - **Client**: `Client.Connect` calls `server/discover` first and uses the result to negotiate a mutually supported version. If discovery fails or the server does not support the latest version, the client falls back to the - legacy `initialize` handshake. + legacy `initialize` handshake. On stdio transports, where an older server may + silently ignore requests before `initialize`, the client also falls back if + discovery goes unanswered for a few seconds. A server advertises and negotiates every protocol version the SDK supports; set [`ServerOptions.SupportedProtocolVersions`](https://pkg.go.dev/github.com/modelcontextprotocol/go-sdk/mcp#ServerOptions) diff --git a/mcp/client.go b/mcp/client.go index 9536bdd4c..68ac9e001 100644 --- a/mcp/client.go +++ b/mcp/client.go @@ -326,7 +326,9 @@ func (c *Client) Connect(ctx context.Context, t Transport, opts *ClientSessionOp // requested version but specifies which versions it supports, we negotiate // a mutually supported version and try again. for range 2 { - discRes, err := c.discover(discoverCtx, cs) + probeCtx, cancelProbe := discoverProbeContext(discoverCtx, cs.mcpConn) + discRes, err := c.discover(probeCtx, cs) + cancelProbe() if err == nil { cs.state.InitializeResult = discRes if hc, ok := cs.mcpConn.(clientConnection); ok { @@ -414,6 +416,34 @@ func (c *Client) Connect(ctx context.Context, t Transport, opts *ClientSessionOp return cs, nil } +var streamDiscoverTimeout = 3 * time.Second // mutable for testing + +// discoverProbeContext bounds the server/discover probe on newline-delimited +// stream connections (stdio, IO and in-memory transports, including custom +// transports that return their connection), where a handshake-era server may +// silently ignore requests it doesn't recognize before initialize. Half of any +// remaining deadline is kept for the initialize fallback. +func discoverProbeContext(ctx context.Context, conn Connection) (context.Context, context.CancelFunc) { + if !isStreamConn(conn) { + return ctx, func() {} + } + timeout := streamDiscoverTimeout + if deadline, ok := ctx.Deadline(); ok { + timeout = min(timeout, time.Until(deadline)/2) + } + return context.WithTimeout(ctx, timeout) +} + +func isStreamConn(conn Connection) bool { + switch conn := conn.(type) { + case *ioConn: + return true + case *loggingConn: + return isStreamConn(conn.delegate) + } + return false +} + // discover sends a SEP-2575 server/discover request to probe the server for // stateless protocol support. func (c *Client) discover(ctx context.Context, cs *ClientSession) (*InitializeResult, error) { diff --git a/mcp/client_test.go b/mcp/client_test.go index afe9ecc6f..70a984902 100644 --- a/mcp/client_test.go +++ b/mcp/client_test.go @@ -7,12 +7,15 @@ package mcp import ( + "bufio" "context" "encoding/json" "fmt" + "io" "log/slog" "sync/atomic" "testing" + "time" "github.com/google/go-cmp/cmp" "github.com/google/go-cmp/cmp/cmpopts" @@ -1094,6 +1097,146 @@ func TestClientConnectDiscover_UnsupportedVersionNegotiation(t *testing.T) { } } +func setStreamDiscoverTimeout(t *testing.T, d time.Duration) { + initial := streamDiscoverTimeout + streamDiscoverTimeout = d + t.Cleanup(func() { streamDiscoverTimeout = initial }) +} + +// serveSilentStdio emulates a handshake-era stdio server that answers only +// initialize and tools/list, silently dropping every other message. +func serveSilentStdio(r io.Reader, w io.Writer) { + sc := bufio.NewScanner(r) + for sc.Scan() { + var msg struct { + ID json.RawMessage `json:"id"` + Method string `json:"method"` + } + if json.Unmarshal(sc.Bytes(), &msg) != nil { + continue + } + var result any + switch msg.Method { + case methodInitialize: + result = map[string]any{ + "protocolVersion": protocolVersion20251125, + "capabilities": map[string]any{"tools": map[string]any{}}, + "serverInfo": map[string]any{"name": "silent", "version": "v1"}, + } + case methodListTools: + result = map[string]any{"tools": []any{}} + default: + continue + } + data, _ := json.Marshal(map[string]any{"jsonrpc": "2.0", "id": msg.ID, "result": result}) + if _, err := w.Write(append(data, '\n')); err != nil { + return + } + } +} + +// delegatingTransport is a user-defined Transport wrapping another one, as an +// instrumentation or filtering decorator would. +type delegatingTransport struct { + Transport +} + +func connectToSilentStdioServer(t *testing.T, ctx context.Context, wrap func(Transport) Transport) { + t.Helper() + cr, sw := io.Pipe() + sr, cw := io.Pipe() + go serveSilentStdio(sr, sw) + + client := NewClient(&Implementation{Name: "client", Version: "v1"}, nil) + cs, err := client.Connect(ctx, wrap(&IOTransport{Reader: cr, Writer: cw}), nil) + if err != nil { + t.Fatalf("client.Connect: %v", err) + } + defer cs.Close() + + if got, want := cs.InitializeResult().ProtocolVersion, protocolVersion20251125; got != want { + t.Errorf("InitializeResult.ProtocolVersion = %q, want %q", got, want) + } + if _, err := cs.ListTools(ctx, nil); err != nil { + t.Errorf("ListTools after fallback initialize: %v", err) + } +} + +// TestClientConnectDiscover_SilentStdioServer verifies that Connect falls back +// to initialize when a stdio server never answers server/discover (#1332). +func TestClientConnectDiscover_SilentStdioServer(t *testing.T) { + setStreamDiscoverTimeout(t, 50*time.Millisecond) + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + connectToSilentStdioServer(t, ctx, func(t Transport) Transport { return t }) +} + +// TestClientConnectDiscover_SilentStdioServerWrapped verifies the fallback when +// the stdio transport is wrapped, either by LoggingTransport or by a +// user-defined Transport. +func TestClientConnectDiscover_SilentStdioServerWrapped(t *testing.T) { + setStreamDiscoverTimeout(t, 50*time.Millisecond) + wrappers := map[string]func(Transport) Transport{ + "logging": func(t Transport) Transport { return &LoggingTransport{Transport: t, Writer: io.Discard} }, + "custom": func(t Transport) Transport { return delegatingTransport{t} }, + } + for name, wrap := range wrappers { + t.Run(name, func(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + connectToSilentStdioServer(t, ctx, wrap) + }) + } +} + +// TestClientConnectDiscover_SilentStdioServerShortDeadline verifies that the +// discover probe leaves part of a short caller deadline for initialize. +func TestClientConnectDiscover_SilentStdioServerShortDeadline(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + connectToSilentStdioServer(t, ctx, func(t Transport) Transport { return t }) +} + +// TestClientConnectDiscover_SlowStdioServer verifies that a modern stdio +// server answering server/discover after the probe timeout still connects +// through the initialize fallback, and its late reply is ignored. +func TestClientConnectDiscover_SlowStdioServer(t *testing.T) { + setStreamDiscoverTimeout(t, 50*time.Millisecond) + ctx := context.Background() + + server := NewServer(&Implementation{Name: "slow-server", Version: "v1"}, nil) + server.AddReceivingMiddleware(func(next MethodHandler) MethodHandler { + return func(ctx context.Context, method string, req Request) (Result, error) { + if method == methodDiscover { + time.Sleep(200 * time.Millisecond) + } + return next(ctx, method, req) + } + }) + + cr, sw := io.Pipe() + sr, cw := io.Pipe() + ss, err := server.Connect(ctx, &IOTransport{Reader: sr, Writer: sw}, nil) + if err != nil { + t.Fatalf("server.Connect: %v", err) + } + defer ss.Close() + + client := NewClient(&Implementation{Name: "client", Version: "v1"}, nil) + cs, err := client.Connect(ctx, &IOTransport{Reader: cr, Writer: cw}, nil) + if err != nil { + t.Fatalf("client.Connect: %v", err) + } + defer cs.Close() + + if got, want := cs.InitializeResult().ProtocolVersion, protocolVersion20251125; got != want { + t.Errorf("InitializeResult.ProtocolVersion = %q, want %q", got, want) + } + if _, err := cs.ListTools(ctx, nil); err != nil { + t.Errorf("ListTools after fallback initialize: %v", err) + } +} + // TestClientRejectsMissingRequiredParams checks that server-to-client messages // whose params are required by the schema are rejected when params is omitted // or null, instead of reaching handlers that dereference them. From fc75b3be0932dfa0294ce1d100de8cc311591f71 Mon Sep 17 00:00:00 2001 From: suryakanta subudhi Date: Tue, 6 Oct 2026 22:05:27 +0530 Subject: [PATCH 2/2] mcp: add ClientSessionOptions.DiscoverTimeout --- docs/protocol.md | 4 +- internal/docs/protocol.src.md | 4 +- mcp/client.go | 36 +++++++++---- mcp/client_test.go | 95 ++++++++++++++++++++--------------- 4 files changed, 86 insertions(+), 53 deletions(-) diff --git a/docs/protocol.md b/docs/protocol.md index a512d6fcf..bb17730e0 100644 --- a/docs/protocol.md +++ b/docs/protocol.md @@ -128,7 +128,9 @@ request. Servers implementing `2026-07-28` MUST implement it. fails or the server does not support the latest version, the client falls back to the legacy `initialize` handshake. On stdio transports, where an older server may silently ignore requests before `initialize`, the client also falls back if - discovery goes unanswered for a few seconds. + discovery goes unanswered for a few seconds. Set + [`ClientSessionOptions.DiscoverTimeout`](https://pkg.go.dev/github.com/modelcontextprotocol/go-sdk/mcp#ClientSessionOptions) + to change that bound or apply it to other transports. A server advertises and negotiates every protocol version the SDK supports; set [`ServerOptions.SupportedProtocolVersions`](https://pkg.go.dev/github.com/modelcontextprotocol/go-sdk/mcp#ServerOptions) diff --git a/internal/docs/protocol.src.md b/internal/docs/protocol.src.md index b4952fda8..bde774759 100644 --- a/internal/docs/protocol.src.md +++ b/internal/docs/protocol.src.md @@ -72,7 +72,9 @@ request. Servers implementing `2026-07-28` MUST implement it. fails or the server does not support the latest version, the client falls back to the legacy `initialize` handshake. On stdio transports, where an older server may silently ignore requests before `initialize`, the client also falls back if - discovery goes unanswered for a few seconds. + discovery goes unanswered for a few seconds. Set + [`ClientSessionOptions.DiscoverTimeout`](https://pkg.go.dev/github.com/modelcontextprotocol/go-sdk/mcp#ClientSessionOptions) + to change that bound or apply it to other transports. A server advertises and negotiates every protocol version the SDK supports; set [`ServerOptions.SupportedProtocolVersions`](https://pkg.go.dev/github.com/modelcontextprotocol/go-sdk/mcp#ServerOptions) diff --git a/mcp/client.go b/mcp/client.go index 68ac9e001..49bebb035 100644 --- a/mcp/client.go +++ b/mcp/client.go @@ -252,6 +252,15 @@ type ClientSessionOptions struct { // discover) request. If empty, the latest supported version is used. // The server may negotiate a different mutually supported version. ProtocolVersion string + // DiscoverTimeout bounds how long Connect waits for a server/discover + // response before falling back to the initialize handshake, since an + // older stdio server may silently ignore discover. + // + // If zero, a 3s bound applies on stdio, IO and in-memory transports, and + // discover is not bounded on other transports. A positive value applies + // on all transports, and a negative value disables the bound. The bound + // never exceeds half of the remaining Connect deadline. + DiscoverTimeout time.Duration } func (c *Client) capabilities(protocolVersion string) *ClientCapabilities { @@ -316,6 +325,10 @@ func (c *Client) Connect(ctx context.Context, t Transport, opts *ClientSessionOp if opts != nil && opts.ProtocolVersion != "" { protocolVersion = opts.ProtocolVersion } + var discoverTimeout time.Duration + if opts != nil { + discoverTimeout = opts.DiscoverTimeout + } if protocolVersion >= protocolVersion20260728 { // Per SEP-2575, try the stateless server/discover RPC first. If the server @@ -326,7 +339,7 @@ func (c *Client) Connect(ctx context.Context, t Transport, opts *ClientSessionOp // requested version but specifies which versions it supports, we negotiate // a mutually supported version and try again. for range 2 { - probeCtx, cancelProbe := discoverProbeContext(discoverCtx, cs.mcpConn) + probeCtx, cancelProbe := discoverProbeContext(discoverCtx, cs.mcpConn, discoverTimeout) discRes, err := c.discover(probeCtx, cs) cancelProbe() if err == nil { @@ -416,18 +429,21 @@ func (c *Client) Connect(ctx context.Context, t Transport, opts *ClientSessionOp return cs, nil } -var streamDiscoverTimeout = 3 * time.Second // mutable for testing +const defaultDiscoverTimeout = 3 * time.Second -// discoverProbeContext bounds the server/discover probe on newline-delimited -// stream connections (stdio, IO and in-memory transports, including custom -// transports that return their connection), where a handshake-era server may -// silently ignore requests it doesn't recognize before initialize. Half of any -// remaining deadline is kept for the initialize fallback. -func discoverProbeContext(ctx context.Context, conn Connection) (context.Context, context.CancelFunc) { - if !isStreamConn(conn) { +// discoverProbeContext bounds the server/discover probe as described by +// [ClientSessionOptions.DiscoverTimeout]. The default applies to +// newline-delimited stream connections (including custom transports that +// return one), where a handshake-era server may silently ignore requests it +// doesn't recognize before initialize. Half of any remaining deadline is kept +// for the initialize fallback. +func discoverProbeContext(ctx context.Context, conn Connection, timeout time.Duration) (context.Context, context.CancelFunc) { + if timeout == 0 && isStreamConn(conn) { + timeout = defaultDiscoverTimeout + } + if timeout <= 0 { return ctx, func() {} } - timeout := streamDiscoverTimeout if deadline, ok := ctx.Deadline(); ok { timeout = min(timeout, time.Until(deadline)/2) } diff --git a/mcp/client_test.go b/mcp/client_test.go index 70a984902..0f41139fa 100644 --- a/mcp/client_test.go +++ b/mcp/client_test.go @@ -1097,12 +1097,6 @@ func TestClientConnectDiscover_UnsupportedVersionNegotiation(t *testing.T) { } } -func setStreamDiscoverTimeout(t *testing.T, d time.Duration) { - initial := streamDiscoverTimeout - streamDiscoverTimeout = d - t.Cleanup(func() { streamDiscoverTimeout = initial }) -} - // serveSilentStdio emulates a handshake-era stdio server that answers only // initialize and tools/list, silently dropping every other message. func serveSilentStdio(r io.Reader, w io.Writer) { @@ -1141,14 +1135,31 @@ type delegatingTransport struct { Transport } -func connectToSilentStdioServer(t *testing.T, ctx context.Context, wrap func(Transport) Transport) { - t.Helper() +// connWrappingTransport is a user-defined Transport that also wraps the +// Connection, which hides the underlying stream connection from Connect. +type connWrappingTransport struct { + Transport +} + +func (t connWrappingTransport) Connect(ctx context.Context) (Connection, error) { + conn, err := t.Transport.Connect(ctx) + if err != nil { + return nil, err + } + return struct{ Connection }{conn}, nil +} + +func silentStdioTransport() *IOTransport { cr, sw := io.Pipe() sr, cw := io.Pipe() go serveSilentStdio(sr, sw) + return &IOTransport{Reader: cr, Writer: cw} +} +func connectWithFallback(t *testing.T, ctx context.Context, tr Transport, opts *ClientSessionOptions) { + t.Helper() client := NewClient(&Implementation{Name: "client", Version: "v1"}, nil) - cs, err := client.Connect(ctx, wrap(&IOTransport{Reader: cr, Writer: cw}), nil) + cs, err := client.Connect(ctx, tr, opts) if err != nil { t.Fatalf("client.Connect: %v", err) } @@ -1165,43 +1176,56 @@ func connectToSilentStdioServer(t *testing.T, ctx context.Context, wrap func(Tra // TestClientConnectDiscover_SilentStdioServer verifies that Connect falls back // to initialize when a stdio server never answers server/discover (#1332). func TestClientConnectDiscover_SilentStdioServer(t *testing.T) { - setStreamDiscoverTimeout(t, 50*time.Millisecond) ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() - connectToSilentStdioServer(t, ctx, func(t Transport) Transport { return t }) + connectWithFallback(t, ctx, silentStdioTransport(), &ClientSessionOptions{DiscoverTimeout: 50 * time.Millisecond}) } -// TestClientConnectDiscover_SilentStdioServerWrapped verifies the fallback when -// the stdio transport is wrapped, either by LoggingTransport or by a -// user-defined Transport. -func TestClientConnectDiscover_SilentStdioServerWrapped(t *testing.T) { - setStreamDiscoverTimeout(t, 50*time.Millisecond) - wrappers := map[string]func(Transport) Transport{ - "logging": func(t Transport) Transport { return &LoggingTransport{Transport: t, Writer: io.Discard} }, - "custom": func(t Transport) Transport { return delegatingTransport{t} }, +// TestClientConnectDiscover_SilentStdioServerShortDeadline verifies that the +// default discover probe leaves part of a short caller deadline for +// initialize, including when the stdio transport is wrapped. +func TestClientConnectDiscover_SilentStdioServerShortDeadline(t *testing.T) { + transports := map[string]func() Transport{ + "plain": func() Transport { return silentStdioTransport() }, + "logging": func() Transport { return &LoggingTransport{Transport: silentStdioTransport(), Writer: io.Discard} }, + "custom": func() Transport { return delegatingTransport{silentStdioTransport()} }, } - for name, wrap := range wrappers { + for name, newTransport := range transports { t.Run(name, func(t *testing.T) { - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + ctx, cancel := context.WithTimeout(context.Background(), time.Second) defer cancel() - connectToSilentStdioServer(t, ctx, wrap) + connectWithFallback(t, ctx, newTransport(), nil) }) } } -// TestClientConnectDiscover_SilentStdioServerShortDeadline verifies that the -// discover probe leaves part of a short caller deadline for initialize. -func TestClientConnectDiscover_SilentStdioServerShortDeadline(t *testing.T) { - ctx, cancel := context.WithTimeout(context.Background(), time.Second) +// TestClientConnectDiscover_ExplicitTimeout verifies that an explicit +// DiscoverTimeout also applies to connections the default doesn't recognize. +func TestClientConnectDiscover_ExplicitTimeout(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() - connectToSilentStdioServer(t, ctx, func(t Transport) Transport { return t }) + tr := connWrappingTransport{silentStdioTransport()} + connectWithFallback(t, ctx, tr, &ClientSessionOptions{DiscoverTimeout: 50 * time.Millisecond}) +} + +// TestClientConnectDiscover_NegativeTimeout verifies that a negative +// DiscoverTimeout disables the bound, so a silent server blocks Connect until +// the caller's deadline. +func TestClientConnectDiscover_NegativeTimeout(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 300*time.Millisecond) + defer cancel() + client := NewClient(&Implementation{Name: "client", Version: "v1"}, nil) + cs, err := client.Connect(ctx, silentStdioTransport(), &ClientSessionOptions{DiscoverTimeout: -1}) + if err == nil { + cs.Close() + t.Fatal("client.Connect succeeded, want an error from the caller's deadline") + } } // TestClientConnectDiscover_SlowStdioServer verifies that a modern stdio // server answering server/discover after the probe timeout still connects // through the initialize fallback, and its late reply is ignored. func TestClientConnectDiscover_SlowStdioServer(t *testing.T) { - setStreamDiscoverTimeout(t, 50*time.Millisecond) ctx := context.Background() server := NewServer(&Implementation{Name: "slow-server", Version: "v1"}, nil) @@ -1222,19 +1246,8 @@ func TestClientConnectDiscover_SlowStdioServer(t *testing.T) { } defer ss.Close() - client := NewClient(&Implementation{Name: "client", Version: "v1"}, nil) - cs, err := client.Connect(ctx, &IOTransport{Reader: cr, Writer: cw}, nil) - if err != nil { - t.Fatalf("client.Connect: %v", err) - } - defer cs.Close() - - if got, want := cs.InitializeResult().ProtocolVersion, protocolVersion20251125; got != want { - t.Errorf("InitializeResult.ProtocolVersion = %q, want %q", got, want) - } - if _, err := cs.ListTools(ctx, nil); err != nil { - t.Errorf("ListTools after fallback initialize: %v", err) - } + tr := &IOTransport{Reader: cr, Writer: cw} + connectWithFallback(t, ctx, tr, &ClientSessionOptions{DiscoverTimeout: 50 * time.Millisecond}) } // TestClientRejectsMissingRequiredParams checks that server-to-client messages