Client¶

natsio.connect `async` ¶

connect(
    *servers: str,
    error_cb: ErrorCallback | None = None,
    options: ConnectOptions | None = None,
    _transport_factory: TransportFactory | None = None,
    **kwargs: Unpack[ConnectKwargs],
) -> Client

Connect to NATS and return a ready Client.

servers and any keyword arguments are folded into a ConnectOptions; pass options= to supply one directly (keyword arguments then override its fields).

natsio.Client ¶

Client(
    options: ConnectOptions | None = None,
    *,
    error_cb: ErrorCallback | None = None,
    _transport_factory: TransportFactory | None = None,
)

A connection to a NATS server or cluster.

Prefer the connect() factory:

async with await natsio.connect("nats://localhost:4222") as nc:
    await nc.publish("greet", b"hello")

close `async` ¶

close() -> None

Close immediately: flush pending writes, drop anything still queued.

drain `async` ¶

drain() -> None

Unsubscribe everything, let queued messages be handled, then close.

Bounded by drain_timeout; the client is closed no matter what.

force_reconnect `async` ¶

force_reconnect() -> None

Deliberately drop the current transport and reconnect immediately.

Pending writes are flushed best-effort, then the session is torn down through the normal lost path — subscriptions replay and buffered publishes survive — but the first reconnect attempt bypasses the backoff and the drop is not counted as a server failure. Disconnected then Reconnected fire as usual. Non-blocking: it returns once the drop is scheduled, not once the connection is back.

Raises ConnectionClosedError if the client is already closed or draining.

jetstream ¶

jetstream(
    *,
    domain: str | None = None,
    api_prefix: str | None = None,
    timeout: float = 5.0,
    publish_async_max_pending: int = 4000,
    publish_async_stall_wait: float = 0.2,
    publish_async_timeout: float | None = None,
) -> JetStreamContext

A JetStream context over this connection.

domain routes the control plane through $JS.<domain>.API (leaf nodes); api_prefix overrides the prefix entirely (exports). The publish_async_* knobs tune the async publish window (max in-flight acks, stall wait when full, optional per-message ack timeout).

events `async` ¶

events() -> AsyncIterator[ConnectionEvent]

Stream connection lifecycle events (connected, disconnected, errors...).

publish `async` ¶

publish(
    subject: str,
    payload: bytes | str = b"",
    *,
    reply: str | None = None,
    headers: HeadersInput | None = None,
    _validate_reply: bool = True,
) -> None

Publish a message. Returns once the frame is buffered, not delivered.

_validate_reply is a private escape hatch: internal callers that build their own reply inbox (request/request_many, JetStream async publish) pass False to skip re-validating a subject they just generated. User-supplied replies are always validated.

flush `async` ¶

flush(timeout: float | None = None) -> None

Round-trip a PING and wait for the PONG.

subscribe(
    subject: str,
    *,
    queue: str | None = None,
    cb: Callback | None = None,
    pending_msgs_limit: int | None = None,
    pending_bytes_limit: int | None = None,
    policy: PendingLimitPolicy = PendingLimitPolicy.DROP_NEW,
) -> Subscription

Subscribe to subject.

Synchronous by design: the SUB frame is buffered on the write path and registration takes effect immediately, so no message can be missed between creating the subscription and starting to consume it.

request `async` ¶

request(
    subject: str,
    payload: bytes | str = b"",
    *,
    timeout: float | None = None,
    headers: HeadersInput | None = None,
) -> Msg

Send a request and await a single reply.

request_many `async` ¶

request_many(
    subject: str,
    payload: bytes | str = b"",
    *,
    timeout: float | None = None,
    max_msgs: int | None = None,
    stall: float | None = None,
    headers: HeadersInput | None = None,
) -> AsyncIterator[Msg]

Send one request and yield every reply (ADR-47 "request many").

Completion is whichever comes first: max_msgs replies, a gap of stall seconds between replies, or the overall timeout. A no-responders status ends the stream without yielding.

stall bounds the gap between replies only — the first reply gets the full timeout.

Raises ConnectionClosedError if the connection closes mid-stream: that result is truncated, not complete.

natsio.ConnectOptions `dataclass` ¶

ConnectOptions(
    *,
    servers: tuple[str, ...] = ("nats://127.0.0.1:4222",),
    name: str | None = None,
    connect_timeout: float = 5.0,
    verbose: bool = False,
    pedantic: bool = False,
    echo: bool = True,
    tls: TLSConfig | None = None,
    user: StrSource | None = None,
    password: StrSource | None = None,
    token: StrSource | None = None,
    nkey_seed: str | None = None,
    nkey_file: str | PathLike[str] | None = None,
    credentials: str | PathLike[str] | None = None,
    authenticator: Authenticator | None = None,
    allow_reconnect: bool = True,
    max_reconnect_attempts: int = 60,
    reconnect_time_wait: float = 2.0,
    reconnect_time_wait_max: float = 8.0,
    reconnect_jitter: float = 0.1,
    reconnect_jitter_tls: float = 1.0,
    no_randomize: bool = False,
    ignore_discovered_servers: bool = False,
    retry_on_failed_connect: bool = False,
    reconnect_buf_size: int = 8 * 1024 * 1024,
    ignore_auth_error_abort: bool = False,
    ping_interval: float = 120.0,
    max_outstanding_pings: int = 2,
    max_pending_size: int = 2 * 1024 * 1024,
    flush_timeout: float = 10.0,
    drain_timeout: float = 30.0,
    request_timeout: float = 5.0,
    inbox_prefix: str = "_INBOX",
    pending_msgs_limit: int = 65536,
    pending_bytes_limit: int = 64 * 1024 * 1024,
    permission_err_on_subscribe: bool = False,
    max_control_line: int = 4096,
    instrumentation: Instrumentation | None = None,
)

natsio.Subscription ¶

Subscription(
    client: Client,
    entry: SubscriptionEntry,
    *,
    callback: Callback | None = None,
    pending_msgs_limit: int,
    pending_bytes_limit: int,
    policy: PendingLimitPolicy,
)

An active subscription.

Consume it as an async iterator (the default), or pass cb= to Client.subscribe() to have messages handed to a callback instead. Usable as an async context manager, which unsubscribes on exit.

delivered `property` ¶

delivered: int

Messages handed to this subscription by the connection.

dropped `property` ¶

dropped: int

Messages discarded because the pending limits were exceeded.

next_msg `async` ¶

next_msg(timeout: float | None = None) -> Msg

Await the next message. Raises TimeoutError on expiry.

unsubscribe `async` ¶

unsubscribe() -> None

Stop delivery immediately and discard anything still queued.

unsubscribe_after `async` ¶

unsubscribe_after(max_msgs: int) -> None

Ask the server to stop delivery after max_msgs total messages.

The count is total deliveries since the subscription was created (the server's UNSUB <sid> <max> contract). When the limit is reached the subscription closes itself: iterators finish, next_msg raises SubscriptionClosedError.

drain `async` ¶

drain() -> None

Stop new delivery, wait for the backlog to be handled, then close.

Unbounded by itself — bound it with Client.drain()'s drain_timeout or your own asyncio.timeout; cancellation is honored, never swallowed.

await ¶

__await__() -> Generator[None, None, Subscription]

await is optional and completes immediately: subscribing does no I/O (the SUB frame is buffered and flushed with everything else). Supported so nats-py muscle memory — sub = await nc.subscribe(...) — works unchanged.

natsio.Msg `dataclass` ¶

Msg(
    subject: str,
    payload: bytes,
    reply: str | None = None,
    headers: Headers | None = None,
    status: InlineStatus | None = None,
    sid: int = 0,
    _client: Client | None = None,
)

A message received from the server.

Compares and hashes by identity: two deliveries are distinct events even when byte-identical (and Headers is intentionally unhashable). Treat an instance as read-only; it is deliberately not frozen (the immutability enforcement is skipped to keep per-message construction cheap on the hot delivery path).

headers is None when the message carried no header block. status is set only for the server's control messages (e.g. a 503 no-responders reply), which carry a status line instead of, or alongside, headers.

data `property` ¶

data: bytes

Alias for payload.

respond `async` ¶

respond(
    payload: bytes | str = b"",
    *,
    headers: HeadersInput | None = None,
) -> None

Publish a reply to this message's reply subject.

Client¶

natsio.connect async ¶

natsio.Client ¶

close async ¶

drain async ¶

force_reconnect async ¶

jetstream ¶

events async ¶

publish async ¶

flush async ¶

subscribe ¶

request async ¶

request_many async ¶

natsio.ConnectOptions dataclass ¶

natsio.Subscription ¶

delivered property ¶

dropped property ¶

next_msg async ¶

unsubscribe async ¶

unsubscribe_after async ¶

drain async ¶

__await__ ¶

natsio.Msg dataclass ¶

data property ¶

respond async ¶

natsio.connect `async` ¶

close `async` ¶

drain `async` ¶

force_reconnect `async` ¶

events `async` ¶

publish `async` ¶

flush `async` ¶

request `async` ¶

request_many `async` ¶

natsio.ConnectOptions `dataclass` ¶

delivered `property` ¶

dropped `property` ¶

next_msg `async` ¶

unsubscribe `async` ¶

unsubscribe_after `async` ¶

drain `async` ¶

await ¶

natsio.Msg `dataclass` ¶

data `property` ¶

respond `async` ¶