How correct migrate on nats-jestream?

[Flosckow]

So, we’re using JetStreamContext for publishing, which returns PubAck, but we need to switch to nats-jetstream to receive PublishAck — how can we do this in the most appropriate way?
Something like that, where .connection is the JetstreamContext.

    @override
    async def publish(self, cmd: "NatsPublishCommand") -> "PubAck":
        payload, content_type = await self.codec.encode(cmd.body, self.serializer)

        headers_to_send = {
            "content-type": content_type or "",
            **cmd.headers_to_publish(js=True),
        }

        return await self.__state.connection.publish(
            subject=cmd.destination,
            payload=payload,
            headers=headers_to_send,
            stream=cmd.stream,
            timeout=cmd.timeout,
            msg_ttl=cmd.msg_ttl,
        )

[DahanItamar]

Replace the connection stored in self.__state.connection with the modern JetStream context. Changing only the return annotation on the legacy context will still return PubAck.

The setup is:

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

client = await connect("nats://localhost:4222")
js = new_jetstream(client)
# Store js in the state used by your publisher.
# Close client when your application shuts down.

For your wrapper:

async def publish(self, cmd: "NatsPublishCommand") -> PublishAck:
    payload, content_type = await self.codec.encode(cmd.body, self.serializer)
    headers_to_send = {
        "content-type": content_type or "",
        **cmd.headers_to_publish(js=True),
    }
    if cmd.stream is not None:
        headers_to_send["Nats-Expected-Stream"] = cmd.stream
    if cmd.msg_ttl is not None:
        headers_to_send["Nats-TTL"] = str(int(cmd.msg_ttl))

    options = {}
    if cmd.timeout is not None:
        options["timeout"] = cmd.timeout
    return await self.__state.connection.publish(
        subject=cmd.destination,
        payload=payload,
        headers=headers_to_send,
        **options,
    )

The new publish() takes neither stream= nor msg_ttl=; they become headers. The integer-seconds TTL conversion above preserves the legacy client's behavior. Keep your message-ID/other headers, and change downstream ack.seq accesses to ack.sequence. Omitting a None timeout uses the new client's default of 5 seconds; if your old context had a custom default, supply that explicitly.

See the official migration guide and publish/ack implementation.

I validated this wrapper with Python 3.13.16, nats-core==0.2.0, nats-jetstream==0.3.0, and a local NATS server 2.15.1: payload/headers, TTL, deduplication, expected-stream rejection, optional parameters, and PublishAck all passed. Use Python 3.13+ for this nats-core version. The test environment installed nats-core[nkeys]==0.2.0 alongside nats-jetstream==0.3.0. Per-message TTL requires server 2.11+ and allow_msg_ttl=True on the stream.

Prepared with OpenAI Codex assistance; verified against the published packages and a local server.