Nats

NATS Core Knowledge

Full Reference: See advanced.md for JetStream patterns (Node.js, Java), security configuration, TLS setup, and clustering.

Deep Knowledge: Use mcp__documentation__fetch_docs with technology: nats for comprehensive documentation.

Quick Start (Docker)

# docker-compose.yml
services:
  nats:
    image: nats:latest
    ports:
      - "4222:4222"   # Client
      - "8222:8222"   # Monitoring
    command: "-js -m 8222"  # Enable JetStream

Core Concepts

Subject Patterns

orders.created     → Specific subject
orders.*           → orders.created, orders.updated (single wildcard)
orders.>           → orders.created, orders.us.east (multi wildcard)

Core NATS (Node.js)

import { connect, StringCodec } from 'nats';

const nc = await connect({ servers: 'localhost:4222' });
const sc = StringCodec();

// Simple publish
nc.publish('orders.created', sc.encode(JSON.stringify(order)));

// Subscribe
const sub = nc.subscribe('orders.*');
for await (const msg of sub) {
  const order = JSON.parse(sc.decode(msg.data));
  console.log(`Received on ${msg.subject}:`, order);
}

// Queue group (load balancing)
const qsub = nc.subscribe('orders.process', { queue: 'workers' });

// Request/Reply
const response = await nc.request('orders.validate', sc.encode(JSON.stringify(order)), {
  timeout: 5000,
});
const result = JSON.parse(sc.decode(response.data));

await nc.drain();

Core NATS (Python)

import nats
import json
import asyncio

async def main():
    nc = await nats.connect("nats://localhost:4222")

    # Publish
    await nc.publish("orders.created", json.dumps(order).encode())

    # Subscribe
    async def message_handler(msg):
        order = json.loads(msg.data.decode())
        print(f"Received: {order}")

    await nc.subscribe("orders.*", cb=message_handler)

    # Request/Reply
    response = await nc.request("orders.validate",
                                json.dumps(order).encode(),
                                timeout=5)
    result = json.loads(response.data.decode())

    await nc.drain()

asyncio.run(main())

Core NATS (Go)

nc, _ := nats.Connect(nats.DefaultURL)
defer nc.Drain()

// Publish
body, _ := json.Marshal(order)
nc.Publish("orders.created", body)

// Subscribe
nc.Subscribe("orders.*", func(msg *nats.Msg) {
    var order Order
    json.Unmarshal(msg.Data, &order)
    processOrder(order)
})

// Queue group
nc.QueueSubscribe("orders.process", "workers", func(msg *nats.Msg) {
    // Load balanced
})

// Request/Reply
response, _ := nc.Request("orders.validate", body, 5*time.Second)

When NOT to Use This Skill

• Long-term event replay required - Kafka provides better retention
• Complex routing patterns - RabbitMQ exchanges are more flexible
• JMS compliance needed - Use ActiveMQ
• AWS-native integration - SQS integrates better

Anti-Patterns

Quick Troubleshooting

Production Checklist

• TLS enabled
• Authentication configured
• Authorization (permissions)
• JetStream replicas >= 3
• Stream limits configured
• Consumer max_deliver set
• Ack timeout configured
• Monitoring dashboards
• Cluster health checks

Reference Documentation

Available topics: basics, jetstream, patterns, production