feat: add client Events() method and anti-fragile STDIO transport - Add Events() method to Client interface and implementation - Add comprehensive test for Events() method access - Implement anti-fragile JSON-RPC filtering in STDIO transport - Add isValidJSONRPC() validation to filter out log messages and noise - Add comprehensive test coverage for filtering behavior - Enhance events_integration example with client event demonstration - Ensure robust communication in noisy environments

Author: localrivet

client/client.go

@@ -227,6 +227,19 @@ type Client interface {
 	// GetSamplingHandler returns the currently registered sampling handler.
 	GetSamplingHandler() SamplingHandler
 
+	// Events returns the events subject for subscribing to client events.
+	//
+	// This provides access to the event system for monitoring client lifecycle,
+	// errors, and other significant events. Subscribers can listen for specific
+	// event types such as initialization, connection changes, and errors.
+	//
+	// Example:
+	//  events := client.Events()
+	//  events.Subscribe(events.TopicClientError, func(event events.ClientErrorEvent) {
+	//      log.Printf("Client error: %s", event.Error)
+	//  })
+	Events() *events.Subject
+
 	// RequestSampling initiates a sampling request to the server.
 	//
 	// This is the unified method for all sampling operations, supporting both
@@ -676,3 +689,8 @@ func (c *clientImpl) sendBatchRequestWithTimeout(requests []map[string]interface
 
 	return result, nil
 }
+
+// Events returns the events subject for subscribing to client events.
+func (c *clientImpl) Events() *events.Subject {
+	return c.events
+}

client/test/v20250326/client_test.go

@@ -262,3 +262,14 @@ func TestRoots_v20250326(t *testing.T) {
 		t.Errorf("Remove root params not as expected: %v", removeParams)
 	}
 }
+
+// TestClientEvents_v20250326 tests that the Events() method returns the events subject
+func TestClientEvents_v20250326(t *testing.T) {
+	c, _ := setupTest(t)
+
+	// Test that Events() method returns a non-nil events subject
+	events := c.Events()
+	if events == nil {
+		t.Fatal("Expected Events() to return non-nil events subject")
+	}
+}

examples/events_integration/main.go

@@ -6,6 +6,7 @@ import (
 	"log/slog"
 	"time"
 
+	"github.com/localrivet/gomcp/client"
 	"github.com/localrivet/gomcp/events"
 	"github.com/localrivet/gomcp/server"
 )
@@ -102,6 +103,31 @@ type RequestFailedEvent struct {
 	Metadata  map[string]any `json:"metadata,omitempty"`
 }
 
+// Client-side event types
+type ClientInitializedEvent struct {
+	ClientName      string         `json:"clientName"`
+	ServerURL       string         `json:"serverUrl"`
+	ProtocolVersion string         `json:"protocolVersion"`
+	InitializedAt   time.Time      `json:"initializedAt"`
+	Capabilities    map[string]any `json:"capabilities"`
+	Metadata        map[string]any `json:"metadata,omitempty"`
+}
+
+type ClientConnectionEvent struct {
+	Status    string         `json:"status"`
+	ServerURL string         `json:"serverUrl"`
+	Timestamp time.Time      `json:"timestamp"`
+	Error     string         `json:"error,omitempty"`
+	Metadata  map[string]any `json:"metadata,omitempty"`
+}
+
+type ClientErrorEvent struct {
+	Operation  string         `json:"operation"`
+	Error      string         `json:"error"`
+	OccurredAt time.Time      `json:"occurredAt"`
+	Metadata   map[string]any `json:"metadata,omitempty"`
+}
+
 func main() {
 	// Create a server with events
 	srv := server.NewServer("events-demo")
@@ -130,18 +156,28 @@ func main() {
 	logger.Info("  💭 prompt execution", "topic", events.TopicPromptExecuted)
 	logger.Info("  ❌ request failures", "topic", events.TopicRequestFailed)
 	logger.Info("")
-	logger.Info("🧪 Testing Event System:")
+	logger.Info("🧪 Testing Server Event System:")
 
-	// Demonstrate event publishing with some test events
+	// Demonstrate server event publishing with some test events
 	demonstrateEvents(srv, logger)
 
+	logger.Info("")
+	logger.Info("🧪 Testing Client Event System:")
+
+	// Demonstrate client events
+	demonstrateClientEvents(logger)
+
 	logger.Info("")
 	logger.Info("✅ Events integration example complete!")
 	logger.Info("💡 In a real MCP server, these events fire automatically when:")
 	logger.Info("   - Server starts and clients connect")
 	logger.Info("   - Tools and resources are registered")
 	logger.Info("   - Clients execute tools, access resources, or use prompts")
 	logger.Info("   - Errors occur during request processing")
+	logger.Info("💡 In a real MCP client, events fire when:")
+	logger.Info("   - Client initializes and connects to servers")
+	logger.Info("   - Connection status changes")
+	logger.Info("   - Errors occur during client operations")
 }
 
 func setupEventSubscriptions(srv server.Server, logger *slog.Logger) {
@@ -422,3 +458,88 @@ func demonstrateEvents(srv server.Server, logger *slog.Logger) {
 	// Give events time to process
 	time.Sleep(50 * time.Millisecond)
 }
+
+func demonstrateClientEvents(logger *slog.Logger) {
+	logger.Info("📢 Demonstrating client events system...")
+
+	// Create a simple client to demonstrate client-side events
+	c, err := client.NewClient("test-client", client.WithStdio())
+	if err != nil {
+		logger.Info("Failed to create client", "error", err)
+		return
+	}
+
+	// Get the client's events subject
+	clientEvents := c.Events()
+
+	// Subscribe to client events (if the topics exist, otherwise demonstrate with custom topics)
+	events.Subscribe[ClientInitializedEvent](clientEvents, "client.initialized",
+		func(ctx context.Context, evt ClientInitializedEvent) error {
+			logger.Info("🚀 Client initialized",
+				"clientName", evt.ClientName,
+				"serverURL", evt.ServerURL,
+				"protocolVersion", evt.ProtocolVersion)
+			return nil
+		})
+
+	events.Subscribe[ClientConnectionEvent](clientEvents, "client.connection",
+		func(ctx context.Context, evt ClientConnectionEvent) error {
+			logger.Info("🔌 Client connection status",
+				"status", evt.Status,
+				"serverURL", evt.ServerURL)
+			return nil
+		})
+
+	events.Subscribe[ClientErrorEvent](clientEvents, "client.error",
+		func(ctx context.Context, evt ClientErrorEvent) error {
+			logger.Info("❌ Client error",
+				"operation", evt.Operation,
+				"error", evt.Error)
+			return nil
+		})
+
+	// Publish test client events
+	testClientEvent := ClientInitializedEvent{
+		ClientName:      "test-client",
+		ServerURL:       "stdio://server",
+		ProtocolVersion: "2025-03-26",
+		InitializedAt:   time.Now(),
+		Capabilities:    map[string]any{"supports": []string{"greet", "calculate"}},
+		Metadata:        make(map[string]any),
+	}
+
+	if err := events.Publish[ClientInitializedEvent](clientEvents, "client.initialized", testClientEvent); err != nil {
+		logger.Info("Failed to publish client initialized event", "error", err)
+	}
+
+	time.Sleep(10 * time.Millisecond)
+
+	testClientConnectionEvent := ClientConnectionEvent{
+		Status:    "connected",
+		ServerURL: "stdio://server",
+		Timestamp: time.Now(),
+		Metadata:  make(map[string]any),
+	}
+
+	if err := events.Publish[ClientConnectionEvent](clientEvents, "client.connection", testClientConnectionEvent); err != nil {
+		logger.Info("Failed to publish client connection event", "error", err)
+	}
+
+	time.Sleep(10 * time.Millisecond)
+
+	testClientErrorEvent := ClientErrorEvent{
+		Operation:  "tool_call",
+		Error:      "Demonstration client error",
+		OccurredAt: time.Now(),
+		Metadata:   make(map[string]any),
+	}
+
+	if err := events.Publish[ClientErrorEvent](clientEvents, "client.error", testClientErrorEvent); err != nil {
+		logger.Info("Failed to publish client error event", "error", err)
+	}
+
+	// Give events time to process
+	time.Sleep(50 * time.Millisecond)
+
+	logger.Info("✅ Client events demonstration complete")
+}

transport/stdio/stdio.go

@@ -6,6 +6,7 @@ package stdio
 
 import (
 	"bufio"
+	"encoding/json"
 	"errors"
 	"io"
 	"os"
@@ -15,6 +16,48 @@ import (
 	"github.com/localrivet/gomcp/transport"
 )
 
+// isValidJSONRPC checks if a message appears to be a valid JSON-RPC message.
+// This provides anti-fragile behavior by filtering out log messages and other noise.
+func isValidJSONRPC(data []byte) bool {
+	// Quick check: must be valid JSON
+	var raw map[string]interface{}
+	if err := json.Unmarshal(data, &raw); err != nil {
+		return false
+	}
+
+	// Must have jsonrpc field with value "2.0"
+	jsonrpc, ok := raw["jsonrpc"]
+	if !ok {
+		return false
+	}
+	if jsonrpcStr, ok := jsonrpc.(string); !ok || jsonrpcStr != "2.0" {
+		return false
+	}
+
+	// Must be one of: request (has method + id), response (has id + result/error), or notification (has method, no id)
+	_, hasMethod := raw["method"]
+	_, hasID := raw["id"]
+	_, hasResult := raw["result"]
+	_, hasError := raw["error"]
+
+	// Request: method + id
+	if hasMethod && hasID {
+		return true
+	}
+
+	// Response: id + (result or error)
+	if hasID && (hasResult || hasError) {
+		return true
+	}
+
+	// Notification: method, no id
+	if hasMethod && !hasID {
+		return true
+	}
+
+	return false
+}
+
 // Transport implements the transport.Transport interface for Standard I/O.
 type Transport struct {
 	transport.BaseTransport
@@ -140,6 +183,19 @@ func (t *Transport) readLoop() {
 				continue
 			}
 
+			// Anti-fragile filtering: only process valid JSON-RPC messages
+			if !isValidJSONRPC([]byte(line)) {
+				// Log filtered message if debug enabled
+				if debugHandler := t.GetDebugHandler(); debugHandler != nil {
+					if len(line) > 100 {
+						debugHandler("stdio transport filtered non-JSON-RPC: " + line[:100] + "...")
+					} else {
+						debugHandler("stdio transport filtered non-JSON-RPC: " + line)
+					}
+				}
+				continue
+			}
+
 			// Log received message if debug enabled
 			if debugHandler := t.GetDebugHandler(); debugHandler != nil {
 				if len(line) > 100 {