3e820ec · crush · git.iko.soy

git.iko.soy/crush

on main help sign in

chore(deps): update mcp-go (#155)

3e820ecCarlos Alexandro Becker1 year ago

go.mod+1−1

compare with main

unifiedsplit

@@ -29,7 +29,7 @@ require (

29 github.com/fsnotify/fsnotify v1.8.0

30 github.com/google/uuid v1.6.0

31 github.com/joho/godotenv v1.5.1

32− github.com/mark3labs/mcp-go v0.32.0

32+ github.com/mark3labs/mcp-go v0.33.0

33 github.com/muesli/termenv v0.16.0

34 github.com/ncruces/go-sqlite3 v0.25.0

35 github.com/nfnt/resize v0.0.0-20180221191011-83c6a9932646

go.sum+2−2

@@ -165,8 +165,8 @@ github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0

165github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw=

166github.com/lucasb-eyer/go-colorful v1.2.0 h1:1nnpGOrhyZZuNyfu1QjKiUICQ74+3FNCN69Aj6K7nkY=

167github.com/lucasb-eyer/go-colorful v1.2.0/go.mod h1:R4dSotOR9KMtayYi1e77YzuveK+i7ruzyGqttikkLy0=

168−github.com/mark3labs/mcp-go v0.32.0 h1:fgwmbfL2gbd67obg57OfV2Dnrhs1HtSdlY/i5fn7MU8=

168+github.com/mark3labs/mcp-go v0.33.0 h1:naxhjnTIs/tyPZmWUZFuG0lDmdA6sUyYGGf3gsHvTCc=

169−github.com/mark3labs/mcp-go v0.32.0/go.mod h1:rXqOudj/djTORU/ThxYx8fqEVj/5pvTuuebQ2RC7uk4=

169+github.com/mark3labs/mcp-go v0.33.0/go.mod h1:rXqOudj/djTORU/ThxYx8fqEVj/5pvTuuebQ2RC7uk4=

170github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY=

171github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=

172github.com/mattn/go-runewidth v0.0.16 h1:E5ScNMtiwvlvB5paMFdw9p4kSQzbXFikJ5SQO6TULQc=

vendor/github.com/mark3labs/mcp-go/client/client.go+101−1

@@ -22,6 +22,7 @@ type Client struct {

22 requestID atomic.Int64

23 clientCapabilities mcp.ClientCapabilities

24 serverCapabilities mcp.ServerCapabilities

25+ samplingHandler SamplingHandler

25}

26}

27type ClientOption func(*Client)

28type ClientOption func(*Client)

@@ -33,6 +34,21 @@ func WithClientCapabilities(capabilities mcp.ClientCapabilities) ClientOption {

33 }

34 }

34}

35}

37+// WithSamplingHandler sets the sampling handler for the client.

38+// When set, the client will declare sampling capability during initialization.

39+func WithSamplingHandler(handler SamplingHandler) ClientOption {

40+ return func(c *Client) {

41+ c.samplingHandler = handler

42+ }

43+}

44+

45+// WithSession assumes a MCP Session has already been initialized

46+func WithSession() ClientOption {

47+ return func(c *Client) {

48+ c.initialized = true

49+ }

50+}

51+

36// NewClient creates a new MCP client with the given transport.

52// NewClient creates a new MCP client with the given transport.

37// Usage:

53// Usage:

38//

54//

@@ -71,6 +87,12 @@ func (c *Client) Start(ctx context.Context) error {

71 handler(notification)

87 handler(notification)

72 }

88 }

73 })

89 })

90+

91+ // Set up request handler for bidirectional communication (e.g., sampling)

92+ if bidirectional, ok := c.transport.(transport.BidirectionalInterface); ok {

93+ bidirectional.SetRequestHandler(c.handleIncomingRequest)

94+ }

95+

74 return nil

96 return nil

75}

97}

@@ -127,6 +149,12 @@ func (c *Client) Initialize(

127 ctx context.Context,

149 ctx context.Context,

128 request mcp.InitializeRequest,

150 request mcp.InitializeRequest,

129) (*mcp.InitializeResult, error) {

151) (*mcp.InitializeResult, error) {

152+ // Merge client capabilities with sampling capability if handler is configured

153+ capabilities := request.Params.Capabilities

154+ if c.samplingHandler != nil {

155+ capabilities.Sampling = &struct{}{}

156+ }

157+

130 // Ensure we send a params object with all required fields

158 // Ensure we send a params object with all required fields

131 params := struct {

159 params := struct {

132 ProtocolVersion string `json:"protocolVersion"`

160 ProtocolVersion string `json:"protocolVersion"`

@@ -135,7 +163,7 @@ func (c *Client) Initialize(

135 }{

163 }{

136 ProtocolVersion: request.Params.ProtocolVersion,

164 ProtocolVersion: request.Params.ProtocolVersion,

137 ClientInfo: request.Params.ClientInfo,

165 ClientInfo: request.Params.ClientInfo,

138− Capabilities: request.Params.Capabilities, // Will be empty struct if not set

166+ Capabilities: capabilities,

139 }

167 }

140

168

141 response, err := c.sendRequest(ctx, "initialize", params)

169 response, err := c.sendRequest(ctx, "initialize", params)

@@ -398,6 +426,64 @@ func (c *Client) Complete(

398 return &result, nil

426 return &result, nil

399}

427}

400

428

429+// handleIncomingRequest processes incoming requests from the server.

430+// This is the main entry point for server-to-client requests like sampling.

431+func (c *Client) handleIncomingRequest(ctx context.Context, request transport.JSONRPCRequest) (*transport.JSONRPCResponse, error) {

432+ switch request.Method {

433+ case string(mcp.MethodSamplingCreateMessage):

434+ return c.handleSamplingRequestTransport(ctx, request)

435+ default:

436+ return nil, fmt.Errorf("unsupported request method: %s", request.Method)

437+ }

438+}

439+

440+// handleSamplingRequestTransport handles sampling requests at the transport level.

441+func (c *Client) handleSamplingRequestTransport(ctx context.Context, request transport.JSONRPCRequest) (*transport.JSONRPCResponse, error) {

442+ if c.samplingHandler == nil {

443+ return nil, fmt.Errorf("no sampling handler configured")

444+ }

445+

446+ // Parse the request parameters

447+ var params mcp.CreateMessageParams

448+ if request.Params != nil {

449+ paramsBytes, err := json.Marshal(request.Params)

450+ if err != nil {

451+ return nil, fmt.Errorf("failed to marshal params: %w", err)

452+ }

453+ if err := json.Unmarshal(paramsBytes, &params); err != nil {

454+ return nil, fmt.Errorf("failed to unmarshal params: %w", err)

455+ }

456+ }

457+

458+ // Create the MCP request

459+ mcpRequest := mcp.CreateMessageRequest{

460+ Request: mcp.Request{

461+ Method: string(mcp.MethodSamplingCreateMessage),

462+ },

463+ CreateMessageParams: params,

464+ }

465+

466+ // Call the sampling handler

467+ result, err := c.samplingHandler.CreateMessage(ctx, mcpRequest)

468+ if err != nil {

469+ return nil, err

470+ }

471+

472+ // Marshal the result

473+ resultBytes, err := json.Marshal(result)

474+ if err != nil {

475+ return nil, fmt.Errorf("failed to marshal result: %w", err)

476+ }

477+

478+ // Create the transport response

479+ response := &transport.JSONRPCResponse{

480+ JSONRPC: mcp.JSONRPC_VERSION,

481+ ID: request.ID,

482+ Result: json.RawMessage(resultBytes),

483+ }

484+

485+ return response, nil

486+}

401func listByPage[T any](

487func listByPage[T any](

402 ctx context.Context,

488 ctx context.Context,

403 client *Client,

489 client *Client,

@@ -432,3 +518,17 @@ func (c *Client) GetServerCapabilities() mcp.ServerCapabilities {

432func (c *Client) GetClientCapabilities() mcp.ClientCapabilities {

518func (c *Client) GetClientCapabilities() mcp.ClientCapabilities {

433 return c.clientCapabilities

519 return c.clientCapabilities

434}

520}

521+

522+// GetSessionId returns the session ID of the transport.

523+// If the transport does not support sessions, it returns an empty string.

524+func (c *Client) GetSessionId() string {

525+ if c.transport == nil {

526+ return ""

527+ }

528+ return c.transport.GetSessionId()

529+}

530+

531+// IsInitialized returns true if the client has been initialized.

532+func (c *Client) IsInitialized() bool {

533+ return c.initialized

534+}

vendor/github.com/mark3labs/mcp-go/client/http.go+6−1

@@ -13,5 +13,10 @@ func NewStreamableHttpClient(baseURL string, options ...transport.StreamableHTTP

13 if err != nil {

14 return nil, fmt.Errorf("failed to create SSE transport: %w", err)

15 }

16+ clientOptions := make([]ClientOption, 0)

17+ sessionID := trans.GetSessionId()

18+ if sessionID != "" {

19+ clientOptions = append(clientOptions, WithSession())

20+ }

16− return NewClient(trans), nil

21+ return NewClient(trans, clientOptions...), nil

17}

22}

vendor/github.com/mark3labs/mcp-go/client/sampling.gonew file+20

@@ -0,0 +1,20 @@

1+package client

3+import (

4+ "context"

6+ "github.com/mark3labs/mcp-go/mcp"

7+)

9+// SamplingHandler defines the interface for handling sampling requests from servers.

10+// Clients can implement this interface to provide LLM sampling capabilities to servers.

11+type SamplingHandler interface {

12+ // CreateMessage handles a sampling request from the server and returns the generated message.

13+ // The implementation should:

14+ // 1. Validate the request parameters

15+ // 2. Optionally prompt the user for approval (human-in-the-loop)

16+ // 3. Select an appropriate model based on preferences

17+ // 4. Generate the response using the selected model

18+ // 5. Return the result with model information and stop reason

19+ CreateMessage(ctx context.Context, request mcp.CreateMessageRequest) (*mcp.CreateMessageResult, error)

20+}

vendor/github.com/mark3labs/mcp-go/client/stdio.go+19−3

@@ -19,10 +19,26 @@ func NewStdioMCPClient(

19 env []string,

20 args ...string,

21) (*Client, error) {

22+ return NewStdioMCPClientWithOptions(command, env, args)

23+}

24+

25+// NewStdioMCPClientWithOptions creates a new stdio-based MCP client that communicates with a subprocess.

26+// It launches the specified command with given arguments and sets up stdin/stdout pipes for communication.

27+// Optional configuration functions can be provided to customize the transport before it starts,

28+// such as setting a custom command function.

29+//

30+// NOTICE: NewStdioMCPClientWithOptions automatically starts the underlying transport.

31+// Don't call the Start method manually.

32+// This is for backward compatibility.

33+func NewStdioMCPClientWithOptions(

34+ command string,

35+ env []string,

36+ args []string,

37+ opts ...transport.StdioOption,

38+) (*Client, error) {

39+ stdioTransport := transport.NewStdioWithOptions(command, env, args, opts...)

23− stdioTransport := transport.NewStdio(command, env, args...)

24− err := stdioTransport.Start(context.Background())

41+ if err := stdioTransport.Start(context.Background()); err != nil {

25− if err != nil {

26 return nil, fmt.Errorf("failed to start stdio transport: %w", err)

42 return nil, fmt.Errorf("failed to start stdio transport: %w", err)

27 }

43 }

vendor/github.com/mark3labs/mcp-go/client/transport/inprocess.go+4

@@ -68,3 +68,7 @@ func (c *InProcessTransport) SetNotificationHandler(handler func(notification mc

68func (*InProcessTransport) Close() error {

69 return nil

70}

71+

72+func (c *InProcessTransport) GetSessionId() string {

73+ return ""

74+}

vendor/github.com/mark3labs/mcp-go/client/transport/interface.go+18−2

@@ -29,6 +29,22 @@ type Interface interface {

30 // Close the connection.

31 Close() error

32+

33+ // GetSessionId returns the session ID of the transport.

34+ GetSessionId() string

35+}

36+

37+// RequestHandler defines a function that handles incoming requests from the server.

38+type RequestHandler func(ctx context.Context, request JSONRPCRequest) (*JSONRPCResponse, error)

39+

40+// BidirectionalInterface extends Interface to support incoming requests from the server.

41+// This is used for features like sampling where the server can send requests to the client.

42+type BidirectionalInterface interface {

43+ Interface

44+

45+ // SetRequestHandler sets the handler for incoming requests from the server.

46+ // The handler should process the request and return a response.

47+ SetRequestHandler(handler RequestHandler)

32}

48}

34type JSONRPCRequest struct {

50type JSONRPCRequest struct {

@@ -41,10 +57,10 @@ type JSONRPCRequest struct {

41type JSONRPCResponse struct {

57type JSONRPCResponse struct {

42 JSONRPC string `json:"jsonrpc"`

58 JSONRPC string `json:"jsonrpc"`

43 ID mcp.RequestId `json:"id"`

59 ID mcp.RequestId `json:"id"`

44− Result json.RawMessage `json:"result"`

60+ Result json.RawMessage `json:"result,omitempty"`

45 Error *struct {

61 Error *struct {

46 Code int `json:"code"`

62 Code int `json:"code"`

47 Message string `json:"message"`

63 Message string `json:"message"`

48 Data json.RawMessage `json:"data"`

64 Data json.RawMessage `json:"data"`

49− } `json:"error"`

65+ } `json:"error,omitempty"`

50}

66}

vendor/github.com/mark3labs/mcp-go/client/transport/sse.go+6

@@ -428,6 +428,12 @@ func (c *SSE) Close() error {

428 return nil

429}

430

431+// GetSessionId returns the session ID of the transport.

432+// Since SSE does not maintain a session ID, it returns an empty string.

433+func (c *SSE) GetSessionId() string {

434+ return ""

435+}

436+

431// SendNotification sends a JSON-RPC notification to the server without expecting a response.

437// SendNotification sends a JSON-RPC notification to the server without expecting a response.

432func (c *SSE) SendNotification(ctx context.Context, notification mcp.JSONRPCNotification) error {

438func (c *SSE) SendNotification(ctx context.Context, notification mcp.JSONRPCNotification) error {

433 if c.endpoint == nil {

439 if c.endpoint == nil {

vendor/github.com/mark3labs/mcp-go/client/transport/stdio.go+191−13

@@ -23,6 +23,7 @@ type Stdio struct {

23 env []string

25 cmd *exec.Cmd

26+ cmdFunc CommandFunc

26 stdin io.WriteCloser

27 stdin io.WriteCloser

27 stdout *bufio.Reader

28 stdout *bufio.Reader

28 stderr io.ReadCloser

29 stderr io.ReadCloser

@@ -31,6 +32,28 @@ type Stdio struct {

31 done chan struct{}

32 done chan struct{}

32 onNotification func(mcp.JSONRPCNotification)

33 onNotification func(mcp.JSONRPCNotification)

33 notifyMu sync.RWMutex

34 notifyMu sync.RWMutex

35+ onRequest RequestHandler

36+ requestMu sync.RWMutex

37+ ctx context.Context

38+ ctxMu sync.RWMutex

39+}

40+

41+// StdioOption defines a function that configures a Stdio transport instance.

42+// Options can be used to customize the behavior of the transport before it starts,

43+// such as setting a custom command function.

44+type StdioOption func(*Stdio)

45+

46+// CommandFunc is a factory function that returns a custom exec.Cmd used to launch the MCP subprocess.

47+// It can be used to apply sandboxing, custom environment control, working directories, etc.

48+type CommandFunc func(ctx context.Context, command string, env []string, args []string) (*exec.Cmd, error)

49+

50+// WithCommandFunc sets a custom command factory function for the stdio transport.

51+// The CommandFunc is responsible for constructing the exec.Cmd used to launch the subprocess,

52+// allowing control over attributes like environment, working directory, and system-level sandboxing.

53+func WithCommandFunc(f CommandFunc) StdioOption {

54+ return func(s *Stdio) {

55+ s.cmdFunc = f

56+ }

34}

57}

36// NewIO returns a new stdio-based transport using existing input, output, and

59// NewIO returns a new stdio-based transport using existing input, output, and

@@ -44,6 +67,7 @@ func NewIO(input io.Reader, output io.WriteCloser, logging io.ReadCloser) *Stdio

45 responses: make(map[string]chan *JSONRPCResponse),

68 responses: make(map[string]chan *JSONRPCResponse),

46 done: make(chan struct{}),

69 done: make(chan struct{}),

70+ ctx: context.Background(),

47 }

71 }

48}

72}

@@ -55,20 +79,43 @@ func NewStdio(

55 env []string,

79 env []string,

56 args ...string,

80 args ...string,

57) *Stdio {

81) *Stdio {

82+ return NewStdioWithOptions(command, env, args)

83+}

85+// NewStdioWithOptions creates a new stdio transport to communicate with a subprocess.

86+// It launches the specified command with given arguments and sets up stdin/stdout pipes for communication.

87+// Returns an error if the subprocess cannot be started or the pipes cannot be created.

88+// Optional configuration functions can be provided to customize the transport before it starts,

89+// such as setting a custom command factory.

90+func NewStdioWithOptions(

91+ command string,

92+ env []string,

93+ args []string,

94+ opts ...StdioOption,

95+) *Stdio {

59− client := &Stdio{

96+ s := &Stdio{

60 command: command,

97 command: command,

61 args: args,

98 args: args,

62 env: env,

99 env: env,

100

64 responses: make(map[string]chan *JSONRPCResponse),

101 responses: make(map[string]chan *JSONRPCResponse),

65 done: make(chan struct{}),

102 done: make(chan struct{}),

103+ ctx: context.Background(),

104+ }

105+

106+ for _, opt := range opts {

107+ opt(s)

66 }

108 }

109

68− return client

110+ return s

69}

111}

112

71func (c *Stdio) Start(ctx context.Context) error {

113func (c *Stdio) Start(ctx context.Context) error {

114+ // Store the context for use in request handling

115+ c.ctxMu.Lock()

116+ c.ctx = ctx

117+ c.ctxMu.Unlock()

118+

72 if err := c.spawnCommand(ctx); err != nil {

119 if err := c.spawnCommand(ctx); err != nil {

73 return err

120 return err

74 }

121 }

@@ -83,18 +130,25 @@ func (c *Stdio) Start(ctx context.Context) error {

83 return nil

130 return nil

84}

131}

132

86−// spawnCommand spawns a new process running c.command.

133+// spawnCommand spawns a new process running the configured command, args, and env.

134+// If an (optional) cmdFunc custom command factory function was configured, it will be used to construct the subprocess;

135+// otherwise, the default behavior uses exec.CommandContext with the merged environment.

136+// Initializes stdin, stdout, and stderr pipes for JSON-RPC communication.

87func (c *Stdio) spawnCommand(ctx context.Context) error {

137func (c *Stdio) spawnCommand(ctx context.Context) error {

88 if c.command == "" {

138 if c.command == "" {

89 return nil

139 return nil

90 }

140 }

141

92− cmd := exec.CommandContext(ctx, c.command, c.args...)

142+ var cmd *exec.Cmd

93−

143+ var err error

94− mergedEnv := os.Environ()

95− mergedEnv = append(mergedEnv, c.env...)

144

97− cmd.Env = mergedEnv

145+ // Standard behavior if no command func present.

146+ if c.cmdFunc == nil {

147+ cmd = exec.CommandContext(ctx, c.command, c.args...)

148+ cmd.Env = append(os.Environ(), c.env...)

149+ } else if cmd, err = c.cmdFunc(ctx, c.command, c.env, c.args); err != nil {

150+ return err

151+ }

152

99 stdin, err := cmd.StdinPipe()

153 stdin, err := cmd.StdinPipe()

100 if err != nil {

154 if err != nil {

@@ -148,6 +202,12 @@ func (c *Stdio) Close() error {

148 return nil

202 return nil

149}

203}

150

204

205+// GetSessionId returns the session ID of the transport.

206+// Since stdio does not maintain a session ID, it returns an empty string.

207+func (c *Stdio) GetSessionId() string {

208+ return ""

209+}

210+

151// SetNotificationHandler sets the handler function to be called when a notification is received.

211// SetNotificationHandler sets the handler function to be called when a notification is received.

152// Only one handler can be set at a time; setting a new one replaces the previous handler.

212// Only one handler can be set at a time; setting a new one replaces the previous handler.

153func (c *Stdio) SetNotificationHandler(

213func (c *Stdio) SetNotificationHandler(

@@ -158,6 +218,14 @@ func (c *Stdio) SetNotificationHandler(

158 c.onNotification = handler

218 c.onNotification = handler

159}

219}

160

220

221+// SetRequestHandler sets the handler function to be called when a request is received from the server.

222+// This enables bidirectional communication for features like sampling.

223+func (c *Stdio) SetRequestHandler(handler RequestHandler) {

224+ c.requestMu.Lock()

225+ defer c.requestMu.Unlock()

226+ c.onRequest = handler

227+}

228+

161// readResponses continuously reads and processes responses from the server's stdout.

229// readResponses continuously reads and processes responses from the server's stdout.

162// It handles both responses to requests and notifications, routing them appropriately.

230// It handles both responses to requests and notifications, routing them appropriately.

163// Runs until the done channel is closed or an error occurs reading from stdout.

231// Runs until the done channel is closed or an error occurs reading from stdout.

@@ -175,13 +243,18 @@ func (c *Stdio) readResponses() {

175 return

243 return

176 }

244 }

177

245

246+ // First try to parse as a generic message to check for ID field

178− var baseMessage JSONRPCResponse

247+ var baseMessage struct {

248+ JSONRPC string `json:"jsonrpc"`

249+ ID *mcp.RequestId `json:"id,omitempty"`

250+ Method string `json:"method,omitempty"`

251+ }

179 if err := json.Unmarshal([]byte(line), &baseMessage); err != nil {

252 if err := json.Unmarshal([]byte(line), &baseMessage); err != nil {

180 continue

253 continue

181 }

254 }

182

255

183− // Handle notification

256+ // If it has a method but no ID, it's a notification

184− if baseMessage.ID.IsNil() {

257+ if baseMessage.Method != "" && baseMessage.ID == nil {

185 var notification mcp.JSONRPCNotification

258 var notification mcp.JSONRPCNotification

186 if err := json.Unmarshal([]byte(line), &notification); err != nil {

259 if err := json.Unmarshal([]byte(line), &notification); err != nil {

187 continue

260 continue

@@ -194,15 +267,30 @@ func (c *Stdio) readResponses() {

194 continue

267 continue

195 }

268 }

196

269

270+ // If it has a method and an ID, it's an incoming request

271+ if baseMessage.Method != "" && baseMessage.ID != nil {

272+ var request JSONRPCRequest

273+ if err := json.Unmarshal([]byte(line), &request); err == nil {

274+ c.handleIncomingRequest(request)

275+ continue

276+ }

277+ }

278+

279+ // Otherwise, it's a response to our request

280+ var response JSONRPCResponse

281+ if err := json.Unmarshal([]byte(line), &response); err != nil {

282+ continue

283+ }

284+

197 // Create string key for map lookup

285 // Create string key for map lookup

198− idKey := baseMessage.ID.String()

286+ idKey := response.ID.String()

199

287

200 c.mu.RLock()

288 c.mu.RLock()

201 ch, exists := c.responses[idKey]

289 ch, exists := c.responses[idKey]

202 c.mu.RUnlock()

290 c.mu.RUnlock()

203

291

204 if exists {

292 if exists {

205− ch <- &baseMessage

293+ ch <- &response

206 c.mu.Lock()

294 c.mu.Lock()

207 delete(c.responses, idKey)

295 delete(c.responses, idKey)

208 c.mu.Unlock()

296 c.mu.Unlock()

@@ -281,6 +369,96 @@ func (c *Stdio) SendNotification(

281 return nil

369 return nil

282}

370}

283

371

372+// handleIncomingRequest processes incoming requests from the server.

373+// It calls the registered request handler and sends the response back to the server.

374+func (c *Stdio) handleIncomingRequest(request JSONRPCRequest) {

375+ c.requestMu.RLock()

376+ handler := c.onRequest

377+ c.requestMu.RUnlock()

378+

379+ if handler == nil {

380+ // Send error response if no handler is configured

381+ errorResponse := JSONRPCResponse{

382+ JSONRPC: mcp.JSONRPC_VERSION,

383+ ID: request.ID,

384+ Error: &struct {

385+ Code int `json:"code"`

386+ Message string `json:"message"`

387+ Data json.RawMessage `json:"data"`

388+ }{

389+ Code: mcp.METHOD_NOT_FOUND,

390+ Message: "No request handler configured",

391+ },

392+ }

393+ c.sendResponse(errorResponse)

394+ return

395+ }

396+

397+ // Handle the request in a goroutine to avoid blocking

398+ go func() {

399+ c.ctxMu.RLock()

400+ ctx := c.ctx

401+ c.ctxMu.RUnlock()

402+

403+ // Check if context is already cancelled before processing

404+ select {

405+ case <-ctx.Done():

406+ errorResponse := JSONRPCResponse{

407+ JSONRPC: mcp.JSONRPC_VERSION,

408+ ID: request.ID,

409+ Error: &struct {

410+ Code int `json:"code"`

411+ Message string `json:"message"`

412+ Data json.RawMessage `json:"data"`

413+ }{

414+ Code: mcp.INTERNAL_ERROR,

415+ Message: ctx.Err().Error(),

416+ },

417+ }

418+ c.sendResponse(errorResponse)

419+ return

420+ default:

421+ }

422+

423+ response, err := handler(ctx, request)

424+

425+ if err != nil {

426+ errorResponse := JSONRPCResponse{

427+ JSONRPC: mcp.JSONRPC_VERSION,

428+ ID: request.ID,

429+ Error: &struct {

430+ Code int `json:"code"`

431+ Message string `json:"message"`

432+ Data json.RawMessage `json:"data"`

433+ }{

434+ Code: mcp.INTERNAL_ERROR,

435+ Message: err.Error(),

436+ },

437+ }

438+ c.sendResponse(errorResponse)

439+ return

440+ }

441+

442+ if response != nil {

443+ c.sendResponse(*response)

444+ }

445+ }()

446+}

447+

448+// sendResponse sends a response back to the server.

449+func (c *Stdio) sendResponse(response JSONRPCResponse) {

450+ responseBytes, err := json.Marshal(response)

451+ if err != nil {

452+ fmt.Printf("Error marshaling response: %v\n", err)

453+ return

454+ }

455+ responseBytes = append(responseBytes, '\n')

456+

457+ if _, err := c.stdin.Write(responseBytes); err != nil {

458+ fmt.Printf("Error writing response: %v\n", err)

459+ }

460+}

461+

284// Stderr returns a reader for the stderr output of the subprocess.

462// Stderr returns a reader for the stderr output of the subprocess.

285// This can be used to capture error messages or logs from the subprocess.

463// This can be used to capture error messages or logs from the subprocess.

286func (c *Stdio) Stderr() io.Reader {

464func (c *Stdio) Stderr() io.Reader {

vendor/github.com/mark3labs/mcp-go/client/transport/streamable_http.go+228−115

@@ -17,10 +17,24 @@ import (

17 "time"

19 "github.com/mark3labs/mcp-go/mcp"

20+ "github.com/mark3labs/mcp-go/util"

20)

21)

22type StreamableHTTPCOption func(*StreamableHTTP)

23type StreamableHTTPCOption func(*StreamableHTTP)

25+// WithContinuousListening enables receiving server-to-client notifications when no request is in flight.

26+// In particular, if you want to receive global notifications from the server (like ToolListChangedNotification),

27+// you should enable this option.

28+//

29+// It will establish a standalone long-live GET HTTP connection to the server.

30+// https://modelcontextprotocol.io/specification/2025-03-26/basic/transports#listening-for-messages-from-the-server

31+// NOTICE: Even enabled, the server may not support this feature.

32+func WithContinuousListening() StreamableHTTPCOption {

33+ return func(sc *StreamableHTTP) {

34+ sc.getListeningEnabled = true

35+ }

36+}

37+

24// WithHTTPClient sets a custom HTTP client on the StreamableHTTP transport.

38// WithHTTPClient sets a custom HTTP client on the StreamableHTTP transport.

25func WithHTTPBasicClient(client *http.Client) StreamableHTTPCOption {

39func WithHTTPBasicClient(client *http.Client) StreamableHTTPCOption {

26 return func(sc *StreamableHTTP) {

40 return func(sc *StreamableHTTP) {

@@ -54,6 +68,19 @@ func WithHTTPOAuth(config OAuthConfig) StreamableHTTPCOption {

54 }

68 }

55}

69}

71+func WithLogger(logger util.Logger) StreamableHTTPCOption {

72+ return func(sc *StreamableHTTP) {

73+ sc.logger = logger

74+ }

75+}

76+

77+// WithSession creates a client with a pre-configured session

78+func WithSession(sessionID string) StreamableHTTPCOption {

79+ return func(sc *StreamableHTTP) {

80+ sc.sessionID.Store(sessionID)

81+ }

82+}

83+

57// StreamableHTTP implements Streamable HTTP transport.

84// StreamableHTTP implements Streamable HTTP transport.

58//

85//

59// It transmits JSON-RPC messages over individual HTTP requests. One message per request.

86// It transmits JSON-RPC messages over individual HTTP requests. One message per request.

@@ -64,19 +91,22 @@ func WithHTTPOAuth(config OAuthConfig) StreamableHTTPCOption {

64//

91//

65// The current implementation does not support the following features:

92// The current implementation does not support the following features:

66// - batching

93// - batching

67−// - continuously listening for server notifications when no request is in flight

68−// (https://modelcontextprotocol.io/specification/2025-03-26/basic/transports#listening-for-messages-from-the-server)

69// - resuming stream

94// - resuming stream

70// (https://modelcontextprotocol.io/specification/2025-03-26/basic/transports#resumability-and-redelivery)

95// (https://modelcontextprotocol.io/specification/2025-03-26/basic/transports#resumability-and-redelivery)

71// - server -> client request

96// - server -> client request

72type StreamableHTTP struct {

97type StreamableHTTP struct {

73− serverURL *url.URL

98+ serverURL *url.URL

74− httpClient *http.Client

99+ httpClient *http.Client

75− headers map[string]string

100+ headers map[string]string

76− headerFunc HTTPHeaderFunc

101+ headerFunc HTTPHeaderFunc

102+ logger util.Logger

103+ getListeningEnabled bool

104

78 sessionID atomic.Value // string

105 sessionID atomic.Value // string

106

107+ initialized chan struct{}

108+ initializedOnce sync.Once

109+

80 notificationHandler func(mcp.JSONRPCNotification)

110 notificationHandler func(mcp.JSONRPCNotification)

81 notifyMu sync.RWMutex

111 notifyMu sync.RWMutex

112

@@ -95,15 +125,19 @@ func NewStreamableHTTP(serverURL string, options ...StreamableHTTPCOption) (*Str

95 }

125 }

126

97 smc := &StreamableHTTP{

127 smc := &StreamableHTTP{

98− serverURL: parsedURL,

128+ serverURL: parsedURL,

99− httpClient: &http.Client{},

129+ httpClient: &http.Client{},

100− headers: make(map[string]string),

130+ headers: make(map[string]string),

131+ closed: make(chan struct{}),

132+ logger: util.DefaultLogger(),

101− closed: make(chan struct{}),

133+ initialized: make(chan struct{}),

102 }

134 }

103 smc.sessionID.Store("") // set initial value to simplify later usage

135 smc.sessionID.Store("") // set initial value to simplify later usage

104

136

105 for _, opt := range options {

137 for _, opt := range options {

138+ if opt != nil {

106− opt(smc)

139+ opt(smc)

140+ }

107 }

141 }

108

142

109 // If OAuth is configured, set the base URL for metadata discovery

143 // If OAuth is configured, set the base URL for metadata discovery

@@ -118,7 +152,20 @@ func NewStreamableHTTP(serverURL string, options ...StreamableHTTPCOption) (*Str

118

152

119// Start initiates the HTTP connection to the server.

153// Start initiates the HTTP connection to the server.

120func (c *StreamableHTTP) Start(ctx context.Context) error {

154func (c *StreamableHTTP) Start(ctx context.Context) error {

121− // For Streamable HTTP, we don't need to establish a persistent connection

155+ // For Streamable HTTP, we don't need to establish a persistent connection by default

156+ if c.getListeningEnabled {

157+ go func() {

158+ select {

159+ case <-c.initialized:

160+ ctx, cancel := c.contextAwareOfClientClose(ctx)

161+ defer cancel()

162+ c.listenForever(ctx)

163+ case <-c.closed:

164+ return

165+ }

166+ }()

167+ }

168+

122 return nil

169 return nil

123}

170}

124

171

@@ -142,13 +189,13 @@ func (c *StreamableHTTP) Close() error {

142 defer cancel()

189 defer cancel()

143 req, err := http.NewRequestWithContext(ctx, http.MethodDelete, c.serverURL.String(), nil)

190 req, err := http.NewRequestWithContext(ctx, http.MethodDelete, c.serverURL.String(), nil)

144 if err != nil {

191 if err != nil {

145− fmt.Printf("failed to create close request\n: %v", err)

192+ c.logger.Errorf("failed to create close request: %v", err)

146 return

193 return

147 }

194 }

148 req.Header.Set(headerKeySessionID, sessionId)

195 req.Header.Set(headerKeySessionID, sessionId)

149 res, err := c.httpClient.Do(req)

196 res, err := c.httpClient.Do(req)

150 if err != nil {

197 if err != nil {

151− fmt.Printf("failed to send close request\n: %v", err)

198+ c.logger.Errorf("failed to send close request: %v", err)

152 return

199 return

153 }

200 }

154 res.Body.Close()

201 res.Body.Close()

@@ -185,77 +232,29 @@ func (c *StreamableHTTP) SendRequest(

185 request JSONRPCRequest,

232 request JSONRPCRequest,

186) (*JSONRPCResponse, error) {

233) (*JSONRPCResponse, error) {

187

234

188− // Create a combined context that could be canceled when the client is closed

189− newCtx, cancel := context.WithCancel(ctx)

190− defer cancel()

191− go func() {

192− select {

193− case <-c.closed:

194− cancel()

195− case <-newCtx.Done():

196− // The original context was canceled, no need to do anything

197− }

198− }()

199− ctx = newCtx

200−

201 // Marshal request

235 // Marshal request

202 requestBody, err := json.Marshal(request)

236 requestBody, err := json.Marshal(request)

203 if err != nil {

237 if err != nil {

204 return nil, fmt.Errorf("failed to marshal request: %w", err)

238 return nil, fmt.Errorf("failed to marshal request: %w", err)

205 }

239 }

206

240

207− // Create HTTP request

241+ ctx, cancel := c.contextAwareOfClientClose(ctx)

208− req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.serverURL.String(), bytes.NewReader(requestBody))

242+ defer cancel()

209− if err != nil {

210− return nil, fmt.Errorf("failed to create request: %w", err)

211− }

212−

213− // Set headers

214− req.Header.Set("Content-Type", "application/json")

215− req.Header.Set("Accept", "application/json, text/event-stream")

216− sessionID := c.sessionID.Load()

217− if sessionID != "" {

218− req.Header.Set(headerKeySessionID, sessionID.(string))

219− }

220− for k, v := range c.headers {

221− req.Header.Set(k, v)

222− }

223−

224− // Add OAuth authorization if configured

225− if c.oauthHandler != nil {

226− authHeader, err := c.oauthHandler.GetAuthorizationHeader(ctx)

227− if err != nil {

228− // If we get an authorization error, return a specific error that can be handled by the client

229− if err.Error() == "no valid token available, authorization required" {

230− return nil, &OAuthAuthorizationRequiredError{

231− Handler: c.oauthHandler,

232− }

233− }

234− return nil, fmt.Errorf("failed to get authorization header: %w", err)

235− }

236− req.Header.Set("Authorization", authHeader)

237− }

238−

239− if c.headerFunc != nil {

240− for k, v := range c.headerFunc(ctx) {

241− req.Header.Set(k, v)

242− }

243− }

244

243

245− // Send request

244+ resp, err := c.sendHTTP(ctx, http.MethodPost, bytes.NewReader(requestBody), "application/json, text/event-stream")

246− resp, err := c.httpClient.Do(req)

247 if err != nil {

245 if err != nil {

246+ if errors.Is(err, ErrSessionTerminated) && request.Method == string(mcp.MethodInitialize) {

247+ // If the request is initialize, should not return a SessionTerminated error

248+ // It should be a genuine endpoint-routing issue.

249+ // ( Fall through to return StatusCode checking. )

250+ } else {

248− return nil, fmt.Errorf("failed to send request: %w", err)

251+ return nil, fmt.Errorf("failed to send request: %w", err)

252+ }

249 }

253 }

250 defer resp.Body.Close()

254 defer resp.Body.Close()

251

255

252 // Check if we got an error response

256 // Check if we got an error response

253 if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusAccepted {

257 if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusAccepted {

254− // handle session closed

255− if resp.StatusCode == http.StatusNotFound {

256− c.sessionID.CompareAndSwap(sessionID, "")

257− return nil, fmt.Errorf("session terminated (404). need to re-initialize")

258− }

259

258

260 // Handle OAuth unauthorized error

259 // Handle OAuth unauthorized error

261 if resp.StatusCode == http.StatusUnauthorized && c.oauthHandler != nil {

260 if resp.StatusCode == http.StatusUnauthorized && c.oauthHandler != nil {

@@ -279,6 +278,10 @@ func (c *StreamableHTTP) SendRequest(

279 if sessionID := resp.Header.Get(headerKeySessionID); sessionID != "" {

278 if sessionID := resp.Header.Get(headerKeySessionID); sessionID != "" {

280 c.sessionID.Store(sessionID)

279 c.sessionID.Store(sessionID)

281 }

280 }

281+

282+ c.initializedOnce.Do(func() {

283+ close(c.initialized)

284+ })

282 }

285 }

283

286

284 // Handle different response types

287 // Handle different response types

@@ -300,16 +303,77 @@ func (c *StreamableHTTP) SendRequest(

300

303

301 case "text/event-stream":

304 case "text/event-stream":

302 // Server is using SSE for streaming responses

305 // Server is using SSE for streaming responses

303− return c.handleSSEResponse(ctx, resp.Body)

306+ return c.handleSSEResponse(ctx, resp.Body, false)

304

307

305 default:

308 default:

306 return nil, fmt.Errorf("unexpected content type: %s", resp.Header.Get("Content-Type"))

309 return nil, fmt.Errorf("unexpected content type: %s", resp.Header.Get("Content-Type"))

307 }

310 }

308}

311}

309

312

313+func (c *StreamableHTTP) sendHTTP(

314+ ctx context.Context,

315+ method string,

316+ body io.Reader,

317+ acceptType string,

318+) (resp *http.Response, err error) {

319+

320+ // Create HTTP request

321+ req, err := http.NewRequestWithContext(ctx, method, c.serverURL.String(), body)

322+ if err != nil {

323+ return nil, fmt.Errorf("failed to create request: %w", err)

324+ }

325+

326+ // Set headers

327+ req.Header.Set("Content-Type", "application/json")

328+ req.Header.Set("Accept", acceptType)

329+ sessionID := c.sessionID.Load().(string)

330+ if sessionID != "" {

331+ req.Header.Set(headerKeySessionID, sessionID)

332+ }

333+ for k, v := range c.headers {

334+ req.Header.Set(k, v)

335+ }

336+

337+ // Add OAuth authorization if configured

338+ if c.oauthHandler != nil {

339+ authHeader, err := c.oauthHandler.GetAuthorizationHeader(ctx)

340+ if err != nil {

341+ // If we get an authorization error, return a specific error that can be handled by the client

342+ if err.Error() == "no valid token available, authorization required" {

343+ return nil, &OAuthAuthorizationRequiredError{

344+ Handler: c.oauthHandler,

345+ }

346+ }

347+ return nil, fmt.Errorf("failed to get authorization header: %w", err)

348+ }

349+ req.Header.Set("Authorization", authHeader)

350+ }

351+

352+ if c.headerFunc != nil {

353+ for k, v := range c.headerFunc(ctx) {

354+ req.Header.Set(k, v)

355+ }

356+ }

357+

358+ // Send request

359+ resp, err = c.httpClient.Do(req)

360+ if err != nil {

361+ return nil, fmt.Errorf("failed to send request: %w", err)

362+ }

363+

364+ // universal handling for session terminated

365+ if resp.StatusCode == http.StatusNotFound {

366+ c.sessionID.CompareAndSwap(sessionID, "")

367+ return nil, ErrSessionTerminated

368+ }

369+

370+ return resp, nil

371+}

372+

310// handleSSEResponse processes an SSE stream for a specific request.

373// handleSSEResponse processes an SSE stream for a specific request.

311// It returns the final result for the request once received, or an error.

374// It returns the final result for the request once received, or an error.

375+// If ignoreResponse is true, it won't return when a response messge is received. This is for continuous listening.

312−func (c *StreamableHTTP) handleSSEResponse(ctx context.Context, reader io.ReadCloser) (*JSONRPCResponse, error) {

376+func (c *StreamableHTTP) handleSSEResponse(ctx context.Context, reader io.ReadCloser, ignoreResponse bool) (*JSONRPCResponse, error) {

313

377

314 // Create a channel for this specific request

378 // Create a channel for this specific request

315 responseChan := make(chan *JSONRPCResponse, 1)

379 responseChan := make(chan *JSONRPCResponse, 1)

@@ -328,7 +392,7 @@ func (c *StreamableHTTP) handleSSEResponse(ctx context.Context, reader io.ReadCl

328

392

329 var message JSONRPCResponse

393 var message JSONRPCResponse

330 if err := json.Unmarshal([]byte(data), &message); err != nil {

394 if err := json.Unmarshal([]byte(data), &message); err != nil {

331− fmt.Printf("failed to unmarshal message: %v\n", err)

395+ c.logger.Errorf("failed to unmarshal message: %v", err)

332 return

396 return

333 }

397 }

334

398

@@ -336,7 +400,7 @@ func (c *StreamableHTTP) handleSSEResponse(ctx context.Context, reader io.ReadCl

336 if message.ID.IsNil() {

400 if message.ID.IsNil() {

337 var notification mcp.JSONRPCNotification

401 var notification mcp.JSONRPCNotification

338 if err := json.Unmarshal([]byte(data), &notification); err != nil {

402 if err := json.Unmarshal([]byte(data), &notification); err != nil {

339− fmt.Printf("failed to unmarshal notification: %v\n", err)

403+ c.logger.Errorf("failed to unmarshal notification: %v", err)

340 return

404 return

341 }

405 }

342 c.notifyMu.RLock()

406 c.notifyMu.RLock()

@@ -347,7 +411,9 @@ func (c *StreamableHTTP) handleSSEResponse(ctx context.Context, reader io.ReadCl

347 return

411 return

348 }

412 }

349

413

414+ if !ignoreResponse {

350− responseChan <- &message

415+ responseChan <- &message

416+ }

351 })

417 })

352 }()

418 }()

353

419

@@ -393,7 +459,7 @@ func (c *StreamableHTTP) readSSE(ctx context.Context, reader io.ReadCloser, hand

393 case <-ctx.Done():

459 case <-ctx.Done():

394 return

460 return

395 default:

461 default:

396− fmt.Printf("SSE stream error: %v\n", err)

462+ c.logger.Errorf("SSE stream error: %v", err)

397 return

463 return

398 }

464 }

399 }

465 }

@@ -432,44 +498,10 @@ func (c *StreamableHTTP) SendNotification(ctx context.Context, notification mcp.

432 }

498 }

433

499

434 // Create HTTP request

500 // Create HTTP request

435− req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.serverURL.String(), bytes.NewReader(requestBody))

501+ ctx, cancel := c.contextAwareOfClientClose(ctx)

436− if err != nil {

502+ defer cancel()

437− return fmt.Errorf("failed to create request: %w", err)

438− }

439−

440− // Set headers

441− req.Header.Set("Content-Type", "application/json")

442− req.Header.Set("Accept", "application/json, text/event-stream")

443− if sessionID := c.sessionID.Load(); sessionID != "" {

444− req.Header.Set(headerKeySessionID, sessionID.(string))

445− }

446− for k, v := range c.headers {

447− req.Header.Set(k, v)

448− }

449−

450− // Add OAuth authorization if configured

451− if c.oauthHandler != nil {

452− authHeader, err := c.oauthHandler.GetAuthorizationHeader(ctx)

453− if err != nil {

454− // If we get an authorization error, return a specific error that can be handled by the client

455− if errors.Is(err, ErrOAuthAuthorizationRequired) {

456− return &OAuthAuthorizationRequiredError{

457− Handler: c.oauthHandler,

458− }

459− }

460− return fmt.Errorf("failed to get authorization header: %w", err)

461− }

462− req.Header.Set("Authorization", authHeader)

463− }

464−

465− if c.headerFunc != nil {

466− for k, v := range c.headerFunc(ctx) {

467− req.Header.Set(k, v)

468− }

469− }

470

503

471− // Send request

504+ resp, err := c.sendHTTP(ctx, http.MethodPost, bytes.NewReader(requestBody), "application/json, text/event-stream")

472− resp, err := c.httpClient.Do(req)

473 if err != nil {

505 if err != nil {

474 return fmt.Errorf("failed to send request: %w", err)

506 return fmt.Errorf("failed to send request: %w", err)

475 }

507 }

@@ -513,3 +545,84 @@ func (c *StreamableHTTP) GetOAuthHandler() *OAuthHandler {

513func (c *StreamableHTTP) IsOAuthEnabled() bool {

545func (c *StreamableHTTP) IsOAuthEnabled() bool {

514 return c.oauthHandler != nil

546 return c.oauthHandler != nil

515}

547}

548+

549+func (c *StreamableHTTP) listenForever(ctx context.Context) {

550+ c.logger.Infof("listening to server forever")

551+ for {

552+ err := c.createGETConnectionToServer(ctx)

553+ if errors.Is(err, ErrGetMethodNotAllowed) {

554+ // server does not support listening

555+ c.logger.Errorf("server does not support listening")

556+ return

557+ }

558+

559+ select {

560+ case <-ctx.Done():

561+ return

562+ default:

563+ }

564+

565+ if err != nil {

566+ c.logger.Errorf("failed to listen to server. retry in 1 second: %v", err)

567+ }

568+ time.Sleep(retryInterval)

569+ }

570+}

571+

572+var (

573+ ErrSessionTerminated = fmt.Errorf("session terminated (404). need to re-initialize")

574+ ErrGetMethodNotAllowed = fmt.Errorf("GET method not allowed")

575+

576+ retryInterval = 1 * time.Second // a variable is convenient for testing

577+)

578+

579+func (c *StreamableHTTP) createGETConnectionToServer(ctx context.Context) error {

580+

581+ resp, err := c.sendHTTP(ctx, http.MethodGet, nil, "text/event-stream")

582+ if err != nil {

583+ return fmt.Errorf("failed to send request: %w", err)

584+ }

585+ defer resp.Body.Close()

586+

587+ // Check if we got an error response

588+ if resp.StatusCode == http.StatusMethodNotAllowed {

589+ return ErrGetMethodNotAllowed

590+ }

591+

592+ if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusAccepted {

593+ body, _ := io.ReadAll(resp.Body)

594+ return fmt.Errorf("request failed with status %d: %s", resp.StatusCode, body)

595+ }

596+

597+ // handle SSE response

598+ contentType := resp.Header.Get("Content-Type")

599+ if contentType != "text/event-stream" {

600+ return fmt.Errorf("unexpected content type: %s", contentType)

601+ }

602+

603+ // When ignoreResponse is true, the function will never return expect context is done.

604+ // NOTICE: Due to the ambiguity of the specification, other SDKs may use the GET connection to transfer the response

605+ // messages. To be more compatible, we should handle this response, however, as the transport layer is message-based,

606+ // currently, there is no convenient way to handle this response.

607+ // So we ignore the response here. It's not a bug, but may be not compatible with other SDKs.

608+ _, err = c.handleSSEResponse(ctx, resp.Body, true)

609+ if err != nil {

610+ return fmt.Errorf("failed to handle SSE response: %w", err)

611+ }

612+

613+ return nil

614+}

615+

616+func (c *StreamableHTTP) contextAwareOfClientClose(ctx context.Context) (context.Context, context.CancelFunc) {

617+ newCtx, cancel := context.WithCancel(ctx)

618+ go func() {

619+ select {

620+ case <-c.closed:

621+ cancel()

622+ case <-newCtx.Done():

623+ // The original context was canceled

624+ cancel()

625+ }

626+ }()

627+ return newCtx, cancel

628+}

vendor/github.com/mark3labs/mcp-go/mcp/tools.go+105−1

@@ -945,7 +945,20 @@ func PropertyNames(schema map[string]any) PropertyOption {

945 }

946}

947

948−// Items defines the schema for array items

948+// Items defines the schema for array items.

949+// Accepts any schema definition for maximum flexibility.

950+//

951+// Example:

952+//

953+// Items(map[string]any{

954+// "type": "object",

955+// "properties": map[string]any{

956+// "name": map[string]any{"type": "string"},

957+// "age": map[string]any{"type": "number"},

958+// },

959+// })

960+//

961+// For simple types, use ItemsString(), ItemsNumber(), ItemsBoolean() instead.

949func Items(schema any) PropertyOption {

962func Items(schema any) PropertyOption {

950 return func(schemaMap map[string]any) {

963 return func(schemaMap map[string]any) {

951 schemaMap["items"] = schema

964 schemaMap["items"] = schema

@@ -972,3 +985,94 @@ func UniqueItems(unique bool) PropertyOption {

972 schema["uniqueItems"] = unique

985 schema["uniqueItems"] = unique

973 }

986 }

974}

987}

988+

989+// WithStringItems configures an array's items to be of type string.

990+//

991+// Supported options: Description(), DefaultString(), Enum(), MaxLength(), MinLength(), Pattern()

992+// Note: Options like Required() are not valid for item schemas and will be ignored.

993+//

994+// Examples:

995+//

996+// mcp.WithArray("tags", mcp.WithStringItems())

997+// mcp.WithArray("colors", mcp.WithStringItems(mcp.Enum("red", "green", "blue")))

998+// mcp.WithArray("names", mcp.WithStringItems(mcp.MinLength(1), mcp.MaxLength(50)))

999+//

1000+// Limitations: Only supports simple string arrays. Use Items() for complex objects.

1001+func WithStringItems(opts ...PropertyOption) PropertyOption {

1002+ return func(schema map[string]any) {

1003+ itemSchema := map[string]any{

1004+ "type": "string",

1005+ }

1006+

1007+ for _, opt := range opts {

1008+ opt(itemSchema)

1009+ }

1010+

1011+ schema["items"] = itemSchema

1012+ }

1013+}

1014+

1015+// WithStringEnumItems configures an array's items to be of type string with a specified enum.

1016+// Example:

1017+//

1018+// mcp.WithArray("priority", mcp.WithStringEnumItems([]string{"low", "medium", "high"}))

1019+//

1020+// Limitations: Only supports string enums. Use WithStringItems(Enum(...)) for more flexibility.

1021+func WithStringEnumItems(values []string) PropertyOption {

1022+ return func(schema map[string]any) {

1023+ schema["items"] = map[string]any{

1024+ "type": "string",

1025+ "enum": values,

1026+ }

1027+ }

1028+}

1029+

1030+// WithNumberItems configures an array's items to be of type number.

1031+//

1032+// Supported options: Description(), DefaultNumber(), Min(), Max(), MultipleOf()

1033+// Note: Options like Required() are not valid for item schemas and will be ignored.

1034+//

1035+// Examples:

1036+//

1037+// mcp.WithArray("scores", mcp.WithNumberItems(mcp.Min(0), mcp.Max(100)))

1038+// mcp.WithArray("prices", mcp.WithNumberItems(mcp.Min(0)))

1039+//

1040+// Limitations: Only supports simple number arrays. Use Items() for complex objects.

1041+func WithNumberItems(opts ...PropertyOption) PropertyOption {

1042+ return func(schema map[string]any) {

1043+ itemSchema := map[string]any{

1044+ "type": "number",

1045+ }

1046+

1047+ for _, opt := range opts {

1048+ opt(itemSchema)

1049+ }

1050+

1051+ schema["items"] = itemSchema

1052+ }

1053+}

1054+

1055+// WithBooleanItems configures an array's items to be of type boolean.

1056+//

1057+// Supported options: Description(), DefaultBool()

1058+// Note: Options like Required() are not valid for item schemas and will be ignored.

1059+//

1060+// Examples:

1061+//

1062+// mcp.WithArray("flags", mcp.WithBooleanItems())

1063+// mcp.WithArray("permissions", mcp.WithBooleanItems(mcp.Description("User permissions")))

1064+//

1065+// Limitations: Only supports simple boolean arrays. Use Items() for complex objects.

1066+func WithBooleanItems(opts ...PropertyOption) PropertyOption {

1067+ return func(schema map[string]any) {

1068+ itemSchema := map[string]any{

1069+ "type": "boolean",

1070+ }

1071+

1072+ for _, opt := range opts {

1073+ opt(itemSchema)

1074+ }

1075+

1076+ schema["items"] = itemSchema

1077+ }

1078+}

vendor/github.com/mark3labs/mcp-go/mcp/types.go+21

@@ -763,6 +763,11 @@ const (

763

764/* Sampling */

765

766+const (

767+ // MethodSamplingCreateMessage allows servers to request LLM completions from clients

768+ MethodSamplingCreateMessage MCPMethod = "sampling/createMessage"

769+)

770+

766// CreateMessageRequest is a request from the server to sample an LLM via the

771// CreateMessageRequest is a request from the server to sample an LLM via the

767// client. The client has full discretion over which model to select. The client

772// client. The client has full discretion over which model to select. The client

768// should also inform the user before beginning sampling, to allow them to inspect

773// should also inform the user before beginning sampling, to allow them to inspect

@@ -865,6 +870,22 @@ type AudioContent struct {

865

870

866func (AudioContent) isContent() {}

871func (AudioContent) isContent() {}

867

872

873+// ResourceLink represents a link to a resource that the client can access.

874+type ResourceLink struct {

875+ Annotated

876+ Type string `json:"type"` // Must be "resource_link"

877+ // The URI of the resource.

878+ URI string `json:"uri"`

879+ // The name of the resource.

880+ Name string `json:"name"`

881+ // The description of the resource.

882+ Description string `json:"description"`

883+ // The MIME type of the resource.

884+ MIMEType string `json:"mimeType"`

885+}

886+

887+func (ResourceLink) isContent() {}

888+

868// EmbeddedResource represents the contents of a resource, embedded into a prompt or tool call result.

889// EmbeddedResource represents the contents of a resource, embedded into a prompt or tool call result.

869//

890//

870// It is up to the client how best to render embedded resources for the

891// It is up to the client how best to render embedded resources for the

vendor/github.com/mark3labs/mcp-go/mcp/utils.go+21

@@ -222,6 +222,17 @@ func NewAudioContent(data, mimeType string) AudioContent {

222 }

223}

224

225+// Helper function to create a new ResourceLink

226+func NewResourceLink(uri, name, description, mimeType string) ResourceLink {

227+ return ResourceLink{

228+ Type: "resource_link",

229+ URI: uri,

230+ Name: name,

231+ Description: description,

232+ MIMEType: mimeType,

233+ }

234+}

235+

225// Helper function to create a new EmbeddedResource

236// Helper function to create a new EmbeddedResource

226func NewEmbeddedResource(resource ResourceContents) EmbeddedResource {

237func NewEmbeddedResource(resource ResourceContents) EmbeddedResource {

227 return EmbeddedResource{

238 return EmbeddedResource{

@@ -476,6 +487,16 @@ func ParseContent(contentMap map[string]any) (Content, error) {

476 }

487 }

477 return NewAudioContent(data, mimeType), nil

488 return NewAudioContent(data, mimeType), nil

478

489

490+ case "resource_link":

491+ uri := ExtractString(contentMap, "uri")

492+ name := ExtractString(contentMap, "name")

493+ description := ExtractString(contentMap, "description")

494+ mimeType := ExtractString(contentMap, "mimeType")

495+ if uri == "" || name == "" {

496+ return nil, fmt.Errorf("resource_link uri or name is missing")

497+ }

498+ return NewResourceLink(uri, name, description, mimeType), nil

499+

479 case "resource":

500 case "resource":

480 resourceMap := ExtractMap(contentMap, "resource")

501 resourceMap := ExtractMap(contentMap, "resource")

481 if resourceMap == nil {

502 if resourceMap == nil {

vendor/github.com/mark3labs/mcp-go/server/sampling.gonew file+37

@@ -0,0 +1,37 @@

1+package server

3+import (

4+ "context"

5+ "fmt"

7+ "github.com/mark3labs/mcp-go/mcp"

8+)

10+// EnableSampling enables sampling capabilities for the server.

11+// This allows the server to send sampling requests to clients that support it.

12+func (s *MCPServer) EnableSampling() {

13+ s.capabilitiesMu.Lock()

14+ defer s.capabilitiesMu.Unlock()

15+}

16+

17+// RequestSampling sends a sampling request to the client.

18+// The client must have declared sampling capability during initialization.

19+func (s *MCPServer) RequestSampling(ctx context.Context, request mcp.CreateMessageRequest) (*mcp.CreateMessageResult, error) {

20+ session := ClientSessionFromContext(ctx)

21+ if session == nil {

22+ return nil, fmt.Errorf("no active session")

23+ }

24+

25+ // Check if the session supports sampling requests

26+ if samplingSession, ok := session.(SessionWithSampling); ok {

27+ return samplingSession.RequestSampling(ctx, request)

28+ }

29+

30+ return nil, fmt.Errorf("session does not support sampling")

31+}

32+

33+// SessionWithSampling extends ClientSession to support sampling requests.

34+type SessionWithSampling interface {

35+ ClientSession

36+ RequestSampling(ctx context.Context, request mcp.CreateMessageRequest) (*mcp.CreateMessageResult, error)

37+}

vendor/github.com/mark3labs/mcp-go/server/stdio.go+174−6

@@ -9,6 +9,7 @@ import (

9 "log"

10 "os"

11 "os/signal"

12+ "sync"

12 "sync/atomic"

13 "sync/atomic"

13 "syscall"

14 "syscall"

@@ -51,10 +52,21 @@ func WithStdioContextFunc(fn StdioContextFunc) StdioOption {

52// stdioSession is a static client session, since stdio has only one client.

53// stdioSession is a static client session, since stdio has only one client.

53type stdioSession struct {

54type stdioSession struct {

54− notifications chan mcp.JSONRPCNotification

55+ notifications chan mcp.JSONRPCNotification

55− initialized atomic.Bool

56+ initialized atomic.Bool

56− loggingLevel atomic.Value

57+ loggingLevel atomic.Value

57− clientInfo atomic.Value // stores session-specific client info

58+ clientInfo atomic.Value // stores session-specific client info

59+ writer io.Writer // for sending requests to client

60+ requestID atomic.Int64 // for generating unique request IDs

61+ mu sync.RWMutex // protects writer

62+ pendingRequests map[int64]chan *samplingResponse // for tracking pending sampling requests

63+ pendingMu sync.RWMutex // protects pendingRequests

64+}

65+

66+// samplingResponse represents a response to a sampling request

67+type samplingResponse struct {

68+ result *mcp.CreateMessageResult

69+ err error

58}

70}

60func (s *stdioSession) SessionID() string {

72func (s *stdioSession) SessionID() string {

@@ -100,14 +112,86 @@ func (s *stdioSession) GetLogLevel() mcp.LoggingLevel {

100 return level.(mcp.LoggingLevel)

112 return level.(mcp.LoggingLevel)

101}

113}

102

114

115+// RequestSampling sends a sampling request to the client and waits for the response.

116+func (s *stdioSession) RequestSampling(ctx context.Context, request mcp.CreateMessageRequest) (*mcp.CreateMessageResult, error) {

117+ s.mu.RLock()

118+ writer := s.writer

119+ s.mu.RUnlock()

120+

121+ if writer == nil {

122+ return nil, fmt.Errorf("no writer available for sending requests")

123+ }

124+

125+ // Generate a unique request ID

126+ id := s.requestID.Add(1)

127+

128+ // Create a response channel for this request

129+ responseChan := make(chan *samplingResponse, 1)

130+ s.pendingMu.Lock()

131+ s.pendingRequests[id] = responseChan

132+ s.pendingMu.Unlock()

133+

134+ // Cleanup function to remove the pending request

135+ cleanup := func() {

136+ s.pendingMu.Lock()

137+ delete(s.pendingRequests, id)

138+ s.pendingMu.Unlock()

139+ }

140+ defer cleanup()

141+

142+ // Create the JSON-RPC request

143+ jsonRPCRequest := struct {

144+ JSONRPC string `json:"jsonrpc"`

145+ ID int64 `json:"id"`

146+ Method string `json:"method"`

147+ Params mcp.CreateMessageParams `json:"params"`

148+ }{

149+ JSONRPC: mcp.JSONRPC_VERSION,

150+ ID: id,

151+ Method: string(mcp.MethodSamplingCreateMessage),

152+ Params: request.CreateMessageParams,

153+ }

154+

155+ // Marshal and send the request

156+ requestBytes, err := json.Marshal(jsonRPCRequest)

157+ if err != nil {

158+ return nil, fmt.Errorf("failed to marshal sampling request: %w", err)

159+ }

160+ requestBytes = append(requestBytes, '\n')

161+

162+ if _, err := writer.Write(requestBytes); err != nil {

163+ return nil, fmt.Errorf("failed to write sampling request: %w", err)

164+ }

165+

166+ // Wait for the response or context cancellation

167+ select {

168+ case <-ctx.Done():

169+ return nil, ctx.Err()

170+ case response := <-responseChan:

171+ if response.err != nil {

172+ return nil, response.err

173+ }

174+ return response.result, nil

175+ }

176+}

177+

178+// SetWriter sets the writer for sending requests to the client.

179+func (s *stdioSession) SetWriter(writer io.Writer) {

180+ s.mu.Lock()

181+ defer s.mu.Unlock()

182+ s.writer = writer

183+}

184+

103var (

185var (

104 _ ClientSession = (*stdioSession)(nil)

186 _ ClientSession = (*stdioSession)(nil)

105 _ SessionWithLogging = (*stdioSession)(nil)

187 _ SessionWithLogging = (*stdioSession)(nil)

106 _ SessionWithClientInfo = (*stdioSession)(nil)

188 _ SessionWithClientInfo = (*stdioSession)(nil)

189+ _ SessionWithSampling = (*stdioSession)(nil)

107)

190)

108

191

109var stdioSessionInstance = stdioSession{

192var stdioSessionInstance = stdioSession{

110− notifications: make(chan mcp.JSONRPCNotification, 100),

193+ notifications: make(chan mcp.JSONRPCNotification, 100),

194+ pendingRequests: make(map[int64]chan *samplingResponse),

111}

195}

112

196

113// NewStdioServer creates a new stdio server wrapper around an MCPServer.

197// NewStdioServer creates a new stdio server wrapper around an MCPServer.

@@ -224,6 +308,9 @@ func (s *StdioServer) Listen(

224 defer s.server.UnregisterSession(ctx, stdioSessionInstance.SessionID())

308 defer s.server.UnregisterSession(ctx, stdioSessionInstance.SessionID())

225 ctx = s.server.WithContext(ctx, &stdioSessionInstance)

309 ctx = s.server.WithContext(ctx, &stdioSessionInstance)

226

310

311+ // Set the writer for sending requests to the client

312+ stdioSessionInstance.SetWriter(stdout)

313+

227 // Add in any custom context.

314 // Add in any custom context.

228 if s.contextFunc != nil {

315 if s.contextFunc != nil {

229 ctx = s.contextFunc(ctx)

316 ctx = s.contextFunc(ctx)

@@ -256,7 +343,29 @@ func (s *StdioServer) processMessage(

256 return s.writeResponse(response, writer)

343 return s.writeResponse(response, writer)

257 }

344 }

258

345

259− // Handle the message using the wrapped server

346+ // Check if this is a response to a sampling request

347+ if s.handleSamplingResponse(rawMessage) {

348+ return nil

349+ }

350+

351+ // Check if this is a tool call that might need sampling (and thus should be processed concurrently)

352+ var baseMessage struct {

353+ Method string `json:"method"`

354+ }

355+ if json.Unmarshal(rawMessage, &baseMessage) == nil && baseMessage.Method == "tools/call" {

356+ // Process tool calls concurrently to avoid blocking on sampling requests

357+ go func() {

358+ response := s.server.HandleMessage(ctx, rawMessage)

359+ if response != nil {

360+ if err := s.writeResponse(response, writer); err != nil {

361+ s.errLogger.Printf("Error writing tool response: %v", err)

362+ }

363+ }

364+ }()

365+ return nil

366+ }

367+

368+ // Handle other messages synchronously

260 response := s.server.HandleMessage(ctx, rawMessage)

369 response := s.server.HandleMessage(ctx, rawMessage)

261

370

262 // Only write response if there is one (not for notifications)

371 // Only write response if there is one (not for notifications)

@@ -269,6 +378,65 @@ func (s *StdioServer) processMessage(

269 return nil

378 return nil

270}

379}

271

380

381+// handleSamplingResponse checks if the message is a response to a sampling request

382+// and routes it to the appropriate pending request channel.

383+func (s *StdioServer) handleSamplingResponse(rawMessage json.RawMessage) bool {

384+ return stdioSessionInstance.handleSamplingResponse(rawMessage)

385+}

386+

387+// handleSamplingResponse handles incoming sampling responses for this session

388+func (s *stdioSession) handleSamplingResponse(rawMessage json.RawMessage) bool {

389+ // Try to parse as a JSON-RPC response

390+ var response struct {

391+ JSONRPC string `json:"jsonrpc"`

392+ ID json.Number `json:"id"`

393+ Result json.RawMessage `json:"result,omitempty"`

394+ Error *struct {

395+ Code int `json:"code"`

396+ Message string `json:"message"`

397+ } `json:"error,omitempty"`

398+ }

399+

400+ if err := json.Unmarshal(rawMessage, &response); err != nil {

401+ return false

402+ }

403+ // Parse the ID as int64

404+ idInt64, err := response.ID.Int64()

405+ if err != nil || (response.Result == nil && response.Error == nil) {

406+ return false

407+ }

408+

409+ // Look for a pending request with this ID

410+ s.pendingMu.RLock()

411+ responseChan, exists := s.pendingRequests[idInt64]

412+ s.pendingMu.RUnlock()

413+

414+ if !exists {

415+ return false

416+ } // Parse and send the response

417+ samplingResp := &samplingResponse{}

418+

419+ if response.Error != nil {

420+ samplingResp.err = fmt.Errorf("sampling request failed: %s", response.Error.Message)

421+ } else {

422+ var result mcp.CreateMessageResult

423+ if err := json.Unmarshal(response.Result, &result); err != nil {

424+ samplingResp.err = fmt.Errorf("failed to unmarshal sampling response: %w", err)

425+ } else {

426+ samplingResp.result = &result

427+ }

428+ }

429+

430+ // Send the response (non-blocking)

431+ select {

432+ case responseChan <- samplingResp:

433+ default:

434+ // Channel is full or closed, ignore

435+ }

436+

437+ return true

438+}

439+

272// writeResponse marshals and writes a JSON-RPC response message followed by a newline.

440// writeResponse marshals and writes a JSON-RPC response message followed by a newline.

273// Returns an error if marshaling or writing fails.

441// Returns an error if marshaling or writing fails.

274func (s *StdioServer) writeResponse(

442func (s *StdioServer) writeResponse(

vendor/github.com/mark3labs/mcp-go/server/streamable_http.go+4−2

@@ -40,7 +40,9 @@ func WithEndpointPath(endpointPath string) StreamableHTTPOption {

40// to StatelessSessionIdManager.

41func WithStateLess(stateLess bool) StreamableHTTPOption {

42 return func(s *StreamableHTTPServer) {

43+ if stateLess {

43− s.sessionIdManager = &StatelessSessionIdManager{}

44+ s.sessionIdManager = &StatelessSessionIdManager{}

45+ }

44 }

46 }

45}

47}

@@ -374,7 +376,7 @@ func (s *StreamableHTTPServer) handleGet(w http.ResponseWriter, r *http.Request)

374 w.Header().Set("Content-Type", "text/event-stream")

376 w.Header().Set("Content-Type", "text/event-stream")

375 w.Header().Set("Cache-Control", "no-cache")

377 w.Header().Set("Cache-Control", "no-cache")

376 w.Header().Set("Connection", "keep-alive")

378 w.Header().Set("Connection", "keep-alive")

377− w.WriteHeader(http.StatusAccepted)

379+ w.WriteHeader(http.StatusOK)

378

380

379 flusher, ok := w.(http.Flusher)

381 flusher, ok := w.(http.Flusher)

380 if !ok {

382 if !ok {

vendor/modules.txt+1−1

@@ -403,7 +403,7 @@ github.com/kylelemons/godebug/pretty

403# github.com/lucasb-eyer/go-colorful v1.2.0

404## explicit; go 1.12

405github.com/lucasb-eyer/go-colorful

406−# github.com/mark3labs/mcp-go v0.32.0

406+# github.com/mark3labs/mcp-go v0.33.0

407## explicit; go 1.23

408github.com/mark3labs/mcp-go/client

409github.com/mark3labs/mcp-go/client/transport