fix: SSE parser now correctly handles events without event field (#376)

* fix: SSE parser now correctly handles events without event field

According to the W3C SSE specification, the event field is optional.
SSE events with only data fields should be processed normally and
treated as "message" events (the default event type).

This fixes issue #369 where the SSE parser incorrectly required both
event and data fields, causing "unexpected nil response" errors when
servers sent valid SSE events with only data fields.

🤖 Generated with opencode
Co-Authored-By: opencode <[email protected]>

* fix: resolve race condition in SSE test

Fixed race condition in TestSSE/SSEEventWithoutEventField where multiple
goroutines were accessing the same HTTP response writer concurrently.
Simplified the test to use channel-based synchronization instead of
shared state between HTTP handlers.

🤖 Generated with opencode
Co-Authored-By: opencode <[email protected]>

* fix: StreamableHTTP SSE parser now correctly handles events without event field

- Fixed SSE parser in StreamableHTTP transport to process events with only data field
- According to W3C SSE specification, the event field is optional
- When no event type is specified, defaults to "message" event type
- Added comprehensive test case to verify the fix works correctly
- Resolves issue #369 where SSE events without event field caused "unexpected nil response" errors

🤖 Generated with opencode
Co-Authored-By: opencode <[email protected]>

---------

Co-authored-by: opencode <[email protected]>

Author: ezynda3

client/transport/sse.go

@@ -194,7 +194,11 @@ func (c *SSE) readSSE(reader io.ReadCloser) {
 		if err != nil {
 			if err == io.EOF {
 				// Process any pending event before exit
-				if event != "" && data != "" {
+				if data != "" {
+					// If no event type is specified, use empty string (default event type)
+					if event == "" {
+						event = "message"
+					}
 					c.handleSSEEvent(event, data)
 				}
 				break
@@ -209,7 +213,11 @@ func (c *SSE) readSSE(reader io.ReadCloser) {
 		line = strings.TrimRight(line, "\r\n")
 		if line == "" {
 			// Empty line means end of event
-			if event != "" && data != "" {
+			if data != "" {
+				// If no event type is specified, use empty string (default event type)
+				if event == "" {
+					event = "message"
+				}
 				c.handleSSEEvent(event, data)
 				event = ""
 				data = ""

client/transport/sse_test.go

@@ -405,6 +405,109 @@ func TestSSE(t *testing.T) {
 		}
 	})
 
+	t.Run("SSEEventWithoutEventField", func(t *testing.T) {
+		// Test that SSE events with only data field (no event field) are processed correctly
+		// This tests the fix for issue #369
+		
+		var messageReceived chan struct{}
+		
+		// Create a custom mock server that sends SSE events without event field
+		sseHandler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			w.Header().Set("Content-Type", "text/event-stream")
+			flusher, ok := w.(http.Flusher)
+			if !ok {
+				http.Error(w, "Streaming unsupported", http.StatusInternalServerError)
+				return
+			}
+
+			// Send initial endpoint event
+			fmt.Fprintf(w, "event: endpoint\ndata: %s\n\n", "/message")
+			flusher.Flush()
+
+			// Wait for message to be received, then send response
+			select {
+			case <-messageReceived:
+				// Send response via SSE WITHOUT event field (only data field)
+				// This should be processed as a "message" event according to SSE spec
+				response := map[string]any{
+					"jsonrpc": "2.0",
+					"id":      1,
+					"result":  "test response without event field",
+				}
+				responseBytes, _ := json.Marshal(response)
+				fmt.Fprintf(w, "data: %s\n\n", responseBytes)
+				flusher.Flush()
+			case <-r.Context().Done():
+				return
+			}
+
+			// Keep connection open
+			<-r.Context().Done()
+		})
+
+		// Create message handler
+		messageHandler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			w.Header().Set("Content-Type", "application/json")
+			w.WriteHeader(http.StatusAccepted)
+			
+			// Signal that message was received
+			close(messageReceived)
+		})
+
+		// Initialize the channel
+		messageReceived = make(chan struct{})
+
+		// Create test server
+		mux := http.NewServeMux()
+		mux.Handle("/", sseHandler)
+		mux.Handle("/message", messageHandler)
+		testServer := httptest.NewServer(mux)
+		defer testServer.Close()
+
+		// Create SSE transport
+		trans, err := NewSSE(testServer.URL)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		// Start the transport
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+
+		err = trans.Start(ctx)
+		if err != nil {
+			t.Fatalf("Failed to start transport: %v", err)
+		}
+		defer trans.Close()
+
+		// Send a request
+		request := JSONRPCRequest{
+			JSONRPC: "2.0",
+			ID:      mcp.NewRequestId(int64(1)),
+			Method:  "test",
+		}
+
+		// This should succeed because the SSE event without event field should be processed
+		response, err := trans.SendRequest(ctx, request)
+		if err != nil {
+			t.Fatalf("SendRequest failed: %v", err)
+		}
+
+		if response == nil {
+			t.Fatal("Expected response, got nil")
+		}
+
+		// Verify the response
+		var result string
+		if err := json.Unmarshal(response.Result, &result); err != nil {
+			t.Fatalf("Failed to unmarshal result: %v", err)
+		}
+
+		if result != "test response without event field" {
+			t.Errorf("Expected 'test response without event field', got '%s'", result)
+		}
+	})
+
 }
 
 func TestSSEErrors(t *testing.T) {

client/transport/streamable_http.go

@@ -380,7 +380,11 @@ func (c *StreamableHTTP) readSSE(ctx context.Context, reader io.ReadCloser, hand
 			if err != nil {
 				if err == io.EOF {
 					// Process any pending event before exit
-					if event != "" && data != "" {
+					if data != "" {
+						// If no event type is specified, use empty string (default event type)
+						if event == "" {
+							event = "message"
+						}
 						handler(event, data)
 					}
 					return
@@ -398,7 +402,11 @@ func (c *StreamableHTTP) readSSE(ctx context.Context, reader io.ReadCloser, hand
 			line = strings.TrimRight(line, "\r\n")
 			if line == "" {
 				// Empty line means end of event
-				if event != "" && data != "" {
+				if data != "" {
+					// If no event type is specified, use empty string (default event type)
+					if event == "" {
+						event = "message"
+					}
 					handler(event, data)
 					event = ""
 					data = ""

client/transport/streamable_http_test.go

@@ -413,6 +413,82 @@ func TestStreamableHTTP(t *testing.T) {
 			t.Errorf("Expected JSONRPC '2.0', got '%s'", responseError.JSONRPC)
 		}
 	})
+
+	t.Run("SSEEventWithoutEventField", func(t *testing.T) {
+		// Test that SSE events with only data field (no event field) are processed correctly
+		// This tests the fix for issue #369
+		
+		// Create a custom mock server that sends SSE events without event field
+		handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			if r.Method != http.MethodPost {
+				http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
+				return
+			}
+
+			// Parse incoming JSON-RPC request
+			var request map[string]any
+			decoder := json.NewDecoder(r.Body)
+			if err := decoder.Decode(&request); err != nil {
+				http.Error(w, fmt.Sprintf("Invalid JSON: %v", err), http.StatusBadRequest)
+				return
+			}
+
+			// Send response via SSE WITHOUT event field (only data field)
+			// This should be processed as a "message" event according to SSE spec
+			w.Header().Set("Content-Type", "text/event-stream")
+			w.WriteHeader(http.StatusOK)
+			
+			response := map[string]any{
+				"jsonrpc": "2.0",
+				"id":      request["id"],
+				"result":  "test response without event field",
+			}
+			responseBytes, _ := json.Marshal(response)
+			// Note: No "event:" field, only "data:" field
+			fmt.Fprintf(w, "data: %s\n\n", responseBytes)
+		})
+
+		// Create test server
+		testServer := httptest.NewServer(handler)
+		defer testServer.Close()
+
+		// Create StreamableHTTP transport
+		trans, err := NewStreamableHTTP(testServer.URL)
+		if err != nil {
+			t.Fatal(err)
+		}
+		defer trans.Close()
+
+		// Send a request
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+
+		request := JSONRPCRequest{
+			JSONRPC: "2.0",
+			ID:      mcp.NewRequestId(int64(1)),
+			Method:  "test",
+		}
+
+		// This should succeed because the SSE event without event field should be processed
+		response, err := trans.SendRequest(ctx, request)
+		if err != nil {
+			t.Fatalf("SendRequest failed: %v", err)
+		}
+
+		if response == nil {
+			t.Fatal("Expected response, got nil")
+		}
+
+		// Verify the response
+		var result string
+		if err := json.Unmarshal(response.Result, &result); err != nil {
+			t.Fatalf("Failed to unmarshal result: %v", err)
+		}
+
+		if result != "test response without event field" {
+			t.Errorf("Expected 'test response without event field', got '%s'", result)
+		}
+	})
 }
 
 func TestStreamableHTTPErrors(t *testing.T) {