Guide · Message Queue

MCP Tools for NATS and JetStream — subjects, streams, and pull consumer delivery

NATS is a cloud-native messaging system with two distinct runtime modes: Core NATS (pub/sub with no persistence — if no subscriber is connected when a message arrives, it is dropped) and JetStream (the persistence layer built on top, providing durable streams, consumers, and Key-Value stores). When you build an MCP tool that monitors or manages NATS — checking consumer lag, inspecting stream state, recovering from processing failures, or reading configuration from the KV store — you must understand the sharp boundary between these two modes: JetStream state (streams, consumers, KV buckets) is queryable via management APIs, while Core NATS pub/sub has no durable state to query. The three non-obvious properties that drive correct NATS tooling are: pull vs push consumer delivery mode (pull is the right choice for MCP tools because it provides natural backpressure and does not require the consumer to be running continuously), ack modes beyond basic ack (nak, term, and in-progress are critical for correct failure handling), and stream retention policies (limits, interest, and workqueue have fundamentally different deletion semantics).

TL;DR

Use the nats Node.js client library for both Core NATS and JetStream. For MCP tools that monitor consumer health, use JetStream's jsm.consumers.info(stream, consumer) and check num_pending (undelivered messages) and num_ack_pending (delivered but unacknowledged). Prefer pull consumers over push consumers in MCP server contexts — pull consumers are durable, support explicit fetch batching, and do not require continuous connection. For stream health, check jsm.streams.info(stream) and inspect state.messages, state.bytes, and state.consumer_count. Register an HTTP health bridge with AliveMCP to detect NATS server unavailability before your application starts failing silently.

Core NATS vs JetStream: knowing which layer to use

Core NATS is a fire-and-forget pub/sub system. A publisher sends a message on a subject; all current subscribers receive it. If no subscriber is connected, the message is gone. Core NATS is appropriate for ephemeral events where delivery to a currently-active subscriber is sufficient: service discovery heartbeats, real-time metrics broadcast, request-reply RPC patterns. It is not appropriate for work queues, event sourcing, or any use case requiring durability.

JetStream adds persistence to NATS by capturing messages published to configured subjects into named streams. JetStream streams are configured with subject filters, retention policies, and storage backends. Consumers are named cursors into a stream, similar to Kafka consumer groups or Pulsar subscriptions.

import { connect, StringCodec, JSONCodec } from 'nats';

const nc = await connect({
  servers: process.env.NATS_URL ?? 'nats://localhost:4222',
  // For TLS:
  // tls: { caFile: '/etc/nats/ca.pem', certFile: '/etc/nats/cert.pem', keyFile: '/etc/nats/key.pem' }
});

const jc = JSONCodec();
const js = nc.jetstream();
const jsm = await nc.jetstreamManager();

// JetStream Manager: for stream and consumer administration
// JetStream client: for publishing to streams and consuming from consumers

// Publish to a JetStream stream (subject must match stream's filter subjects)
async function publishToStream(subject, payload) {
  const ack = await js.publish(subject, jc.encode(payload));
  // ack.stream: which stream captured the message
  // ack.seq: sequence number assigned by the stream
  // ack.duplicate: true if this message was a duplicate (idempotent publish with msg_id header)
  return { stream: ack.stream, seq: ack.seq, duplicate: ack.duplicate };
}

// For idempotent publish (deduplication within stream's duplicate window):
async function publishIdempotent(subject, payload, msgId) {
  const hdrs = headers();
  hdrs.set('Nats-Msg-Id', msgId); // unique ID for deduplication
  const ack = await js.publish(subject, jc.encode(payload), { headers: hdrs });
  return ack;
}

The NATS server exposes a monitoring HTTP port (default 8222) with JSON endpoints for server health inspection without requiring a NATS connection:

async function getNATSServerHealth(monitoringUrl = 'http://localhost:8222') {
  const [varz, jsz] = await Promise.all([
    fetch(`${monitoringUrl}/varz`).then(r => r.json()),
    fetch(`${monitoringUrl}/jsz`).then(r => r.json()).catch(() => null)
  ]);

  return {
    server_name: varz.server_name,
    version: varz.version,
    uptime: varz.uptime,
    connections: varz.connections,
    subscriptions: varz.subscriptions,
    slow_consumers: varz.slow_consumers, // subscribers that can't keep up with Core NATS delivery
    jetstream_enabled: jsz != null,
    jetstream: jsz ? {
      memory_bytes: jsz.memory,
      store_bytes: jsz.store,
      accounts: jsz.accounts,
      streams: jsz.streams,
      consumers: jsz.consumers,
    } : null,
  };
}

Stream configuration: retention policies and their deletion semantics

JetStream streams have three retention policies that determine when messages are deleted:

Limits retention (default) — messages are kept until limits are reached: max_msgs, max_bytes, or max_age. The oldest messages are deleted when any limit is exceeded. Independent of consumer acknowledgment — acknowledged messages stay in the stream until limits force deletion. Use for event log or audit trail use cases where messages should be available for replay.
Interest retention — messages are kept while at least one consumer has not acknowledged them. Once all consumers on the stream have acknowledged a message, it is deleted regardless of age or size limits. Use for fan-out scenarios where you want to guarantee all consumers receive a message, but do not need long-term retention.
WorkQueue retention — messages are deleted as soon as any one consumer acknowledges them. This implements queue semantics: one message is processed by one consumer, then deleted. Unlike Limits/Interest, WorkQueue streams require that each message be consumed by exactly one consumer. Use for task queues where parallel workers compete for messages.

// Create a WorkQueue stream (replaces traditional message queue)
async function createWorkQueueStream(streamName, subjects) {
  await jsm.streams.add({
    name: streamName,
    subjects: subjects, // e.g. ['orders.>', 'payments.>'] — '>' matches any suffix
    retention: 'workqueue',
    storage: 'file',       // 'file' (persistent) or 'memory' (non-persistent)
    replicas: 3,           // number of NATS cluster replicas for durability
    max_age: 24 * 60 * 60 * 1e9, // 24 hours in nanoseconds
    discard: 'old',        // when limits reached: drop oldest ('old') or reject new ('new')
    duplicate_window: 2 * 60 * 1e9, // 2-minute deduplication window in nanoseconds
  });
}

// Inspect stream state
async function getStreamHealth(streamName) {
  const info = await jsm.streams.info(streamName);
  return {
    name: info.config.name,
    subjects: info.config.subjects,
    retention: info.config.retention,
    state: {
      messages: info.state.messages,
      bytes: info.state.bytes,
      first_seq: info.state.first_seq,
      last_seq: info.state.last_seq,
      consumer_count: info.state.consumer_count,
      num_subjects: info.state.num_subjects,
      num_deleted: info.state.num_deleted, // gaps due to deleted messages
    },
    limits: {
      max_msgs: info.config.max_msgs,
      max_bytes: info.config.max_bytes,
      max_age_ns: info.config.max_age,
    },
  };
}

Subject filters use NATS wildcard notation. * matches exactly one token; > matches one or more tokens. A stream with subjects ['orders.*'] captures orders.created and orders.paid but not orders.item.added. A stream with subjects ['orders.>'] captures any subject starting with orders.. Multiple subjects in one stream allow a single consumer to process events from logically related subject hierarchies without creating separate streams.

Pull consumers: the right delivery model for MCP tools

JetStream has two consumer delivery modes. Push consumers have the server deliver messages to a push subject — the consumer subscribes to that subject and receives messages as the server sends them. Pull consumers require the consumer to explicitly fetch messages via a Fetch call with a batch size and timeout — messages are only delivered when explicitly requested.

Pull consumers are almost always correct for MCP tools: they work even when the MCP server has intermittent connectivity, they support controlled batch sizes that match processing capacity, and they naturally prevent overwhelming a slow consumer (the consumer fetches only as fast as it can process). Push consumers with continuous high-rate delivery can overwhelm consumers that can't keep up, leading to NATS flagging them as slow consumers and dropping messages.

// Create a durable pull consumer
async function createPullConsumer(streamName, consumerName, filterSubject = null) {
  await jsm.consumers.add(streamName, {
    durable_name: consumerName,   // durable = persists across consumer restarts
    deliver_policy: 'all',        // start from beginning; 'new' = only new messages
    ack_policy: 'explicit',       // must ack each message; 'all' acks all before it
    ack_wait: 30e9,               // 30 seconds in nanoseconds; must ack before timeout
    max_deliver: 5,               // max redelivery attempts before marking as terminal
    filter_subject: filterSubject, // optional: consume only matching subjects from stream
    // For WorkQueue streams, each message goes to one consumer; pull is appropriate
  });
}

// Fetch and process messages in controlled batches
async function processBatch(streamName, consumerName, batchSize = 10) {
  const consumer = await js.consumers.get(streamName, consumerName);
  const messages = await consumer.fetch({
    max_messages: batchSize,
    expires: 5000 // 5-second timeout if fewer than batchSize messages available
  });

  const results = [];
  for await (const msg of messages) {
    try {
      const payload = JSONCodec().decode(msg.data);
      await processMessage(payload, msg.subject, msg.seq);

      msg.ack(); // acknowledge: message processed successfully, do not redeliver
      results.push({ seq: msg.seq, status: 'acked' });
    } catch (err) {
      if (isTransientError(err)) {
        // Nak with delay: redeliver after 30 seconds (do not redeliver immediately)
        msg.nak(30000);
        results.push({ seq: msg.seq, status: 'nacked', retry_in_ms: 30000 });
      } else {
        // Term: permanent failure, do not redeliver
        // Message lands in a "dead letter" stream if configured via max_deliver + discard policy
        msg.term();
        results.push({ seq: msg.seq, status: 'terminated' });
      }
    }
  }
  return results;
}

Ack modes: ack, nak, term, and in-progress

NATS JetStream has four acknowledgment modes that control message redelivery differently from other messaging systems:

ack() — message processed successfully. NATS advances the consumer's cursor past this sequence and does not redeliver it. For WorkQueue retention, this also deletes the message from the stream.
nak(delay?) — processing failed, redeliver after optional delay. Without a delay, NATS redelivers immediately (useful for transient errors that resolve quickly). With a delay in milliseconds (msg.nak(30000)), NATS waits before redelivering. Each nak increments the delivery count — when count reaches max_deliver, the next action depends on configuration.
term() — processing failed permanently, do not redeliver under any circumstances. This terminates delivery regardless of remaining max_deliver count. Use for messages with invalid schema, duplicate detection, or processing failures that will never succeed regardless of retry. The message sequence is marked as terminal in the stream's consumer state — it does not move to a dead letter queue automatically (NATS has no native DLQ; implement one by publishing to a separate stream in a term handler).
working() / in-progress() — heartbeat: processing is ongoing, extend the ack deadline. Send this periodically for long-running operations to prevent NATS from redelivering the message due to ack timeout expiry. Each working() call resets the ack_wait timer.

async function processWithHeartbeat(msg, longRunningProcessor) {
  const jc = JSONCodec();
  const payload = jc.decode(msg.data);

  // Start periodic working() heartbeat
  const heartbeatInterval = setInterval(() => {
    msg.working(); // "still processing, don't redeliver yet"
  }, 10000); // every 10 seconds

  try {
    await longRunningProcessor(payload);
    msg.ack();
  } catch (err) {
    if (isPermanentError(err)) {
      // Publish to dead letter stream before term()
      await js.publish('deadletter.failed', jc.encode({
        original_subject: msg.subject,
        original_seq: msg.seq,
        error: err.message,
        payload,
      }));
      msg.term();
    } else {
      msg.nak(60000); // retry in 1 minute
    }
  } finally {
    clearInterval(heartbeatInterval);
  }
}

Consumer lag monitoring: num_pending and num_ack_pending

JetStream consumer info exposes the two key consumer lag metrics:

async function getConsumerLag(streamName, consumerName) {
  const info = await jsm.consumers.info(streamName, consumerName);

  return {
    stream: streamName,
    consumer: consumerName,
    durable: info.config.durable_name,
    filter_subject: info.config.filter_subject,

    // num_pending: messages in the stream not yet delivered to this consumer
    // This is the backlog — how many messages are waiting to be fetched
    num_pending: info.num_pending,

    // num_ack_pending: messages delivered but not yet acknowledged
    // These are in-flight (fetched but processing not confirmed)
    num_ack_pending: info.num_ack_pending,

    // num_redelivered: messages that have been redelivered at least once
    // High value indicates persistent consumer failures
    num_redelivered: info.num_redelivered,

    // num_waiting: number of pending fetch requests (pull consumers)
    // Non-zero means consumers are waiting for messages — queue is draining
    num_waiting: info.num_waiting,

    // Last activity timestamps
    delivered_consumer_seq: info.delivered.consumer_seq,
    delivered_stream_seq: info.delivered.stream_seq,
    ack_floor_consumer_seq: info.ack_floor.consumer_seq,
    ack_floor_stream_seq: info.ack_floor.stream_seq,
    // Lag = delivered_stream_seq - ack_floor_stream_seq
    explicit_ack_lag: info.delivered.stream_seq - info.ack_floor.stream_seq,
  };
}

// List all consumers and their lag for a stream
async function auditStreamConsumers(streamName) {
  const consumers = await jsm.consumers.list(streamName).next();
  return Promise.all(
    consumers.map(c => getConsumerLag(streamName, c.name))
  );
}

The difference between delivered.stream_seq and ack_floor.stream_seq is the explicit ack lag — how many sequence numbers have been delivered but not yet had their acks committed back to the server. For a healthy consumer actively processing messages, this should be small (bounded by batch size). For a stalled consumer, this grows without bound until ack_wait expires and NATS redelivers.

Key-Value store: a distinct JetStream primitive for configuration and state

JetStream's Key-Value store is a distinct abstraction over streams — a bucket is a stream, and each KV entry is a subject within that stream. Unlike a raw stream, the KV API exposes get, put, delete, and watch operations with revision-based optimistic concurrency control. For MCP tools, KV store is useful for storing and watching configuration state, feature flags, or distributed coordination primitives:

// Create a KV bucket
async function createKVBucket(bucketName, ttlSeconds = 0) {
  const kv = await js.views.kv(bucketName, {
    history: 10,          // keep last 10 revisions per key
    ttl: ttlSeconds * 1e9, // per-key TTL in nanoseconds; 0 = no TTL
    storage: 'file',
    replicas: 3,
  });
  return kv;
}

// Read and write KV entries
async function kvOperations(bucketName) {
  const kv = await js.views.kv(bucketName);

  // Put (create or update)
  await kv.put('config.max_workers', JSONCodec().encode({ value: 10 }));

  // Get with revision for optimistic concurrency
  const entry = await kv.get('config.max_workers');
  if (!entry) throw new Error('Key not found');
  const value = JSONCodec().decode(entry.value);
  const revision = entry.revision;

  // Update only if revision matches (optimistic lock)
  await kv.update('config.max_workers', JSONCodec().encode({ value: 20 }), revision);

  // Watch for changes (reactive configuration)
  const watcher = await kv.watch({ key: 'config.>' }); // '>' = any suffix
  for await (const update of watcher) {
    if (update.operation === 'PUT') {
      console.log('Config updated:', update.key, JSONCodec().decode(update.value));
    } else if (update.operation === 'DEL') {
      console.log('Config deleted:', update.key);
    }
  }
}

KV watch operations return both current values (at startup) and subsequent changes — the watcher delivers a complete snapshot of the bucket's current state before switching to real-time change notification. This makes it safe to initialize configuration from the KV store at MCP server startup without a separate "read all keys then watch" race condition.

Frequently asked questions

What happens to Core NATS messages when no subscriber is connected?

Core NATS messages are dropped immediately if no subscriber is connected to the subject when the message is published. There is no buffering, persistence, or retry at the Core NATS level. This is by design — Core NATS prioritizes low latency over durability, and its delivery guarantee is "at most once" (best-effort, no guarantee of delivery). For MCP tools that need guaranteed delivery — task queues, event streams, workflow triggers — use JetStream with a stream configured to capture the relevant subjects. The MCP tool publishes to a JetStream-captured subject (the stream is created first), and even if no consumer is running, the message persists in the stream until consumed. Never use Core NATS pub/sub for work queue patterns — missed messages during consumer restarts will cause silent processing gaps.

What is the difference between nak() and term() in JetStream?

nak() tells NATS that message processing failed but should be retried. NATS will redeliver the message after an optional delay, incrementing the delivery count each time. When the delivery count reaches the consumer's max_deliver setting, further delivery stops (the consumer skips the message). term() tells NATS to permanently abandon this message — it will not be redelivered regardless of how many delivery attempts remain. Use nak() for transient errors (database connection failure, network timeout, dependency unavailable) that may resolve on retry. Use term() for permanent errors (message schema invalid, business rule violation, ID already processed) where retrying will always fail. A common mistake is using nak() for permanent errors — the consumer will exhaust all max_deliver attempts unnecessarily, during which no other messages on the same key (for KeyFiltered consumers) may process. Always implement a dead letter handler that publishes to a separate stream before calling term() so permanent failures can be reviewed.

How does NATS Key-Value store compare to Redis for MCP tool configuration storage?

Both provide fast key-value reads and writes, but differ in operational and semantic properties. NATS KV: built on JetStream streams, revision history per key, watch operations with at-least-once delivery, storage backed by the NATS cluster (no additional infrastructure), replicas via NATS clustering. Redis: mature ecosystem, richer data structures (sorted sets, streams, pub/sub), Lua scripting, better suited for high-frequency cache-hit patterns. For MCP tools already using NATS, the KV store eliminates a Redis dependency for simple configuration and coordination use cases. For pure caching (frequently read, infrequently written, TTL-based expiry, high read throughput), Redis is more appropriate. For reactive configuration (watch for changes, optimistic concurrency updates, revision history for audit), NATS KV is a natural fit. The watch semantics in NATS KV (delivers full snapshot on start, then changes) are simpler to implement correctly than Redis's keyspace notifications (requires enabling notifications in redis.conf and handling potential missed events).

How do I implement a dead letter queue pattern in NATS JetStream?

NATS JetStream does not have a built-in dead letter queue like RabbitMQ's dead-letter exchange. Implement DLQ semantics explicitly: (1) Create a separate stream (e.g., DEADLETTER) for failed messages. (2) In your consumer's term() handler, publish the original message with failure metadata to a subject captured by the DEADLETTER stream before calling term(). (3) Create a separate consumer on the DEADLETTER stream for human review or automated retry logic. The metadata to include: original subject, original stream sequence number, error message, failure timestamp, original payload. NATS 2.10+ added support for dead_letter stream configuration on consumers (experimental), which automatically forwards messages that exceed max_deliver to a configured subject — check your NATS server version for availability. Until that feature is stable, the explicit publish-to-DL-stream pattern is the reliable approach.

What monitoring endpoints does the NATS server expose without authentication?

The NATS server monitoring port (default 8222) exposes several JSON endpoints accessible without authentication (if the monitoring interface is enabled in server config): /varz — server configuration and general statistics (connections, subscriptions, bytes in/out, slow consumer count); /connz — current client connections with subject subscriptions; /subsz — subscription tree (subject hierarchy and subscriber counts); /jsz — JetStream summary (streams count, consumers count, memory and storage bytes); /jsz?streams=true — per-stream details; /jsz?accounts=true&consumers=true — full consumer state. The /healthz endpoint returns HTTP 200 with {"status":"ok"} if the server is healthy. Unlike the NATS client protocol (port 4222), the monitoring port is read-only HTTP and does not allow publishing or subscribing. For production deployments, the monitoring port is typically restricted to internal network access or not exposed at all — use the NATS server's Prometheus metrics exporter (/metrics) for time-series collection.