Add notification system to MCP server

The MCP server subscribes to the global event stream at startup,
filtering out self-events by actor name. Notifications from other
actors are buffered (up to 50) and piggybacked onto tool responses
as a text block listing what changed since the last tool call.

- Notifier with ring buffer, concurrent-safe Drain()
- Human-readable notification summaries (created, transitioned, etc.)
- withNotifications wrapper applied to all tool handlers
- Actor identity resolved at startup via /me for self-filtering
- Graceful fallback: if /me fails or no events arrive, tools work normally

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: bricef

cmd/taskflow-mcp/main.go

@@ -29,7 +29,22 @@ func main() {
 	}
 
 	client := httpclient.New(serverURL, apiKey)
-	server := tfmcp.NewServer(client)
+
+	// Discover self identity for notification filtering.
+	actor, err := client.WhoAmI()
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "Warning: could not resolve actor identity: %v\n", err)
+	}
+
+	// Start event subscription for notifications.
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	var notifier *tfmcp.Notifier
+	if actor.Name != "" {
+		notifier = tfmcp.NewNotifier(ctx, client, actor.Name)
+	}
+
+	server := tfmcp.NewServer(client, notifier)
 
 	if err := server.Run(context.Background(), &gomcp.StdioTransport{}); err != nil {
 		fmt.Fprintf(os.Stderr, "MCP server error: %v\n", err)

internal/mcp/notifications.go

@@ -0,0 +1,141 @@
+package mcp
+
+import (
+	"context"
+	"fmt"
+	"sync"
+
+	"github.com/bricef/taskflow/internal/eventbus"
+	"github.com/bricef/taskflow/internal/httpclient"
+)
+
+const maxPendingNotifications = 50
+
+// Notification is a summary of a domain event from another actor.
+type Notification struct {
+	Event     string `json:"event"`
+	Board     string `json:"board"`
+	Task      string `json:"task,omitempty"`
+	Actor     string `json:"actor"`
+	Summary   string `json:"summary"`
+	Timestamp string `json:"timestamp"`
+}
+
+// Notifier subscribes to the global event stream and buffers notifications
+// from other actors. Pending notifications are retrieved and cleared on each
+// tool call via Drain().
+type Notifier struct {
+	mu       sync.Mutex
+	pending  []Notification
+	actorName string
+}
+
+// NewNotifier starts a background event subscription and buffers notifications
+// from actors other than selfName. Cancel the context to stop.
+func NewNotifier(ctx context.Context, client *httpclient.Client, selfName string) *Notifier {
+	n := &Notifier{actorName: selfName}
+
+	stream := client.Subscribe(ctx, httpclient.SubscribeOptions{})
+	go func() {
+		for {
+			select {
+			case evt, ok := <-stream.Events:
+				if !ok {
+					return
+				}
+				if evt.Actor.Name == n.actorName {
+					continue // skip self-events
+				}
+				n.add(evt)
+			case <-stream.Errors:
+				// Reconnection is handled by httpclient.Subscribe.
+				// We just ignore errors here.
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
+
+	return n
+}
+
+func (n *Notifier) add(evt eventbus.Event) {
+	n.mu.Lock()
+	defer n.mu.Unlock()
+
+	task := ""
+	snap := evt.After
+	if snap == nil {
+		snap = evt.Before
+	}
+	if snap != nil {
+		task = snap.Ref
+	}
+
+	notif := Notification{
+		Event:     evt.Type,
+		Board:     evt.Board.Slug,
+		Task:      task,
+		Actor:     evt.Actor.Name,
+		Summary:   formatNotification(evt),
+		Timestamp: evt.Timestamp.Format("2006-01-02T15:04:05Z"),
+	}
+
+	n.pending = append(n.pending, notif)
+	if len(n.pending) > maxPendingNotifications {
+		n.pending = n.pending[len(n.pending)-maxPendingNotifications:]
+	}
+}
+
+// Drain returns all pending notifications and clears the buffer.
+func (n *Notifier) Drain() []Notification {
+	n.mu.Lock()
+	defer n.mu.Unlock()
+
+	if len(n.pending) == 0 {
+		return nil
+	}
+	result := n.pending
+	n.pending = nil
+	return result
+}
+
+func formatNotification(evt eventbus.Event) string {
+	actor := evt.Actor.Name
+	if actor == "" {
+		actor = "system"
+	}
+
+	snap := evt.After
+	if snap == nil {
+		snap = evt.Before
+	}
+
+	switch evt.Type {
+	case eventbus.EventTaskCreated:
+		if snap != nil {
+			return fmt.Sprintf("%s created %s %q", actor, snap.Ref, snap.Title)
+		}
+	case eventbus.EventTaskTransitioned:
+		if evt.Before != nil && snap != nil {
+			return fmt.Sprintf("%s moved %s from %s to %s", actor, snap.Ref, evt.Before.State, snap.State)
+		}
+	case eventbus.EventTaskAssigned:
+		if snap != nil && snap.Assignee != nil {
+			return fmt.Sprintf("%s assigned %s to %s", actor, snap.Ref, *snap.Assignee)
+		}
+	case eventbus.EventTaskCommented:
+		if snap != nil {
+			return fmt.Sprintf("%s commented on %s", actor, snap.Ref)
+		}
+	case eventbus.EventTaskDeleted:
+		if evt.Before != nil {
+			return fmt.Sprintf("%s deleted %s", actor, evt.Before.Ref)
+		}
+	}
+
+	if snap != nil {
+		return fmt.Sprintf("%s: %s on %s", actor, evt.Type, snap.Ref)
+	}
+	return fmt.Sprintf("%s: %s", actor, evt.Type)
+}

internal/mcp/server.go

@@ -17,15 +17,16 @@ import (
 )
 
 // NewServer creates an MCP server with all TaskFlow resources and tools registered.
-func NewServer(client *httpclient.Client) *gomcp.Server {
+// If notifier is non-nil, tool responses include pending notifications from other actors.
+func NewServer(client *httpclient.Client, notifier *Notifier) *gomcp.Server {
 	server := gomcp.NewServer(&gomcp.Implementation{
 		Name:    "taskflow",
 		Version: "1.0.0",
 	}, nil)
 
 	registerResources(server, client)
-	registerTools(server, client)
-	registerWhoAmI(server, client)
+	registerTools(server, client, notifier)
+	registerWhoAmI(server, client, notifier)
 
 	return server
 }
@@ -78,7 +79,7 @@ func resourceHandler(client *httpclient.Client, res model.Resource) gomcp.Resour
 
 // --- Tools ---
 
-func registerTools(server *gomcp.Server, client *httpclient.Client) {
+func registerTools(server *gomcp.Server, client *httpclient.Client, notifier *Notifier) {
 	for _, op := range model.Operations() {
 		op := op
 		server.AddTool(
@@ -87,7 +88,7 @@ func registerTools(server *gomcp.Server, client *httpclient.Client) {
 				Description: descriptionOf(op.Summary, op.Description),
 				InputSchema: buildInputSchema(op),
 			},
-			toolHandler(client, op),
+			withNotifications(notifier, toolHandler(client, op)),
 		)
 	}
 }
@@ -154,14 +155,14 @@ func toolHandler(client *httpclient.Client, op model.Operation) gomcp.ToolHandle
 
 // --- WhoAmI ---
 
-func registerWhoAmI(server *gomcp.Server, client *httpclient.Client) {
+func registerWhoAmI(server *gomcp.Server, client *httpclient.Client, notifier *Notifier) {
 	server.AddTool(
 		&gomcp.Tool{
 			Name:        "whoami",
 			Description: "Returns your actor identity (name, role, type) as determined by your API key.",
 			InputSchema: map[string]any{"type": "object", "properties": map[string]any{}},
 		},
-		func(ctx context.Context, req *gomcp.CallToolRequest) (*gomcp.CallToolResult, error) {
+		withNotifications(notifier, func(ctx context.Context, req *gomcp.CallToolRequest) (*gomcp.CallToolResult, error) {
 			actor, err := client.WithContext(ctx).WhoAmI()
 			if err != nil {
 				return &gomcp.CallToolResult{
@@ -173,10 +174,41 @@ func registerWhoAmI(server *gomcp.Server, client *httpclient.Client) {
 			return &gomcp.CallToolResult{
 				Content: []gomcp.Content{&gomcp.TextContent{Text: string(data)}},
 			}, nil
-		},
+		}),
 	)
 }
 
+// --- Notification piggyback ---
+
+// withNotifications wraps a tool handler to append pending notifications
+// to the response. If notifier is nil, the handler is returned unchanged.
+func withNotifications(notifier *Notifier, handler gomcp.ToolHandler) gomcp.ToolHandler {
+	if notifier == nil {
+		return handler
+	}
+	return func(ctx context.Context, req *gomcp.CallToolRequest) (*gomcp.CallToolResult, error) {
+		result, err := handler(ctx, req)
+		if err != nil || result == nil {
+			return result, err
+		}
+
+		notifications := notifier.Drain()
+		if len(notifications) == 0 {
+			return result, nil
+		}
+
+		// Append notifications as a separate text block.
+		var summary strings.Builder
+		summary.WriteString(fmt.Sprintf("\n--- %d notification(s) from other actors ---\n", len(notifications)))
+		for _, n := range notifications {
+			summary.WriteString(fmt.Sprintf("  [%s] %s\n", n.Timestamp, n.Summary))
+		}
+
+		result.Content = append(result.Content, &gomcp.TextContent{Text: summary.String()})
+		return result, nil
+	}
+}
+
 // --- Schema generation ---
 
 func buildInputSchema(op model.Operation) map[string]any {