iap_messenger

PyPI - License

The IA Parc inference plugin allows developers to easily integrate their inference pipeline into IA Parc's production module.

Installation

pip install iaparc-inference

Usage

If your inference pipeline support batching:

from iap_messenger import MsgListener, Message

# Define a callback to query your inference pipeline
# To load your model only once it is recommended to use a class:
class MyModel:
    def __init__(self, model_path: str):
        ## Load your model in pytorch, tensorflow or any other backend
    
    def batch_query(msgs: list[Message]) -> list[Message]:
        ''' execute your pipeline on a batch input
            Note:   "parameters" is an optional argument.
                    It can be used to handle URL's query parameters
                    It's a list of key(string)/value(string) dictionaries
        '''

if __name__ == '__main__':
    # Initiate your model class
    my_model = MyModel("path/to/my/model")

    # Initiate IAParc listener
    listener = MsgListener(my_model.batch_query)
    # Start the listener
    listener.run()

If your inference pipeline do not support batching:

from iap_messenger import MsgListener, Message

# Define a callback to query your inference pipeline
# To load your model only once it is recommended to use a class:
class MyModel:
    def __init__(self, model_path: str):
        ## Load your model in pytorch, tensorflow or any other backend
    
    def single_query(msg: Message) -> Message:
        ''' execute your pipeline on a single input
            Note:   "parameters" is an optional argument.
                    It can be used to handle URL's query parameters
                    It's a key(string)/value(string) dictionary
        '''

if __name__ == '__main__':
    # Initiate your model class
    my_model = MyModel("path/to/my/model")

    # Initiate IAParc listener
    listener = MsgListener(my_model.single_query, batch=1)  # Note that batch size is forced to 1 here
    # Start the listener
    listener.run()

Message consumption and replicas

Each input link is served by one durable JetStream consumer named {queue}-{link}, shared by every replica of the worker. Since the IAP_CONSUMER_MODE release the consumer is a pull consumer: a replica only fetches a message (or a batch) when it is free, so nothing ever waits in the buffer of a busy or dead replica while another one is idle, and a crash only leaves the batch in flight to be redelivered after ack_wait. Older versions used a push consumer with a queue group, where the server spreads messages across replicas at random and up to max_ack_pending messages could sit in a dead pod until ack_wait expired.

Variable	Default	Role
`IAP_CONSUMER_MODE`	`auto`	`auto` uses pull unless the existing consumer is a push consumer that still has subscribers bound (workers of a push-only version), in which case it joins it in push mode. `pull` forces the migration: the push consumer is deleted and recreated as pull, and bound push subscribers stop receiving messages until restarted. `push` keeps the historical behaviour. The mode chosen for each link is logged at startup.
`IAP_ACK_WAIT_SECONDS`	`300`	JetStream `ack_wait`. Long callbacks are kept alive with `in_progress` heartbeats, so this only bounds two things: how long a NATS outage may last before in-flight work is redelivered, and how long a message held by a crashed replica waits before another one gets it.
`IAP_PULL_IDLE_TIMEOUT_SECONDS`	`30`	Longest an idle pull fetch blocks. After each idle timeout the worker checks that its consumer still exists and rebuilds the subscription if the API recreated the stream.
`IAP_SHUTDOWN_DDL_GRACE_SECONDS`	`30`	On shutdown, how long to keep serving large replies from the local DDL store so the next component can still fetch them.

Upgrading a pipeline from push to pull

The stream is a WorkQueue, which refuses two consumers on the same subjects, so a pipeline is either all push or all pull for a given link. auto makes a rolling update safe: new pods join the push consumer while old pods are bound to it, nothing is delivered twice and no old pod breaks. Once every pod of the queue runs a pull-capable version, the migration happens on its own the next time no pod is bound (scale to zero, NATS restart), or immediately with a rollout restart after setting IAP_CONSUMER_MODE=pull. Note that after a NATS restart, a pod of the old version reconnecting to a consumer that has just been migrated fails to subscribe and exits; the rollout replaces it.

Graceful shutdown

On SIGTERM (or SIGINT) the listener stops fetching, lets the task or batch in flight complete and acks it, hands back with a NAK any message already pushed to the process but not started (push mode), waits up to IAP_SHUTDOWN_DDL_GRACE_SECONDS for large replies to be fetched downstream, then exits with code 0. Set the pod's terminationGracePeriodSeconds above the longest inference plus that grace, otherwise Kubernetes kills the pod mid-task and the message is redelivered after ack_wait. A second signal exits at once.

Large payloads (DDL)

Payloads above 8 MB never travel through NATS: the sender keeps them in its local DDL store and the message only carries a uuid@host:port pointer that the recipient resolves over ZMQ. Environment variables:

Variable	Role
`POD_IP`	Enables DDL (the address advertised to recipients). Without it, payloads fall back to the NATS object store.
`ZMQ_PORT`	DDL server port (default `5555`).
`DDL_STORAGE`	Directory for the blob store. Strongly recommended in production: blobs then never transit through the `multiprocessing.Manager` pipe. Mount it on a disk-backed `emptyDir` — a `medium: Memory` volume is counted against the container's memory limit.

The reply format is negotiated per request: a client that can read a two-frame reply ([status], [payload]) asks with the MP: prefix and the server then hands the stored bytes to ZMQ without copying them; any older peer keeps getting the historical single frame. No configuration and no deployment order are required — mixed versions of workers and inferences-api interoperate.

Features

Dynamic batching
Autoscalling
Support both synchronous and asynchronous queries
Data agnostic

License

This project is licensed under the Apache License Version 2.0 - see the Apache LICENSE file for details.

Metadata

Release files for iap-messenger 2.1.0

For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.

Source distribution (sdist)

Source distribution for iap-messenger 2.1.0
File	Size	Uploaded
iap_messenger-2.1.0.tar.gz	118.0 kB	Sep 17, 2026	Details

Built distribution (wheel)

Table of built distributions (wheels) for iap-messenger 2.1.0
File	Interpreter	ABI	Platform	Reset
iap_messenger-2.1.0-cp310-abi3-manylinux_2_17_x86_64.manylinux2014_x86_64.whl 678.4 kB Sep 17, 2026	CPython 3.10	abi3	Linux glibc 2.17+ x86-64	Details

Total release size: 796.4 kB

Release files / iap_messenger-2.1.0.tar.gz

Download URL	iap_messenger-2.1.0.tar.gz
Size	118.0 kB
Tags	Source
SHA-256 checksum How to use checksums	`214ae997946d0c83ade32487bedc5acc1249a7ea78b0d7df33df3bdbb0711055`
BLAKE2b-256 checksum How to use checksums	`9f9f8e84de209b63524062ceab10978f74c847b3aed576b6398b9ed3d9f1bccf`
Upload date	Sep 17, 2026
Uploaded using Trusted Publishing? What is trusted publishing?	No
Uploaded via	`twine/7.0.0 CPython/3.10.21`

Release files / iap_messenger-2.1.0-cp310-abi3-manylinux_2_17_x86_64.manylinux2014_x86_64.whl

Download URL	iap_messenger-2.1.0-cp310-abi3-manylinux_2_17_x86_64.manylinux2014_x86_64.whl
Size	678.4 kB
Tags	CPython 3.10 Linux glibc 2.17+ x86-64 abi3
SHA-256 checksum How to use checksums	`3dafe48365f71db080b1aaf0fe7a91a88cfa7e8b181fb38225ecdf76e33bcf9a`
BLAKE2b-256 checksum How to use checksums	`7effe92bc19792f32c2cf54c54402d41836e4a9a69a53ce3bf6920efc1a71cc1`
Upload date	Sep 17, 2026
Uploaded using Trusted Publishing? What is trusted publishing?	No
Uploaded via	`twine/7.0.0 CPython/3.10.21`