NATS Agent Patterns

Production-tested NATS JetStream patterns for AI agent communication.

Built by GenBrain AI -- the company behind agent.ceo, where these patterns power production AI agent teams.

What This Is

A collection of messaging patterns extracted from a real production system where AI agents collaborate as teams. Each pattern is a self-contained Python class that wraps NATS JetStream with agent-friendly abstractions: JSON serialization, agent addressing, durable delivery, and failure handling.

graph LR
    subgraph Agents
        A1[Manager Agent]
        A2[Worker Agent 1]
        A3[Worker Agent 2]
        A4[Monitor Agent]
    end

    subgraph NATS JetStream
        RR[Request/Reply]
        TQ[Task Queue]
        EB[Event Bus]
        IN[Agent Inboxes]
        DLQ[Dead Letter Queue]
    end

    A1 -->|assign task| TQ
    TQ -->|consume| A2
    TQ -->|consume| A3
    A2 -->|reply result| RR
    RR -->|response| A1
    A4 -->|publish alert| EB
    EB -->|subscribe| A1
    EB -->|subscribe| A2
    A1 -->|send message| IN
    IN -->|deliver| A3
    TQ -.->|failed tasks| DLQ

Loading

Patterns

Pattern	Use Case	Durability
Request/Reply	Synchronous agent-to-agent calls	Ephemeral
Task Queue	Async work distribution with load balancing	Durable (JetStream)
Pub/Sub EventBus	Event broadcasting, decoupled reactions	Durable (JetStream)
Agent Inbox	Guaranteed per-agent message delivery	Durable (JetStream)
SLA Tracking	Monitor response time compliance	In-memory
Dead Letter Queue	Handle persistent processing failures	Durable (JetStream)

Quick Start

pip install nats-agent-patterns

Or install from source:

git clone https://github.com/genbrain-ai/nats-agent-patterns.git
cd nats-agent-patterns
pip install -e ".[dev]"

Start a NATS server with JetStream:

# Install: https://docs.nats.io/running-a-nats-service/introduction/installation
nats-server --jetstream

Send a message to an agent's inbox:

import asyncio
import nats
from patterns import AgentInbox

async def main():
    nc = await nats.connect("nats://localhost:4222")
    inbox = AgentInbox(nc, agent_id="worker-1", org="myorg")
    await inbox.setup()

    await inbox.send(
        "worker-1",
        subject="New task",
        body={"action": "review", "pr": 42},
        priority=1,
    )

    messages = await inbox.peek(limit=5)
    for msg in messages:
        print(f"From {msg.from_agent}: {msg.subject}")

    await nc.drain()

asyncio.run(main())

Pattern Examples

Request/Reply -- Synchronous Agent Calls

One agent asks another a question and waits for the answer.

from patterns import RequestReplyClient, RequestReplyServer, AgentRequest

# Worker agent: serve requests
server = RequestReplyServer(nc, agent_id="worker-1", org="myorg")

async def handle(request: AgentRequest) -> dict:
    return {"status": "idle", "tasks_completed": 42}

await server.serve(handle)

# Manager agent: send request
client = RequestReplyClient(nc, agent_id="manager", org="myorg")
reply = await client.request("worker-1", {"action": "status"}, timeout=5.0)
print(reply.payload)  # {"status": "idle", "tasks_completed": 42}

Task Queue -- Distributed Work Processing

Enqueue tasks for workers with automatic retry and load balancing.

from patterns import TaskQueue, Task

queue = TaskQueue(nc, stream_name="CODE_REVIEWS", org="myorg")
await queue.setup()

# Enqueue a task
await queue.enqueue(Task(
    type="review",
    payload={"pr_number": 42, "repo": "api"},
    priority=1,
))

# Worker processes tasks
async def review_handler(task: Task):
    print(f"Reviewing PR #{task.payload['pr_number']}")

worker = await queue.worker(review_handler, durable_name="reviewer-1")
await worker.process_batch()

Pub/Sub EventBus -- Event Broadcasting

Publish events that multiple agents can react to independently.

from patterns import EventBus, Event

bus = EventBus(nc, org="myorg", source_agent="deployer", stream_name="EVENTS")
await bus.setup()

# Publish an event
await bus.publish_event("deployment.completed", {
    "service": "api",
    "version": "2.1.0",
    "environment": "production",
})

# Another agent subscribes
async def on_deploy(event: Event):
    print(f"{event.source_agent} deployed {event.data['service']}")

await bus.subscribe_events(["deployment.*"], handler=on_deploy, durable_name="notifier")

Agent Inbox -- Guaranteed Delivery

Send messages to a specific agent with persistence and priority.

from patterns import AgentInbox

inbox = AgentInbox(nc, agent_id="cto", org="myorg")
await inbox.setup()

# Send a high-priority message
await inbox.send(
    "cto",
    subject="Architecture review needed",
    body={"pr": 123, "deadline": "2024-03-15"},
    priority=2,  # urgent
)

# Agent processes inbox
async def handle_message(msg):
    print(f"[{msg.priority}] {msg.subject}: {msg.body}")

processed = await inbox.receive(handler=handle_message, batch_size=10)

SLA Tracking -- Monitor Response Times

Track whether agents respond within acceptable timeframes.

from patterns import SLATracker

tracker = SLATracker()

# Start tracking when a request is sent
tracker.track_request("req-123", deadline_seconds=30.0)

# Record when the response arrives
elapsed = tracker.record_completion("req-123")
print(f"Response took {elapsed:.2f}s")

# Check for violations
violations = tracker.check_violations()
for v in violations:
    print(f"SLA VIOLATION: {v.request_id} exceeded deadline by {v.exceeded_by:.1f}s")

# Get aggregate metrics
metrics = tracker.get_metrics()
print(f"p95 latency: {metrics.p95_seconds:.2f}s")

Dead Letter Queue -- Handle Failures

Capture failed messages for debugging and replay after fixing the issue.

from patterns import DeadLetterQueue

dlq = DeadLetterQueue(nc, org="myorg")
await dlq.setup()

# Route a failed message to DLQ
await dlq.on_failure(
    message={"task_id": "abc", "type": "analyze"},
    error="Model timeout after 30s",
    retry_count=3,
    original_subject="myorg.tasks.analyze",
)

# Inspect what failed
entries = await dlq.inspect(limit=10)
for entry in entries:
    print(f"Failed: {entry.error} (retried {entry.retry_count}x)")

# Replay after fixing the bug
successes, failures = await dlq.replay(fixed_handler)
print(f"Replayed: {successes} ok, {failures} still failing")

Running the Examples

Each example is a standalone script that connects to a local NATS server:

# Start NATS with JetStream
nats-server --jetstream

# In another terminal
python examples/multi_agent_chat.py
python examples/task_delegation.py
python examples/event_driven_agents.py
python examples/resilient_delivery.py

Documentation

• Why NATS for AI Agents? -- Comparison with Kafka, RabbitMQ, and Redis
• Patterns Guide -- When to use which pattern, with decision flowchart
• Production Tips -- Cluster sizing, TLS, monitoring, and operational advice

Running Tests

pip install -e ".[dev]"
pytest tests/ -v

Tests use mocked NATS connections -- no running NATS server required.

agent.ceo

agent.ceo is a production implementation of these patterns, orchestrating AI agent teams at scale. Agents like CTO, DevOps, and Fullstack collaborate through NATS JetStream -- delegating tasks, broadcasting events, tracking SLAs, and recovering from failures using exactly these patterns.

Try it free -- agent.ceo

Contributing

Contributions are welcome. Please:

1. Fork the repository
2. Create a feature branch (git checkout -b feature/my-pattern)
3. Write tests for any new patterns
4. Ensure all tests pass (pytest tests/ -v)
5. Submit a pull request

License

MIT -- see LICENSE for details.

---

Built with care by GenBrain AI.