Greyhorse NATS library

Greyhorse framework library for NATS/JetStream support (async only -- nats-py has no sync client).

The primary API is the pieces, not a ready-made module:

Piece	Role
NatsFragment	material -- builds the engine
NatsBorder	lifecycle -- start/stop plus a real RTT liveness probe
NatsAccess	access -- hands out a connection window and a JetStream window
NatsEngine	the resource itself
NatsModule	ready-made single-connection floor, sugar over the three pieces above

An application lists the ones it needs on its own Module, alongside pieces from any other storage/messaging library -- no subclassing, no multiple inheritance. See examples/.

NatsStreams/NatsKv/NatsObjects are a separate, second layer: plain, stateless facades a consumer builds by hand over a live JetStreamContext (reached through NatsJsCtx). They add nothing to the DI graph -- their only job is turning nats-py exceptions into Result[T, NatsError], so nothing raises for an expected failure (a missing stream/bucket/key/object, a timed-out request, a broker that is down).

How to build

• Install the project

  uv python pin 3.14

  uv venv

  uv sync

  source .venv/bin/activate

• Format code commands

  ruff check --unsafe-fixes --fix

  ruff format

  mypy greyhorse_nats

• Run tests

  pytest

Usage

Every snippet below is a runnable program, checked against a live broker. The longer, commented versions live in examples/ and are executed by the test suite, so they cannot rot silently.

One engine, one connection

NatsModule is the ready-made bundle for the single-broker case. The config reaches the engine's constructor through the same args={Type: value} door every greyhorse.strand resource uses -- there is no nats-specific wiring.

import asyncio

from greyhorse.run import wrap_async
from greyhorse.strand import running
from greyhorse_nats import NatsConf, NatsConnCtx, NatsModule


async def main() -> None:
    conf = NatsConf(servers='nats://localhost:4222')

    with running(NatsModule, args={NatsConf: conf}) as module:
        conn_ctx = module.get(NatsConnCtx).unwrap()
        async with conn_ctx as nc:
            sub = await nc.subscribe('greet')
            await nc.publish('greet', b'hello')
            msg = await sub.next_msg(timeout=2)
            print(msg.data.decode())
            await sub.unsubscribe()


asyncio.run(wrap_async(main))

The engine connects when the module starts and drains when it stops; nc is an ordinary nats.aio.client.Client borrowed for the length of the async with block. See examples/01_minimal.py.

Always drive the program through greyhorse.run.wrap_async (or run), never a bare asyncio.run(main()) around application code. A NATS client and its subscriptions belong to the loop they were created on, and the border's invoke_sync bridge has to land on that same loop. Skip the wrapper and the module comes up, works, and then fails at shutdown with Event loop is closed -- an error the package recognises and explains, but one you never need to see.

Request/reply

Same connection, both roles. request() needs a live subscriber, so flush before you rely on one being registered server-side.

async with conn_ctx as nc:
    sub = await nc.subscribe('rpc.echo')
    await nc.flush()  # the server now knows about the subscription

    async def responder() -> None:
        request = await sub.next_msg(timeout=5)
        await request.respond(b'pong')

    task = asyncio.create_task(responder())
    reply = await nc.request('rpc.echo', b'ping', timeout=5)
    await task
    print(reply.data.decode())
    await sub.unsubscribe()

JetStream

NatsJsCtx is the second product of the same engine -- a JetStreamContext over the same connection, not a second one. NatsStreams is a plain facade over it: every method answers with Result, so a missing stream or a timed out fetch is a value, not an exception.

from greyhorse_nats import NatsJsCtx, NatsStreams

async with module.get(NatsJsCtx).unwrap() as js:
    streams = NatsStreams(js)

    await streams.ensure_stream('ORDERS', ['orders.>'])
    ack = await streams.publish('orders.new', b'order-1')
    print(ack.unwrap().stream, ack.unwrap().seq)

    pulled = await streams.pull('orders.new', durable='worker', stream='ORDERS')
    fetched = await streams.fetch(pulled.unwrap(), batch=1, timeout=5)
    for msg in fetched.unwrap():
        print(msg.data.decode())
        await msg.ack()

    missing = await streams.stream_info('NO_SUCH_STREAM')
    print(missing.is_err())  # True -- NatsError.NotFound, nothing raised

An idle fetch that simply had nothing to deliver is Ok([]), while a broker that stopped answering is Err(NatsError.Unavailable) -- the two look identical at the nats-py level, and telling them apart is most of why this facade exists. See examples/03_jetstream.py.

Key-value and object storage

from greyhorse_nats import NatsKv, NatsObjects

async with module.get(NatsJsCtx).unwrap() as js:
    kv = NatsKv(js, 'settings')
    await kv.ensure()
    await kv.put('greeting', b'hello kv')
    print((await kv.get('greeting')).unwrap().value.decode())
    print((await kv.get('no-such-key')).is_err())  # True

    objects = NatsObjects(js, 'reports')
    await objects.ensure()
    await objects.put('daily.txt', b'contents')
    print((await objects.get('daily.txt')).unwrap().data.decode())

A facade binds a bucket by NAME and remembers when the server says that bucket is gone: after a delete it refuses until you call ensure() again, rather than quietly recreating what an administrator removed. The limits of that promise are spelled out under JetStream facades below. See examples/04_kv_objects.py.

A consumer that knows nothing about greyhorse

The class that uses NATS takes a context by TYPE. NatsConnCtx is an alias for AsyncShared[Client] -- no marker, no base class, no import from this package in the domain code:

class PingService:
    def __init__(self, conn: NatsConnCtx) -> None:
        self._conn = conn

    async def ping(self) -> str:
        async with self._conn as nc:
            reply = await nc.request('rpc.echo', b'ping', timeout=5)
            return reply.data.decode()

The declaration is the only place that names this library, and it says what the component takes and what it hands out:

class PingComponent(Component):
    imports: ClassVar = Shared[NatsEngine]
    providers: ClassVar = NatsAccess
    exports: ClassVar = PingService

Two components declaring providers = NatsAccess over the SAME engine each get their own provider instance and their own borrow -- one connection, several independent consumers. See examples/02_component.py, and examples/05_multi_storage.py for two engines on one floor without subclassing anything.

Configuration from the environment

NatsConf above is built in code. A deployed service usually wants it from the environment instead -- that is NatsSettings, a pydantic_settings model reading NATS_-prefixed variables (and a .env file):

from greyhorse_nats import NatsSettings

conf = NatsSettings().engine_conf()   # NATS_HOST, NATS_PORT, NATS_USER, ...
conf = NatsSettings().engine_conf(connect_timeout_seconds=5)   # tune anything

engine_conf() is the only supported way to turn settings into a config. settings.servers carries broker locations only -- no credentials live in a DSN any more -- so hand-building NatsConf(servers=settings.servers, ...) silently drops every credential variable -- NATS_USER, NATS_PASSWORD, NATS_PASSWORD_FILE, NATS_TOKEN, NATS_CREDENTIALS_FILE, NATS_NKEYS_SEED_FILE -- and connects unauthenticated. Keyword arguments pass through to NatsConf for the tuning knobs that have no env var of their own (timeouts, pool sizes, no_echo).

variable	default	notes
NATS_SERVERS	assembled from host/port	JSON list, bare DSN, or multi-host DSN -- see below
NATS_HOST / NATS_PORT	localhost / 4222	used only when NATS_SERVERS is unset
NATS_SCHEME	nats	tls, ws, wss also accepted -- see the ws:///wss:// teardown note below
NATS_NAME	greyhorse	the client name the broker reports
NATS_USER / NATS_PASSWORD	unset	carried as config FIELDS, never inside the DSN
NATS_PASSWORD_FILE	unset	read from disk, so the secret never sits in an env var
NATS_TOKEN	unset	alternative to user/password
NATS_CREDENTIALS_FILE	unset	nkeys/JWT creds file; needs the nkeys extra
NATS_NKEYS_SEED_FILE	unset	nkeys seed file; needs the nkeys extra

NATS_SERVERS takes any of three shapes:

value	result
["nats://a:4222","nats://b:4333"]	two servers (the JSON list pydantic imposes on a list-typed setting)
nats://a:4222	one server -- a bare DSN needs no JSON
nats://a:4222,b:4333	ONE multi-host DSN, NATS's own cluster spelling
empty or unset	falls back to NATS_HOST/NATS_PORT

ws:///wss:// servers are accepted and work identically to nats:/// tls:// for connecting and serving requests. They carry one narrower teardown guarantee: if the loop that owns the connection has already died by the time stop() runs on it (the ordinary shutdown order for a threaded gateway), the local last-resort fallback that still recovers a nats:///tls:// connection in that situation cannot reach a WebSocket connection's socket while it is still genuinely open (it lives inside aiohttp's own pool, not a dependency of this package) -- stop() raises RuntimeError rather than reporting a teardown it did not do. Every other shutdown path -- including a threaded gateway whose owning loop is still alive when stop() runs -- closes a ws:///wss:// connection exactly as cleanly as nats://.

One thing that is not scheme-specific, though it reads like it should be: if the broker connection had already been lost and nats-py's own reconnect logic had already closed the old transport before stop() got there (the driver leaves the client RECONNECTING, not CLOSED, while it keeps retrying, and keeps pointing at the closed transport), stop() recognises that and does not raise -- on nats:///tls:// exactly as on ws:///wss://. stop() raises only when a real connection may still be open and there is no loop-free way to reach it, never merely because of which transport class happens to be attached.

A DSN naming no host (nats://, or nats://$BROKER with the variable unset) is refused, not quietly turned into nats://localhost:4222 the way pydantic would have it -- otherwise a templating slip points the service at whatever broker happens to run locally, and nothing looks wrong.

NATS_PASSWORD_FILE exists for the deployment shape where secrets arrive as mounted files (Kubernetes, Docker secrets) rather than environment variables. Setting it makes the file mandatory: if it is missing, unreadable or empty, engine_conf() raises instead of quietly connecting unauthenticated, so a missing mount fails at configuration time rather than surfacing later as a server-side auth error.

Credentials never live in a DSN. NatsConf.servers refuses one that carries userinfo, because a DSN is a plain value pydantic serializes verbatim from model_dump()/model_dump_json() -- no amount of repr work reaches those. Pass credentials as fields instead:

NatsConf(servers='nats://broker:4222', user='alice', password='...')   # user/password
NatsConf(servers='nats://broker:4222', token='...')                    # token
NatsConf(servers='nats://broker:4222', credentials_file='/run/secrets/app.creds')  # needs [nkeys]

Exactly one auth mode per config, and the model enforces it. The modes are not additive: nats-py picks one through an elif chain, so a config carrying two would have the loser silently dropped -- and the loser is not always the same one, because the nkeys branch is taken only when the server offers a nonce. Rather than ship a credential whose effect depends on which server answered, these are rejected when the config is built:

refused	because
user without password, or password without user	NATS authenticates with the pair; neither half is sent alone
token with a complete user+password pair	user/password wins, the token is dropped
credentials_file with nkeys_seed_file	the seed is never read
credentials_file/nkeys_seed_file with user/password	which one wins depends on the server
NATS_TOKEN with NATS_USER + NATS_PASSWORD_FILE †	same clash, one step earlier
NATS_PASSWORD_FILE without NATS_USER †	the resolved password would never be sent
NATS_PASSWORD_FILE with NATS_CREDENTIALS_FILE/NATS_NKEYS_SEED_FILE †	same clash as password, one step earlier

A token beside HALF a pair is not the token clash: there nats-py sends the token and nothing else, so the incomplete pair is the defect, and that is what the error names. Saying "your token is ignored" would send you to delete the one credential that works.

An empty value is not a value: NATS_TOKEN='' (or ' ') means unset, not "authenticate with an empty token". Without that, a blank variable from an unfilled template connected anonymously and looked configured.

† NatsSettings only — NatsConf has no password_file field, so these three are caught by engine_conf() before the file is opened, and reported as the mode clash they are rather than as a file it could not read.

A token alongside credentials_file/nkeys_seed_file is supported -- nats-py sends both, for auth callouts.

credentials_file and nkeys_seed_file need the signing support, which is an optional extra: pip install greyhorse-nats[nkeys]. Without it the config is refused with that instruction, rather than failing later inside nats.connect() with a bare ModuleNotFoundError.

password and token are SecretStr, so they render as ********** in every repr, dump and traceback and yield their real value only where the connection is actually made. A bare username in a URL counts as a credential too -- nats-py reads it as an auth token, not as a user name.

Not to be confused with NATS_TEST_URI below: that one is a test-suite gate and has nothing to do with NatsSettings.

Who owns reconnect

An outage can be repaired two ways, and the package supports both without choosing between them: the nats-py driver's own reconnect loop (allow_reconnect, max_reconnect_attempts, reconnect_time_wait_seconds on NatsConf -- True/60/2 by default), or the border's repair (NatsBorder.check() drives NatsEngine.is_alive(), a real RTT probe; a failing check feeds light repair at the border and, on a permanent failure, heavy repair at the controller). With the default allow_reconnect=True the two genuinely overlap: the driver may already be reconnecting underneath the same connection while check() still reports the broker unreachable.

Keep the default. Driver-level reconnect is cheaper -- it resumes the same connection and its subscriptions rather than tearing the engine down -- and the border's repair still stands as the outer net for the case the driver gives up (max_reconnect_attempts exhausted). Setting allow_reconnect=False hands the whole job to the border instead, and turns every outage, however short, into a full rebuild of the engine; a consumer who wants that path can additionally bound the controller's own escalation with a ControllerRestartPolicy on the resource (Resource(NatsEngine, operators=NatsBorder, policy=ControllerRestartPolicy(...))). A consumer who does not want to think about it should leave allow_reconnect at its default of True.

JetStream facades

js_ctx = module.get(NatsJsCtx).unwrap()
async with js_ctx as js:
    streams = NatsStreams(js)
    match await streams.ensure_stream('ORDERS', ['orders.>']):
        case Ok(info):
            ...
        case Err(err):
            ...  # NatsError -- never an exception

See examples/03_jetstream.py and examples/04_kv_objects.py.

A KV/object facade is bound to a bucket NAME, not to a bucket. nats-py's handle carries the name and the JetStream context; nothing in it refers to the bucket instance. Two consequences worth knowing before you rely on one:

• A bucket deleted and recreated under the same name is adopted silently by an existing facade -- ordinary calls keep succeeding, against the new bucket. Only NatsObjects.list() catches it, because that path already asks the server and can compare identity for free. Checking everywhere would mean a round-trip on every put and every get, which is not a price this library pays. After an administrative delete, build a new facade rather than trusting an old one.
• A bucket deleted and left deleted is remembered, and the facade then refuses until you call ensure() deliberately -- but only the calls that can prove it do the remembering. keys, history, put, purge and list surface a bucket-level error and latch; get (and info/delete on the object side) raise exactly what an ordinary missing key or object raises, so they report NotFound correctly and record nothing.

Tests

Most of the suite (config parsing, plan()-only DI checks) needs no server at all. The rest needs a running NATS broker with JetStream enabled:

``docker compose -f tests/docker-compose.yml up -d --wait``

They are gated behind the NATS_TEST_URI environment variable; without it they skip with a clear reason instead of erroring on "connection refused".

``NATS_TEST_URI=nats://localhost:54222 pytest``

tests/docker-compose.yml brings up the broker on non-default ports (54222/58222, container name greyhorse-nats-throwaway) so it never collides with a NATS instance already running on the standard 4222/8222. Both host ports are overridable -- NATS_CLIENT_PORT and NATS_MONITOR_PORT -- for the case those two are taken as well.

Afterwards, take the broker down and its volume with it:

``docker compose -f tests/docker-compose.yml down -v``