bot/skills/hanzo-pubsub/SKILL.md

name, description, metadata

name	description	metadata
hanzo-pubsub	High-performance event streaming and message queues with Hanzo PubSub (NATS). Pub/sub, persistent streams, consumer groups, exactly-once delivery.	bot requires install bins python3 id kind package label pip pip nats-py Install NATS Client (pip)

Hanzo PubSub — Event Streaming

pip install nats-py

NATS-based high-performance event streaming: pub/sub, persistent streams, consumer groups, exactly-once delivery. 10M+ messages/sec.

Quick Start

import nats

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

# Simple pub/sub
await nc.publish("events.user.created", b'{"user_id": "123"}')

# Subscribe
async def handler(msg):
    print(f"Received: {msg.data.decode()}")

sub = await nc.subscribe("events.>", cb=handler)

JetStream (Persistent Streams)

js = nc.jetstream()

# Create persistent stream
await js.add_stream(name="ORDERS", subjects=["orders.*"])

# Publish to stream
ack = await js.publish("orders.new", b'{"item": "widget", "qty": 5}')
print(f"Sequence: {ack.seq}")

# Subscribe with durable consumer
sub = await js.subscribe("orders.*", durable="order-processor")
async for msg in sub.messages:
    print(f"Order: {msg.data.decode()}")
    await msg.ack()

Consumer Groups

# Create pull-based consumer group
psub = await js.pull_subscribe("orders.*", durable="workers")

# Fetch batch of messages
msgs = await psub.fetch(batch=10, timeout=5)
for msg in msgs:
    process(msg.data)
    await msg.ack()

Key-Value Store (Built-in)

kv = await js.create_key_value(bucket="config")

await kv.put("db.host", b"localhost")
entry = await kv.get("db.host")
print(entry.value.decode())

# Watch for changes
watcher = await kv.watchall()
async for entry in watcher:
    print(f"{entry.key} = {entry.value}")

Object Store

obs = await js.create_object_store(bucket="artifacts")

# Store object
await obs.put("model.bin", open("model.bin", "rb"))

# Retrieve
data = await obs.get("model.bin")

Request/Reply

# Service
async def handler(msg):
    await msg.respond(b"pong")

await nc.subscribe("ping", cb=handler)

# Client
response = await nc.request("ping", b"", timeout=5)
print(response.data.decode())  # "pong"

Ports

• Client: 4222
• Monitoring: 8222

Environment Variables

NATS_URL=nats://localhost:4222
NATS_TOKEN=...                  # Auth token