🎛️ Real-Time Collaborative Agents: SSE Event Bus and Live Dashboard with MCP

---

This is Part 13 of the AI Engineering with TypeScript series.

Prerequisites: Part 4 — Streaming Agents · Part 5 — Production MCP Server · Part 6 — Multi-Tenant Sessions Stack: Node.js 20+ · TypeScript 5.x · Express 5 · Server-Sent Events · EventEmitter · React (dashboard)

---

🗺️ What we'll cover

In Parts 3–4 you built agents that stream output to a single CLI user. In Parts 5–6 you made the server multi-tenant. But there is still a blind spot: when three agents are simultaneously calling tools on your MCP server, you have no live view of what is happening. You cannot see which agent is mid-call, which tool is slow, or which session just errored.

In Part 13 we wire up a server-side event bus inside the MCP server that emits an event on every significant action — tool call start, tool call done, tool call error, session open, session close. We then expose a GET /events SSE endpoint that any observer (dashboard, monitoring tool, another service) can subscribe to and receive a live stream of those events.

By the end you will have:

• 📡 An internal EventBus built on Node's EventEmitter — zero dependencies, zero overhead
• 🔌 A GET /events SSE endpoint that broadcasts all agent activity to connected observers
• 🎛️ A live React dashboard that subscribes to the SSE stream and renders tool calls, sessions, and errors in real time
• 🔒 Auth-gated SSE — only admin tokens can subscribe to the event stream
• 📊 Session and tool-call stats computed in real time from the event stream
• 🔁 Reconnection handling — clients automatically resume after network drops using Last-Event-ID

---

📡 Part 1: The Internal Event Bus

The event bus is the heart of the system. Every tool call, session lifecycle event, and error is published here. Observers subscribe and react. The bus itself is a thin wrapper around Node's built-in EventEmitter with typed events:

// src/events/bus.ts
import { EventEmitter } from "events";

export type AgentEventType =
  | "tool:start"
  | "tool:done"
  | "tool:error"
  | "session:open"
  | "session:close"
  | "session:expired";

export interface AgentEvent {
  id: string;                 // monotonically increasing event ID for SSE Last-Event-ID
  type: AgentEventType;
  sessionId: string;
  tenantId: string;
  timestamp: string;
  data: Record<string, unknown>;
}

class EventBus extends EventEmitter {
  private counter = 0;

  publish(
    type: AgentEventType,
    sessionId: string,
    tenantId: string,
    data: Record<string, unknown>
  ): AgentEvent {
    const event: AgentEvent = {
      id: String(++this.counter),
      type,
      sessionId,
      tenantId,
      timestamp: new Date().toISOString(),
      data,
    };

    this.emit("event", event);
    return event;
  }
}

// Singleton — one bus for the entire server process
export const eventBus = new EventBus();
// Allow many SSE subscribers without Node warning about memory leaks
eventBus.setMaxListeners(200);

One bus instance, exported as a singleton. Every part of the server imports eventBus and calls eventBus.publish(...). Observers call eventBus.on("event", handler). No Redis pub/sub, no message broker — just in-process events. This is the right starting point; you add a broker (Redis pub/sub, NATS) only when you scale to multiple server instances and need cross-process delivery. 💡

---

🔧 Part 2: Emitting Events from Tool Calls

Update your tool execution wrapper to publish bus events before and after every tool call:

// src/utils/instrument-tool.ts (updated from Part 7)
import { eventBus } from "../events/bus.js";

export async function instrumentedToolCall<T>(
  toolName: string,
  sessionId: string,
  tenantId: string,
  fn: () => Promise<T>
): Promise<T> {
  const start = Date.now();

  eventBus.publish("tool:start", sessionId, tenantId, {
    tool: toolName,
  });

  try {
    const result = await fn();

    eventBus.publish("tool:done", sessionId, tenantId, {
      tool: toolName,
      durationMs: Date.now() - start,
    });

    return result;
  } catch (err) {
    eventBus.publish("tool:error", sessionId, tenantId, {
      tool: toolName,
      durationMs: Date.now() - start,
      error: err instanceof Error ? err.message : String(err),
    });

    throw err;
  }
}

And emit session lifecycle events from the session management layer:

// src/session-store.ts (updated)
import { eventBus } from "./events/bus.js";

// In createSession:
export async function createSession(sessionId: string, tenantId: string) {
  // ... existing session creation logic ...
  eventBus.publish("session:open", sessionId, tenantId, { sessionId });
  return state;
}

// In deleteSession:
export async function deleteSession(sessionId: string) {
  const state = await getSession(sessionId);
  if (state) {
    eventBus.publish("session:close", sessionId, state.tenantId, { sessionId });
  }
  await redis.del(KEY(sessionId));
}

// In evictExpired:
// After deleting a stale session:
eventBus.publish("session:expired", sessionId, state.tenantId, {
  sessionId,
  idleMs: Date.now() - new Date(state.lastActiveAt).getTime(),
});

Two lines per lifecycle event. No performance impact — EventEmitter.emit() is synchronous and fast. ✅

---

🔌 Part 3: The SSE Endpoint

SSE (Server-Sent Events) is a W3C standard for one-way real-time streams from server to client over a plain HTTP connection. It is simpler than WebSockets for this use case because we only need server-to-client push — no bidirectional communication required.

The wire format is plain text:

data: {"type":"tool:done","sessionId":"abc-123",...}\n\n
id: 42\n\n
event: tool:done\n\n

// src/routes/events.ts
import type { Request, Response } from "express";
import { eventBus, type AgentEvent } from "../events/bus.js";

// Separate admin token set — only ops/dashboard clients get event stream access
const ADMIN_TOKENS = new Set(
  (process.env.ADMIN_TOKENS ?? "").split(",").filter(Boolean)
);

export function handleEventStream(req: Request, res: Response): void {
  const token = (req.headers.authorization ?? "").replace("Bearer ", "").trim();

  if (!ADMIN_TOKENS.has(token)) {
    res.status(403).json({ error: "forbidden" });
    return;
  }

  // SSE headers
  res.setHeader("Content-Type", "text/event-stream");
  res.setHeader("Cache-Control", "no-cache");
  res.setHeader("Connection", "keep-alive");
  res.setHeader("X-Accel-Buffering", "no");   // disable nginx buffering
  res.flushHeaders();

  // Send a heartbeat comment every 15s to keep the connection alive through proxies
  const heartbeat = setInterval(() => {
    res.write(": heartbeat\n\n");
  }, 15_000);

  // Replay events since Last-Event-ID (so clients resume after reconnect)
  const lastId = parseInt(req.headers["last-event-id"] as string ?? "0", 10);

  // Flush buffered events newer than lastId
  for (const event of recentEvents) {
    if (parseInt(event.id, 10) > lastId) {
      writeEvent(res, event);
    }
  }

  // Subscribe to live events
  const handler = (event: AgentEvent) => {
    writeEvent(res, event);
    bufferEvent(event);
  };

  eventBus.on("event", handler);

  // Clean up on client disconnect
  req.on("close", () => {
    clearInterval(heartbeat);
    eventBus.off("event", handler);
  });
}

function writeEvent(res: Response, event: AgentEvent): void {
  res.write(`id: ${event.id}\n`);
  res.write(`event: ${event.type}\n`);
  res.write(`data: ${JSON.stringify(event)}\n\n`);
}

// Rolling buffer of the last 500 events for reconnect replay
const recentEvents: AgentEvent[] = [];
const MAX_BUFFER = 500;

function bufferEvent(event: AgentEvent): void {
  recentEvents.push(event);
  if (recentEvents.length > MAX_BUFFER) {
    recentEvents.shift();
  }
}

Mount it in server.ts:

import { handleEventStream } from "./routes/events.js";

app.get("/events", handleEventStream);

The Last-Event-ID replay is the critical production detail. When a dashboard client's network drops and reconnects, the browser automatically sends the last event ID it received. Your server replays everything since that ID, so the dashboard never misses an event. Without this, every network blip leaves a gap in your event history. 🔁

---

📊 Part 4: Real-Time Stats Aggregator

Keep a live stats snapshot updated by the event bus — no database queries needed for the dashboard:

// src/events/stats.ts
import { eventBus, type AgentEvent } from "./bus.js";

export interface LiveStats {
  activeSessions: number;
  totalToolCalls: number;
  errorCount: number;
  toolCallsInFlight: number;
  toolLatencies: number[];    // last 100 durations for p99 calc
  recentEvents: AgentEvent[]; // last 50 events for activity feed
}

const stats: LiveStats = {
  activeSessions: 0,
  totalToolCalls: 0,
  errorCount: 0,
  toolCallsInFlight: 0,
  toolLatencies: [],
  recentEvents: [],
};

eventBus.on("event", (event: AgentEvent) => {
  switch (event.type) {
    case "session:open":
      stats.activeSessions++;
      break;
    case "session:close":
    case "session:expired":
      stats.activeSessions = Math.max(0, stats.activeSessions - 1);
      break;
    case "tool:start":
      stats.toolCallsInFlight++;
      stats.totalToolCalls++;
      break;
    case "tool:done":
      stats.toolCallsInFlight = Math.max(0, stats.toolCallsInFlight - 1);
      if (typeof event.data.durationMs === "number") {
        stats.toolLatencies.push(event.data.durationMs);
        if (stats.toolLatencies.length > 100) stats.toolLatencies.shift();
      }
      break;
    case "tool:error":
      stats.toolCallsInFlight = Math.max(0, stats.toolCallsInFlight - 1);
      stats.errorCount++;
      break;
  }

  stats.recentEvents.unshift(event);
  if (stats.recentEvents.length > 50) stats.recentEvents.pop();
});

export function getStats(): LiveStats & { p99LatencyMs: number } {
  const sorted = [...stats.toolLatencies].sort((a, b) => a - b);
  const p99 = sorted[Math.floor(sorted.length * 0.99)] ?? 0;
  return { ...stats, p99LatencyMs: p99 };
}

Expose it as a JSON endpoint for the dashboard's initial load:

// In server.ts
import { getStats } from "./events/stats.js";

app.get("/stats", adminAuth, (_req, res) => {
  res.json(getStats());
});

---

🖥️ Part 5: The Live React Dashboard

A self-contained React component that connects to your SSE stream and renders live agent activity. Drop this into any React app or serve it as a standalone HTML file:

// dashboard/LiveDashboard.tsx
import { useState, useEffect, useRef } from "react";

interface AgentEvent {
  id: string;
  type: string;
  sessionId: string;
  tenantId: string;
  timestamp: string;
  data: Record<string, unknown>;
}

interface Stats {
  activeSessions: number;
  totalToolCalls: number;
  errorCount: number;
  toolCallsInFlight: number;
  p99LatencyMs: number;
}

const EVENT_URL = import.meta.env.VITE_MCP_SERVER_URL + "/events";
const STATS_URL = import.meta.env.VITE_MCP_SERVER_URL + "/stats";
const ADMIN_TOKEN = import.meta.env.VITE_ADMIN_TOKEN;

function eventColour(type: string): string {
  if (type === "tool:done") return "#22c55e";
  if (type === "tool:error") return "#ef4444";
  if (type === "tool:start") return "#f59e0b";
  if (type.startsWith("session:")) return "#60a5fa";
  return "#94a3b8";
}

function eventIcon(type: string): string {
  if (type === "tool:done") return "✓";
  if (type === "tool:error") return "✗";
  if (type === "tool:start") return "●";
  if (type === "session:open") return "→";
  if (type === "session:close") return "←";
  return "·";
}

export default function LiveDashboard() {
  const [events, setEvents] = useState<AgentEvent[]>([]);
  const [stats, setStats] = useState<Stats | null>(null);
  const [connected, setConnected] = useState(false);
  const lastIdRef = useRef("0");

  // Load initial stats
  useEffect(() => {
    fetch(STATS_URL, {
      headers: { Authorization: `Bearer ${ADMIN_TOKEN}` },
    })
      .then((r) => r.json())
      .then(setStats)
      .catch(console.error);
  }, []);

  // SSE connection with auto-reconnect
  useEffect(() => {
    let es: EventSource;
    let reconnectTimer: ReturnType<typeof setTimeout>;

    function connect() {
      const url = new URL(EVENT_URL);
      es = new EventSource(url.toString(), {
        // EventSource doesn't support custom headers natively
        // use a query param for the admin token
      });

      es.onopen = () => setConnected(true);

      es.addEventListener("tool:start", handleEvent);
      es.addEventListener("tool:done", handleEvent);
      es.addEventListener("tool:error", handleEvent);
      es.addEventListener("session:open", handleEvent);
      es.addEventListener("session:close", handleEvent);
      es.addEventListener("session:expired", handleEvent);

      es.onerror = () => {
        setConnected(false);
        es.close();
        reconnectTimer = setTimeout(connect, 3000);
      };
    }

    function handleEvent(e: MessageEvent) {
      const event: AgentEvent = JSON.parse(e.data);
      lastIdRef.current = event.id;

      setEvents((prev) => [event, ...prev].slice(0, 200));

      // Update stats from event stream (no poll needed)
      setStats((prev) => {
        if (!prev) return prev;
        const next = { ...prev };
        if (event.type === "session:open") next.activeSessions++;
        if (event.type === "session:close" || event.type === "session:expired")
          next.activeSessions = Math.max(0, next.activeSessions - 1);
        if (event.type === "tool:start") {
          next.totalToolCalls++;
          next.toolCallsInFlight = (next.toolCallsInFlight ?? 0) + 1;
        }
        if (event.type === "tool:done")
          next.toolCallsInFlight = Math.max(0, (next.toolCallsInFlight ?? 0) - 1);
        if (event.type === "tool:error") {
          next.errorCount++;
          next.toolCallsInFlight = Math.max(0, (next.toolCallsInFlight ?? 0) - 1);
        }
        return next;
      });
    }

    connect();

    return () => {
      clearTimeout(reconnectTimer);
      es?.close();
    };
  }, []);

  return (
    <div style={{ background: "#06080f", minHeight: "100vh", color: "#e2e8f0", fontFamily: "monospace", padding: "24px" }}>
      {/* Header */}
      <div style={{ display: "flex", alignItems: "center", gap: "12px", marginBottom: "24px" }}>
        <h1 style={{ margin: 0, fontSize: "20px", color: "#f1f5f9" }}>MCP Agent Dashboard</h1>
        <span style={{
          padding: "2px 10px",
          borderRadius: "999px",
          fontSize: "11px",
          background: connected ? "#052a10" : "#1a0808",
          color: connected ? "#22c55e" : "#ef4444",
          border: `1px solid ${connected ? "#22c55e" : "#ef4444"}`,
        }}>
          {connected ? "● LIVE" : "○ RECONNECTING"}
        </span>
      </div>

      {/* Stats row */}
      {stats && (
        <div style={{ display: "grid", gridTemplateColumns: "repeat(5, 1fr)", gap: "12px", marginBottom: "24px" }}>
          {[
            { label: "Active Sessions", value: stats.activeSessions, colour: "#60a5fa" },
            { label: "Total Tool Calls", value: stats.totalToolCalls, colour: "#22c55e" },
            { label: "In Flight", value: stats.toolCallsInFlight, colour: "#f59e0b" },
            { label: "Errors", value: stats.errorCount, colour: "#ef4444" },
            { label: "p99 Latency", value: `${stats.p99LatencyMs}ms`, colour: "#a855f7" },
          ].map((s) => (
            <div key={s.label} style={{ background: "#0a0e18", border: "1px solid #1e3a5f", borderRadius: "8px", padding: "14px" }}>
              <div style={{ fontSize: "11px", color: "#64748b", marginBottom: "6px" }}>{s.label}</div>
              <div style={{ fontSize: "22px", fontWeight: 700, color: s.colour }}>{s.value}</div>
            </div>
          ))}
        </div>
      )}

      {/* Event feed */}
      <div style={{ background: "#080b14", border: "1px solid #1e3a5f", borderRadius: "8px", overflow: "hidden" }}>
        <div style={{ padding: "10px 16px", borderBottom: "1px solid #1e3a5f", fontSize: "11px", color: "#64748b" }}>
          LIVE EVENT FEED ({events.length} events)
        </div>
        <div style={{ maxHeight: "480px", overflowY: "auto" }}>
          {events.length === 0 && (
            <div style={{ padding: "20px 16px", color: "#334155", fontSize: "12px" }}>
              Waiting for agent activity...
            </div>
          )}
          {events.map((event) => (
            <div key={event.id} style={{
              display: "grid",
              gridTemplateColumns: "40px 140px 180px 1fr auto",
              gap: "12px",
              padding: "8px 16px",
              borderBottom: "1px solid #0f1828",
              fontSize: "12px",
              alignItems: "center",
            }}>
              <span style={{ color: eventColour(event.type), fontWeight: 700 }}>
                {eventIcon(event.type)}
              </span>
              <span style={{ color: "#475569" }}>
                {new Date(event.timestamp).toLocaleTimeString()}
              </span>
              <span style={{ color: eventColour(event.type) }}>{event.type}</span>
              <span style={{ color: "#94a3b8", overflow: "hidden", textOverflow: "ellipsis", whiteSpace: "nowrap" }}>
                {event.data.tool
                  ? `\({event.data.tool}\){event.data.durationMs ? ` (${event.data.durationMs}ms)` : ""}`
                  : event.data.error
                  ? String(event.data.error)
                  : `session ${event.sessionId.slice(0, 8)}`}
              </span>
              <span style={{ color: "#334155", fontSize: "10px" }}>
                {event.sessionId.slice(0, 8)}
              </span>
            </div>
          ))}
        </div>
      </div>
    </div>
  );
}

---

🔒 Part 6: Securing the SSE Endpoint

EventSource in the browser does not support custom Authorization headers. There are two common workarounds:

Option A — Query parameter token (simpler, less ideal):

// Client
const es = new EventSource(`/events?token=${ADMIN_TOKEN}`);

// Server — read from query param
const token = req.query.token as string;

Tokens in URLs appear in server logs and browser history. Acceptable for internal dashboards on a private network; not ideal for public-facing services. 🚨

Option B — Cookie-based auth (recommended for browsers):

// Client: first POST to /auth/session to get a session cookie
await fetch("/auth/session", {
  method: "POST",
  headers: { Authorization: `Bearer ${ADMIN_TOKEN}` },
  credentials: "include",
});

// Then open SSE — browser sends cookie automatically
const es = new EventSource("/events", { withCredentials: true });

// Server: read cookie instead of Authorization header
import cookieParser from "cookie-parser";
app.use(cookieParser(process.env.COOKIE_SECRET));

export function adminAuth(req: Request, res: Response, next: NextFunction) {
  const sessionToken = req.cookies?.admin_session;
  if (!ADMIN_SESSIONS.has(sessionToken)) {
    res.status(403).json({ error: "forbidden" });
    return;
  }
  next();
}

For an internal ops dashboard, Option A is pragmatic. For any user-facing product, use Option B. 🔐

---

🔁 Part 7: Reconnection and Missed Events

The browser's EventSource API automatically reconnects on network drops using Last-Event-ID. Here is the full flow:

Client opens:  GET /events
Server sends:  id: 1\ndata: {...}\n\n
               id: 2\ndata: {...}\n\n
               id: 3\ndata: {...}\n\n

Network drops ───────────────────────────

Client reconnects: GET /events
                   Last-Event-ID: 3      ← browser adds this automatically

Server replays:    id: 4\ndata: {...}\n\n   ← only events since id 3
                   id: 5\ndata: {...}\n\n

The 500-event rolling buffer in your SSE handler covers reconnects that happen within minutes. For longer gaps (server restart, extended outage), the buffer is empty and the client gets a "connected" event with current stats rather than a backfill. This is acceptable — the dashboard is a real-time view, not an audit log. Your audit log lives in pino structured logs (Part 7). 📂

---

🐳 Part 8: Serving the Dashboard from the Same Process

For simplicity, serve the built React dashboard as static files from Express:

// In server.ts
import { fileURLToPath } from "url";
import path from "path";
import express from "express";

const __dirname = path.dirname(fileURLToPath(import.meta.url));

// Serve built dashboard — run: npm run build:dashboard first
app.use(
  "/dashboard",
  express.static(path.join(__dirname, "../../dashboard/dist"))
);

// SPA fallback
app.get("/dashboard/*", (_req, res) => {
  res.sendFile(path.join(__dirname, "../../dashboard/dist/index.html"));
});

Open http://localhost:3000/dashboard and you see the live event feed. No separate deployment needed. 🎉

For production, put the dashboard behind Nginx with a separate subdomain (dashboard.your-mcp-server.com) so you can apply stricter access controls — IP allowlisting, VPN-only access — without affecting the MCP API routes.

---

🎬 Part 9: Putting It All Together — Live Trace

Here is what the dashboard shows when three agents run simultaneously:

● 10:31:02  session:open      session a1b2c3d4
● 10:31:02  session:open      session e5f6g7h8
● 10:31:03  tool:start        search_knowledge_base  (a1b2c3d4)
● 10:31:03  tool:start        get_current_weather    (e5f6g7h8)
● 10:31:04  session:open      session i9j0k1l2
● 10:31:04  tool:start        index_document         (i9j0k1l2)
✓ 10:31:05  tool:done         get_current_weather    842ms  (e5f6g7h8)
✓ 10:31:05  tool:done         search_knowledge_base  1203ms (a1b2c3d4)
● 10:31:05  tool:start        search_knowledge_base  (a1b2c3d4)  ← second call
✗ 10:31:06  tool:error        search_knowledge_base  "No results" (a1b2c3d4)
✓ 10:31:08  tool:done         index_document         4123ms (i9j0k1l2)
← 10:31:10  session:close     session e5f6g7h8

In 8 seconds you can see: which sessions are active, which tools are being called, how long each takes, where errors occurred, and which sessions have closed. This is the operational visibility that was completely absent in a single-session CLI agent. 🔍

---

💡 Key Takeaways

EventEmitter is enough for a single process. Do not reach for Redis pub/sub or NATS until you have multiple server instances. The in-process event bus is synchronous, zero-latency, and requires zero infrastructure.

SSE beats WebSockets for server-to-client push. SSE is HTTP/1.1 compatible, survives proxies and load balancers without configuration, auto-reconnects with Last-Event-ID, and has a native browser API. Use WebSockets only when you need bidirectional communication.

The rolling event buffer is your reconnect safety net. Without it, every network blip leaves a gap. 500 events at typical load covers several minutes of reconnect window — tune based on your event rate.

Heartbeat comments keep connections alive through proxies. A : heartbeat\n\n comment (no data: prefix, invisible to listeners) every 15 seconds prevents idle connection timeouts in nginx, AWS ALB, and most corporate proxies.

Update stats from the event stream, not from polling. The dashboard in Part 5 updates stats state directly from incoming SSE events rather than polling /stats every second. This eliminates polling entirely — the UI is always up to date within milliseconds of each event. ⚡

---

🎯 Summary

In Part 13 you added real-time operational visibility to the MCP server:

• 📡 EventBus — typed, singleton EventEmitter that every part of the server publishes to
• 🔧 Event emission — tool:start, tool:done, tool:error, session:open, session:close, session:expired published automatically via instrumentedToolCall
• 🔌 GET /events SSE endpoint — admin-gated, with heartbeat keepalive and Last-Event-ID replay
• 📊 LiveStats aggregator — in-process stats updated from event stream, zero database queries
• 🖥️ React dashboard — live event feed with colour-coded rows, stats cards, and auto-reconnect
• 🔒 Cookie-based SSE auth — browser-compatible auth without tokens in URLs

In Part 14 we will add agent-to-agent communication — one MCP agent that can spawn sub-agents, delegate tasks, and aggregate results, building a true multi-agent orchestration system. 🤖🤖

---

📚 Further Reading

• 📡 Server-Sent Events — MDN
• 🔁 EventSource reconnection with Last-Event-ID
• ⚡ Node.js EventEmitter docs
• 🖥️ React hooks and SSE
• 💬 Part 4: Streaming Agents and Interactive CLI
• 🏗️ Part 6: Multi-Tenant MCP Sessions