`token` parameter ignored when using `user_credentials` (nkeys authentication)

[denizkenan]

Observed behavior

Summary

When connecting to NATS with both user_credentials (nkeys/JWT authentication) and a token parameter, the Python client silently ignores the token and does not include it in the CONNECT message sent to the server. This breaks authentication workflows that rely on authorization callouts where the token is needed to identify the connecting user.

Actual Behavior

The token parameter is completely ignored when user_credentials is provided. The CONNECT message only includes the JWT authentication fields (jwt, sig, nkey) but omits the auth_token field entirely.

Expected behavior

When both user_credentials and token parameters are provided to nats.connect(), the CONNECT message sent to the NATS server should include both:

• The JWT authentication (via jwt and sig fields)
• The token (via auth_token field)

This is how the official NATS clients for Go and JavaScript behave.

Server and client version

Server(Embedded): v2.11.3 // github.com/nats-io/nats-server/v2
Client(nats-py): 2.11.0

Host environment

No response

Steps to reproduce

Server is configured with auth_callout. Connecting to AUTH account with basic sentinel creds.

import asyncio
import nats

async def main():
    nc = await nats.connect(
        servers=["nats://localhost:4222"],
        user_credentials="/path/to/sentinel.creds",
        token="my-auth-token"  # This token is silently ignored
    )
    print("Connected")
    await nc.close()

asyncio.run(main())

Workaround

Created a custom TokenAwareNatsClient class that extends the standard NATS client and overrides _connect_command() to ensure the token is always included in the CONNECT message when provided, even when using nkey authentication.

Implementation

File: nats_token_client.py

"""Custom NATS client that forwards auth tokens alongside nkey credentials."""

from __future__ import annotations
import json
from typing import Final
from nats.aio.client import Client as NatsClient

_CONNECT_PREFIX: Final[bytes] = b"CONNECT "
_CONNECT_SUFFIX: Final[bytes] = b"\r\n"


class TokenAwareNatsClient(NatsClient):
    """A NATS client that keeps auth tokens in CONNECT when using nkeys."""

    def _connect_command(self) -> bytes:
        connect_cmd = super()._connect_command()
        token = self.options.get("token")
        if not token:
            return connect_cmd

        if not connect_cmd.startswith(_CONNECT_PREFIX) or not connect_cmd.endswith(
            _CONNECT_SUFFIX
        ):
            return connect_cmd

        payload = connect_cmd[len(_CONNECT_PREFIX) : -len(_CONNECT_SUFFIX)]
        try:
            options = json.loads(payload.decode())
        except json.JSONDecodeError:
            return connect_cmd

        if options.get("auth_token"):
            return connect_cmd

        options["auth_token"] = token
        patched_payload = json.dumps(options, sort_keys=True).encode()
        return _CONNECT_PREFIX + patched_payload + _CONNECT_SUFFIX

Usage

Replace:

from nats import NATS

nc = NATS()
await nc.connect(
    NATS_URL,
    user_credentials=NATS_CREDS_FILE,
    token=NATS_APP_TOKEN
)

With:

from nats_token_client import TokenAwareNatsClient

nc = TokenAwareNatsClient()
await nc.connect(
    NATS_URL,
    user_credentials=NATS_CREDS_FILE,
    token=NATS_APP_TOKEN
)