NATS Python Client

The NATS client provides several core messaging capabilities:

• Publish/subscribe
• Request/reply
• Queue groups
• Multi-value message headers

The NATS Key-Value client provides the following capabilities:

• Bucket Management: Create, update, and delete buckets.
• Key Operations: Put, get, create, update, delete, and purge keys.
• Concurrency & History: Optimistic concurrency via revisions and per-key history.
• Lifecycle & Monitoring: Per-key TTL, watching for changes with key pattern filters, and listing keys and bucket statuses.

The nats-jetstream client provides the following capabilities:

• Stream Management: Create, update, and delete streams.
• Publishing: Publish messages with acknowledgements.
• Consumers: Support for pull and ordered consumers.
• Message Retrieval: Fetch, message streams, and single-message next().
• Direct Access: Direct message access via get_message and get_last_message_for_subject.
• Metadata: Access account info and list streams/consumers.

Modern JetStream uses Stream and Consumer objects. Instead of passing the stream name to every method on the JetStream context, you create a Stream object (via create_stream or get_stream) which then exposes its own methods for operations like get_message, purge, and delete_message.

from nats.jetstream import StreamConfig

stream = await js.create_stream(StreamConfig(name="ORDERS", subjects=["orders.*"]))
info = await stream.get_info()
msg = await stream.get_message(42)
await stream.purge(filter="orders.tmp")
await stream.delete_message(17)

In nats.client, subscriptions are handled as async iterators rather than using a callback function. To respond to a message, you must explicitly publish to the message.reply field, as message.respond() is no longer available.

Handling Slow Consumers: Subscription delivery is queued (default 65,536 msgs / 64 MiB). If the queue fills, a SlowConsumerError is reported via the error callback. You can increase these limits per subscription:

await client.subscribe("firehose", max_pending_messages=None, max_pending_bytes=None)

Usage Pattern:

# Before (nats.aio)
async def handler(msg):
    await msg.respond(b"ok")

await nc.subscribe("greet.*", cb=handler)

# After (nats.client)
subscription = await client.subscribe("greet.*")
async for message in subscription:
    if message.reply:
        await client.publish(message.reply, b"ok")

To run the documentation examples provided in this repository, ensure you are using Python 3.13+. You can synchronize the environment and run a specific example (like basics_publish.py) from the workspace root using uv.

uv sync
uv run python examples/docs/basics_publish.py

Install the NATS core client package using pip.

pip install nats-core

While many examples connect to the public demo.nats.io server, the JetStream example (jetstream_basic.py) requires a local NATS server with JetStream enabled running on 127.0.0.1:4222.

nats-server -js

In the modern nats.jetstream client, JetStream is no longer attached to the connection object via nc.jetstream(). Instead, you must use the nats.jetstream.new module-level function to explicitly create a context from a client connection. Note that the timeout parameter is no longer part of the context creation; pass timeout= directly to individual API calls where supported.

from nats.client import connect
from nats.jetstream import new as new_jetstream

client = await connect("nats://localhost:4222")
js = new_jetstream(client, domain="hub")

For NATS v2.0 authentication, you can use NKEYS and JWT credentials. First, ensure you have installed the nkeys extra:

pip install nats-py[nkeys]

Then, provide the path to your secret credentials file using the user_credentials parameter in nats.connect().

await nats.connect("tls://connect.ngs.global:4222", user_credentials="/path/to/secret.creds")