amgi-pulsar

amgi-pulsar is an AMGI compatible server project, supporting running AMGI applications against Apache Pulsar topics.

The server consumes messages from a shared subscription, which is created if it does not exist. The subscription type can be configured with consumer_type; Shared (the default) provides no ordering guarantees between consumers, while Exclusive, Failover and KeyShared are also available. Messages are only acknowledged when the application sends a message.ack event; a message.nack event triggers a redelivery. Messages that remain unacknowledged for longer than unacked_messages_timeout_ms (30000 by default) are also redelivered.

Installation

pip install amgi-pulsar==0.47.0

Example

This example uses AsyncFast:

from dataclasses import dataclass

from amgi_pulsar import run
from asyncfast import AsyncFast

app = AsyncFast()


@dataclass
class Order:
    item_ids: list[str]


@app.channel("order.topic")
async def order_topic(order: Order) -> None:
    # Makes an order
    ...


if __name__ == "__main__":
    run(app, "order.topic")

Or the application could be run via the commandline:

asyncfast run amgi-pulsar main:app order.topic

Contact

For questions or suggestions, please contact jack.burridge@mail.com.

License

Copyright 2025 AMGI