Event Sourcing Integration

Event sourcing systems store every state change as an immutable event — OrderPlaced, PaymentReceived, ItemShipped. These event streams are a natural fit for MCP subscriptions because the change detection problem is already solved: the event store IS the change log. Instead of polling a database for differences, the MCP server tails the event stream and emits notifications when new events arrive for subscribed resources.

The mapping from events to MCP resources requires a projection layer. An event like OrderPlaced { orderId: "123", customerId: "456" } should trigger notifications for multiple resource URIs: the specific order (orders://123), the customer's order list (customers://456/orders), and possibly an aggregate resource (orders://summary). The projection maps each event type to the set of URIs it affects. This is the same projection pattern used in CQRS read models, applied to MCP notification routing.

interface DomainEvent {
  type: string;
  aggregateId: string;
  data: Record<string, unknown>;
  timestamp: number;
  sequence: number;
}

// Map event types to affected MCP resource URIs
const EVENT_PROJECTIONS: Record<string, (event: DomainEvent) => string[]> = {
  OrderPlaced: (e) => [
    `orders://${e.aggregateId}`,
    `customers://${e.data.customerId}/orders`,
    "orders://summary",
  ],
  PaymentReceived: (e) => [
    `orders://${e.aggregateId}`,
    `payments://recent`,
  ],
  ItemShipped: (e) => [
    `orders://${e.aggregateId}`,
    `shipments://active`,
  ],
};

// Tail the event stream and emit MCP notifications
async function bridgeEventStream(stream: AsyncIterable<DomainEvent>) {
  for await (const event of stream) {
    const projection = EVENT_PROJECTIONS[event.type];
    if (!projection) continue;

    const affectedUris = projection(event);
    for (const uri of affectedUris) {
      subscriptionManager.onChange(uri); // Debounced notification
    }
  }
}