NATS Client for Python

Warnings

• breaking In `v2.11.0`, Key-Value (KV) keys validation became enabled by default. Previously invalid keys might now cause errors unless explicitly opted out.
  Fix: If you require using keys that do not conform to the new validation rules, pass `validate_keys=False` to the relevant KV methods (e.g., `kv.put(key, value, validate_keys=False)`).
• breaking The library was renamed from `asyncio-nats-client` to `nats-py` in `v2.0.0`. Key API changes include `subscribe()` no longer returning a Subscription ID directly (it returns a `Subscription` object), and error class names were updated to follow PEP-8 conventions (e.g., `ErrSlowConsumer` became `SlowConsumerError`).
  Fix: Update imports from `asyncio_nats_client` to `nats`. Adjust code to handle `Subscription` objects for subscriptions and use new error class names. Old style errors are subclasses of new ones, so existing `except` blocks might still catch them.
• gotcha `nats-py` is an `asyncio`-native client. All core operations are asynchronous and must be `await`ed within `async def` functions. Attempting to call these directly from synchronous code will result in runtime errors.
  Fix: Ensure all NATS operations are performed within an `async def` function and executed using `asyncio.run(your_async_main_function())` or integrated into an existing `asyncio` event loop. Do not block the event loop with long-running synchronous tasks; offload them if necessary.
• gotcha Proper connection lifecycle management is crucial. Connections should be explicitly closed or drained, especially when using a `nats.connect()` without `async with`. Use `await nc.close()` or `await nc.drain()` to ensure all pending messages are sent and resources are released gracefully.
  Fix: Always use `async with await nats.connect(...) as nc:` for robust connection management, or ensure `await nc.close()` (or `await nc.drain()`) is called in a `finally` block or before application exit for explicit connection handling.
• gotcha The NATS server clustering protocol is incompatible between NATS server v1 and v2. While `nats-py` is compatible with both server versions, a rolling upgrade of a NATS server cluster from v1 to v2 (or vice-versa) can lead to split-brain scenarios and client connection issues during the transition period.
  Fix: If you manage the NATS server, plan server upgrades carefully. Consider a blue/green deployment strategy or a full cluster shutdown/restart if a rolling upgrade is not feasible or causes issues. Consult NATS server documentation for recommended upgrade paths.
• gotcha In `v2.13.0`, the `token` parameter in `nats.connect()` was enhanced to accept a callable, enabling dynamic token refresh on reconnect. If you were previously relying on static token behavior for reconnections, be aware of this new capability.
  Fix: If your authentication tokens are short-lived, consider updating your `connect` call to pass a callable function to the `token` parameter. This function will be invoked on each connection attempt, including reconnections, to fetch a fresh token.

Install

• pip install nats-py Core client
• pip install nats-py[nkeys] With NKeys/JWT support (optional)

Imports

• nats.connect

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

  While `from nats.aio.client import Client` is technically possible for advanced usage, `import nats` and using `nats.connect()` is the idiomatic and recommended way to establish a connection for most use cases since v2.0.0.

Quickstart

This quickstart demonstrates how to connect to a NATS server, subscribe to a subject with an asynchronous callback, publish a message, and ensure graceful shutdown using `async with`. Replace `nats://localhost:4222` with your NATS server address or set the `NATS_URL` environment variable.

import asyncio
import nats
import os

async def main():
    nats_url = os.environ.get("NATS_URL", "nats://localhost:4222")
    try:
        # Connect to NATS. Use async with for graceful connection management.
        async with await nats.connect(nats_url) as nc:
            print(f"Connected to NATS at {nats_url}")

            # Simple asynchronous message handler
            async def message_handler(msg):
                subject = msg.subject
                data = msg.data.decode()
                print(f"Received on '{subject}': {data}")

            # Subscribe to a subject
            sub = await nc.subscribe("foo", cb=message_handler)
            print("Subscribed to 'foo'")

            # Publish a message
            await nc.publish("foo", b'Hello NATS from Python!')
            print("Published 'Hello NATS from Python!' to 'foo'")

            # Wait for a short period to allow message processing
            await asyncio.sleep(1) 

            # Unsubscribe and drain connection
            await sub.unsubscribe()
            await nc.drain()
            print("Drained connection and unsubscribed.")

    except nats.errors.NoServersError:
        print(f"Could not connect to NATS at {nats_url}. Is the server running?")
    except Exception as e:
        print(f"An error occurred: {e}")

if __name__ == '__main__':
    # Run the main asynchronous function
    asyncio.run(main())