diff --git a/README.md b/README.md index ded7586..22a3fed 100644 --- a/README.md +++ b/README.md @@ -73,7 +73,7 @@ func main() { client := mcp.NewClient(&mcp.Implementation{Name: "mcp-client", Version: "v1.0.0"}, nil) // Connect to a server over stdin/stdout - transport := mcp.NewCommandTransport(exec.Command("myserver")) + transport := &mcp.CommandTransport{Command: exec.Command("myserver")} session, err := client.Connect(ctx, transport, nil) if err != nil { log.Fatal(err) @@ -127,7 +127,7 @@ func main() { mcp.AddTool(server, &mcp.Tool{Name: "greet", Description: "say hi"}, SayHi) // Run the server over stdin/stdout, until the client disconnects - if err := server.Run(context.Background(), mcp.NewStdioTransport()); err != nil { + if err := server.Run(context.Background(), &mcp.StdioTransport{}); err != nil { log.Fatal(err) } } diff --git a/design/design.md b/design/design.md index 93dc552..65a2b61 100644 --- a/design/design.md +++ b/design/design.md @@ -100,11 +100,7 @@ The `CommandTransport` is the client side of the stdio transport, and connects b ```go // A CommandTransport is a [Transport] that runs a command and communicates // with it over stdin/stdout, using newline-delimited JSON. -type CommandTransport struct { /* unexported fields */ } - -// NewCommandTransport returns a [CommandTransport] that runs the given command -// and communicates with it over stdin/stdout. -func NewCommandTransport(cmd *exec.Command) *CommandTransport +type CommandTransport struct { Command *exec.Command } // Connect starts the command, and connects to it over stdin/stdout. func (*CommandTransport) Connect(ctx context.Context) (Connection, error) { @@ -115,9 +111,7 @@ The `StdioTransport` is the server side of the stdio transport, and connects by ```go // A StdioTransport is a [Transport] that communicates using newline-delimited // JSON over stdin/stdout. -type StdioTransport struct { /* unexported fields */ } - -func NewStdioTransport() *StdioTransport +type StdioTransport struct { } func (t *StdioTransport) Connect(context.Context) (Connection, error) ``` @@ -128,6 +122,8 @@ The HTTP transport APIs are even more asymmetrical. Since connections are initia Importantly, since they serve many connections, the HTTP handlers must accept a callback to get an MCP server for each new session. As described below, MCP servers can optionally connect to multiple clients. This allows customization of per-session servers: if the MCP server is stateless, the user can return the same MCP server for each connection. On the other hand, if any per-session customization is required, it is possible by returning a different `Server` instance for each connection. +Both the SSE and Streamable HTTP server transports are http.Handlers which serve messages to their associated connection. Consequently, they can be connected at most once. + ```go // SSEHTTPHandler is an http.Handler that serves SSE-based MCP sessions as defined by // the 2024-11-05 version of the MCP protocol. @@ -153,26 +149,10 @@ By default, the SSE handler creates messages endpoints with the `?sessionId=...` ```go // A SSEServerTransport is a logical SSE session created through a hanging GET // request. -// -// When connected, it returns the following [Connection] implementation: -// - Writes are SSE 'message' events to the GET response. -// - Reads are received from POSTs to the session endpoint, via -// [SSEServerTransport.ServeHTTP]. -// - Close terminates the hanging GET. -type SSEServerTransport struct { /* ... */ } - -// NewSSEServerTransport creates a new SSE transport for the given messages -// endpoint, and hanging GET response. -// -// Use [SSEServerTransport.Connect] to initiate the flow of messages. -// -// The transport is itself an [http.Handler]. It is the caller's responsibility -// to ensure that the resulting transport serves HTTP requests on the given -// session endpoint. -// -// Most callers should instead use an [SSEHandler], which transparently handles -// the delegation to SSEServerTransports. -func NewSSEServerTransport(endpoint string, w http.ResponseWriter) *SSEServerTransport +type SSEServerTransport struct { + Endpoint string + Response http.ResponseWriter +} // ServeHTTP handles POST requests to the transport endpoint. func (*SSEServerTransport) ServeHTTP(w http.ResponseWriter, req *http.Request) @@ -185,20 +165,14 @@ func (*SSEServerTransport) Connect(context.Context) (Connection, error) The SSE client transport is simpler, and hopefully self-explanatory. ```go -type SSEClientTransport struct { /* ... */ } - -// SSEClientTransportOptions provides options for the [NewSSEClientTransport] -// constructor. -type SSEClientTransportOptions struct { +type SSEClientTransport struct { + // Endpoint is the SSE endpoint to connect to. + Endpoint string // HTTPClient is the client to use for making HTTP requests. If nil, // http.DefaultClient is used. HTTPClient *http.Client } -// NewSSEClientTransport returns a new client transport that connects to the -// SSE server at the provided URL. -func NewSSEClientTransport(url string, opts *SSEClientTransportOptions) (*SSEClientTransport, error) - // Connect connects through the client endpoint. func (*SSEClientTransport) Connect(ctx context.Context) (Connection, error) ``` @@ -218,23 +192,22 @@ func (*StreamableHTTPHandler) Close() error // session ID, not an endpoint, along with the HTTP response for the request // that created the session. It is the caller's responsibility to delegate // requests to this session. -type StreamableServerTransport struct { /* ... */ } -func NewStreamableServerTransport(sessionID string) *StreamableServerTransport +type StreamableServerTransport struct { + // SessionID is the ID of this session. + SessionID string + // Storage for events, to enable stream resumption. + EventStore EventStore +} func (*StreamableServerTransport) ServeHTTP(w http.ResponseWriter, req *http.Request) func (*StreamableServerTransport) Connect(context.Context) (Connection, error) // The streamable client handles reconnection transparently to the user. -type StreamableClientTransport struct { /* ... */ } - -// StreamableClientTransportOptions provides options for the -// [NewStreamableClientTransport] constructor. -type StreamableClientTransportOptions struct { - // HTTPClient is the client to use for making HTTP requests. If nil, - // http.DefaultClient is used. - HTTPClient *http.Client +type StreamableClientTransport struct { + Endpoint string + HTTPClient *http.Client + ReconnectOptions *StreamableReconnectOptions } -func NewStreamableClientTransport(url string, opts *StreamableClientTransportOptions) *StreamableClientTransport func (*StreamableClientTransport) Connect(context.Context) (Connection, error) ``` @@ -257,8 +230,10 @@ func NewInMemoryTransports() (*InMemoryTransport, *InMemoryTransport) // A LoggingTransport is a [Transport] that delegates to another transport, // writing RPC logs to an io.Writer. -type LoggingTransport struct { /* ... */ } -func NewLoggingTransport(delegate Transport, w io.Writer) *LoggingTransport +type LoggingTransport struct { + Delegate Transport + Writer io.Writer +} ``` ### Protocol types @@ -358,7 +333,9 @@ Here's an example of these APIs from the client side: ```go client := mcp.NewClient(&mcp.Implementation{Name:"mcp-client", Version:"v1.0.0"}, nil) // Connect to a server over stdin/stdout -transport := mcp.NewCommandTransport(exec.Command("myserver")) +transport := &mcp.CommandTransport{ + Command: exec.Command("myserver"}, +} session, err := client.Connect(ctx, transport) if err != nil { ... } // Call a tool on the server. @@ -374,7 +351,7 @@ A server that can handle that client call would look like this: server := mcp.NewServer(&mcp.Implementation{Name:"greeter", Version:"v1.0.0"}, nil) mcp.AddTool(server, &mcp.Tool{Name: "greet", Description: "say hi"}, SayHi) // Run the server over stdin/stdout, until the client disconnects. -if err := server.Run(context.Background(), mcp.NewStdioTransport()); err != nil { +if err := server.Run(context.Background(), &mcp.StdioTransport{}); err != nil { log.Fatal(err) } ``` diff --git a/examples/client/listfeatures/main.go b/examples/client/listfeatures/main.go index 00aa459..755a4f9 100644 --- a/examples/client/listfeatures/main.go +++ b/examples/client/listfeatures/main.go @@ -41,7 +41,7 @@ func main() { ctx := context.Background() cmd := exec.Command(args[0], args[1:]...) client := mcp.NewClient(&mcp.Implementation{Name: "mcp-client", Version: "v1.0.0"}, nil) - cs, err := client.Connect(ctx, mcp.NewCommandTransport(cmd), nil) + cs, err := client.Connect(ctx, &mcp.CommandTransport{Command: cmd}, nil) if err != nil { log.Fatal(err) } diff --git a/examples/server/hello/main.go b/examples/server/hello/main.go index f01a6c9..00fb37a 100644 --- a/examples/server/hello/main.go +++ b/examples/server/hello/main.go @@ -58,7 +58,7 @@ func main() { log.Printf("MCP handler listening at %s", *httpAddr) http.ListenAndServe(*httpAddr, handler) } else { - t := mcp.NewLoggingTransport(mcp.NewStdioTransport(), os.Stderr) + t := &mcp.LoggingTransport{Transport: &mcp.StdioTransport{}, Writer: os.Stderr} if err := server.Run(context.Background(), t); err != nil { log.Printf("Server failed: %v", err) } diff --git a/examples/server/memory/main.go b/examples/server/memory/main.go index 61ab106..99d109a 100644 --- a/examples/server/memory/main.go +++ b/examples/server/memory/main.go @@ -137,7 +137,7 @@ func main() { log.Printf("MCP handler listening at %s", *httpAddr) http.ListenAndServe(*httpAddr, handler) } else { - t := mcp.NewLoggingTransport(mcp.NewStdioTransport(), os.Stderr) + t := &mcp.LoggingTransport{Transport: &mcp.StdioTransport{}, Writer: os.Stderr} if err := server.Run(context.Background(), t); err != nil { log.Printf("Server failed: %v", err) } diff --git a/examples/server/sequentialthinking/main.go b/examples/server/sequentialthinking/main.go index 4830bb0..e9cb594 100644 --- a/examples/server/sequentialthinking/main.go +++ b/examples/server/sequentialthinking/main.go @@ -536,7 +536,7 @@ func main() { log.Fatal(err) } } else { - t := mcp.NewLoggingTransport(mcp.NewStdioTransport(), os.Stderr) + t := &mcp.LoggingTransport{Transport: &mcp.StdioTransport{}, Writer: os.Stderr} if err := server.Run(context.Background(), t); err != nil { log.Printf("Server failed: %v", err) } diff --git a/internal/readme/client/client.go b/internal/readme/client/client.go index 57ec54f..e2794f8 100644 --- a/internal/readme/client/client.go +++ b/internal/readme/client/client.go @@ -20,7 +20,7 @@ func main() { client := mcp.NewClient(&mcp.Implementation{Name: "mcp-client", Version: "v1.0.0"}, nil) // Connect to a server over stdin/stdout - transport := mcp.NewCommandTransport(exec.Command("myserver")) + transport := &mcp.CommandTransport{Command: exec.Command("myserver")} session, err := client.Connect(ctx, transport, nil) if err != nil { log.Fatal(err) diff --git a/internal/readme/server/server.go b/internal/readme/server/server.go index 8901e77..3746e19 100644 --- a/internal/readme/server/server.go +++ b/internal/readme/server/server.go @@ -28,7 +28,7 @@ func main() { mcp.AddTool(server, &mcp.Tool{Name: "greet", Description: "say hi"}, SayHi) // Run the server over stdin/stdout, until the client disconnects - if err := server.Run(context.Background(), mcp.NewStdioTransport()); err != nil { + if err := server.Run(context.Background(), &mcp.StdioTransport{}); err != nil { log.Fatal(err) } } diff --git a/mcp/cmd.go b/mcp/cmd.go index 163bb0c..5ec8c9e 100644 --- a/mcp/cmd.go +++ b/mcp/cmd.go @@ -16,7 +16,7 @@ import ( // A CommandTransport is a [Transport] that runs a command and communicates // with it over stdin/stdout, using newline-delimited JSON. type CommandTransport struct { - cmd *exec.Cmd + Command *exec.Cmd } // NewCommandTransport returns a [CommandTransport] that runs the given command @@ -24,25 +24,29 @@ type CommandTransport struct { // // The resulting transport takes ownership of the command, starting it during // [CommandTransport.Connect], and stopping it when the connection is closed. +// +// Deprecated: use a CommandTransport literal. +// +//go:fix inline func NewCommandTransport(cmd *exec.Cmd) *CommandTransport { - return &CommandTransport{cmd} + return &CommandTransport{Command: cmd} } // Connect starts the command, and connects to it over stdin/stdout. func (t *CommandTransport) Connect(ctx context.Context) (Connection, error) { - stdout, err := t.cmd.StdoutPipe() + stdout, err := t.Command.StdoutPipe() if err != nil { return nil, err } stdout = io.NopCloser(stdout) // close the connection by closing stdin, not stdout - stdin, err := t.cmd.StdinPipe() + stdin, err := t.Command.StdinPipe() if err != nil { return nil, err } - if err := t.cmd.Start(); err != nil { + if err := t.Command.Start(); err != nil { return nil, err } - return newIOConn(&pipeRWC{t.cmd, stdout, stdin}), nil + return newIOConn(&pipeRWC{t.Command, stdout, stdin}), nil } // A pipeRWC is an io.ReadWriteCloser that communicates with a subprocess over diff --git a/mcp/cmd_test.go b/mcp/cmd_test.go index a021bc8..6c3a1a7 100644 --- a/mcp/cmd_test.go +++ b/mcp/cmd_test.go @@ -49,7 +49,7 @@ func runServer() { server := mcp.NewServer(testImpl, nil) mcp.AddTool(server, &mcp.Tool{Name: "greet", Description: "say hi"}, SayHi) - if err := server.Run(ctx, mcp.NewStdioTransport()); err != nil { + if err := server.Run(ctx, &mcp.StdioTransport{}); err != nil { log.Fatal(err) } } @@ -59,7 +59,7 @@ func runCancelContextServer() { defer done() server := mcp.NewServer(testImpl, nil) - if err := server.Run(ctx, mcp.NewStdioTransport()); err != nil { + if err := server.Run(ctx, &mcp.StdioTransport{}); err != nil { log.Fatal(err) } } @@ -116,7 +116,7 @@ func TestServerInterrupt(t *testing.T) { cmd := createServerCommand(t, "default") client := mcp.NewClient(testImpl, nil) - _, err := client.Connect(ctx, mcp.NewCommandTransport(cmd), nil) + _, err := client.Connect(ctx, &mcp.CommandTransport{Command: cmd}, nil) if err != nil { t.Fatal(err) } @@ -198,7 +198,7 @@ func TestCmdTransport(t *testing.T) { cmd := createServerCommand(t, "default") client := mcp.NewClient(&mcp.Implementation{Name: "client", Version: "v0.0.1"}, nil) - session, err := client.Connect(ctx, mcp.NewCommandTransport(cmd), nil) + session, err := client.Connect(ctx, &mcp.CommandTransport{Command: cmd}, nil) if err != nil { t.Fatal(err) } diff --git a/mcp/mcp_test.go b/mcp/mcp_test.go index 9e9a6a3..9e3eccd 100644 --- a/mcp/mcp_test.go +++ b/mcp/mcp_test.go @@ -774,7 +774,7 @@ func TestNoJSONNull(t *testing.T) { // Collect logs, to sanity check that we don't write JSON null anywhere. var logbuf safeBuffer - ct = NewLoggingTransport(ct, &logbuf) + ct = &LoggingTransport{Transport: ct, Writer: &logbuf} s := NewServer(testImpl, nil) ss, err := s.Connect(ctx, st, nil) diff --git a/mcp/sse.go b/mcp/sse.go index f74a3fb..b7f0d4e 100644 --- a/mcp/sse.go +++ b/mcp/sse.go @@ -74,47 +74,66 @@ func NewSSEHandler(getServer func(request *http.Request) *Server) *SSEHandler { // A SSEServerTransport is a logical SSE session created through a hanging GET // request. // +// Use [SSEServerTransport.Connect] to initiate the flow of messages. +// // When connected, it returns the following [Connection] implementation: // - Writes are SSE 'message' events to the GET response. // - Reads are received from POSTs to the session endpoint, via // [SSEServerTransport.ServeHTTP]. // - Close terminates the hanging GET. -type SSEServerTransport struct { - endpoint string - incoming chan jsonrpc.Message // queue of incoming messages; never closed - - // We must guard both pushes to the incoming queue and writes to the response - // writer, because incoming POST requests are arbitrarily concurrent and we - // need to ensure we don't write push to the queue, or write to the - // ResponseWriter, after the session GET request exits. - mu sync.Mutex - w http.ResponseWriter // the hanging response body - closed bool // set when the stream is closed - done chan struct{} // closed when the connection is closed -} - -// NewSSEServerTransport creates a new SSE transport for the given messages -// endpoint, and hanging GET response. -// -// Use [SSEServerTransport.Connect] to initiate the flow of messages. // // The transport is itself an [http.Handler]. It is the caller's responsibility // to ensure that the resulting transport serves HTTP requests on the given // session endpoint. // +// Each SSEServerTransport may be connected (via [Server.Connect]) at most +// once, since [SSEServerTransport.ServeHTTP] serves messages to the connected +// session. +// // Most callers should instead use an [SSEHandler], which transparently handles // the delegation to SSEServerTransports. +type SSEServerTransport struct { + // Endpoint is the endpoint for this session, where the client can POST + // messages. + Endpoint string + + // Response is the hanging response body to the incoming GET request. + Response http.ResponseWriter + + // incoming is the queue of incoming messages. + // It is never closed, and by convention, incoming is non-nil if and only if + // the transport is connected. + incoming chan jsonrpc.Message + + // We must guard both pushes to the incoming queue and writes to the response + // writer, because incoming POST requests are arbitrarily concurrent and we + // need to ensure we don't write push to the queue, or write to the + // ResponseWriter, after the session GET request exits. + mu sync.Mutex // also guards writes to Response + closed bool // set when the stream is closed + done chan struct{} // closed when the connection is closed +} + +// NewSSEServerTransport creates a new SSE transport for the given messages +// endpoint, and hanging GET response. +// +// Deprecated: use an SSEServerTransport literal. +// +//go:fix inline func NewSSEServerTransport(endpoint string, w http.ResponseWriter) *SSEServerTransport { return &SSEServerTransport{ - endpoint: endpoint, - w: w, - incoming: make(chan jsonrpc.Message, 100), - done: make(chan struct{}), + Endpoint: endpoint, + Response: w, } } // ServeHTTP handles POST requests to the transport endpoint. func (t *SSEServerTransport) ServeHTTP(w http.ResponseWriter, req *http.Request) { + if t.incoming == nil { + http.Error(w, "session not connected", http.StatusInternalServerError) + return + } + // Read and parse the message. data, err := io.ReadAll(req.Body) if err != nil { @@ -146,12 +165,15 @@ func (t *SSEServerTransport) ServeHTTP(w http.ResponseWriter, req *http.Request) // Connect sends the 'endpoint' event to the client. // See [SSEServerTransport] for more details on the [Connection] implementation. func (t *SSEServerTransport) Connect(context.Context) (Connection, error) { - t.mu.Lock() - _, err := writeEvent(t.w, Event{ + if t.incoming != nil { + return nil, fmt.Errorf("already connected") + } + t.incoming = make(chan jsonrpc.Message, 100) + t.done = make(chan struct{}) + _, err := writeEvent(t.Response, Event{ Name: "endpoint", - Data: []byte(t.endpoint), + Data: []byte(t.Endpoint), }) - t.mu.Unlock() if err != nil { return nil, err } @@ -203,7 +225,7 @@ func (h *SSEHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) { return } - transport := NewSSEServerTransport(endpoint.RequestURI(), w) + transport := &SSEServerTransport{Endpoint: endpoint.RequestURI(), Response: w} // The session is terminated when the request exits. h.mu.Lock() @@ -279,7 +301,7 @@ func (s *sseServerConn) Write(ctx context.Context, msg jsonrpc.Message) error { return io.EOF } - _, err = writeEvent(s.t.w, Event{Name: "message", Data: data}) + _, err = writeEvent(s.t.Response, Event{Name: "message", Data: data}) return err } @@ -304,12 +326,18 @@ func (s *sseServerConn) Close() error { // // https://modelcontextprotocol.io/specification/2024-11-05/basic/transports type SSEClientTransport struct { - sseEndpoint *url.URL - opts SSEClientTransportOptions + // Endpoint is the SSE endpoint to connect to. + Endpoint string + + // HTTPClient is the client to use for making HTTP requests. If nil, + // http.DefaultClient is used. + HTTPClient *http.Client } // SSEClientTransportOptions provides options for the [NewSSEClientTransport] // constructor. +// +// Deprecated: use an SSEClientTransport literal. type SSEClientTransportOptions struct { // HTTPClient is the client to use for making HTTP requests. If nil, // http.DefaultClient is used. @@ -319,28 +347,28 @@ type SSEClientTransportOptions struct { // NewSSEClientTransport returns a new client transport that connects to the // SSE server at the provided URL. // -// NewSSEClientTransport panics if the given URL is invalid. -func NewSSEClientTransport(baseURL string, opts *SSEClientTransportOptions) *SSEClientTransport { - url, err := url.Parse(baseURL) - if err != nil { - panic(fmt.Sprintf("invalid base url: %v", err)) - } - t := &SSEClientTransport{ - sseEndpoint: url, - } +// Deprecated: use an SSEClientTransport literal. +// +//go:fix inline +func NewSSEClientTransport(endpoint string, opts *SSEClientTransportOptions) *SSEClientTransport { + t := &SSEClientTransport{Endpoint: endpoint} if opts != nil { - t.opts = *opts + t.HTTPClient = opts.HTTPClient } return t } // Connect connects through the client endpoint. func (c *SSEClientTransport) Connect(ctx context.Context) (Connection, error) { - req, err := http.NewRequestWithContext(ctx, "GET", c.sseEndpoint.String(), nil) + parsedURL, err := url.Parse(c.Endpoint) + if err != nil { + return nil, fmt.Errorf("invalid endpoint: %v", err) + } + req, err := http.NewRequestWithContext(ctx, "GET", c.Endpoint, nil) if err != nil { return nil, err } - httpClient := c.opts.HTTPClient + httpClient := c.HTTPClient if httpClient == nil { httpClient = http.DefaultClient } @@ -362,7 +390,7 @@ func (c *SSEClientTransport) Connect(ctx context.Context) (Connection, error) { return nil, fmt.Errorf("first event is %q, want %q", evt.Name, "endpoint") } raw := string(evt.Data) - return c.sseEndpoint.Parse(raw) + return parsedURL.Parse(raw) }() if err != nil { resp.Body.Close() @@ -372,7 +400,6 @@ func (c *SSEClientTransport) Connect(ctx context.Context) (Connection, error) { // From here on, the stream takes ownership of resp.Body. s := &sseClientConn{ client: httpClient, - sseEndpoint: c.sseEndpoint, msgEndpoint: msgEndpoint, incoming: make(chan []byte, 100), body: resp.Body, @@ -404,7 +431,6 @@ func (c *SSEClientTransport) Connect(ctx context.Context) (Connection, error) { // - Close terminates the GET request. type sseClientConn struct { client *http.Client // HTTP client to use for requests - sseEndpoint *url.URL // SSE endpoint for the GET msgEndpoint *url.URL // session endpoint for POSTs incoming chan []byte // queue of incoming messages diff --git a/mcp/sse_example_test.go b/mcp/sse_example_test.go index 9a7c8ae..cf1e75d 100644 --- a/mcp/sse_example_test.go +++ b/mcp/sse_example_test.go @@ -35,7 +35,7 @@ func ExampleSSEHandler() { defer httpServer.Close() ctx := context.Background() - transport := mcp.NewSSEClientTransport(httpServer.URL, nil) + transport := &mcp.SSEClientTransport{Endpoint: httpServer.URL} client := mcp.NewClient(&mcp.Implementation{Name: "test", Version: "v1.0.0"}, nil) cs, err := client.Connect(ctx, transport, nil) if err != nil { diff --git a/mcp/sse_test.go b/mcp/sse_test.go index 79cfacc..408e92e 100644 --- a/mcp/sse_test.go +++ b/mcp/sse_test.go @@ -44,9 +44,10 @@ func TestSSEServer(t *testing.T) { }), } - clientTransport := NewSSEClientTransport(httpServer.URL, &SSEClientTransportOptions{ + clientTransport := &SSEClientTransport{ + Endpoint: httpServer.URL, HTTPClient: customClient, - }) + } c := NewClient(testImpl, nil) cs, err := c.Connect(ctx, clientTransport, nil) diff --git a/mcp/streamable.go b/mcp/streamable.go index 72ac7e8..048b99a 100644 --- a/mcp/streamable.go +++ b/mcp/streamable.go @@ -38,8 +38,8 @@ type StreamableHTTPHandler struct { getServer func(*http.Request) *Server opts StreamableHTTPOptions - sessionsMu sync.Mutex - sessions map[string]*StreamableServerTransport // keyed by IDs (from Mcp-Session-Id header) + mu sync.Mutex + transports map[string]*StreamableServerTransport // keyed by IDs (from Mcp-Session-Id header) } // StreamableHTTPOptions configures the StreamableHTTPHandler. @@ -50,12 +50,10 @@ type StreamableHTTPOptions struct { // meaning it is not persisted and no session validation is performed. GetSessionID func() string - // TODO: support configurable session ID generation (?) // TODO: support session retention (?) - // transportOptions sets the streamable server transport options to use when - // establishing a new session. - transportOptions *StreamableServerTransportOptions + // jsonResponse is forwarded to StreamableServerTransport.jsonResponse. + jsonResponse bool } // NewStreamableHTTPHandler returns a new [StreamableHTTPHandler]. @@ -65,8 +63,8 @@ type StreamableHTTPOptions struct { // If getServer returns nil, a 400 Bad Request will be served. func NewStreamableHTTPHandler(getServer func(*http.Request) *Server, opts *StreamableHTTPOptions) *StreamableHTTPHandler { h := &StreamableHTTPHandler{ - getServer: getServer, - sessions: make(map[string]*StreamableServerTransport), + getServer: getServer, + transports: make(map[string]*StreamableServerTransport), } if opts != nil { h.opts = *opts @@ -85,12 +83,12 @@ func NewStreamableHTTPHandler(getServer func(*http.Request) *Server, opts *Strea // Should we allow passing in a session store? That would allow the handler to // be stateless. func (h *StreamableHTTPHandler) closeAll() { - h.sessionsMu.Lock() - defer h.sessionsMu.Unlock() - for _, s := range h.sessions { - s.Close() + h.mu.Lock() + defer h.mu.Unlock() + for _, s := range h.transports { + s.connection.Close() } - h.sessions = nil + h.transports = nil } func (h *StreamableHTTPHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) { @@ -119,9 +117,9 @@ func (h *StreamableHTTPHandler) ServeHTTP(w http.ResponseWriter, req *http.Reque var session *StreamableServerTransport if id := req.Header.Get(sessionIDHeader); id != "" { - h.sessionsMu.Lock() - session, _ = h.sessions[id] - h.sessionsMu.Unlock() + h.mu.Lock() + session, _ = h.transports[id] + h.mu.Unlock() if session == nil { http.Error(w, "session not found", http.StatusNotFound) return @@ -136,10 +134,10 @@ func (h *StreamableHTTPHandler) ServeHTTP(w http.ResponseWriter, req *http.Reque http.Error(w, "DELETE requires an Mcp-Session-Id header", http.StatusBadRequest) return } - h.sessionsMu.Lock() - delete(h.sessions, session.sessionID) - h.sessionsMu.Unlock() - session.Close() + h.mu.Lock() + delete(h.transports, session.SessionID) + h.mu.Unlock() + session.connection.Close() w.WriteHeader(http.StatusNoContent) return } @@ -164,7 +162,7 @@ func (h *StreamableHTTPHandler) ServeHTTP(w http.ResponseWriter, req *http.Reque return } sessionID := h.opts.GetSessionID() - s := NewStreamableServerTransport(sessionID, h.opts.transportOptions) + s := &StreamableServerTransport{SessionID: sessionID, jsonResponse: h.opts.jsonResponse} // To support stateless mode, we initialize the session with a default // state, so that it doesn't reject subsequent requests. @@ -189,9 +187,9 @@ func (h *StreamableHTTPHandler) ServeHTTP(w http.ResponseWriter, req *http.Reque // Stateless mode: close the session when the request exits. defer ss.Close() // close the fake session after handling the request } else { - h.sessionsMu.Lock() - h.sessions[s.sessionID] = s - h.sessionsMu.Unlock() + h.mu.Lock() + h.transports[s.SessionID] = s + h.mu.Unlock() } session = s } @@ -199,10 +197,34 @@ func (h *StreamableHTTPHandler) ServeHTTP(w http.ResponseWriter, req *http.Reque session.ServeHTTP(w, req) } +// StreamableServerTransportOptions configures the stramable server transport. +// +// Deprecated: use a StreamableServerTransport literal. type StreamableServerTransportOptions struct { // Storage for events, to enable stream resumption. // If nil, a [MemoryEventStore] with the default maximum size will be used. EventStore EventStore +} + +// A StreamableServerTransport implements the server side of the MCP streamable +// transport. +// +// Each StreamableServerTransport may be connected (via [Server.Connect]) at +// most once, since [StreamableServerTransport.ServeHTTP] serves messages to +// the connected session. +type StreamableServerTransport struct { + // SessionID is the ID of this session. + // + // If SessionID is the empty string, this is a 'stateless' session, which has + // limited ability to communicate with the client. Otherwise, the session ID + // must be globally unique, that is, different from any other session ID + // anywhere, past and future. (We recommend using a crypto random number + // generator to produce one, as with [crypto/rand.Text].) + SessionID string + + // Storage for events, to enable stream resumption. + // If nil, a [MemoryEventStore] with the default maximum size will be used. + EventStore EventStore // jsonResponse, if set, tells the server to prefer to respond to requests // using application/json responses rather than text/event-stream. @@ -212,22 +234,36 @@ type StreamableServerTransportOptions struct { // requests made within the context of a server request will be sent to the // hanging GET request, if any. jsonResponse bool + + // connection is non-nil if and only if the transport has been connected. + connection *streamableServerConn } // NewStreamableServerTransport returns a new [StreamableServerTransport] with // the given session ID and options. -// The session ID must be globally unique, that is, different from any other -// session ID anywhere, past and future. (We recommend using a crypto random number -// generator to produce one, as with [crypto/rand.Text].) // -// A StreamableServerTransport implements the server-side of the streamable -// transport. +// Deprecated: use a StreamableServerTransport literal. +// +//go:fix inline. func NewStreamableServerTransport(sessionID string, opts *StreamableServerTransportOptions) *StreamableServerTransport { - if opts == nil { - opts = &StreamableServerTransportOptions{} - } t := &StreamableServerTransport{ - sessionID: sessionID, + SessionID: sessionID, + } + if opts != nil { + t.EventStore = opts.EventStore + } + return t +} + +// Connect implements the [Transport] interface. +func (t *StreamableServerTransport) Connect(context.Context) (Connection, error) { + if t.connection != nil { + return nil, fmt.Errorf("transport already connected") + } + t.connection = &streamableServerConn{ + sessionID: t.SessionID, + eventStore: t.EventStore, + jsonResponse: t.jsonResponse, incoming: make(chan jsonrpc.Message, 10), done: make(chan struct{}), streams: make(map[StreamID]*stream), @@ -237,29 +273,23 @@ func NewStreamableServerTransport(sessionID string, opts *StreamableServerTransp // // It is always text/event-stream, since it must carry arbitrarily many // messages. - t.streams[0] = newStream(0, false) - if opts != nil { - t.opts = *opts + t.connection.streams[0] = newStream(0, false) + if t.connection.eventStore == nil { + t.connection.eventStore = NewMemoryEventStore(nil) } - if t.opts.EventStore == nil { - t.opts.EventStore = NewMemoryEventStore(nil) - } - return t + return t.connection, nil } -func (t *StreamableServerTransport) SessionID() string { - return t.sessionID -} +type streamableServerConn struct { + sessionID string + jsonResponse bool + eventStore EventStore -// A StreamableServerTransport implements the [Transport] interface for a -// single session. -type StreamableServerTransport struct { lastStreamID atomic.Int64 // last stream ID used, atomically incremented - sessionID string - opts StreamableServerTransportOptions - incoming chan jsonrpc.Message // messages from the client to the server - done chan struct{} + opts StreamableServerTransportOptions + incoming chan jsonrpc.Message // messages from the client to the server + done chan struct{} mu sync.Mutex // Sessions are closed exactly once. @@ -289,6 +319,10 @@ type StreamableServerTransport struct { requestStreams map[jsonrpc.ID]StreamID } +func (c *streamableServerConn) SessionID() string { + return c.sessionID +} + // A stream is a single logical stream of SSE events within a server session. // A stream begins with a client request, or with a client GET that has // no Last-Event-ID header. @@ -349,13 +383,6 @@ func signalChanPtr() *chan struct{} { // [ServerSession]. type StreamID int64 -// Connect implements the [Transport] interface. -// -// TODO(rfindley): Connect should return a new object. (Why?) -func (s *StreamableServerTransport) Connect(context.Context) (Connection, error) { - return s, nil -} - // We track the incoming request ID inside the handler context using // idContextValue, so that notifications and server->client calls that occur in // the course of handling incoming requests are correlated with the incoming @@ -381,15 +408,20 @@ type idContextKey struct{} // ServeHTTP handles a single HTTP request for the session. func (t *StreamableServerTransport) ServeHTTP(w http.ResponseWriter, req *http.Request) { + if t.connection == nil { + http.Error(w, "transport not connected", http.StatusInternalServerError) + return + } switch req.Method { case http.MethodGet: - t.serveGET(w, req) + t.connection.serveGET(w, req) case http.MethodPost: - t.servePOST(w, req) + t.connection.servePOST(w, req) default: // Should not be reached, as this is checked in StreamableHTTPHandler.ServeHTTP. w.Header().Set("Allow", "GET, POST") http.Error(w, "unsupported method", http.StatusMethodNotAllowed) + return } } @@ -397,7 +429,7 @@ func (t *StreamableServerTransport) ServeHTTP(w http.ResponseWriter, req *http.R // message parsed from the Last-Event-ID header. // // It returns an HTTP status code and error message. -func (t *StreamableServerTransport) serveGET(w http.ResponseWriter, req *http.Request) { +func (c *streamableServerConn) serveGET(w http.ResponseWriter, req *http.Request) { // connID 0 corresponds to the default GET request. id := StreamID(0) // By default, we haven't seen a last index. Since indices start at 0, we represent @@ -414,9 +446,9 @@ func (t *StreamableServerTransport) serveGET(w http.ResponseWriter, req *http.Re } } - t.mu.Lock() - stream, ok := t.streams[id] - t.mu.Unlock() + c.mu.Lock() + stream, ok := c.streams[id] + c.mu.Unlock() if !ok { http.Error(w, "unknown stream", http.StatusBadRequest) return @@ -428,7 +460,7 @@ func (t *StreamableServerTransport) serveGET(w http.ResponseWriter, req *http.Re } defer stream.signal.Store(nil) persistent := id == 0 // Only the special stream 0 is a hanging get. - t.respondSSE(stream, w, req, lastIdx, persistent) + c.respondSSE(stream, w, req, lastIdx, persistent) } // servePOST handles an incoming message, and replies with either an outgoing @@ -436,7 +468,7 @@ func (t *StreamableServerTransport) serveGET(w http.ResponseWriter, req *http.Re // jsonResponse option is set. // // It returns an HTTP status code and error message. -func (t *StreamableServerTransport) servePOST(w http.ResponseWriter, req *http.Request) { +func (c *streamableServerConn) servePOST(w http.ResponseWriter, req *http.Request) { if len(req.Header.Values("Last-Event-ID")) > 0 { http.Error(w, "can't send Last-Event-ID for POST request", http.StatusBadRequest) return @@ -485,20 +517,20 @@ func (t *StreamableServerTransport) servePOST(w http.ResponseWriter, req *http.R // notifications or server->client requests made in the course of handling. // Update accounting for this incoming payload. if len(requests) > 0 { - stream = newStream(StreamID(t.lastStreamID.Add(1)), t.opts.jsonResponse) - t.mu.Lock() - t.streams[stream.id] = stream + stream = newStream(StreamID(c.lastStreamID.Add(1)), c.jsonResponse) + c.mu.Lock() + c.streams[stream.id] = stream stream.requests = requests for reqID := range requests { - t.requestStreams[reqID] = stream.id + c.requestStreams[reqID] = stream.id } - t.mu.Unlock() + c.mu.Unlock() stream.signal.Store(signalChanPtr()) } // Publish incoming messages. for _, msg := range incoming { - t.incoming <- msg + c.incoming <- msg } if stream == nil { @@ -507,22 +539,22 @@ func (t *StreamableServerTransport) servePOST(w http.ResponseWriter, req *http.R } if stream.jsonResponse { - t.respondJSON(stream, w, req) + c.respondJSON(stream, w, req) } else { - t.respondSSE(stream, w, req, -1, false) + c.respondSSE(stream, w, req, -1, false) } } -func (t *StreamableServerTransport) respondJSON(stream *stream, w http.ResponseWriter, req *http.Request) { +func (c *streamableServerConn) respondJSON(stream *stream, w http.ResponseWriter, req *http.Request) { w.Header().Set("Cache-Control", "no-cache, no-transform") w.Header().Set("Content-Type", "application/json") - if t.sessionID != "" { - w.Header().Set(sessionIDHeader, t.sessionID) + if c.sessionID != "" { + w.Header().Set(sessionIDHeader, c.sessionID) } var msgs []json.RawMessage ctx := req.Context() - for msg, ok := range t.messages(ctx, stream, false) { + for msg, ok := range c.messages(ctx, stream, false) { if !ok { if ctx.Err() != nil { w.WriteHeader(http.StatusNoContent) @@ -550,15 +582,15 @@ func (t *StreamableServerTransport) respondJSON(stream *stream, w http.ResponseW } // lastIndex is the index of the last seen event if resuming, else -1. -func (t *StreamableServerTransport) respondSSE(stream *stream, w http.ResponseWriter, req *http.Request, lastIndex int, persistent bool) { +func (c *streamableServerConn) respondSSE(stream *stream, w http.ResponseWriter, req *http.Request, lastIndex int, persistent bool) { writes := 0 // Accept checked in [StreamableHTTPHandler] w.Header().Set("Cache-Control", "no-cache, no-transform") w.Header().Set("Content-Type", "text/event-stream") // Accept checked in [StreamableHTTPHandler] w.Header().Set("Connection", "keep-alive") - if t.sessionID != "" { - w.Header().Set(sessionIDHeader, t.sessionID) + if c.sessionID != "" { + w.Header().Set(sessionIDHeader, c.sessionID) } // write one event containing data. @@ -588,7 +620,7 @@ func (t *StreamableServerTransport) respondSSE(stream *stream, w http.ResponseWr if lastIndex >= 0 { // Resume. - for data, err := range t.opts.EventStore.After(req.Context(), t.SessionID(), stream.id, lastIndex) { + for data, err := range c.eventStore.After(req.Context(), c.SessionID(), stream.id, lastIndex) { if err != nil { // TODO: reevaluate these status codes. // Maybe distinguish between storage errors, which are 500s, and missing @@ -610,7 +642,7 @@ func (t *StreamableServerTransport) respondSSE(stream *stream, w http.ResponseWr // Repeatedly collect pending outgoing events and send them. ctx := req.Context() - for msg, ok := range t.messages(ctx, stream, persistent) { + for msg, ok := range c.messages(ctx, stream, persistent) { if !ok { if ctx.Err() != nil && writes == 0 { // This probably doesn't matter, but respond with NoContent if the client disconnected. @@ -620,7 +652,7 @@ func (t *StreamableServerTransport) respondSSE(stream *stream, w http.ResponseWr } return } - if err := t.opts.EventStore.Append(req.Context(), t.SessionID(), stream.id, msg); err != nil { + if err := c.eventStore.Append(req.Context(), c.SessionID(), stream.id, msg); err != nil { errorf(http.StatusInternalServerError, "storing event: %v", err.Error()) return } @@ -638,14 +670,14 @@ func (t *StreamableServerTransport) respondSSE(stream *stream, w http.ResponseWr // If the stream did not terminate normally, it is either because ctx was // cancelled, or the connection is closed: check the ctx.Err() to differentiate // these cases. -func (t *StreamableServerTransport) messages(ctx context.Context, stream *stream, persistent bool) iter.Seq2[json.RawMessage, bool] { +func (c *streamableServerConn) messages(ctx context.Context, stream *stream, persistent bool) iter.Seq2[json.RawMessage, bool] { return func(yield func(json.RawMessage, bool) bool) { for { - t.mu.Lock() + c.mu.Lock() outgoing := stream.outgoing stream.outgoing = nil nOutstanding := len(stream.requests) - t.mu.Unlock() + c.mu.Unlock() for _, data := range outgoing { if !yield(data, true) { @@ -665,7 +697,7 @@ func (t *StreamableServerTransport) messages(ctx context.Context, stream *stream select { case <-*stream.signal.Load(): // there are new outgoing messages // return to top of loop - case <-t.done: // session is closed + case <-c.done: // session is closed yield(nil, false) return case <-ctx.Done(): @@ -708,22 +740,22 @@ func parseEventID(eventID string) (sid StreamID, idx int, ok bool) { } // Read implements the [Connection] interface. -func (t *StreamableServerTransport) Read(ctx context.Context) (jsonrpc.Message, error) { +func (c *streamableServerConn) Read(ctx context.Context) (jsonrpc.Message, error) { select { case <-ctx.Done(): return nil, ctx.Err() - case msg, ok := <-t.incoming: + case msg, ok := <-c.incoming: if !ok { return nil, io.EOF } return msg, nil - case <-t.done: + case <-c.done: return nil, io.EOF } } // Write implements the [Connection] interface. -func (t *StreamableServerTransport) Write(ctx context.Context, msg jsonrpc.Message) error { +func (c *streamableServerConn) Write(ctx context.Context, msg jsonrpc.Message) error { // Find the incoming request that this write relates to, if any. var forRequest jsonrpc.ID isResponse := false @@ -746,9 +778,9 @@ func (t *StreamableServerTransport) Write(ctx context.Context, msg jsonrpc.Messa // connection 0. var forStream StreamID if forRequest.IsValid() { - t.mu.Lock() - forStream = t.requestStreams[forRequest] - t.mu.Unlock() + c.mu.Lock() + forStream = c.requestStreams[forRequest] + c.mu.Unlock() } data, err := jsonrpc2.EncodeMessage(msg) @@ -756,13 +788,13 @@ func (t *StreamableServerTransport) Write(ctx context.Context, msg jsonrpc.Messa return err } - t.mu.Lock() - defer t.mu.Unlock() - if t.isDone { + c.mu.Lock() + defer c.mu.Unlock() + if c.isDone { return errors.New("session is closed") } - stream := t.streams[forStream] + stream := c.streams[forStream] if stream == nil { return fmt.Errorf("no stream with ID %d", forStream) } @@ -775,7 +807,7 @@ func (t *StreamableServerTransport) Write(ctx context.Context, msg jsonrpc.Messa // TODO(rfindley): either of these, particularly the first, might be // considered a bug in the server. Report it through a side-channel? if len(stream.requests) == 0 && forStream != 0 || stream.jsonResponse && !isResponse { - stream = t.streams[0] + stream = c.streams[0] } // TODO: if there is nothing to send these messages to (as would happen, for example, if forConn == 0 @@ -798,15 +830,15 @@ func (t *StreamableServerTransport) Write(ctx context.Context, msg jsonrpc.Messa } // Close implements the [Connection] interface. -func (t *StreamableServerTransport) Close() error { - t.mu.Lock() - defer t.mu.Unlock() - if !t.isDone { - t.isDone = true - close(t.done) +func (c *streamableServerConn) Close() error { + c.mu.Lock() + defer c.mu.Unlock() + if !c.isDone { + c.isDone = true + close(c.done) // TODO: find a way to plumb a context here, or an event store with a long-running // close operation can take arbitrary time. Alternative: impose a fixed timeout here. - return t.opts.EventStore.SessionClosed(context.TODO(), t.sessionID) + return c.eventStore.SessionClosed(context.TODO(), c.sessionID) } return nil } @@ -815,8 +847,9 @@ func (t *StreamableServerTransport) Close() error { // endpoint serving the streamable HTTP transport defined by the 2025-03-26 // version of the spec. type StreamableClientTransport struct { - url string - opts StreamableClientTransportOptions + Endpoint string + HTTPClient *http.Client + ReconnectOptions *StreamableReconnectOptions } // StreamableReconnectOptions defines parameters for client reconnect attempts. @@ -847,6 +880,8 @@ var DefaultReconnectOptions = &StreamableReconnectOptions{ // StreamableClientTransportOptions provides options for the // [NewStreamableClientTransport] constructor. +// +// Deprecated: use a StremableClientTransport literal. type StreamableClientTransportOptions struct { // HTTPClient is the client to use for making HTTP requests. If nil, // http.DefaultClient is used. @@ -856,10 +891,15 @@ type StreamableClientTransportOptions struct { // NewStreamableClientTransport returns a new client transport that connects to // the streamable HTTP server at the provided URL. +// +// Deprecated: use a StreamableClientTransport literal. +// +//go:fix inline func NewStreamableClientTransport(url string, opts *StreamableClientTransportOptions) *StreamableClientTransport { - t := &StreamableClientTransport{url: url} + t := &StreamableClientTransport{Endpoint: url} if opts != nil { - t.opts = *opts + t.HTTPClient = opts.HTTPClient + t.ReconnectOptions = opts.ReconnectOptions } return t } @@ -873,11 +913,11 @@ func NewStreamableClientTransport(url string, opts *StreamableClientTransportOpt // When closed, the connection issues a DELETE request to terminate the logical // session. func (t *StreamableClientTransport) Connect(ctx context.Context) (Connection, error) { - client := t.opts.HTTPClient + client := t.HTTPClient if client == nil { client = http.DefaultClient } - reconnOpts := t.opts.ReconnectOptions + reconnOpts := t.ReconnectOptions if reconnOpts == nil { reconnOpts = DefaultReconnectOptions } @@ -886,7 +926,7 @@ func (t *StreamableClientTransport) Connect(ctx context.Context) (Connection, er // cancelling its blocking network operations, which prevents hangs on exit. connCtx, cancel := context.WithCancel(context.Background()) conn := &streamableClientConn{ - url: t.url, + url: t.Endpoint, client: client, incoming: make(chan []byte, 100), done: make(chan struct{}), diff --git a/mcp/streamable_test.go b/mcp/streamable_test.go index 5480393..ba7af9f 100644 --- a/mcp/streamable_test.go +++ b/mcp/streamable_test.go @@ -44,7 +44,7 @@ func TestStreamableTransports(t *testing.T) { // 2. Start an httptest.Server with the StreamableHTTPHandler, wrapped in a // cookie-checking middleware. handler := NewStreamableHTTPHandler(func(req *http.Request) *Server { return server }, &StreamableHTTPOptions{ - transportOptions: &StreamableServerTransportOptions{jsonResponse: useJSON}, + jsonResponse: useJSON, }) var ( @@ -77,9 +77,10 @@ func TestStreamableTransports(t *testing.T) { } jar.SetCookies(u, []*http.Cookie{{Name: "test-cookie", Value: "test-value"}}) httpClient := &http.Client{Jar: jar} - transport := NewStreamableClientTransport(httpServer.URL, &StreamableClientTransportOptions{ + transport := &StreamableClientTransport{ + Endpoint: httpServer.URL, HTTPClient: httpClient, - }) + } client := NewClient(testImpl, nil) session, err := client.Connect(ctx, transport, nil) if err != nil { @@ -173,7 +174,7 @@ func TestClientReplay(t *testing.T) { notifications <- params.Message }, }) - clientSession, err := client.Connect(ctx, NewStreamableClientTransport(proxy.URL, nil), nil) + clientSession, err := client.Connect(ctx, &StreamableClientTransport{Endpoint: proxy.URL}, nil) if err != nil { t.Fatalf("client.Connect() failed: %v", err) } @@ -239,7 +240,7 @@ func TestServerInitiatedSSE(t *testing.T) { notifications <- "toolListChanged" }, }) - clientSession, err := client.Connect(ctx, NewStreamableClientTransport(httpServer.URL, nil), nil) + clientSession, err := client.Connect(ctx, &StreamableClientTransport{Endpoint: httpServer.URL}, nil) if err != nil { t.Fatalf("client.Connect() failed: %v", err) } @@ -765,7 +766,7 @@ func TestStreamableClientTransportApplicationJSON(t *testing.T) { httpServer := httptest.NewServer(http.HandlerFunc(serverHandler)) defer httpServer.Close() - transport := NewStreamableClientTransport(httpServer.URL, nil) + transport := &StreamableClientTransport{Endpoint: httpServer.URL} client := NewClient(testImpl, nil) session, err := client.Connect(ctx, transport, nil) if err != nil { diff --git a/mcp/transport.go b/mcp/transport.go index f45d7d2..76b7998 100644 --- a/mcp/transport.go +++ b/mcp/transport.go @@ -87,36 +87,39 @@ type serverConnection interface { // A StdioTransport is a [Transport] that communicates over stdin/stdout using // newline-delimited JSON. type StdioTransport struct { - ioTransport } -// An ioTransport is a [Transport] that communicates using newline-delimited -// JSON over an io.ReadWriteCloser. -type ioTransport struct { - rwc io.ReadWriteCloser -} - -func (t *ioTransport) Connect(context.Context) (Connection, error) { - return newIOConn(t.rwc), nil +// Connect implements the [Transport] interface. +func (*StdioTransport) Connect(context.Context) (Connection, error) { + return newIOConn(rwc{os.Stdin, os.Stdout}), nil } // NewStdioTransport constructs a transport that communicates over // stdin/stdout. +// +// Deprecated: use a StdioTransport literal. +// +//go:fix inline func NewStdioTransport() *StdioTransport { - return &StdioTransport{ioTransport{rwc{os.Stdin, os.Stdout}}} + return &StdioTransport{} } // An InMemoryTransport is a [Transport] that communicates over an in-memory // network connection, using newline-delimited JSON. type InMemoryTransport struct { - ioTransport + rwc io.ReadWriteCloser } -// NewInMemoryTransports returns two InMemoryTransports that connect to each +// Connect implements the [Transport] interface. +func (t *InMemoryTransport) Connect(context.Context) (Connection, error) { + return newIOConn(t.rwc), nil +} + +// NewInMemoryTransports returns two [InMemoryTransports] that connect to each // other. func NewInMemoryTransports() (*InMemoryTransport, *InMemoryTransport) { c1, c2 := net.Pipe() - return &InMemoryTransport{ioTransport{c1}}, &InMemoryTransport{ioTransport{c2}} + return &InMemoryTransport{c1}, &InMemoryTransport{c2} } type binder[T handler, State any] interface { @@ -208,24 +211,28 @@ func call(ctx context.Context, conn *jsonrpc2.Connection, method string, params // A LoggingTransport is a [Transport] that delegates to another transport, // writing RPC logs to an io.Writer. type LoggingTransport struct { - delegate Transport - w io.Writer + Transport Transport + Writer io.Writer } // NewLoggingTransport creates a new LoggingTransport that delegates to the // provided transport, writing RPC logs to the provided io.Writer. +// +// Deprecated: use a LoggingTransport literal. +// +//go:fix inline func NewLoggingTransport(delegate Transport, w io.Writer) *LoggingTransport { - return &LoggingTransport{delegate, w} + return &LoggingTransport{Transport: delegate, Writer: w} } // Connect connects the underlying transport, returning a [Connection] that writes // logs to the configured destination. func (t *LoggingTransport) Connect(ctx context.Context) (Connection, error) { - delegate, err := t.delegate.Connect(ctx) + delegate, err := t.Transport.Connect(ctx) if err != nil { return nil, err } - return &loggingConn{delegate, t.w}, nil + return &loggingConn{delegate, t.Writer}, nil } type loggingConn struct {