AliveMCP

Guide · Real-time / WebSocket

MCP Tools for Centrifugo — subscription JWTs, channel namespaces, history recovery, cluster presence

Three Centrifugo behaviours catch developers building MCP real-time tools: subscription tokens must include the exact channel string in the channel claim — a connection JWT (which grants general access) is not sufficient to subscribe to a private channel; Centrifugo uses a separate subscription JWT per channel with a channel claim matching the channel name exactly (including the namespace prefix); a missing or mismatched channel claim returns a "permission denied" error with error code 103 even when the connection JWT is valid; subscription tokens expire independently of the connection — the client must implement a getToken callback that returns a fresh subscription token before the current one expires; Centrifugo disconnects the subscription (not the connection) when the token's exp claim passes, and the default refresh happens at 75% of TTL elapsed; if the callback returns an error, the subscription enters a failed state; and channel history in a Centrifugo cluster is node-local by default — in a multi-node Centrifugo cluster without Redis broker, history stored on node A is not visible when querying from node B; to get cluster-wide consistent history, configure the Redis broker (broker: redis) which centralises history storage, or use the Centrifugo PRO history API that aggregates across nodes.

TL;DR

Publish server-side: POST /api with {"method":"publish","params":{"channel":"ns:ch","data":{}}} and Authorization: apikey YOUR_KEY. Subscription JWT: sign {"sub":"user_id","channel":"ns:ch","exp":...} with CENTRIFUGO_TOKEN_SECRET. History: subscribe with { since: { epoch, offset }, recover: true }. Health: GET /health → {"status":"ok"}.

Centrifugo architecture — server, client, and API layers

Centrifugo is a self-hosted real-time messaging server. Your MCP server publishes messages to Centrifugo's HTTP or gRPC API; Centrifugo then fans out to all clients subscribed to that channel. This decoupled architecture means your MCP server does not maintain WebSocket connections to clients — Centrifugo manages all client connections and subscriptions. Your server only needs to be able to make HTTP POST requests to the Centrifugo API.

// Centrifugo server-side API client
class CentrifugoClient {
  private baseUrl: string;
  private apiKey: string;

  constructor() {
    this.baseUrl = process.env.CENTRIFUGO_URL!; // e.g. 'http://centrifugo:8000'
    this.apiKey = process.env.CENTRIFUGO_API_KEY!;
  }

  private async call(method: string, params: object): Promise {
    const response = await fetch(`${this.baseUrl}/api`, {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        'Authorization': `apikey ${this.apiKey}`,
        'X-API-Key': this.apiKey, // alternative header supported by some versions
      },
      body: JSON.stringify({ method, params }),
    });

    if (!response.ok) {
      const text = await response.text();
      throw new Error(`Centrifugo API error ${response.status}: ${text}`);
    }

    const result = await response.json() as { error?: { code: number; message: string }; result?: unknown };
    if (result.error) {
      throw new Error(`Centrifugo method error ${result.error.code}: ${result.error.message}`);
    }
    return result.result;
  }

  async publish(channel: string, data: unknown): Promise {
    await this.call('publish', { channel, data });
  }

  async broadcast(channels: string[], data: unknown): Promise {
    await this.call('broadcast', { channels, data });
  }

  async presence(channel: string): Promise {
    return this.call('presence', { channel }) as Promise;
  }

  async history(channel: string, limit = 100, since?: { epoch: string; offset: number }): Promise {
    return this.call('history', {
      channel,
      limit,
      ...(since ? { since, reverse: false } : {}),
    }) as Promise;
  }

  async disconnect(user: string): Promise {
    await this.call('disconnect', { user });
  }
}

export const centrifugo = new CentrifugoClient();

		
The Centrifugo API returns a top-level error field even on HTTP 200 responses — always check for it. HTTP 200 with {"error":{"code":103,"message":"permission denied"}} is an application-level error, not an HTTP error. The SDK wrapping above handles this distinction consistently.

	

	

		
Connection and subscription JWTs

		
Centrifugo uses two distinct JWT types: a connection JWT (issued at client connect time) and per-channel subscription JWTs (issued for each private channel subscription). The connection JWT authenticates the user to the Centrifugo server. The subscription JWT authorises access to a specific channel. Both are signed with the same secret (CENTRIFUGO_TOKEN_SECRET in config) but carry different claims.


		
import jwt from 'jsonwebtoken';

const CENTRIFUGO_SECRET = process.env.CENTRIFUGO_TOKEN_SECRET!;

// Connection JWT — authenticates the client to Centrifugo
// The client passes this when establishing the WebSocket connection
function issueConnectionToken(userId: string): string {
  return jwt.sign(
    {
      sub: userId,       // required: user ID (string)
      // info: { name: 'Alice' },  // optional: arbitrary user metadata
    },
    CENTRIFUGO_SECRET,
    { expiresIn: '1h' }
  );
}

// Subscription JWT — authorises access to a SPECIFIC private channel
// Private channels in Centrifugo require this (channels that start with '$' or
// channels in namespaces configured with 'protected: true')
function issueSubscriptionToken(userId: string, channel: string): string {
  return jwt.sign(
    {
      sub: userId,
      channel,           // REQUIRED for subscription JWT — exact channel string
      // info: { role: 'viewer' },  // optional: channel-specific user info
    },
    CENTRIFUGO_SECRET,
    { expiresIn: '30m' }  // shorter TTL for subscription tokens
  );
}

// Token endpoint — clients call this to get both token types
// app.post('/api/centrifugo/token', authRequired, async (req, res) => {
//   const userId = req.user.id;
//   const connectionToken = issueConnectionToken(userId);
//   res.json({ token: connectionToken });
// });

// Subscription token endpoint — called per private channel subscription
// app.post('/api/centrifugo/subscription-token', authRequired, async (req, res) => {
//   const { channel } = req.body as { channel: string };
//   const userId = req.user.id;
//
//   // Validate that the user can access this channel
//   const allowed = await userCanSubscribe(userId, channel);
//   if (!allowed) return res.status(403).json({ error: 'forbidden' });
//
//   const token = issueSubscriptionToken(userId, channel);
//   res.json({ token });
// });


		
The channel claim must be the complete channel name including its namespace prefix. If your channel is agent:session:abc123, the claim must be exactly "channel": "agent:session:abc123". A token with "channel": "session:abc123" (wrong namespace) returns error code 103. Wildcard channel claims ("channel": "agent:*") are not supported in standard Centrifugo — each channel needs its own token.

	


	

		
Channel namespaces — configuration and routing

		
Centrifugo channels are prefixed with a namespace: namespace:channel-name. The colon separates the namespace from the channel slug. Namespaces are defined in Centrifugo's configuration file and control per-namespace behaviour: history retention, presence tracking, publish permissions, and whether subscriptions require JWT tokens. A channel in an unconfigured namespace returns error code 102 (unknown channel).


		
// Centrifugo config.json (server configuration)
// {
//   "token_hmac_secret_key": "your-secret",
//   "api_key": "your-api-key",
//   "namespaces": [
//     {
//       "name": "agent",
//       "history_size": 100,
//       "history_ttl": "24h",
//       "presence": true,
//       "presence_idle_timeout": "60s",
//       "protected": true  // requires subscription JWT for all channels in this namespace
//     },
//     {
//       "name": "public",
//       "history_size": 0,     // no history
//       "presence": false,     // no presence
//       "protected": false     // anyone can subscribe (only connection JWT required)
//     },
//     {
//       "name": "broadcast",
//       "history_size": 10,
//       "history_ttl": "5m",
//       "presence": false,
//       "allow_publish_for_client": true  // clients can publish (without going through your server)
//     }
//   ]
// }

// Validate channel format before publishing
function parseChannel(channel: string): { namespace: string; slug: string } {
  const colonIndex = channel.indexOf(':');
  if (colonIndex === -1) {
    // No namespace prefix — channel belongs to the root namespace
    // Root namespace capabilities depend on Centrifugo config
    return { namespace: '', slug: channel };
  }
  return {
    namespace: channel.slice(0, colonIndex),
    slug: channel.slice(colonIndex + 1),
  };
}

// Build channel names consistently
function agentChannel(sessionId: string): string {
  return `agent:session-${sessionId}`;
}

function publicBroadcastChannel(topic: string): string {
  return `public:${topic}`;
}


		
The presence_idle_timeout controls when inactive clients (those who haven't pinged recently) are removed from presence. The default Centrifugo ping interval is 25 seconds — set presence_idle_timeout to at least 2x the client ping interval to avoid spurious presence removals during temporary network hiccups.

	


	

		
History and gap recovery

		
Centrifugo's channel history allows clients to recover messages missed during a disconnection. Each message in history has an offset (monotonically increasing integer per channel) and an epoch (a string that resets when Centrifugo restarts or clears history). The client must store the epoch and offset of the last received message and pass them when reconnecting to request only the gap.


		
// Server: check what messages a client missed since their last known position
async function getHistorySince(
  channel: string,
  epoch: string,
  offset: number
): Promise<{ messages: unknown[]; recovered: boolean; topOffset: number; topEpoch: string }> {
  type HistoryResult = {
    publications: Array<{ data: unknown; offset: number }>;
    epoch: string;
    offset: number;
  };

  const result = await centrifugo.history(channel, 100, { epoch, offset }) as HistoryResult;

  return {
    messages: result.publications ?? [],
    recovered: result.publications !== null,  // null means epoch mismatch — full state needed
    topOffset: result.offset,
    topEpoch: result.epoch,
  };
}

// Client-side recovery pattern (centrifuge-js client library)
// const sub = centrifuge.newSubscription('agent:session-abc', {
//   // On initial connect, recover from the last seen position
//   since: lastSeenPosition,  // { epoch: '...', offset: 42 }
//   recoverable: true,
//
//   // getToken is called when the subscription token is about to expire
//   // Centrifugo refreshes at 75% of TTL elapsed
//   getToken: async (ctx) => {
//     const response = await fetch('/api/centrifugo/subscription-token', {
//       method: 'POST',
//       headers: { 'Content-Type': 'application/json' },
//       body: JSON.stringify({ channel: ctx.channel }),
//     });
//     const { token } = await response.json();
//     return token;
//   },
// });
//
// sub.on('subscribed', (ctx) => {
//   if (ctx.wasRecovering && !ctx.recovered) {
//     // Gap was too large or epoch changed — fetch full application state
//     fetchFullState();
//   }
// });
//
// // Update the stored position on every received publication
// sub.on('publication', (ctx) => {
//   lastSeenPosition = { epoch: ctx.streamPosition.epoch, offset: ctx.streamPosition.offset };
//   handleMessage(ctx.data);
// });


		
When epoch mismatch occurs (Centrifugo restarted and history was cleared, or the history TTL expired), the history API returns a response indicating the client cannot recover from the specified position. The client must detect this case and perform a full state fetch from your application layer rather than assuming it can fill the gap from Centrifugo history alone.

	


	

		
Presence in single-node vs clustered deployments

		
Centrifugo presence tracks which clients are subscribed to a channel. In a single-node deployment, presence is maintained in memory and returned accurately. In a clustered deployment (multiple Centrifugo nodes behind a load balancer), each node only knows about clients connected to it — the HTTP API presence query hits only the node that receives the request. To get cluster-wide presence, either use the Redis broker (which centralises presence state) or query all nodes and merge results.


		
// Single-node presence query
async function getChannelPresence(channel: string) {
  type PresenceResult = {
    presence: Record;
  };
  const result = await centrifugo.presence(channel) as PresenceResult;
  const members = Object.values(result.presence ?? {});
  return {
    channel,
    count: members.length,
    members: members.map(m => ({ clientId: m.client, userId: m.user })),
  };
}

// Cluster-aware presence query (when Redis broker is NOT configured)
// Requires knowing all Centrifugo node addresses
async function getClusterPresence(
  channel: string,
  nodeUrls: string[]
): Promise<{ count: number; members: Array<{ clientId: string; userId: string }> }> {
  const results = await Promise.allSettled(
    nodeUrls.map(async (url) => {
      const response = await fetch(`${url}/api`, {
        method: 'POST',
        headers: {
          'Content-Type': 'application/json',
          'Authorization': `apikey ${process.env.CENTRIFUGO_API_KEY}`,
        },
        body: JSON.stringify({ method: 'presence', params: { channel } }),
      });
      return response.json() as Promise<{ result: { presence: Record } }>;
    })
  );

  const seenClients = new Set();
  const allMembers: Array<{ clientId: string; userId: string }> = [];

  for (const r of results) {
    if (r.status === 'fulfilled' && r.value.result?.presence) {
      for (const [clientId, member] of Object.entries(r.value.result.presence)) {
        if (!seenClients.has(clientId)) {
          seenClients.add(clientId);
          allMembers.push({ clientId, userId: member.user });
        }
      }
    }
  }

  return { count: allMembers.length, members: allMembers };
}


		
The simplest path to cluster-wide presence consistency is configuring the Redis broker in Centrifugo's config: "broker": "redis" with "redis_address": "redis:6379". With Redis broker enabled, all presence data is stored in Redis and the presence API returns accurate counts regardless of which node handles the request.

	


	

		
Health probe — API and node info

		
Centrifugo exposes a dedicated health endpoint at /health and a node info endpoint at /api (via the info method). The health endpoint does not require API key authentication and returns {"status":"ok"} when the node is healthy. The info endpoint returns detailed node metrics including channel count, client count, and uptime.


		
// Health check — no auth required
async function checkCentrifugoHealth(): Promise {
  try {
    const response = await fetch(`${process.env.CENTRIFUGO_URL}/health`);
    const body = await response.json() as { status?: string };
    return {
      healthy: response.ok && body.status === 'ok',
      httpStatus: response.status,
      status: body.status,
    };
  } catch (err) {
    return { healthy: false, error: String(err) };
  }
}

// Detailed node info — requires API key
async function getCentrifugoInfo(): Promise {
  type InfoResult = {
    nodes: Array<{
      uid: string;
      name: string;
      version: string;
      num_clients: number;
      num_users: number;
      num_channels: number;
      uptime: number;
    }>;
  };

  try {
    const result = await centrifugo['call']('info', {}) as InfoResult;
    return {
      nodeCount: result.nodes.length,
      totalClients: result.nodes.reduce((sum, n) => sum + n.num_clients, 0),
      totalChannels: result.nodes.reduce((sum, n) => sum + n.num_channels, 0),
      nodes: result.nodes.map(n => ({
        name: n.name,
        version: n.version,
        clients: n.num_clients,
        channels: n.num_channels,
        uptimeSeconds: n.uptime,
      })),
    };
  } catch (err) {
    return { error: String(err) };
  }
}

// Startup probe
async function runStartupCheck(): Promise {
  const health = await checkCentrifugoHealth();
  if (!('healthy' in health) || !health.healthy) {
    console.error('Centrifugo health check failed:', health);
    process.exit(1);
  }
  console.log('Centrifugo health OK:', health);
}

		
AliveMCP monitors Centrifugo-backed MCP servers by polling /health and tracking publish latency via the HTTP API. A rising publish latency (over 100ms) without a corresponding increase in channel count typically indicates Redis broker connectivity issues — the Centrifugo node is waiting on Redis for history or presence writes.

	

	

		
Related guides