feat(streamable_http): elicitation request

Author: Ghosthell

Author: JBUinfo

client/client.go

@@ -478,6 +478,8 @@ func (c *Client) handleIncomingRequest(ctx context.Context, request transport.JS
 		return c.handleSamplingRequestTransport(ctx, request)
 	case string(mcp.MethodElicitationCreate):
 		return c.handleElicitationRequestTransport(ctx, request)
+	case string(mcp.MethodPing):
+		return c.handlePingRequestTransport(ctx, request)
 	default:
 		return nil, fmt.Errorf("unsupported request method: %s", request.Method)
 	}
@@ -579,6 +581,15 @@ func (c *Client) handleElicitationRequestTransport(ctx context.Context, request
 	return response, nil
 }
 
+func (c *Client) handlePingRequestTransport(ctx context.Context, request transport.JSONRPCRequest) (*transport.JSONRPCResponse, error) {
+	b, _ := json.Marshal(&mcp.EmptyResult{})
+	return &transport.JSONRPCResponse{
+		JSONRPC: mcp.JSONRPC_VERSION,
+		ID:      request.ID,
+		Result:  b,
+	}, nil
+}
+
 func listByPage[T any](
 	ctx context.Context,
 	client *Client,

server/streamable_http.go

@@ -473,6 +473,21 @@ func (s *StreamableHTTPServer) handleGet(w http.ResponseWriter, r *http.Request)
 				case <-done:
 					return
 				}
+			case elicitationReq := <-session.elicitationRequestChan:
+				// Send elicitation request to client via SSE
+				jsonrpcRequest := mcp.JSONRPCRequest{
+					JSONRPC: "2.0",
+					ID:      mcp.NewRequestId(elicitationReq.requestID),
+					Request: mcp.Request{
+						Method: string(mcp.MethodElicitationCreate),
+					},
+					Params: elicitationReq.request.Params,
+				}
+				select {
+				case writeChan <- jsonrpcRequest:
+				case <-done:
+					return
+				}
 			case <-done:
 				return
 			}
@@ -612,12 +627,6 @@ func (s *StreamableHTTPServer) handleSamplingResponse(w http.ResponseWriter, r *
 		}
 	} else if responseMessage.Result != nil {
 		// Parse result
-		var result mcp.CreateMessageResult
-		if err := json.Unmarshal(responseMessage.Result, &result); err != nil {
-			response.err = fmt.Errorf("failed to parse sampling result: %v", err)
-		} else {
-			response.result = &result
-		}
 	} else {
 		response.err = fmt.Errorf("sampling response has neither result nor error")
 	}
@@ -764,10 +773,17 @@ type samplingRequestItem struct {
 
 type samplingResponseItem struct {
 	requestID int64
-	result    *mcp.CreateMessageResult
+	result    json.RawMessage
 	err       error
 }
 
+// Elicitation support types for HTTP transport
+type elicitationRequestItem struct {
+	requestID int64
+	request   mcp.ElicitationRequest
+	response  chan samplingResponseItem
+}
+
 // streamableHttpSession is a session for streamable-http transport
 // When in POST handlers(request/notification), it's ephemeral, and only exists in the life of the request handler.
 // When in GET handlers(listening), it's a real session, and will be registered in the MCP server.
@@ -779,18 +795,21 @@ type streamableHttpSession struct {
 	logLevels           *sessionLogLevelsStore
 
 	// Sampling support for bidirectional communication
-	samplingRequestChan chan samplingRequestItem // server -> client sampling requests
-	samplingRequests    sync.Map                 // requestID -> pending sampling request context
-	requestIDCounter    atomic.Int64             // for generating unique request IDs
+	samplingRequestChan    chan samplingRequestItem    // server -> client sampling requests
+	elicitationRequestChan chan elicitationRequestItem // server -> client elicitation requests
+
+	samplingRequests sync.Map     // requestID -> pending sampling request context
+	requestIDCounter atomic.Int64 // for generating unique request IDs
 }
 
 func newStreamableHttpSession(sessionID string, toolStore *sessionToolsStore, levels *sessionLogLevelsStore) *streamableHttpSession {
 	s := &streamableHttpSession{
-		sessionID:           sessionID,
-		notificationChannel: make(chan mcp.JSONRPCNotification, 100),
-		tools:               toolStore,
-		logLevels:           levels,
-		samplingRequestChan: make(chan samplingRequestItem, 10),
+		sessionID:              sessionID,
+		notificationChannel:    make(chan mcp.JSONRPCNotification, 100),
+		tools:                  toolStore,
+		logLevels:              levels,
+		samplingRequestChan:    make(chan samplingRequestItem, 10),
+		elicitationRequestChan: make(chan elicitationRequestItem, 10),
 	}
 	return s
 }
@@ -877,13 +896,63 @@ func (s *streamableHttpSession) RequestSampling(ctx context.Context, request mcp
 		if response.err != nil {
 			return nil, response.err
 		}
-		return response.result, nil
+		var result mcp.CreateMessageResult
+		if err := json.Unmarshal(response.result, &result); err != nil {
+			return nil, fmt.Errorf("failed to unmarshal sampling response: %v", err)
+		}
+		return &result, nil
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	}
+}
+
+// RequestElicitation implements SessionWithElicitation interface for HTTP transport
+func (s *streamableHttpSession) RequestElicitation(ctx context.Context, request mcp.ElicitationRequest) (*mcp.ElicitationResult, error) {
+	// Generate unique request ID
+	requestID := s.requestIDCounter.Add(1)
+
+	// Create response channel for this specific request
+	responseChan := make(chan samplingResponseItem, 1)
+
+	// Create the sampling request item
+	elicitationRequest := elicitationRequestItem{
+		requestID: requestID,
+		request:   request,
+		response:  responseChan,
+	}
+
+	// Store the pending request
+	s.samplingRequests.Store(requestID, responseChan)
+	defer s.samplingRequests.Delete(requestID)
+
+	// Send the sampling request via the channel (non-blocking)
+	select {
+	case s.elicitationRequestChan <- elicitationRequest:
+		// Request queued successfully
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	default:
+		return nil, fmt.Errorf("elicitation request queue is full - server overloaded")
+	}
+
+	// Wait for response or context cancellation
+	select {
+	case response := <-responseChan:
+		if response.err != nil {
+			return nil, response.err
+		}
+		var result mcp.ElicitationResult
+		if err := json.Unmarshal(response.result, &result); err != nil {
+			return nil, fmt.Errorf("failed to unmarshal elicitation response: %v", err)
+		}
+		return &result, nil
 	case <-ctx.Done():
 		return nil, ctx.Err()
 	}
 }
 
 var _ SessionWithSampling = (*streamableHttpSession)(nil)
+var _ SessionWithElicitation = (*streamableHttpSession)(nil)
 
 // --- session id manager ---
 
@@ -952,6 +1021,7 @@ func NewTestStreamableHTTPServer(server *MCPServer, opts ...StreamableHTTPOption
 //   - null
 //   - empty object: {}
 //   - empty array: []
+//
 // It also treats nil/whitespace-only input as empty.
 // It does NOT treat 0, false, "" or non-empty composites as empty.
 func isJSONEmpty(data json.RawMessage) bool {